forked from erp-dev/erp
84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
from ninja.security import HttpBearer
|
||
from rest_framework_simplejwt.authentication import JWTAuthentication
|
||
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ==================== 自定义 JWT 认证 for Django Ninja ====================
|
||
|
||
class JWTAuth(HttpBearer):
|
||
"""
|
||
Django Ninja 的 JWT 认证类
|
||
兼容 rest_framework_simplejwt
|
||
|
||
与 DRF 使用相同的认证逻辑，验证 JWT Token 并获取用户
|
||
"""
|
||
def authenticate(self, request, token):
|
||
jwt_authenticator = JWTAuthentication()
|
||
try:
|
||
# 验证 token 并获取用户
|
||
validated_token = jwt_authenticator.get_validated_token(token)
|
||
user = jwt_authenticator.get_user(validated_token)
|
||
|
||
# 记录认证成功
|
||
logger.debug(f"JWT 认证成功: user={user.username}")
|
||
|
||
return user
|
||
except (InvalidToken, TokenError) as e:
|
||
logger.warning(f"JWT 认证失败: {str(e)}")
|
||
return None
|
||
|
||
|
||
class JWTAuthWithEmployee(HttpBearer):
|
||
"""
|
||
Django Ninja 的 JWT 认证类（要求必须有员工身份）
|
||
|
||
与 CustomTokenObtainPairView 的逻辑一致：
|
||
- 验证 JWT Token
|
||
- 验证用户必须有关联的员工信息
|
||
"""
|
||
|
||
def __call__(self, request):
|
||
"""
|
||
重写 __call__ 方法以添加调试日志
|
||
这个方法会被 Ninja 调用来执行认证
|
||
"""
|
||
logger.info(f"[JWTAuthWithEmployee.__call__] 认证被调用")
|
||
logger.info(f"[JWTAuthWithEmployee.__call__] Authorization Header: {request.META.get('HTTP_AUTHORIZATION', 'None')[:50]}...")
|
||
|
||
# 调用父类的 __call__ 方法
|
||
result = super().__call__(request)
|
||
logger.info(f"[JWTAuthWithEmployee.__call__] 认证结果: {result}")
|
||
return result
|
||
|
||
def authenticate(self, request, token):
|
||
logger.info(f"[JWTAuthWithEmployee.authenticate] 开始认证，token 前 20 字符: {token[:20] if token else 'None'}...")
|
||
|
||
if not token:
|
||
logger.warning(f"[JWTAuthWithEmployee.authenticate] Token 为空")
|
||
return None
|
||
|
||
jwt_authenticator = JWTAuthentication()
|
||
try:
|
||
# 验证 token 并获取用户
|
||
validated_token = jwt_authenticator.get_validated_token(token)
|
||
user = jwt_authenticator.get_user(validated_token)
|
||
|
||
logger.info(f"[JWTAuthWithEmployee.authenticate] Token 验证成功，用户: {user.username}")
|
||
|
||
# 检查用户是否有员工身份（与登录逻辑一致）
|
||
if not hasattr(user, 'employee'):
|
||
logger.warning(f"[JWTAuthWithEmployee.authenticate] 用户 {user.username} 没有关联的员工信息")
|
||
return None # 返回 None 会导致 401 Unauthorized
|
||
|
||
logger.info(f"[JWTAuthWithEmployee.authenticate] 认证成功: user={user.username}, employee={user.employee.name}")
|
||
return user
|
||
|
||
except (InvalidToken, TokenError) as e:
|
||
logger.warning(f"[JWTAuthWithEmployee.authenticate] Token 验证失败: {str(e)}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"[JWTAuthWithEmployee.authenticate] 认证异常: {str(e)}", exc_info=True)
|
||
return None
|