如何在django频道上使用令牌认证对websocket进行身份验证？

Question

我们想为我们的websockets使用django-channels,但我们也需要进行身份验证.我们有一个运行django-rest-framework的rest api,我们使用令牌来验证用户,但是django-channels似乎没有内置相同的功能.

Answer 1

对于Django-Channels 2,您可以编写自定义身份验证中间件 https://gist.github.com/rluts/22e05ed8f53f97bdd02eafdf38f3d60a

token_auth.py:

from channels.auth import AuthMiddlewareStack
from rest_framework.authtoken.models import Token
from django.contrib.auth.models import AnonymousUser


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

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

    def __call__(self, scope):
        headers = dict(scope['headers'])
        if 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
            except Token.DoesNotExist:
                scope['user'] = AnonymousUser()
        return self.inner(scope)

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

routing.py:

from django.urls import path

from channels.http import AsgiHandler
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack

from yourapp.consumers import SocketCostumer
from yourapp.token_auth import TokenAuthMiddlewareStack

application = ProtocolTypeRouter({
    "websocket": TokenAuthMiddlewareStack(
        URLRouter([
            path("socket/", SocketCostumer),
        ]),
    ),

})

Answer 2

此答案适用于频道1.

您可以在此github问题中找到所有信息:https: //github.com/django/channels/issues/510#issuecomment-288677354

我将在此总结讨论.

1. 将此mixin复制到您的项目中:https: //gist.github.com/leonardoo/9574251b3c7eefccd84fc38905110ce4

2. 应用装饰器 ws_connect

通过对/auth-tokendjango-rest-framework中的视图的早期身份验证请求,在应用程序中接收令牌.我们使用查询字符串将令牌发送回django-channels.如果你没有使用django-rest-framework,你可以用你自己的方式使用查询字符串.阅读mixin以了解如何获得它.

3. 在使用mixin之后,正确的令牌与升级/连接请求一起使用,该消息将具有如下示例中的用户.如您所见,我们已has_permission()在User模型上实现,因此它可以只检查其实例.如果没有令牌或令牌无效,则消息上将没有用户.

    #  get_group, get_group_category and get_id are specific to the way we named
    #  things in our implementation but I've included them for completeness.
    #  We use the URL `wss://www.website.com/ws/app_1234?token=3a5s4er34srd32`

    def get_group(message):
        return message.content['path'].strip('/').replace('ws/', '', 1)


    def get_group_category(group):
        partition = group.rpartition('_')

        if partition[0]:
            return partition[0]
        else:
            return group


    def get_id(group):
        return group.rpartition('_')[2]


    def accept_connection(message, group):
        message.reply_channel.send({'accept': True})
        Group(group).add(message.reply_channel)


    #  here in connect_app we access the user on message
    #  that has been set by @rest_token_user

    def connect_app(message, group):
        if message.user.has_permission(pk=get_id(group)):
            accept_connection(message, group)


    @rest_token_user
    def ws_connect(message):
        group = get_group(message) # returns 'app_1234'
        category = get_group_category(group) # returns 'app'

        if category == 'app':
            connect_app(message, group)


    # sends the message contents to everyone in the same group

    def ws_message(message):
        Group(get_group(message)).send({'text': message.content['text']})


    # removes this connection from its group. In this setup a
    # connection wil only ever have one group.

    def ws_disconnect(message):
        Group(get_group(message)).discard(message.reply_channel)

感谢github用户leonardoo分享他的mixin.

Answer 3

以下 Django-Channels 2 中间件验证由djangorestframework-jwt生成的JWT。

令牌可以通过 djangorestframework-jwt http API 设置，如果JWT_AUTH_COOKIE定义，它也会被发送给 WebSocket 连接。

设置.py

JWT_AUTH = {
    'JWT_AUTH_COOKIE': 'JWT',     # the cookie will also be sent on WebSocket connections
}

路由.py：

from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path
from json_token_auth import JsonTokenAuthMiddlewareStack
from yourapp.consumers import SocketCostumer

application = ProtocolTypeRouter({
    "websocket": JsonTokenAuthMiddlewareStack(
        URLRouter([
            path("socket/", SocketCostumer),
        ]),
    ),

})

json_token_auth.py

from http import cookies

from channels.auth import AuthMiddlewareStack
from django.contrib.auth.models import AnonymousUser
from django.db import close_old_connections
from rest_framework_jwt.authentication import BaseJSONWebTokenAuthentication


class JsonWebTokenAuthenticationFromScope(BaseJSONWebTokenAuthentication):
    """
    Extracts the JWT from a channel scope (instead of an http request)
    """

    def get_jwt_value(self, scope):
        try:
            cookie = next(x for x in scope['headers'] if x[0].decode('utf-8') == 'cookie')[1].decode('utf-8')
            return cookies.SimpleCookie(cookie)['JWT'].value
        except:
            return None


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

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

    def __call__(self, scope):

        try:
            # Close old database connections to prevent usage of timed out connections
            close_old_connections()

            user, jwt_value = JsonWebTokenAuthenticationFromScope().authenticate(scope)
            scope['user'] = user
        except:
            scope['user'] = AnonymousUser()

        return self.inner(scope)


def JsonTokenAuthMiddlewareStack(inner):
    return JsonTokenAuthMiddleware(AuthMiddlewareStack(inner))

Answer 4

如果您使用的是Django Channels 3，您可以使用以下代码：https : //gist.github.com/AliRn76/1fb99688315bedb2bf32fc4af0e50157

中间件.py

from django.contrib.auth.models import AnonymousUser
from rest_framework.authtoken.models import Token
from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware

@database_sync_to_async
def get_user(token_key):
    try:
        token = Token.objects.get(key=token_key)
        return token.user
    except Token.DoesNotExist:
        return AnonymousUser()

class TokenAuthMiddleware(BaseMiddleware):
    def __init__(self, inner):
        super().__init__(inner)

    async def __call__(self, scope, receive, send):
        try:
            token_key = (dict((x.split('=') for x in scope['query_string'].decode().split("&")))).get('token', None)
        except ValueError:
            token_key = None
        scope['user'] = AnonymousUser() if token_key is None else await get_user(token_key)
        return await super().__call__(scope, receive, send)

路由.py

from channels.security.websocket import AllowedHostsOriginValidator
from channels.routing import ProtocolTypeRouter, URLRouter
from .middleware import TokenAuthMiddleware
from main.consumers import MainConsumer
from django.conf.urls import url

application = ProtocolTypeRouter({
        'websocket': AllowedHostsOriginValidator(
            TokenAuthMiddleware(
                URLRouter(
                    [
                        url(r"^main/$", MainConsumer.as_asgi()),
                    ]
                )
            )
        )
    })

Answer 5

channels-auth-token-middlewares在将Simple JWT与Django REST Framework结合使用时，提供QueryStringSimpleJWTAuthTokenMiddleware了开箱即用的令牌身份验证支持。

更新INSTALLED_APPS：

INSTALLED_APPS = [
    # base django apps (django.contrib.auth is required)
    # other apps this one depends on (like rest_framework if it's necessary)
    'channels_auth_token_middlewares',
    # custom apps
]

插入QueryStringSimpleJWTAuthTokenMiddleware到您的ASGIapplication堆栈中：

application = ProtocolTypeRouter(
    {
        "http": django_asgi_app,
        "websocket": AllowedHostsOriginValidator(
            QueryStringSimpleJWTAuthTokenMiddleware(
                URLRouter(...),
            ),
        ),
    }
)

客户端将其 JWT 令牌传递到token查询参数中：

from websocket import create_connection

token = "EXAMPLE_TOKEN"
ws = create_connection(f"ws://127.0.0.1/ws/?token={token}")

经过身份验证的User（或者AnonymousUser如果 JWT 无效）将被填充到传递给"user"的密钥中scopeConsumer

class MyAsyncCommunicator(AsyncWebsocketConsumer):
    async def connect(self) -> None:
        user = self.scope["user"]
        # Validate user before accepting the Websocket Connection
        # For example:
        if not user.is_authenticated or user.is_anonymous:
            # Handle unauthorized.