Django Channels 2：两种身份验证方法

Posted 2023-02-16

【中文标题】Django Channels 2：两种身份验证方法

【英文标题】：Django Channels 2: two authentication methods

【发布时间】：2019-05-14 20:32:28

【问题描述】：

如何为我的 django 频道应用程序实现多种身份验证方法？ 我的意思是，我有一个基于令牌的身份验证和一个 AuthMiddlewareStack。

我如何维护这两个身份验证，以便为我的内部应用程序 websocket 数据消费者/通知/等管理会话中的 django 身份验证...并为第三方应用程序采用 websocket 令牌身份验证？

这是我的routing.py：

application = ProtocolTypeRouter(
    # (http->django views is added by default)
    'websocket': AuthMiddlewareStack(
        URLRouter(
            app.routing.websocket_test
        )
    ),
    "websocket": TokenAuthMiddlewareStack(
        URLRouter([
            app.routing.websocket_test
        ]),
    )
)

谢谢。

【问题讨论】：

您可以将它们堆叠在一起，即将一个身份验证堆栈包装在另一个中。喜欢FInakMiddleqareStack = TokenMiddlewareStack(AuthMiddlewareStack)。然后在你的路由器中使用FInalMiddlewareStack

感谢@Ken4scholars，我昨天已经解决了我作为答案发布的实现。

【参考方案1】：

我已经解决了实现通用身份验证中间件堆栈的问题，Django Channels 的文档缺少这些示例。

无论如何，我已经实现了一个混合身份验证堆栈，它支持：会话、令牌标头和令牌 urlpath 身份验证方法，完美运行：

WSTokenAuthMiddleware.py：

from urllib.parse import parse_qs

from channels.auth import AuthMiddlewareStack
from django.utils import timezone

from rest_framework.authtoken.models import Token
from django.contrib.auth.models import AnonymousUser
from django.db import close_old_connections


class WSTokenAuthMiddleware:
    """
    Token [Querystring/Header] authorization middleware for Django Channels 2
    """

    def __init__(self, inner):
        self.inner = inner

    def __call__(self, scope):
        query_string = parse_qs(scope['query_string']) #Used for querystring token url auth
        headers = dict(scope['headers']) #Used for headers token url auth
        if b'token' in query_string:
            try:
                token_key = query_string[b'token'][0].decode()
                token = Token.objects.get(key=token_key)
                scope['user'] = token.user
                close_old_connections()
            except ApiToken.DoesNotExist:
                scope['user'] = AnonymousUser()
        elif b'authorization' in headers:
            try:
                token_name, token_key = headers[b'authorization'].decode().split()
                if token_name == 'Token':
                    token = Token.objects.get(key=token_key)
                    scope['user'] = token.user
                    close_old_connections()
            except ApiToken.DoesNotExist:
                scope['user'] = AnonymousUser()
        else:
            pass #Session auth or anonymus

        return self.inner(scope)

UniversalAuthMiddlewareStack = lambda inner: WSTokenAuthMiddleware(AuthMiddlewareStack(inner))

Routing.py：

application = ProtocolTypeRouter(
    "websocket": UniversalAuthMiddlewareStack(
        URLRouter(
            ...
        ),
    ),
)