了解 django 频道 - QueryAuthMiddleware

Question

如何编写用户的自定义身份验证，通过 ws:// 协议连接到聊天？该用户在 Django 应用程序的另一端，他是移动用户，通过移动应用程序中的 ws:// 连接 websocket。我尝试使用 chrome 扩展测试 websocket，它无法连接到我的 websocket。我认为是因为身份验证。

在 Django 频道文档中它说：

如果您有自定义身份验证方案，则可以编写自定义中间件来解析详细信息并将用户对象（或您需要的任何其他对象）放入您的作用域。中间件被编写为可调用的，它接受一个 ASGI 应用程序和包装它以返回另一个 ASGI 应用程序。大多数身份验证只能在作用域上完成，因此您需要做的就是覆盖采用作用域的初始构造函数，而不是事件运行协程。 这是一个简单的中间件示例，它只从查询字符串中取出用户 ID 并使用它： 相同的原则可以应用于通过非 HTTP 协议进行身份验证；例如，您可能希望使用来自聊天协议的某人的聊天用户名将其转换为用户。

from django.db import close_old_connections

class QueryAuthMiddleware:
    def __init__(self, inner):
        # Store the ASGI application we were passed
        self.inner = inner

    def __call__(self, scope):
        # Look up user from query string (you should also do things like
        # check it's a valid user ID, or if scope["user"] is already populated)
        user = User.objects.get(id=int(scope["query_string"]))
        close_old_connections()
        # Return the inner application directly and let it run everything else
        return self.inner(dict(scope, user=user))

我需要做什么查询？我对该用户一无所知，实际上是匿名用户。

请帮帮我。

Answer 1

在此示例代码中，您可能必须使用以下命令打开 websocket 连接：

ws://SERVER:PORT/PATH?1

? 之后的所有内容都是查询字符串。在您的示例代码中，您的 query_string 必须是用户 ID，例如 1。

您可以更改代码以使用不同的查询字符串。例如，您可以使用：

from urllib.parse import parse_qs
from django.db import close_old_connections

class QueryAuthMiddleware:
    def __init__(self, inner):
        # Store the ASGI application we were passed
        self.inner = inner

    def __call__(self, scope):
        # Look up user from query string (you should also do things like
        # check it's a valid user ID, or if scope["user"] is already populated)

        query_string = parse_qs(self.scope['query_string'])
        if b'user_id' in query_string:
            user = User.objects.get(id=int(query_string[b'user_id'][0]))
            close_old_connections()
        else:
            user = AnonymousUser
        # Return the inner application directly and let it run everything else
        return self.inner(dict(scope, user=user))

现在，你可以使用这个 uri：

ws://SERVER:PORT/PATH?user_id=1

您还必须确保具有该 ID 的用户存在于数据库中。您还必须编写实际的身份验证代码。每个用户都可以使用任意用户 ID 连接到此应用程序。不需要密码或身份验证令牌。

Answer 2

我也遇到了同样的问题，经过一番研究，找到了下面代码 sn-p 的解决方案：

假设您定义了User 类。您希望在与 ws 建立连接时通过查询发送查询对用户进行身份验证。

我将通过频道安装和配置假设您已成功安装频道并配置。

QueryAuthMiddleware类如下图：

from channels.auth import AuthMiddlewareStack
from django.conf import LazySettings
from urllib import parse
from rest_socket.models import User
from urllib.parse import parse_qs
from urllib.parse import unquote, urlparse

from channels.auth import AuthMiddlewareStack

from django.contrib.auth.models import AnonymousUser

class QueryAuthMiddleware:
    """
    QueryAuthMiddleware authorization 
    """

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

    def __call__(self, scope):
        query_string = parse_qs(scope['query_string']) #Used for query string token url auth
        headers = dict(scope['headers']) #Used for headers token url auth

        print("query", query_string)
        if b'user' in query_string:
            try:
                user = query_string[b'user'][0].decode()
                print("user", user)
                existing = User.objects.filter(id=user).last()
                print(existing)
                if existing:
                    print("existinguser")
                    scope['user'] = existing

                else:
                    scope["user"]  ="no user found" 

            except User.DoesNotExist:
                pass
        return self.inner(scope)


QueryAuthMiddlewareStack = lambda inner: QueryAuthMiddleware(AuthMiddlewareStack(inner))

你的 routing.py 应该是这样的：

from channels.security.websocket import AllowedHostsOriginValidator
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import OriginValidator
from django.urls import path
import your_app.routing
from project.utils.auth import QueryAuthMiddlewareStack

application = ProtocolTypeRouter({
    # (http->django views is added by default)
    'websocket': QueryAuthMiddlewareStack(
        URLRouter(
            your_app.routing.websocket_urlpatterns
        )
    ),
})

应用内部路由：

websocket_urlpatterns = [
    path("tasks/", consumers.Task)
]

以及您与 django-channels 的 ws 连接：

<script>
    var notification;

    if (location.protocol === 'https:') {
        notification = new WebSocket('wss://' + "window.location.host" + "/tasks"+ "/?user=id");

        console.log("with htpps")
    }
    else {
        notification = new WebSocket('ws://' + window.location.host +  "/tasks"+ "/?userr=id");

        console.log("htpp")
    }

    notification.onopen = function open() {
        console.log('notification connection created for NotificationWebsocket.');
    };

    notification.onmessage = function message(event) {
        var data = JSON.parse(event.data);
        console.log("Socket response from NotificationWebsocket => ", data);
    };

    if (notification.readyState === WebSocket.OPEN) {
        notification.onopen();
    }
</script>