使用 Django OAuth2 Toolkit 以编程方式生成访问令牌

Posted 2023-03-10

【中文标题】使用 Django OAuth2 Toolkit 以编程方式生成访问令牌

【英文标题】：Generating an access token programatically with Django OAuth2 Toolkit

【发布时间】：2015-06-10 03:04:57

【问题描述】：

我正在使用 Python Social Auth 和 Django OAuth Toolkit 来管理我的用户帐户并限制对我的 REST API 的访问。

我可以使用常规

为使用我的应用手动注册的用户创建令牌

curl -X POST -d "grant_type=password&username=<user_name>&password=<password>" -u"<client_id>:<client_secret>" http://localhost:8000/o/token/

但是当我通过访问令牌向 PSA 注册用户时，我想为我自己的应用程序创建一个 OAuth2 Toolkit 令牌并将其作为 JSON 返回给客户端，以便它可以使用它来通过我的 API 发出请求。

目前，我仅使用 oauthlib 中的generate_token 生成令牌，这是一种好的做法吗？我应该考虑其他因素吗？

from oauthlib.common import generate_token

...

@psa('social:complete')
def register_by_access_token(request, backend):
    # This view expects an access_token GET parameter, if it's needed,
    # request.backend and request.strategy will be loaded with the current
    # backend and strategy.
    token = request.GET.get('access_token')
    user = request.backend.do_auth(token)

    if user:
        login(request, user)
        app = Application.objects.get(name="myapp")

        # We delete the old one
        try:
            old = AccessToken.objects.get(user=user, application=app)
        except:
            pass
        else:
            old.delete()

        # We create a new one
        tok = generate_token()

        AccessToken.objects.get_or_create(user=user,
                                          application=app,
                                          expires=now() + timedelta(days=365),
                                          token=tok)

        return "OK" # I will eventually return JSON with the token
    else:
        return "ERROR"

【问题讨论】：

非常有趣的问题。我没有找到任何其他有用的答案，所以我在您的博客文章中工作

我建议你看看github.com/PhilipGarnero/django-rest-framework-social-oauth2，它可以帮助你以一种干净的方式做到这一点。

【参考方案1】：

我最近使用https://github.com/PhilipGarnero/django-rest-framework-social-oauth2 来达到这个目的，就像用户Felix D. 建议的那样。下面是我的实现：

class TokenHandler:
    application = Application.objects.filter(name=APPLICATION_NAME).values('client_id', 'client_secret')

    def handle_token(self, request):
        """
        Gets the latest token (to access my API) and if it's expired, check to see if the social token has expired.
        If the social token has expired, then the user must log back in to access the API. If it hasn't expired, 
        (my) token is refreshed.
        """
        try:
            token_list = AccessToken.objects.filter(user=request.user)\
                .order_by('-id').values('token', 'expires')
            if token_list[0]['expires'] < datetime.now(timezone.utc):
                if not self.social_token_is_expired(request):
                    token = self.refresh_token(request)
                else:
                    token = 'no_valid_token'
            else:
                token = token_list[0]['token']
        except IndexError:  # happens where there are no old tokens to check
            token = self.convert_social_token(request)
        except TypeError:  # happens when an anonymous user attempts to get a token for the API

            token = 'no_valid_token'
        return token

    def convert_social_token(self, request):
        grant_type = 'convert_token'
        client_id = self.application[0]['client_id']
        client_secret = self.application[0]['client_secret']
        try:
            user_social_auth = request.user.social_auth.filter(user=request.user).values('provider', 'extra_data')[0]
            backend = user_social_auth['provider']
            token = user_social_auth['extra_data']['access_token']
            url = get_base_url(request) + reverse('convert_token')
            fields = 'client_id': client_id, 'client_secret': client_secret, 'grant_type': grant_type,
                      'backend': backend,
                      'token': token
            if backend == 'azuread-oauth2':
                fields['id_token'] = user_social_auth['extra_data']['id_token']
            response = requests.post(url, data=fields)
            response_dict = json.loads(response.text)
        except IndexError:
            return 'error': 'You must use an OAuth account to access the API.'
        except UserSocialAuth.DoesNotExist:
            return 'error': 'You must use an OAuth account to access the API.'
        return response_dict['access_token']

    def refresh_token(self, request):
        grant_type = 'refresh_token'
        client_id = self.application[0]['client_id']
        client_secret = self.application[0]['client_secret']
        try:
            refresh_token_object = RefreshToken.objects.filter(user=request.user).order_by('-id').values('token')[0]
            token = refresh_token_object['token']
            url = get_base_url(request) + reverse('token')
            fields = 'client_id': client_id, 'client_secret': client_secret, 'grant_type': grant_type,
                      'refresh_token': token
            response = requests.post(url, data=fields)
            response_dict = json.loads(response.text)
        except RefreshToken.DoesNotExist:
            return 'error': 'You must use an OAuth account to access the API.'

        return response_dict['access_token']

    @staticmethod
    def social_token_is_expired(request):
        user_social_auth = UserSocialAuth.objects.filter(user=request.user).values('provider', 'extra_data')[0]
        try:
            return float(user_social_auth['extra_data']['expires_on']) <= datetime.now().timestamp()
        except KeyError:  # social API did not provide an expiration
            return True  # if our token is expired and social API did not provide a time, we do consider them expired