python - 决定何时使用 Python Social Auth 刷新 OAUTH2 token

Question

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

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

我找到了 here刷新访问 token 非常容易，但我不明白如何决定何时刷新。

简单的答案可能是“当它不再工作时”，意思是当我从后端收到 HTTP 401 时。此解决方案的问题在于它效率不高，而且我只能假设我收到了 401，因为 token 已过期。

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


{
“范围”:“读写”，
“到期”:36000，
"refresh_token": "xxxxxxxxxxxxx",
“access_token”:“xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx”，
“token_type”:“不记名”
}


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

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

编辑:我刚找到 this comment这似乎是相关的，但我不明白如何将这个新功能插入管道以便在 token 刷新期间工作。

Answer 1

我终于明白了这一点。我最初感到困惑的原因是因为实际上有两种情况:

1. 当用户登录后，管道基本上就会执行。
2. 当 token 刷新时调用用户社交身份验证方法 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

还是在后端类中，我添加了一个额外的字段来提取来自服务器的 expires_in 字段。

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

此时，我有了访问 token 创建时的时间戳 (updated_at) 及其有效秒数 (expires_in)。

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

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

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)

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