68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
"""自定义 JWT 认证 — 向后兼容旧 token 的 user_id claim"""
|
||
|
||
import hmac
|
||
|
||
from rest_framework_simplejwt.authentication import JWTAuthentication
|
||
from rest_framework_simplejwt.exceptions import InvalidToken, AuthenticationFailed
|
||
from rest_framework_simplejwt.settings import api_settings
|
||
from django.utils.translation import gettext_lazy as _
|
||
|
||
|
||
class CompatibleJWTAuthentication(JWTAuthentication):
|
||
"""
|
||
兼容旧 token 的 JWT 认证类。
|
||
|
||
当前配置（settings.SIMPLE_JWT）：
|
||
- USER_ID_CLAIM = 'UserUUID'
|
||
- USER_ID_FIELD = 'UserUUID'
|
||
|
||
旧 token（默认 SimpleJWT 配置时生成）payload 中使用 'user_id' claim，
|
||
值为 user.pk（即 UserUUID 的字符串形式）。
|
||
|
||
此类先尝试当前配置的 USER_ID_CLAIM（'UserUUID'），找不到时回退到旧
|
||
'user_id' claim，两者均通过 UserUUID 字段查库。
|
||
"""
|
||
|
||
def get_user(self, validated_token):
|
||
# 1. 先尝试当前配置的 USER_ID_CLAIM
|
||
user_id = validated_token.get(api_settings.USER_ID_CLAIM)
|
||
lookup_field = api_settings.USER_ID_FIELD
|
||
|
||
if user_id is None:
|
||
# 2. 回退：尝试旧 token 中的 'user_id' claim
|
||
user_id = validated_token.get('user_id')
|
||
if user_id is None:
|
||
raise InvalidToken(
|
||
{'detail': _('令牌未包含用户标识符'), 'code': 'token_not_valid'}
|
||
)
|
||
# 旧 token 的 user_id 来自 user.pk（即 UserUUID 字符串），
|
||
# 需要用 UserUUID 字段查库
|
||
lookup_field = 'UserUUID'
|
||
|
||
try:
|
||
user = self.user_model.objects.get(**{lookup_field: user_id})
|
||
except self.user_model.DoesNotExist:
|
||
raise AuthenticationFailed(_('用户不存在'), code='user_not_found')
|
||
except (ValueError, TypeError):
|
||
raise InvalidToken(
|
||
{'detail': _('令牌中的用户标识符格式无效'), 'code': 'token_not_valid'}
|
||
)
|
||
|
||
if not user.is_active:
|
||
raise AuthenticationFailed(_('用户已被禁用'), code='user_inactive')
|
||
|
||
if api_settings.CHECK_REVOKE_TOKEN:
|
||
# 注意：get_md5_hash_password 使用 MD5 哈希，这是 simplejwt 库的默认行为。
|
||
# 此处仅用于令牌吊销校验（判断密码是否变更），不用于密码存储。
|
||
# 如需更安全的哈希算法，需升级 simplejwt 版本或自定义校验逻辑。
|
||
from rest_framework_simplejwt.utils import get_md5_hash_password
|
||
if not hmac.compare_digest(
|
||
str(validated_token.get(api_settings.REVOKE_TOKEN_CLAIM, '')),
|
||
str(get_md5_hash_password(user.UserPassword))
|
||
):
|
||
raise AuthenticationFailed(
|
||
_('用户密码已更改'), code='password_changed'
|
||
)
|
||
|
||
return user
|