决定何时使用 Python 社交身份验证刷新 OAUTH2 令牌

Decide when to refresh OAUTH2 token with Python Social Auth

我认为这主要是关于最佳实践的问题。

我有一个 OAUTH2 提供商,只要刷新令牌,它就会颁发访问令牌(有效期为 10 小时)。

我发现 here 刷​​新访问令牌非常容易,但我不明白如何决定何时刷新。

最简单的答案可能是 "when it does not work any more",意思是当我从后端收到 HTTP 401 时。 这个解决方案的问题是它效率不高,而且我只能假设我收到了 401,因为令牌已过期。

在我的 Django 应用程序中,我发现 user social auth 有一个 Extra data 字段,其中包含如下内容:

{ "scope": "read write", "expires": 36000, "refresh_token": "xxxxxxxxxxxxx", "access_token": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", "token_type": "Bearer" }

但我不确定如何使用 expires 字段。

所以我的问题是:我如何知道访问令牌是否已过期并且我需要刷新它?

编辑: 我刚刚发现 this comment 似乎相关,但我不明白如何将这个新函数插入管道以便在令牌刷新期间工作。

我终于想通了。 我一开始很困惑的原因是因为实际上有两种情况:

  1. 当用户登录后,管道基本上就会执行。
  2. 刷新令牌时调用用户社交身份验证方法refresh_token

解决第一种情况

我为管道创建了一个新函数:

def set_last_update(details, *args, **kwargs):  # pylint: disable=unused-argument
    """
    Pipeline function to add extra information about when the social auth
    profile has been updated.
    Args:
        details (dict): dictionary of informations about the user
    Returns:
        dict: updated details dictionary
    """
    details['updated_at'] = datetime.utcnow().timestamp()
    return details

在设置中,我将其添加到 load_extra_data

之前的管道中
SOCIAL_AUTH_PIPELINE = (
    'social.pipeline.social_auth.social_details',
    'social.pipeline.social_auth.social_uid',
    'social.pipeline.social_auth.auth_allowed',
    'social.pipeline.social_auth.social_user',
    'social.pipeline.user.get_username',
    'social.pipeline.user.create_user',
    'social.pipeline.social_auth.associate_user',
    # the following custom pipeline func goes before load_extra_data
    'backends.pipeline_api.set_last_update',
    'social.pipeline.social_auth.load_extra_data',
    'social.pipeline.user.user_details',
    'backends.pipeline_api.update_profile_from_edx',
    'backends.pipeline_api.update_from_linkedin',
)

并且,仍然在设置中,我在额外数据中添加了新字段。

SOCIAL_AUTH_EDXORG_EXTRA_DATA = ['updated_at']

第二种情况:

我重写了后端的 refresh_token 方法以添加额外的字段。

def refresh_token(self, token, *args, **kwargs):
    """
    Overridden method to add extra info during refresh token.
    Args:
        token (str): valid refresh token
    Returns:
        dict of information about the user
    """
    response = super(EdxOrgOAuth2, self).refresh_token(token, *args, **kwargs)
    response['updated_at'] = datetime.utcnow().timestamp()
    return response

仍然在后端 class,我添加了一个额外的字段来提取来自服务器的 expires_in 字段。

EXTRA_DATA = [
    ('refresh_token', 'refresh_token', True),
    ('expires_in', 'expires_in'),
    ('token_type', 'token_type', True),
    ('scope', 'scope'),
]

此时我有访问令牌创建时的时间戳 (updated_at) 和有效秒数 (expires_in)。

注意:updated_at 是一个近似值,因为它是在客户端而非提供商服务器上创建的。

现在唯一缺少的是检查是否到了刷新访问令牌的功能。

def _send_refresh_request(user_social):
    """
    Private function that refresh an user access token
    """
    strategy = load_strategy()
    try:
        user_social.refresh_token(strategy)
    except HTTPError as exc:
        if exc.response.status_code in (400, 401,):
            raise InvalidCredentialStored(
                message='Received a {} status code from the OAUTH server'.format(
                    exc.response.status_code),
                http_status_code=exc.response.status_code
            )
        raise


def refresh_user_token(user_social):
    """
    Utility function to refresh the access token if is (almost) expired
    Args:
        user_social (UserSocialAuth): a user social auth instance
    """
    try:
        last_update = datetime.fromtimestamp(user_social.extra_data.get('updated_at'))
        expires_in = timedelta(seconds=user_social.extra_data.get('expires_in'))
    except TypeError:
        _send_refresh_request(user_social)
        return
    # small error margin of 5 minutes to be safe
    error_margin = timedelta(minutes=5)
    if datetime.utcnow() - last_update >= expires_in - error_margin:
        _send_refresh_request(user_social)

我希望这对其他人有所帮助。

目前,extra_data 字段现在有一个 auth_time。您可以将其与 expires 一起使用来确定 access_token 的有效性:

if (social.extra_data['auth_time'] + social.extra_data['expires'] - 10) <= int(time.time()):
    from social_django.utils import load_strategy
    strategy = load_strategy()
    social.refresh_token(strategy)

额外的“10”秒是为了防止 access_token 可能在执行更多代码之前过期的竞争条件。

这个问题给出了更多细节: