Django Channels 2 websockets 多个 AuthMiddlewareStacks答案

【问题标题】：Django Channels 2 websockets multiple AuthMiddlewareStacksDjango Channels 2 websockets 多个 AuthMiddlewareStacks
【发布时间】：2022-02-01 03:04:10
【问题描述】：

尝试使用带有 django 频道的 websocket 制作一些应用程序。我有 2 个 websocket 客户端 - 一个是 web interface/js 应用程序，另一个是 python 应用程序。我想要对他们（他们的消费者）有不同的授权要求（最好是使用 AuthMiddlewareStacks 的一种方式）

我该如何实现呢？在文档中找不到答案https://channels.readthedocs.io/en/latest/topics/routing.html

这是一个“草图”。（routing.py 不会这样工作）。如果相关，我使用 DRF、DRF-JWT、django 频道 2。

appmain.routing.py

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

app.routing.py

websocket_cli_urlpatterns = [
    path('ws/app/<str:var1>/<str:var2>/', consumers.CliConsumer),
]

websocket_web_urlpatterns = [
    path('ws/app/<str:var1>/', consumers.WebConsumer),
]

谢谢！

【问题讨论】：

我也面临同样的问题。你的问题解决了吗？
并非如此。我最终在 websocket 消费者中处理授权（部分）

标签： python django django-rest-framework django-channels

【解决方案1】：

编写您自己的身份验证处理程序，该处理程序结合了您要使用的身份验证处理程序的逻辑。（您可以同时调用它们的 call 方法，并且只有在两者都失败时才会失败。）从 GET 参数或 http 标头中处理令牌，以便 URL 相同。

【讨论】：

【解决方案2】：

# middleware.py
from urllib.parse import parse_qs

import jwt
from channels.auth import AuthMiddlewareStack
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.db import close_old_connections
from django.utils.translation import ugettext as _
from rest_framework import exceptions
from rest_framework_jwt.authentication import jwt_decode_handler, jwt_get_username_from_payload


class JWTAuthMiddleware():
    """
    Token authorization middleware for Django Channels 2
    """

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

    def __call__(self, scope):
        query_dict = {k: v[0] for k, v in parse_qs(scope["query_string"].decode()).items()}

        jwt_value = query_dict.get('token', None)
        if jwt_value:
            try:
                try:
                    payload = jwt_decode_handler(jwt_value)
                except jwt.ExpiredSignature:
                    msg = _('Signature has expired.')
                    raise exceptions.AuthenticationFailed(msg)
                except jwt.DecodeError:
                    msg = _('Error decoding signature.')
                    raise exceptions.AuthenticationFailed(msg)
                except jwt.InvalidTokenError:
                    raise exceptions.AuthenticationFailed()

                scope['user'] = self.authenticate_credentials(payload)

            except exceptions.AuthenticationFailed:
                scope['user'] = AnonymousUser()

        return self.inner(scope)

    def authenticate_credentials(self, payload):
        """
        Returns an active user that matches the payload's user id and email.
        """
        User = get_user_model()
        username = jwt_get_username_from_payload(payload)

        if not username:
            msg = _('Invalid payload.')
            raise exceptions.AuthenticationFailed(msg)

        try:
            user = User.objects.get_by_natural_key(username)
        except User.DoesNotExist:
            msg = _('Invalid signature.')
            raise exceptions.AuthenticationFailed(msg)

        if not user.is_active:
            msg = _('User account is disabled.')
            raise exceptions.AuthenticationFailed(msg)

        close_old_connections()
        return user


JWTAuthMiddlewareStack = lambda inner: JWTAuthMiddleware(AuthMiddlewareStack(inner))



#routing.py
from channels.routing import ProtocolTypeRouter, URLRouter
from core.middleware import JWTAuthMiddlewareStack

application = ProtocolTypeRouter({
        'websocket':
            JWTAuthMiddlewareStack(
                URLRouter(apps.myapp.routing.websocket_urlpatterns)
            )

    })

【讨论】：

它是关于什么的？也许你没有仔细阅读问题

【解决方案3】：

我这样做了，它奏效了：

# consumers.py

from channels.auth import AuthMiddlewareStack

class PrinterWSAuthMiddleWare:
    """
    Token authorization middleware for Django Channels 2
    """

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

    def __call__(self, scope):
        close_old_connections()

        headers = dict(scope['headers'])
        if b'authorization' in headers:
            try:
                token_name, token_key = headers[b'authorization'].decode().split()
                if token_name == 'bearer':
                    printer = Printer.objects.select_related('user').get(auth_token=token_key)
                    scope['user'] = printer
                else:
                    scope['user'] = None
            except ObjectDoesNotExist:
                scope['user'] = None
        return self.inner(scope)

TokenAuthMiddlewareStack = lambda inner: PrinterWSAuthMiddleWare(AuthMiddlewareStack(inner))

# routing.py

...
    'websocket': TokenAuthMiddlewareStack(
        URLRouter(
            api.ws_routing.websocket_urlpatterns
        )
    ),
...

没有记录，但AuthMiddlewareStack 在后台检查 cookie 和会话，如果其中任何一个有效，则设置范围。

【讨论】：

【解决方案4】：

只有在将自定义中间件中__call__ 方法的签名更改为异步后，我才设法使其工作：

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

    async def __call__(self, scope, receive, send):
        # Do custom auth
        return await self.inner(scope, receive, send)


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

【讨论】：