如何在Postman中为原生WebSocket连接传递Bearer Token

之前尝试失效的核心原因 #

• 自定义Authorization请求头方案：浏览器环境下原生WebSocket构造函数不支持自定义请求头是W3C标准的硬性限制，和Django Channels逻辑无关，该方案从API层面就无法实现，非浏览器环境（如Node.js的ws库）才支持自定义头配置。
• URL拼接查询参数方案：思路本身完全可行，失效原因是Django Channels默认的AuthMiddlewareStack不会主动解析URL查询字符串内的token凭证，没有配套修改服务端鉴权逻辑的情况下，服务端自然拿不到传入的参数。
• 连接建立后发token消息方案：思路本身可行，失效有两个原因：一是Channels默认在握手完成、connect()方法执行时就会完成权限校验，连接建立后再发消息时鉴权流程已经结束；二是你尝试发送的40{"token":token}是Channels内部通信的帧格式，客户端直接发送该格式会被服务端解析失败。

可直接落地的生产级方案 #

方案1：URL查询参数传参（适配所有原生WebSocket场景，实现成本最低） #

客户端连接代码

直接将token作为查询参数拼接在连接地址末尾，参数名可自定义，和服务端解析逻辑保持一致即可：

const socket = new WebSocket(`ws://localhost:8000/ws/chat/1/?token=${替换为你的实际BearerToken值}`);

服务端自定义鉴权中间件

新建中间件文件，实现查询参数解析、token校验逻辑：

# channels_auth_middleware.py
from urllib.parse import parse_qs
from channels.middleware import BaseMiddleware
from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
from rest_framework.authtoken.models import Token  # 如果你用JWT，替换为JWT校验逻辑即可

class TokenQueryAuthMiddleware(BaseMiddleware):
    async def __call__(self, scope, receive, send):
        # 解析握手阶段的URL查询参数
        query_params = parse_qs(scope["query_string"].decode())
        token = query_params.get("token", [None])[0]
        scope["user"] = AnonymousUser()
        
        if token:
            try:
                token_obj = await database_sync_to_async(Token.objects.get)(key=token)
                scope["user"] = token_obj.user
            except Token.DoesNotExist:
                scope["user"] = AnonymousUser()
        
        return await super().__call__(scope, receive, send)

def TokenAuthMiddlewareStack(inner):
    return TokenQueryAuthMiddleware(inner)

替换ASGI配置中的默认鉴权中间件

修改项目asgi.py文件，将默认的AuthMiddlewareStack替换为自定义的中间件栈：

# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from .channels_auth_middleware import TokenAuthMiddlewareStack
import chat.routing

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "替换为你的项目.settings")

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AllowedHostsOriginValidator(
        TokenAuthMiddlewareStack(
            URLRouter(chat.routing.websocket_urlpatterns)
        )
    ),
})

配置完成后，在Consumer内可直接通过self.scope["user"]获取鉴权后的用户对象，在connect()方法中判断如果是匿名用户直接调用await self.close()拒绝连接即可。

方案2：连接建立后首条消息传参（避免token出现在URL中，降低日志泄露风险） #

如果担心URL传参可能被代理、服务器日志记录导致token泄露，可采用首条消息鉴权的方案，需要调整Consumer的鉴权时序：

客户端连接代码

连接建立成功后第一时间发送鉴权消息，不要插入其他逻辑：

const socket = new WebSocket("ws://localhost:8000/ws/chat/1/");
socket.onopen = () => {
  // 直接发送标准JSON格式的鉴权消息即可，不需要加额外帧前缀
  socket.send(JSON.stringify({
    action: "auth",
    token: "替换为你的实际BearerToken值"
  }));
};

服务端Consumer鉴权逻辑调整

改写Consumer的连接和消息处理逻辑，将鉴权时机从连接建立阶段延后到收到首条鉴权消息阶段：

from channels.generic.websocket import AsyncJsonWebsocketConsumer
from channels.db import database_sync_to_async
from rest_framework.authtoken.models import Token
from django.contrib.auth.models import AnonymousUser

class ChatConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        self.room_id = self.scope["url_route"]["kwargs"]["room_id"]
        self.room_group = f"chat_{self.room_id}"
        self.is_authorized = False
        # 先接受连接，但默认标记为未授权
        await self.accept()

    async def receive_json(self, content, **kwargs):
        # 处理鉴权消息
        if content.get("action") == "auth" and not self.is_authorized:
            token = content.get("token")
            try:
                token_obj = await database_sync_to_async(Token.objects.get)(key=token)
                self.scope["user"] = token_obj.user
                self.is_authorized = True
                # 鉴权通过后再加入聊天群组、开放消息收发权限
                await self.channel_layer.group_add(self.room_group, self.channel_name)
                await self.send_json({"code": 0, "msg": "auth success"})
            except Token.DoesNotExist:
                await self.send_json({"code": 401, "msg": "invalid token"})
                await self.close()
            return
        
        # 未授权状态下发送非鉴权消息直接断开连接
        if not self.is_authorized:
            await self.close()
            return
        
        # 后续写正常的聊天消息处理逻辑即可

    async def disconnect(self, close_code):
        if self.is_authorized:
            await self.channel_layer.group_discard(self.room_group, self.channel_name)

补充说明 #

• 如果你使用JWT而非DRF原生Token，只需要将上述代码中查询Token表的逻辑替换为JWT解析、校验逻辑即可，整体流程不需要调整。
• 生产环境使用URL传参方案时，必须使用wss://协议加密连接，避免token明文传输泄露。

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