Django API网关自定义中间件无法获取已认证用户，权限校验失败

问题背景 #

我需要基于Django搭建一个API网关微服务，核心需求是校验微服务1请求微服务2的URL是否具备权限，但网关项目中并未配置微服务2的URL。尝试了两种方案均存在问题：

• 自定义权限方案：权限校验逻辑正常，但由于网关没有对应URL的路由配置，校验通过后仍返回404响应（该URL实际存在于微服务2）。
• 自定义中间件方案：无法获取已认证用户，导致无法校验请求用户的权限，认证方式采用django-rest-framework-simplejwt实现的JWT认证。

相关配置与代码 #

Settings.py #

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': (
        'rest_framework_simplejwt.authentication.JWTAuthentication',
    )
}

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
    'core.middleware.ReverseProxyMiddleware',
]

BY_PASS_URLS = ["/api/token/", "/api/token/refresh/", "/admin"]

custom_middleware.py #

class ReverseProxyMiddleware(MiddlewareMixin):
    def __init__(self, get_response):
        self.get_response = get_response

    def process_request(self, request):

        print(request.user.is_authenticated)
        
        for each_path in settings.BY_PASS_URLS:
            if each_path in request.path:
                response = self.get_response(request)
                return response
        
        if self.check_permission(request):
            response = self.get_response(request)
            return response
        else:
            return HttpResponseForbidden("Permission denied")

    def check_permission(self, request):
         
        perm = APICheckPermission()
        return perm.has_permission(request)

日志输出 #

2023-12-09 17:43:34 False AnonymousUser
2023-12-09 17:43:34 Forbidden: /some_url
2023-12-09 17:43:34 [09/Dec/2023 17:43:34] "GET /some_url HTTP/1.1" 403 17

响应结果 #

Permission denied

models.py #

class APIEndpoint(models.Model):
    url = models.CharField(max_length=255, unique=True)
    http_method = models.CharField(max_length=10)
    groups = models.ManyToManyField(Group)

    def __str__(self):
        return f"{self.http_method} {self.url}"

custom_permission.py #

class APICheckPermission(permissions.BasePermission):
    """
    Global permission check for blocked IPs.
    """

    def has_permission(self, request, view):
        print("permission check")
        url = request.path
        method = request.method

        print(url, method, request.user)
        if request.user.groups.filter(apiendpoint__url__icontains=url, apiendpoint__http_method__icontains=method).exists():
            return True
        return False

urls.py #

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
    path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
]

问题原因分析 #

1. 自定义权限方案的404问题：DRF的权限校验逻辑触发于视图执行前，但如果网关未配置对应URL的路由，Django会直接返回404响应，根本无法走到权限校验后的转发环节。
2. 中间件无法获取认证用户的问题：JWT认证是DRF视图层的逻辑，自定义中间件执行时DRF尚未解析token，因此request.user仍为匿名用户。

解决方案 #

方案1：改进中间件，手动解析JWT获取用户 #

在中间件中手动调用simplejwt的认证逻辑，提前解析token并绑定用户到request对象：

修改custom_middleware.py：

from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework.exceptions import AuthenticationFailed

class ReverseProxyMiddleware(MiddlewareMixin):
    def __init__(self, get_response):
        self.get_response = get_response
        self.jwt_auth = JWTAuthentication()

    def process_request(self, request):
        # 跳过无需校验的URL
        for each_path in settings.BY_PASS_URLS:
            if each_path in request.path:
                return self.get_response(request)
        
        # 手动执行JWT认证
        try:
            user, token = self.jwt_auth.authenticate(request)
            request.user = user
            request.auth = token
        except AuthenticationFailed:
            return HttpResponseForbidden("Invalid or missing token")
        
        # 校验权限
        if self.check_permission(request):
            # 此处添加转发到微服务2的逻辑，示例：
            # import requests
            # target_url = f"http://微服务2地址{request.path}"
            # response = requests.request(request.method, target_url, headers=request.headers, data=request.body)
            # return HttpResponse(response.content, status=response.status_code, headers=dict(response.headers))
            return self.get_response(request)
        else:
            return HttpResponseForbidden("Permission denied")

    def check_permission(self, request):
        perm = APICheckPermission()
        # 中间件无view参数，传入None
        return perm.has_permission(request, None)

修改custom_permission.py，允许view参数为空：

class APICheckPermission(permissions.BasePermission):
    def has_permission(self, request, view=None):
        url = request.path
        method = request.method

        if request.user.is_authenticated and request.user.groups.filter(
            apiendpoint__url__icontains=url, 
            apiendpoint__http_method__icontains=method
        ).exists():
            return True
        return False

方案2：配置通配路由+视图转发 #

在网关中配置通配路由，通过统一视图处理权限校验和请求转发，既利用DRF的认证/权限系统，又解决404问题：

修改urls.py：

from django.contrib import admin
from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
from core.views import GatewayProxyView

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
    path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
    # 通配所有未匹配的路径
    path('<path:path>', GatewayProxyView.as_view()),
    # 匹配根路径
    path('', GatewayProxyView.as_view()),
]

创建core/views.py：

import requests
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated
from core.permissions import APICheckPermission
from django.http import HttpResponse

class GatewayProxyView(APIView):
    permission_classes = [IsAuthenticated, APICheckPermission]

    def dispatch(self, request, *args, **kwargs):
        # 转发请求到微服务2
        target_service_url = "http://微服务2实际地址"
        target_url = f"{target_service_url}/{kwargs.get('path', '')}"
        
        # 过滤Django特定请求头
        headers = {k: v for k, v in request.headers.items() if k != 'Host'}
        
        # 转发请求并返回响应
        response = requests.request(
            method=request.method,
            url=target_url,
            headers=headers,
            data=request.body,
            params=request.query_params
        )
        
        return HttpResponse(
            content=response.content,
            status=response.status_code,
            headers=dict(response.headers)
        )

注意事项 #

• 转发请求时需处理好请求头、请求体和参数，避免信息丢失。
• 权限校验中的apiendpoint__url__icontains匹配逻辑可根据需求优化为精确匹配或正则匹配。
• 中间件方案需完善异常处理，比如token过期、无效等场景。

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