Django迁移至rest_framework_simplejwt：自定义TokenBackend及动态密钥问题

一、不使用JWK_URL从S3动态加载密钥 #

完全可以实现，核心是通过自定义TokenBackend重写密钥获取逻辑，从S3拉取PEM文件内容：

1. 编写S3密钥获取工具函数 #

import boto3
from botocore.exceptions import ClientError
from django.core.cache import cache

def get_s3_key(bucket_name, key_path):
    # 加入缓存避免频繁S3请求，可根据需求调整缓存时长
    cache_key = f"s3_jwt_{key_path}"
    cached_key = cache.get(cache_key)
    if cached_key:
        return cached_key

    s3 = boto3.client('s3')
    try:
        response = s3.get_object(Bucket=bucket_name, Key=key_path)
        key_content = response['Body'].read().decode('utf-8')
        cache.set(cache_key, key_content, 3600)  # 缓存1小时
        return key_content
    except ClientError as e:
        raise ValueError(f"Failed to fetch key from S3: {str(e)}")

2. 自定义TokenBackend并重写密钥方法 #

继承simplejwt默认的TokenBackend，替换密钥获取逻辑：

from rest_framework_simplejwt.backends import TokenBackend

class S3TokenBackend(TokenBackend):
    def get_signing_key(self):
        # 替换为你的S3桶名和私钥路径
        return get_s3_key('your-s3-bucket', 'jwt/private-key.pem')
    
    def get_verifying_key(self):
        # 替换为你的S3桶名和公钥路径
        return get_s3_key('your-s3-bucket', 'jwt/public-key.pem')

3. 配置simplejwt使用自定义后端 #

在settings.py中指定自定义后端：

SIMPLE_JWT = {
    # 保留其他原有配置
    'TOKEN_BACKEND': 'your_app.backends.S3TokenBackend',
}

二、自定义编解码逻辑替代原JWT_DECODE_HANDLER/JWT_ENCODE_HANDLER #

通过继承TokenBackend并重写encode()和decode()方法，实现自定义编解码逻辑：

1. 编写带自定义逻辑的TokenBackend #

from rest_framework_simplejwt.backends import TokenBackend
from rest_framework_simplejwt.exceptions import InvalidToken

class CustomTokenBackend(TokenBackend):
    def encode(self, payload):
        # 编码前添加自定义负载字段或修改现有字段
        payload['app_version'] = 'v1.0.0'
        # 调用父类方法完成实际JWT编码
        return super().encode(payload)
    
    def decode(self, token, verify=True):
        # 先调用父类方法完成基础解码
        payload = super().decode(token, verify)
        # 解码后验证自定义字段或修改负载
        if 'app_version' not in payload:
            raise InvalidToken("Token missing required app version field")
        # 可对负载做后续处理
        payload['is_valid_version'] = payload['app_version'].startswith('v1')
        return payload

2. 合并S3密钥与自定义编解码（可选） #

如果需要同时实现两种功能，合并到同一个类即可：

class S3CustomTokenBackend(TokenBackend):
    def get_signing_key(self):
        return get_s3_key('your-s3-bucket', 'jwt/private-key.pem')
    
    def get_verifying_key(self):
        return get_s3_key('your-s3-bucket', 'jwt/public-key.pem')
    
    def encode(self, payload):
        payload['app_version'] = 'v1.0.0'
        return super().encode(payload)
    
    def decode(self, token, verify=True):
        payload = super().decode(token, verify)
        if 'app_version' not in payload:
            raise InvalidToken("Token missing required app version field")
        payload['is_valid_version'] = payload['app_version'].startswith('v1')
        return payload

3. 配置启用 #

同样在settings.py中指定该类：

SIMPLE_JWT = {
    # 其他配置...
    'TOKEN_BACKEND': 'your_app.backends.S3CustomTokenBackend',
}

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