Vue3前端通过DCRF连接Django4 Consumer的WebSocket鉴权问题

问题背景 #

我有一个Django 4.x项目，采用Token鉴权（通过djoser管理），HTTP请求用fetch/axios在请求头传递Token能正常完成鉴权。但使用Django Channels REST Framework（DCRF）配合Demultiplexer处理WebSocket路由时，consumer.py中自定义操作里的self.scope.get('user')始终返回AnonymousUser，导致鉴权失败返回"Not allowed"提示。将consumer的permission_classes设置为permissions.AllowAny虽能让所有操作正常运行，但存在安全风险，希望在前后端分离的架构下实现WebSocket连接的用户鉴权。

解决方案 #

1. 编写自定义Token鉴权中间件 #

在项目根目录创建middleware/websocket_auth.py，实现从WebSocket连接中提取并验证Token，关联到当前用户：

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

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

    async def __call__(self, scope, receive, send):
        # 从URL查询参数中获取Token
        query_string = scope.get('query_string', b'').decode()
        token = None
        if query_string:
            params = dict(param.split('=') for param in query_string.split('&'))
            token = params.get('token')
        
        # 也支持从请求头提取（比如前端自定义header传递）
        # headers = dict(scope['headers'])
        # if b'authorization' in headers:
        #     auth_header = headers[b'authorization'].decode()
        #     if auth_header.startswith('Token '):
        #         token = auth_header.split(' ')[1]

        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()
        else:
            scope['user'] = AnonymousUser()
        
        return await self.inner(scope, receive, send)

2. 修改ASGI配置，替换鉴权中间件 #

更新asgi.py，将默认的AuthMiddlewareStack替换为自定义的TokenAuthMiddleware：

#ASGI config for InterAct project.

import os

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

from django.urls import re_path
from operation.consumers import DMultiplexer

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

django_application = get_asgi_application()

application = ProtocolTypeRouter(
    {
        "http": django_application,
        "websocket": AllowedHostsOriginValidator(
            TokenAuthMiddleware(URLRouter([
                            re_path(r"^ws/$", DMultiplexer.as_asgi()),
                            ]))
        ),
    }
)

3. 前端修改WebSocket连接逻辑，传递Token #

更新Vuex Store的action，在WebSocket连接URL中带上Token参数：

async openEndpoint(state,params){
  console.log(params);
  // 从状态中获取用户Token（根据你的实际存储位置调整）
  const token = state.user.token;
  // 构建带Token的WebSocket连接地址
  let ws = new WebSocket(`ws://localhost:8000/ws/?token=${token}`);
  let endpoint=params.endpoint
  
  ws.onopen=async ()=>{
    console.log('connection established');
    console.log(endpoint);
    
    let data={
      stream:endpoint,
      payload:{
        action:'subscribe_to_groups_feed',
        request_id: new Date().getTime()
      }
    }
    
    ws.onmessage=(ws_event)=>{
      let response=JSON.parse(ws_event.data)
      console.log(response);
      console.log(response.payload.errors);
    }
    ws.send(JSON.stringify(data))
  }
}

4. 恢复Consumer的权限配置 #

将各个Consumer的permission_classes改回permissions.IsAuthenticated，以启用鉴权校验：

class Groups_Consumer(
        ListModelMixin,
        PatchModelMixin,
        UpdateModelMixin,
        CreateModelMixin,
        DeleteModelMixin,
        ObserverModelInstanceMixin,
        GenericAsyncAPIConsumer):

    permission_classes = [permissions.IsAuthenticated]
    queryset=Group.objects.all()
    serializer_class=GroupSerializer
    
    # 其他方法保持不变

验证效果 #

启动项目后，前端通过带Token的URL连接WebSocket，consumer中的self.scope.get('user')即可获取到当前登录用户，所有操作将在鉴权通过后执行，无需开放AllowAny权限。

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