Apollo客户端订阅传递的JWT令牌如何被Django Channels中间件处理

解决方案 #

方案1：动态拼接令牌到WebSocket查询字符串（适合临时测试） #

可以将令牌拼接到WebSocket连接的查询参数中，中间件可以直接读取该参数，支持动态获取最新令牌。

客户端修改

调整WebSocketLink配置，开启lazy模式，将uri改为动态生成的函数，每次建立连接时自动读取最新的令牌拼接：

const wsLink = new WebSocketLink({
  // uri改为函数，每次新建连接时执行获取最新令牌
  uri: async () => {
    const baseWsUri = process.env.VUE_APP_WS_GRAPHQL_URL || "ws://0.0.0.0:8000/ws/graphql/"
    const { value: authStr } = await Storage.get({ key: "auth" });
    if (!authStr) return baseWsUri
    const auth = JSON.parse(authStr)
    return `${baseWsUri}?token=${encodeURIComponent(auth.token)}`
  },
  options: {
    reconnect: true,
    lazy: true, // 延迟初始化连接，有订阅请求时才建立连接
  },
});

如果用户登录/刷新令牌后需要立刻更新连接，手动调用wsLink.subscriptionClient.close()触发自动重连即可，新连接会携带最新的令牌。

后端中间件修改

在中间件中解析查询字符串获取令牌：

from urllib.parse import parse_qs
from django.contrib.auth.models import AnonymousUser

class JwtAuthMiddleware(BaseMiddleware):
    def __init__(self, inner):
        self.inner = inner

    async def __call__(self, scope, receive, send):
        close_old_connections()
        scope["user"] = AnonymousUser()
        
        # 解析查询参数
        query_str = scope.get("query_string", b"").decode()
        query_params = parse_qs(query_str)
        token = query_params.get("token", [None])[0]
        
        if token:
            try:
                # 替换为你自己的JWT解码校验逻辑
                decoded_data = jwt_decode(token)
                scope["user"] = await get_user(email=decoded_data["email"])
            except Exception:
                pass
        
        return await super().__call__(scope, receive, send)

注意：该方案令牌会出现在URL中，可能被服务日志记录，生产环境使用需要做好日志脱敏，更推荐使用方案2。

方案2：在Consumer层完成认证（生产环境推荐） #

connectionParams本身就是设计用来传递WebSocket连接的元数据，直接在Consumer层处理认证更符合协议规范，无需修改现有客户端代码。

后端修改

不用调整现有中间件，直接重写你的MyGraphqlWsConsumer的connect方法，读取connection_params完成认证：

from channels_graphql_ws import GraphqlWsConsumer
from django.contrib.auth.models import AnonymousUser

class MyGraphqlWsConsumer(GraphqlWsConsumer):
    # 保留你原来的schema等配置
    async def connect(self):
        await self.accept()
        # 读取Apollo传递的connectionParams
        token = self.connection_params.get("token")
        self.scope["user"] = AnonymousUser()
        
        if token:
            try:
                # 替换为你自己的JWT解码校验逻辑
                decoded_data = jwt_decode(token)
                self.scope["user"] = await get_user(email=decoded_data["email"])
            except Exception:
                # 认证失败可以选择直接关闭连接
                # await self.close(code=4401)
                pass
        
        # 执行父类原有逻辑
        await super().connect()

该方案优势：

• 无需修改现有客户端代码，你已经配置的connectionParams可以直接复用
• 令牌不会出现在URL中，安全性更高
• 每次重连都会自动获取最新的令牌，符合动态取值的需求
• 符合GraphQL over WebSocket的协议设计规范

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