如何在 django 通道上使用 Knox 令牌身份验证来验证 websocket 连接?

How do you authenticate a websocket connection with Knox token authentication on django channels?

我知道您可以编写自定义身份验证中间件以在 Django 通道 2 中使用。 这适用于 Django 的内置令牌身份验证,但使用 django-rest-knox 令牌是另一回事。 Knox 以加密形式存储其令牌,因此它不像通过查找令牌从数据库中检索用户那样简单。 请帮忙。

想通了!

from knox.auth import TokenAuthentication
...
knoxAuth = TokenAuthentication();
user, auth_token = knoxAuth.authenticate_credentials(tokenString.encode(HTTP_HEADER_ENCODING))
scope['user'] = user

将以上代码与:https://gist.github.com/rluts/22e05ed8f53f97bdd02eafdf38f3d60a

整合

为了能够使用令牌验证来验证用户,您必须使用 cookie,headers 您可以使用 WS 发送是有限的,您还必须实现自己的“TokenAuthMiddleware”来处理 cookie .对于通道 2,您还必须正确处理对数据库的访问,下面是如何做到这一点:

from    channels.auth                   import  AuthMiddlewareStack
from    channels.db                     import  database_sync_to_async
from    knox.auth                       import  TokenAuthentication
from    django.contrib.auth.models      import  AnonymousUser
from    django.db                       import  close_old_connections
from    rest_framework.exceptions       import  AuthenticationFailed

import  re

class TokenAuthMiddlewareInstance   :
    def __init__            (
        #
                self        ,
                scope       ,
                middleware  ,
            ):
        self.middleware = middleware
        self.scope      = dict(scope)
        self.inner      = self.middleware.inner
    async def __call__      (
        #
                self    ,
                receive ,
                send    ,
            ):
        self.scope['user']  = AnonymousUser()
        cookie              = dict(self.scope.get('headers',{})).get(b'cookie')

        if      cookie  :
            token   = re.findall(r'X-Authorization=(\w*)', cookie.decode('ascii'))
            if      len(token)  :
                self.scope['user']   = await self._g_user(token)

        inner = self.inner(self.scope)
        return await inner(receive, send)
    
    @database_sync_to_async
    def _g_user (
        #
                self    ,
                token   ,
            ):
        try                                 :
            token_key       = token[0]
            user, token     = TokenAuthentication().authenticate_credentials(token_key.encode('ascii'))
            close_old_connections()
            return user
        except AuthenticationFailed as e    :
            return AnonymousUser()
class TokenAuthMiddleware           :
    def __init__    (
        #
                self    ,
                inner   ,
            ):
        self.inner = inner

    def __call__    (
        #
                self    ,
                scope   ,
            ):
        return TokenAuthMiddlewareInstance(scope, self)
TokenAuthMiddlewareStack            = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner))