"""自定义 JWT 认证 — 向后兼容旧 token 的 user_id claim""" from django.core.cache import cache 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 认证类。 旧 token(USER_ID_CLAIM='user_id' 时生成)payload 中使用 'user_id' claim, 值为 user.id(UUID 字符串)。 新 token(USER_ID_CLAIM='UserUID' 时生成)payload 中使用 'UserUID' claim, 值为 user.UserUID(7位业务ID)。 此类先尝试新 claim,找不到时回退到旧 claim,并用对应字段查库。 用户对象缓存 300 秒(Redis),避免每个 API 都打一次 MySQL。 """ def get_user(self, validated_token): user_id = validated_token.get(api_settings.USER_ID_CLAIM) lookup_field = api_settings.USER_ID_FIELD if user_id is None: user_id = validated_token.get('user_id') if user_id is None: raise InvalidToken( {'detail': _('令牌未包含用户标识符'), 'code': 'token_not_valid'} ) lookup_field = 'UserUUID' cache_key = f'jwt_user:{lookup_field}:{user_id}' user = cache.get(cache_key) if user is None: try: user = self.user_model.objects.get(**{lookup_field: user_id}) except self.user_model.DoesNotExist: raise AuthenticationFailed(_('用户不存在'), code='user_not_found') cache.set(cache_key, user, 300) if not user.is_active: raise AuthenticationFailed(_('用户已被禁用'), code='user_inactive') if api_settings.CHECK_REVOKE_TOKEN: from rest_framework_simplejwt.utils import get_md5_hash_password if validated_token.get( api_settings.REVOKE_TOKEN_CLAIM ) != get_md5_hash_password(user.password): raise AuthenticationFailed( _('用户密码已更改'), code='password_changed' ) return user