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

Question

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

Answer 1

弄清楚了！

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

Answer 2

为了能够使用令牌身份验证对用户进行身份验证，您必须使用 cookie，您可以使用 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))