Django Channels携带有效JWT仍未认证，WebSocket连接被拒

项目配置 #

• Django版本：4.1.4
• Django Channels版本：4.0.0
• ASGI服务器：Daphne
• 自定义用户模型：BasicUserProfile（在auth_token字段存储JWT令牌）

相关代码 #

1. JWT认证中间件 CustomAuthMiddleware #

import jwt
from datetime import datetime
from channels.middleware.base import BaseMiddleware
from authentication.models import BasicUserProfile
from django.contrib.auth.models import AnonymousUser
from django.conf import settings
from channels.db import database_sync_to_async

class CustomAuthMiddleware(BaseMiddleware):
    async def populate_scope(self, scope):
        user = scope.get('user', None)

        if user is None:
            token = self.get_token_from_headers(scope)
            if token:
                user = await self.get_user_by_token(token)
            else:
                user = AnonymousUser()

        scope['user'] = user

    def get_token_from_headers(self, scope):
        headers = dict(scope.get('headers', []))
        token = headers.get(b'authorization', None)
        if token:
            token_str = token.decode()
            if token_str.startswith("Bearer "):
                return token_str[len("Bearer "):]
        return None

    @database_sync_to_async
    def get_user_by_token(self, token):
        try:
            decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
            if decoded_token.get('exp') and decoded_token['exp'] < datetime.utcnow().timestamp():
                return AnonymousUser()

            user_profile = BasicUserProfile.objects.get(auth_token=token)
            return user_profile.user
        except (jwt.ExpiredSignatureError, jwt.DecodeError, BasicUserProfile.DoesNotExist):
            return AnonymousUser()

2. WebSocket消费者 ChatConsumer #

import json
import logging
from channels.generic.websocket import AsyncWebsocketConsumer
from authentication.models import BasicUserProfile
from .models import Chat
from .serializers import ChatSerializer
from django.contrib.auth.models import AnonymousUser

logger = logging.getLogger(__name__)

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        current_user = self.scope.get('user', None)

        if current_user is None or not current_user.is_authenticated:
            logger.warning("Unauthenticated user attempted to connect.")
            await self.close(code=4000)
            return

        receiver_username = self.scope['url_route']['kwargs']['username']

        try:
            current_profile = await self.get_user_profile(current_user.id)
            receiver_profile = await self.get_receiver_profile(receiver_username)
        except BasicUserProfile.DoesNotExist:
            logger.error(f"Profile not found for user {current_user.id} or receiver {receiver_username}")
            await self.close(code=4001)
            return

        self.room_name = f'{min(current_profile.id, receiver_profile.id)}_{max(current_profile.id, receiver_profile.id)}'
        self.room_group_name = f'chat_{self.room_name}'

        self.sender_profile = current_profile
        self.receiver_profile = receiver_profile

        await self.channel_layer.group_add(self.room_group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        if hasattr(self, 'room_group_name'):
            await self.channel_layer.group_discard(self.room_group_name, self.channel_name)
            logger.debug(f"User {self.scope['user'].username} disconnected from {self.room_group_name}")

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data.get('message', '')
        sender_username = data.get('senderUsername', '')

        sender_profile = await self.get_user_profile_by_username(sender_username)
        if not sender_profile:
            logger.error(f"Sender profile for {sender_username} not found.")
            return

        chat_message = await self.save_message(sender_profile, message)
        chat_history = await self.get_chat_history()

        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message,
                'senderUsername': sender_username,
                'chat_history': chat_history,
            }
        )

    async def chat_message(self, event):
        message = event['message']
        sender_username = event['senderUsername']
        chat_history = event['chat_history']

        await self.send(text_data=json.dumps({
            'message': message,
            'senderUsername': sender_username,
            'chat_history': chat_history,
        }))

    @database_sync_to_async
    def get_user_profile(self, user_id):
        return BasicUserProfile.objects.get(user__id=user_id)

    @database_sync_to_async
    def get_receiver_profile(self, username):
        return BasicUserProfile.objects.get(user__username=username)

    @database_sync_to_async
    def get_user_profile_by_username(self, username):
        return BasicUserProfile.objects.get(user__username=username)

    @database_sync_to_async
    def save_message(self, sender, message):
        chat = Chat.objects.create(sender=sender, content=message, receiver=self.receiver_profile)
        return chat

    @database_sync_to_async
    def get_chat_history(self):
        chat_history_feed = Chat.objects.filter(sender=self.sender_profile, receiver=self.receiver_profile) | \
                             Chat.objects.filter(sender=self.receiver_profile, receiver=self.sender_profile)
        chat_serializer = ChatSerializer(chat_history_feed, many=True)
        return chat_serializer.data

遇到的问题 #

WebSocket请求已在Authorization头中携带有效Bearer格式JWT令牌，但连接始终被拒绝，日志提示："Unauthenticated user attempted to connect."

已尝试操作 #

• 验证令牌已正确传递到请求头
• 检查CustomAuthMiddleware的JWT解码和用户获取逻辑
• 确认JWT在普通API请求中可正常认证
• 调试中间件确认令牌被接收并解码

预期结果 #

WebSocket连接应通过JWT令牌完成用户认证，允许用户连接并进行实时聊天

排查与解决方案 #

1. 中间件执行顺序错误 #

Django Channels的中间件顺序决定了执行优先级，默认内置认证中间件可能先将scope['user']设为AnonymousUser，导致你的中间件中user is None的判断不触发。

修复方式：
修改ASGI配置，确保自定义认证中间件包裹在最外层：

# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from your_app.routing import websocket_urlpatterns
from your_app.middleware import CustomAuthMiddleware

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

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": CustomAuthMiddleware(URLRouter(websocket_urlpatterns)),
})

2. populate_scope方法的局限性 #

BaseMiddleware的populate_scope仅在scope初始化时执行一次，对于WebSocket连接的生命周期来说不够可靠。改用__call__方法处理每个请求：

修复后的中间件核心代码：

class CustomAuthMiddleware(BaseMiddleware):
    async def __call__(self, scope, receive, send):
        # 优先处理认证逻辑
        token = self.get_token_from_headers(scope)
        scope['user'] = await self.get_user_by_token(token) if token else AnonymousUser()
        # 传递请求到下一个中间件
        return await super().__call__(scope, receive, send)

    # get_token_from_headers和get_user_by_token方法保持不变

3. JWT令牌匹配逻辑验证 #

确认数据库中BasicUserProfile的auth_token字段存储的是原始JWT字符串，而非解码后的内容。同时在get_user_by_token中添加日志调试：

@database_sync_to_async
def get_user_by_token(self, token):
    try:
        decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        print(f"解码后的令牌: {decoded_token}")
        if decoded_token.get('exp') and decoded_token['exp'] < datetime.utcnow().timestamp():
            print("令牌已过期")
            return AnonymousUser()

        user_profile = BasicUserProfile.objects.get(auth_token=token)
        print(f"匹配到用户: {user_profile.user.username}")
        return user_profile.user
    except Exception as e:
        print(f"认证失败原因: {str(e)}")
        return AnonymousUser()

4. 用户模型is_authenticated属性检查 #

确保BasicUserProfile.user对应的用户模型（无论是Django内置User还是自定义模型）的is_authenticated属性正确返回True。自定义用户模型需继承AbstractBaseUser并正确实现相关属性。

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