问 如何在django通道上使用令牌身份验证来验证websocket？

EN

下面的Django-Channel 2中间件对 djangorestframework-jwt 生成的JWT进行身份验证。

令牌可以通过djangorestframework-jwt进行设置，如果 JWT_AUTH_COOKIE WebSocket 被定义为 ，则还会发送给 connections 。

settings.py

JWT_AUTH = {
    'JWT_AUTH_COOKIE': 'JWT',     # the cookie will also be sent on WebSocket connections
}

routing.py：

from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path
from json_token_auth import JsonTokenAuthMiddlewareStack
from yourapp.consumers import SocketCostumer
application = ProtocolTypeRouter({
    "websocket": JsonTokenAuthMiddlewareStack(
        URLRouter([
            path("socket/", SocketCostumer),
})

json_token_auth.py

from http import cookies
from channels.auth import AuthMiddlewareStack
from django.contrib.auth.models import AnonymousUser
from django.db import close_old_connections
from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication
class JsonWebTokenAuthenticationFromScope(BaseJSONWebTokenAuthentication):
    Extracts the JWT from a channel scope (instead of an http request)
    def get_jwt_value(self, scope):
            cookie = next(x for x in scope['headers'] if x[0].decode('utf-8') == 'cookie')[1].decode('utf-8')
            return cookies.SimpleCookie(cookie)['JWT'].value
        except:
            return None
class JsonTokenAuthMiddleware(BaseJSONWebTokenAuthentication):
    Token authorization middleware for Django Channels 2
    def __init__(self, inner):
        self.inner = inner
    def __call__(self, scope):
            # Close old database connections to prevent usage of timed out connections
            close_old_connections()
            user, jwt_value = JsonWebTokenAuthenticationFromScope().authenticate(scope)
            scope['user'] = user
        except:
            scope['user'] = AnonymousUser()