import threading from datetime import datetime, timedelta from pathlib import Path from zoneinfo import ZoneInfo from Crypto.PublicKey import RSA from django.conf import settings from jwt import JWT, jwk_from_pem from jwt.utils import get_int_from_datetime from rest_framework import status from rest_framework.decorators import action from rest_framework.request import Request from rest_framework.response import Response from rest_framework.viewsets import ViewSet from django_jwt_token.jwt认证接口 import Jwt身份认证 private_key_pem_私钥 = Path( settings.BASE_DIR ).joinpath( "django_jwt_token/private.pem" ).read_bytes( ) class 用户登陆接口ViewSets( ViewSet ): lock = threading.Lock( ) instance = JWT( ) # 创建对象 @action( detail=False, methods=[ "post" ], url_path="登陆" ) def 登陆( self, request: Request ): """""" message = { 'iss': 'my-app', 'sub': 'user123', 'name': '张三', 'role': 'admin', 'iat': get_int_from_datetime( datetime.now( tz=ZoneInfo( key="utc" ) ) ), 'exp': get_int_from_datetime( datetime.now( tz=ZoneInfo( key="utc" ) ) + timedelta( hours=1 ) ), } # ✅ 正确:使用私钥进行签名 私钥对象 = RSA.import_key( extern_key=private_key_pem_私钥, passphrase=settings.SECRET_KEY ).export_key( ) signing_key = jwk_from_pem( pem_content=私钥对象 ) token = self.instance.encode( payload=message, key=signing_key, alg='RS256' ) # 注意算法名改为 RS256 print( f"生成的 JWT: {token}" ) return Response( data=token, status=status.HTTP_200_OK, headers={ "Authorization": token, "Access-Control-Allow-Origin": "*" } ) @action( detail=False, methods=[ "post" ], url_path="验证用户身份", authentication_classes=[ Jwt身份认证 ] ) def 验证用户身份( self, request: Request ): """""" # 模拟一个可能导致错误的操作 # ✅ 正确:使用公钥进行验证 print( "只是打印测试一下" ) print( request.user ) print( request.auth ) return Response( data={ "code": 200, "msg": "Success", "data": "" }, )