在Django中间件更新请求Header时遭遇JWT未授权问题

问题场景 #

使用Nuxt作为前端框架，后端基于Django实现JWT认证。需求是当access token过期但refresh token仍有效时，自动校验并刷新access token。编写了JWTRefreshMiddleware中间件尝试更新请求的Authorization Header，但调用带JWTAuth认证的/profile接口时仍返回未授权。尝试直接赋值request.headers["Authorization"]时触发错误：TypeError: 'HttpHeaders' object does not support item assignment。

问题代码 #

自定义刷新中间件 #

class JWTRefreshMiddleware:

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request: HttpRequest):
        if request.headers.get("Authorization"):
            try:
                AccessToken(
                    str(request.headers.get("Authorization")).split(" ")[1], verify=True
                )
            except TokenError:
                if refresh_token_cookie := request.COOKIES.get("refresh_token"):
                    try:
                        refresh_token = RefreshToken(refresh_token_cookie, verify=True)
                        user = CustomUser.objects.get(
                            id=refresh_token.payload.get("user_id")
                        )
                        new_access_token = str(AccessToken.for_user(user))

                        request.META["HTTP_AUTHORIZATION"] = (
                            f"Bearer {new_access_token}"
                        )

                    except TokenError:
                        pass

        response: HttpResponse = self.get_response(request)

        return response

受保护接口 #

from ninja_jwt.authentication import JWTAuth

router = Router()

@router.get("/profile", response={200: UserProfileResponse}, auth=JWTAuth())
def get_user_profile(request):
    # 业务逻辑代码

问题原因分析 #

1. 中间件执行顺序错误：如果JWTRefreshMiddleware在Django自带的AuthenticationMiddleware或JWTAuth认证逻辑之后执行，此时认证已经完成，修改request.META不会影响后续的认证校验。
2. 异常捕获范围过宽：原代码捕获所有TokenError，包括无效签名、格式错误等非过期场景，导致不必要的刷新逻辑触发，同时可能忽略了部分token异常的处理。
3. Authorization头解析不严谨：未校验Authorization头的格式是否为Bearer <token>，直接split可能导致索引错误。

解决方案 #

1. 调整中间件执行顺序 #

在settings.py的MIDDLEWARE列表中，将自定义刷新中间件放在AuthenticationMiddleware之前，确保在认证逻辑执行前完成token刷新：

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    # 自定义刷新中间件放在此处，优先执行
    'your_app_name.middleware.JWTRefreshMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

2. 优化中间件逻辑 #

细化异常捕获范围，严谨解析Authorization头，并处理用户不存在的情况：

from django.http import HttpRequest, HttpResponse
from ninja_jwt.tokens import AccessToken, RefreshToken, TokenError, ExpiredSignatureError
from your_app.models import CustomUser

class JWTRefreshMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request: HttpRequest):
        auth_header = request.headers.get("Authorization")
        # 仅处理Bearer格式的Authorization头
        if auth_header and auth_header.startswith("Bearer "):
            token = auth_header.split(" ")[1]
            try:
                # 验证access token是否有效
                AccessToken(token, verify=True)
            except ExpiredSignatureError:
                # 仅当access token过期时尝试刷新
                refresh_token_cookie = request.COOKIES.get("refresh_token")
                if refresh_token_cookie:
                    try:
                        refresh_token = RefreshToken(refresh_token_cookie, verify=True)
                        user_id = refresh_token.payload.get("user_id")
                        user = CustomUser.objects.get(id=user_id)
                        new_access_token = str(AccessToken.for_user(user))
                        # 更新META中的Authorization头，供后续认证使用
                        request.META["HTTP_AUTHORIZATION"] = f"Bearer {new_access_token}"
                    except (TokenError, CustomUser.DoesNotExist):
                        # 处理refresh token无效或用户不存在的情况，不做操作
                        pass
            except TokenError:
                # 处理其他token错误（如无效签名、篡改等），不做操作
                pass

        response = self.get_response(request)
        return response

3. 确保Refresh Token的安全设置 #

登录时设置refresh token的cookie为HttpOnly，避免前端XSS攻击：

# 示例登录视图中的cookie设置
response = HttpResponse(...)
refresh_token = RefreshToken.for_user(user)
response.set_cookie(
    'refresh_token',
    str(refresh_token),
    httponly=True,
    secure=True,  # 生产环境开启HTTPS时启用
    samesite='Lax',
    max_age=refresh_token.lifetime.total_seconds()
)

验证方法 #

在中间件中添加打印语句，确认request.META["HTTP_AUTHORIZATION"]是否被正确更新：

# 在设置META后添加
print(f"Updated Authorization: {request.META.get('HTTP_AUTHORIZATION')}")

同时在get_user_profile视图中打印请求的META信息，确认认证时能获取到新的access token。

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