Django Channels身份验证最佳实践及安全风险咨询

当前环境：Django 4.1.4、djoser 2.1.0、channels 4.0.0

我按照文档推荐实现了自定义中间件，在WebSocket连接时通过查询字符串传递用户ID来验证用户，已经能成功获取用户并检查user.is_authenticated，但用户对象不会自动出现在WebSocket的scope里。

我不确定这种方式的安全性：虽然文档说这种方法不安全，但我已经做了用户认证检查，觉得应该已经做好了安全防护。另外我觉得用djoser生成的token来验证更好，但不知道怎么在WebSocket请求里发送请求头，所以考虑把token替代用户ID放到查询字符串里，想了解相关的最佳实践。

相关代码 #

前端传递用户ID的代码 #

websocket.value = new WebSocket(`ws://127.0.0.1:8000/ws/marketwatch/? ${authStore.userId}`)

自定义认证中间件（middleware.py） #

from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist


@database_sync_to_async
def get_user(user_id):
    User = get_user_model()
    try:
        user = User.objects.get(id=user_id)
    except ObjectDoesNotExist:
        return AnonymousUser()
    else:
        if user.is_authenticated:
            return user
        else:
            return AnonymousUser()

class QueryAuthMiddleware:
    
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        
        scope['user'] = await get_user(int(scope["query_string"].decode()))

        return await self.app(scope, receive, send)

ASGI 应用配置（consumers.py） #

import os

from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from channels.security.websocket import AllowedHostsOriginValidator

from api.middleware import QueryAuthMiddleware
from .routing import ws_urlpatterns

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'api.settings')

application = ProtocolTypeRouter({
    'http':get_asgi_application(),
    'websocket': AllowedHostsOriginValidator(
        QueryAuthMiddleware(
            URLRouter(ws_urlpatterns)
        )
    )
})

最佳实践方案 #

1. 修复用户无法进入scope的问题 #

你当前的中间件已经手动给scope['user']赋值了，理论上在消费者里可以通过self.scope['user']直接访问。如果拿不到用户，核心问题是查询字符串解析错误：你的前端代码里查询字符串带了空格（? ${authStore.userId}），导致解析出的字符串包含空格，转int时会报错。

修正代码：

• 前端调整查询字符串格式：

websocket.value = new WebSocket(`ws://127.0.0.1:8000/ws/marketwatch/?user_id=${authStore.userId}`)

• 中间件修改解析逻辑，提取指定参数：

async def __call__(self, scope, receive, send):
    query_string = scope["query_string"].decode()
    # 解析查询参数键值对
    params = dict(qc.split('=') for qc in query_string.split('&') if qc)
    user_id = params.get('user_id')
    
    if user_id and user_id.isdigit():
        scope['user'] = await get_user(int(user_id))
    else:
        scope['user'] = AnonymousUser()
        
    return await self.app(scope, receive, send)

2. 查询字符串传用户ID的安全风险 #

即便你做了user.is_authenticated检查，这种方式依然存在致命风险：

• 用户ID属于低敏感度信息，极易被猜测、窃取，攻击者拿到其他用户ID就能冒充建立连接
• 查询字符串会被记录在服务器日志、代理日志中，容易泄露用户信息

必须换成基于token的认证方式。

3. WebSocket中使用Djoser Token的可行方案 #

WebSocket标准不支持自定义请求头，常用的两种token传递方式如下：

方案一：查询字符串传递Token（跨域/前后端分离场景）

生产环境必须使用wss://（加密WebSocket），防止token明文泄露。

• 前端传递token：

// 假设authStore存储了Djoser生成的access token
websocket.value = new WebSocket(`wss://your-domain.com/ws/marketwatch/?token=${authStore.accessToken}`)

• 中间件验证token（Djoser默认依赖simplejwt，直接用其解析逻辑）：

from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
from django.contrib.auth import get_user_model
from rest_framework_simplejwt.tokens import AccessToken
from rest_framework.exceptions import AuthenticationFailed

User = get_user_model()

@database_sync_to_async
def get_user_from_token(token):
    try:
        access_token = AccessToken(token)
        user = User.objects.get(id=access_token['user_id'])
        return user if user.is_authenticated else AnonymousUser()
    except (AuthenticationFailed, User.DoesNotExist, ValueError):
        return AnonymousUser()

class TokenAuthMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        query_string = scope["query_string"].decode()
        params = dict(qc.split('=') for qc in query_string.split('&') if qc)
        token = params.get('token')
        
        scope['user'] = await get_user_from_token(token) if token else AnonymousUser()
        return await self.app(scope, receive, send)

• 更新consumers.py中的中间件为TokenAuthMiddleware即可。

方案二：Cookie传递Token（同域场景）

如果前后端同域，可以配置Djoser将token存储在HttpOnly Cookie中，然后直接使用Channels内置的AuthMiddlewareStack自动获取认证用户，无需自定义中间件。

4. 额外安全加固建议 #

• 生产环境强制使用wss://，禁止明文WebSocket连接
• 给WebSocket连接设置token过期重验机制，定期刷新认证状态
• 可选：在中间件中添加IP校验，限制请求来源为用户常用IP

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