53 lines
2.0 KiB
Python
53 lines
2.0 KiB
Python
"""
|
||
自定义 JWT 认证：兼容旧版 token（claim 为 user_id）与新版（claim 为 yonghuid）。
|
||
"""
|
||
from django.utils.translation import gettext_lazy as _
|
||
from rest_framework_simplejwt.authentication import JWTAuthentication
|
||
from rest_framework_simplejwt.exceptions import AuthenticationFailed, InvalidToken
|
||
from rest_framework_simplejwt.settings import api_settings
|
||
|
||
|
||
class YonghuidJWTAuthentication(JWTAuthentication):
|
||
"""签发与校验均以 yonghuid 为准，同时兼容历史 user_id claim。"""
|
||
|
||
def get_user(self, validated_token):
|
||
user_id, lookup_field = self._resolve_user_identity(validated_token)
|
||
if user_id is None:
|
||
raise InvalidToken(_('Token contained no recognizable user identification'))
|
||
|
||
try:
|
||
return self.user_model.objects.get(**{lookup_field: user_id})
|
||
except self.user_model.DoesNotExist as exc:
|
||
raise AuthenticationFailed(_('User not found'), code='user_not_found') from exc
|
||
|
||
def _resolve_user_identity(self, validated_token):
|
||
claim = api_settings.USER_ID_CLAIM
|
||
field = api_settings.USER_ID_FIELD
|
||
|
||
if claim in validated_token:
|
||
return str(validated_token[claim]), field
|
||
|
||
legacy_id = validated_token.get('user_id')
|
||
if legacy_id is None:
|
||
return None, field
|
||
|
||
legacy_id = str(legacy_id)
|
||
# 旧 token 可能写入主键 id，需回退按 id 查询
|
||
if field != 'id' and legacy_id.isdigit():
|
||
if self.user_model.objects.filter(id=int(legacy_id)).exists():
|
||
return legacy_id, 'id'
|
||
return legacy_id, field
|
||
|
||
|
||
class OptionalYonghuidJWTAuthentication(YonghuidJWTAuthentication):
|
||
"""可选登录：有合法 token 则识别用户，无/坏 token 当匿名，不打断 AllowAny 接口。"""
|
||
|
||
def authenticate(self, request):
|
||
header = self.get_header(request)
|
||
if header is None:
|
||
return None
|
||
try:
|
||
return super().authenticate(request)
|
||
except (InvalidToken, AuthenticationFailed):
|
||
return None
|