Django处理接口端点请求前验证AWS Cognito令牌有效性的方案咨询

第一步：实现DRF自定义认证类 #

DRF的认证体系天生兼容类视图和函数视图两种模式，只需要实现标准的自定义认证类，就能在两种视图中复用同一套鉴权逻辑。你可以在应用目录下新建authentication.py文件，写入以下代码：

from rest_framework import authentication
from rest_framework.exceptions import AuthenticationFailed
from .core.api import jwt
from django.contrib.auth import get_user_model

User = get_user_model()

class CognitoJWTAuthentication(authentication.BaseAuthentication):
    def authenticate(self, request):
        # 提取请求头中的Authorization字段
        auth_header = request.META.get('HTTP_AUTHORIZATION')
        if not auth_header:
            return None
        # 校验Bearer格式合法性
        parts = auth_header.split()
        if parts[0].lower() != 'bearer' or len(parts) != 2:
            raise AuthenticationFailed('Authorization头格式错误，应为 Bearer [token]')
        token = parts[1]
        # 调用Cognito JWT校验逻辑
        try:
            decoded_payload = jwt.decode_cognito_jwt(token)
        except Exception as e:
            raise AuthenticationFailed(f'无效JWT: {str(e)}')
        
        # 可选逻辑：将Cognito用户和Django本地用户关联，不需要可直接返回(user, None)
        cognito_user_id = decoded_payload.get('sub')
        if not cognito_user_id:
            raise AuthenticationFailed('JWT缺少用户标识')
        # 不存在则自动创建对应用户，可根据业务逻辑调整
        user, created = User.objects.get_or_create(username=cognito_user_id)
        return (user, None)

第二步：完善JWT校验的合规性 #

你当前的校验逻辑存在安全风险，decode_cognito_jwt方法必须完成以下校验才能保证鉴权合规：

• 必须用AWS Cognito用户池公开的JWK公钥校验JWT签名，防止伪造令牌
• 必须校验exp过期时间字段，拒绝过期令牌
• 必须校验iss发行人字段，值为你的Cognito用户池URL（格式为https://cognito-idp.<区域>.amazonaws.com/<用户池ID>）
• 必须校验aud受众字段，值为你的AWS Cognito客户端ID
• 必须校验token_use字段，确认令牌类型符合预期（id令牌/访问令牌）

第三步：全局配置认证规则 #

在项目settings.py中添加DRF全局配置，这样所有视图默认都会走Cognito鉴权，不需要逐个接口添加代码：

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        '你的应用名.authentication.CognitoJWTAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated', # 强制要求用户处于登录状态
    ]
}

第四步：视图适配 #

全局配置后两种视图默认都会自动应用鉴权逻辑，仅需要对豁免鉴权的接口单独配置即可：

APIView类视图适配 #

from rest_framework.permissions import AllowAny

class LoginView(APIView):
    # 仅当该接口需要豁免鉴权时添加该配置，比如登录回调、token校验接口
    permission_classes = [AllowAny]
    def post(self, request):
        # 你的原有逻辑，注意修正笔误：status.Http_200_OK 应为全大写的 status.HTTP_200_OK
        token = request.META['HTTP_AUTHORIZATION'].split(' ')[1]
        try:
            res = jwt.decode_cognito_jwt(token)
            return Response(status=status.HTTP_200_OK)
        except:
            return Response("Invalid JWT", status=status.HTTP_401_UNAUTHORIZED)

@api_view装饰器修饰的函数视图适配 #

@api_view(['GET'])
@swagger_auto_schema(
    operation_description="Get Goals joined by User"
)
def get_goals_by_user(request, user_id):
    # 鉴权通过后可以直接通过request.user拿到当前登录的用户对象
    print("当前登录Cognito用户ID：", request.user.username)
    try:
        goals_query = JoinGoal.objects.filter(
            joiner_id=user_id).values_list('goal_id', flat=True)
        goals_list = list(goals_query)
        data = list(Goal.objects.filter(
            pk__in=goals_list).values('description', 'uuid'))
        response_data = dict(goals=data)
        return JsonResponse(response_data, status=status.HTTP_200_OK)
    except JoinGoal.DoesNotExist:
        return Response(dict(error=does_not_exist_msg(JoinGoal.__name__, 'joiner_id', user_id)), status=status.HTTP_400_BAD_REQUEST)

内容的提问来源于stack exchange，提问作者user12314098