Django通道JWT身份验证

Question

我正在尝试访问Consumer.py中作用域中的用户，并从数据库中获取一些与用户相关的模型。但是，我用来验证所有websocket连接的AuthMiddlewareStack似乎不能正常工作。

当我在REST框架中使用JWT Tokens django-rest-framework-simplejwt进行身份验证时，能够在django通道中对websocket连接进行身份验证的最佳/安全方法是什么？

Answer 1

我也有同样的问题。首先，您不能使用django通道进行JWT身份验证，因为您只能通过通道发送query string，并且不能设置头参数或http协议之类的东西(特别是当您使用JavaScript作为客户端时)。出于安全考虑，我不想将令牌作为查询字符串发送(因为每个人都可以看到它)。所以我在这里解释我的解决方案，也许它也能解决你的问题。我创建了一个用于在套接字中注册的API，在该API中，我返回了一个票证(uuid类型)作为响应，并且在同一API中，我根据用户缓存了这个票证：

class RegisterFilterAPIView(APIView):
    """
        get:
            API view for retrieving ticket uuid.
    """
    authentication_classes = (JWTAuthentication,)
    permission_classes = (IsAuthenticatedOrReadOnly,)

    def get(self, request, *args, **kwargs):
        ticket_uuid = str(uuid4())

        if request.user.is_anonymous:
            cache.set(ticket_uuid, False, TICKET_EXPIRE_TIME)
        else:
            # You can set any condition based on logged in user here
            cache.set(ticket_uuid, some_conditions, TICKET_EXPIRE_TIME)

        return Response({'ticket_uuid': ticket_uuid})

在这一部分之后，我将这个票据作为查询字符串发送到我的套接字，如下所示：

var endpoint = 'ws://your/socket/endpoint/?ticket_uuid=some_ticket';
var newSocket = new WebSocket(endpoint);

newSocket.onmessage = function (e) {
    console.log("message", e)
};
newSocket.onopen = function (e) {
    console.log("open", e);
};
newSocket.onerror = function (e) {
    console.log("error", e)
};
newSocket.onclose = function (e) {
    console.log("close", e)
};

请注意，上面的代码是用JS编写的，因此您应该根据自己的需求将其更改为其他代码。最后，在我的使用者中，我处理了这个在我的注册API中创建的票据：

from urllib.parse import parse_qsl
from django.core.cache import cache
from channels.generic.websocket import AsyncJsonWebsocketConsumer


class FilterConsumer(AsyncJsonWebsocketConsumer):

    async def websocket_connect(self, event):
        try:
            query_string = self.scope['query_string'].decode('utf-8')
            query_params = dict(parse_qsl(query_string))
            ticket_uuid = query_params.get('ticket_uuid')
            self.scope['has_ticket'] = cache.get(ticket_uuid)
            if not cache.delete(ticket_uuid): # I destroyed ticket for performance and security purposes
                raise Exception('ticket not found')
        except:
            await self.close()
            return

        await self.accept()

现在你有了一个安全的注册API (比如获取令牌API)，你可以基于你的JWT令牌生成一个令牌，但是要确保你的服务器支持缓存后端服务。您也可以根据您的票证值在您的websocket连接方法中设置self.scope['user']。我希望这能解决你的问题。

Answer 2

通道3身份验证不同于通道2您必须创建自己的身份验证中间件，首先要创建一个文件channelsmiddleware.py

"""General web socket middlewares
"""

from channels.db import database_sync_to_async
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from rest_framework_simplejwt.tokens import UntypedToken
from rest_framework_simplejwt.authentication import JWTTokenUserAuthentication
from rest_framework_simplejwt.state import User
from channels.middleware import BaseMiddleware
from channels.auth import AuthMiddlewareStack
from django.db import close_old_connections
from urllib.parse import parse_qs
from jwt import decode as jwt_decode
from django.conf import settings
@database_sync_to_async
def get_user(validated_token):
    try:
        user = get_user_model().objects.get(id=validated_token["user_id"])
        # return get_user_model().objects.get(id=toke_id)
        print(f"{user}")
        return user
   
    except User.DoesNotExist:
        return AnonymousUser()



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

    async def __call__(self, scope, receive, send):
       # Close old database connections to prevent usage of timed out connections
        close_old_connections()

        # Get the token
        token = parse_qs(scope["query_string"].decode("utf8"))["token"][0]

        # Try to authenticate the user
        try:
            # This will automatically validate the token and raise an error if token is invalid
            UntypedToken(token)
        except (InvalidToken, TokenError) as e:
            # Token is invalid
            print(e)
            return None
        else:
            #  Then token is valid, decode it
            decoded_data = jwt_decode(token, settings.SECRET_KEY, algorithms=["HS256"])
            print(decoded_data)
            # Will return a dictionary like -
            # {
            #     "token_type": "access",
            #     "exp": 1568770772,
            #     "jti": "5c15e80d65b04c20ad34d77b6703251b",
            #     "user_id": 6
            # }

            # Get the user using ID
            scope["user"] = await get_user(validated_token=decoded_data)
        return await super().__call__(scope, receive, send)


def JwtAuthMiddlewareStack(inner):
    return JwtAuthMiddleware(AuthMiddlewareStack(inner))

您可以像这样将其导入到消费者的routing.py或asgi.py文件中

"""
ASGI config for config project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/
"""

import os
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.core.asgi import get_asgi_application
from channels.security.websocket import AllowedHostsOriginValidator
from chat.consumers import ChatConsumer
from django.urls import path, re_path
from .channelsmiddleware import JwtAuthMiddlewareStack

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")

application = ProtocolTypeRouter(
    {
        "http": get_asgi_application(),
        "websocket": AllowedHostsOriginValidator(
            JwtAuthMiddlewareStack(
                URLRouter(
                    [
                        #path(),your routes here 
                    ]
                )
            ),
        ),
    }
)