59

我们想为我们的 websocket 使用 django-channels，但我们也需要进行身份验证。我们有一个使用 django-rest-framework 运行的 rest api，在那里我们使用令牌来验证用户，但相同的功能似乎没有内置到 django-channels 中。

4

7 回答 7

59

对于 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))

路由.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),
        ]),
    ),

})
于 2018-03-19T21:03:55.303 回答
14

如果您使用的是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()),
                    ]
                )
            )
        )
    })
于 2020-12-24T10:47:56.417 回答
11

此答案对通道 1 有效。

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

我将在这里总结讨论。

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

  2. 将装饰器应用于ws_connect

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

  1. 使用 mixin 并在升级/连接请求中使用正确的令牌后，消息将具有如下示例中的用户。正如你所看到的，我们已经has_permission()在User模型上实现了，所以它可以检查它的实例。如果没有令牌或令牌无效，则消息上不会有用户。
    # get_group、get_group_category 和 get_id 是特定于我们命名的方式
    # 我们实现中的东西，但为了完整起见，我将它们包括在内。
    # 我们使用 URL `wss://www.website.com/ws/app_1234?token=3a5s4er34srd32`

    def get_group（消息）：
        返回 message.content['path'].strip('/').replace('ws/', '', 1)


    def get_group_category（组）：
        分区 = group.rpartition('_')

        如果分区 [0]：
            返回分区[0]
        别的：
            返回组


    def get_id（组）：
        返回 group.rpartition('_')[2]


    def accept_connection（消息，组）：
        message.reply_channel.send({'accept': True})
        Group(group).add(message.reply_channel)


    # 在 connect_app 中，我们通过消息访问用户
    # 已由@rest_token_user 设置

    def connect_app（消息，组）：
        如果 message.user.has_permission(pk=get_id(group)):
            接受连接（消息，组）


    @rest_token_user
    def ws_connect（消息）：
        group = get_group(message) # 返回 'app_1234'
        category = get_group_category(group) # 返回 'app'

        如果类别 == '应用程序'：
            connect_app（消息，组）


    # 将消息内容发送给同一组中的每个人

    def ws_message（消息）：
        Group(get_group(message)).send({'text': message.content['text']})


    # 从其组中删除此连接。在此设置中
    # 连接将永远只有一个组。

    def ws_disconnect（消息）：
        Group(get_group(message)).discard(message.reply_channel)

感谢 github 用户 leonardoo 分享了他的 mixin。

于 2017-04-13T12:52:46.667 回答
11

以下 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))
于 2019-06-02T21:34:06.353 回答
0

我相信在查询字符串中发送令牌甚至可以在 HTTPS 协议中暴露令牌。为了解决此类问题，我使用了以下步骤：

  1. 创建一个基于令牌的 REST API 端点，该端点创建临时会话并用此响应session_key（此会话设置为在 2 分钟内到期）

    login(request,request.user)#Create session with this user
request.session.set_expiry(2*60)#Make this session expire in 2Mins
return Response({'session_key':request.session.session_key})

  2. session_key在通道参数的查询参数中使用它

我知道有一个额外的 API 调用，但我相信它比在 URL 字符串中发送令牌更安全。

编辑：这只是解决此问题的另一种方法，正如评论中所讨论的，get 参数仅在 http 协议的 url 中公开，无论如何都应该避免。

于 2017-09-21T16:26:19.883 回答
0

关于频道 1.x

正如这里已经指出的那样，leonardoo 的 mixin 是最简单的方法： https ://gist.github.com/leonardoo/9574251b3c7eefccd84fc38905110ce4

但是，我认为弄清楚 mixin 在做什么和不做什么有点令人困惑，所以我会尽量说明这一点：

在寻找使用原生 django 通道装饰器访问 message.user 的方法时，您必须像这样实现它：

@channel_session_user_from_http
def ws_connect(message):
  print(message.user)
  pass

@channel_session_user
def ws_receive(message):
  print(message.user)
  pass

@channel_session_user
def ws_disconnect(message):
  print(message.user)
  pass

Channels 通过对用户进行身份验证、创建 http_session 然后将 http_session 转换为 channel_session 来做到这一点，它使用回复通道而不是 cookie 来识别客户端。所有这些都在channel_session_user_from_http中完成。查看频道源代码以获取更多详细信息： https ://github.com/django/channels/blob/1.x/channels/sessions.py

然而，leonardoo 的装饰器rest_token_user不会创建通道会话，它只是将用户存储在 ws_connect 的消息对象中。由于令牌不会在 ws_receive 中再次发送，并且消息对象也不可用，因此为了使用户也进入 ws_receive 和 ws_disconnect，您必须自己将其存储在会话中。这将是一种简单的方法：

@rest_token_user #Set message.user
@channel_session #Create a channel session
def ws_connect(message):
    message.channel_session['userId'] = message.user.id
    message.channel_session.save()
    pass

@channel_session
def ws_receive(message):
    message.user = User.objects.get(id = message.channel_session['userId'])
    pass

@channel_session
def ws_disconnect(message):
    message.user = User.objects.get(id = message.channel_session['userId'])
    pass
于 2018-04-05T08:09:30.727 回答
0 
from rest_framework_simplejwt.tokens import UntypedToken
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from jwt import decode as jwt_decode
from urllib.parse import parse_qs
from django.contrib.auth import get_user_model
from channels.db import database_sync_to_async
from django.conf import settings


@database_sync_to_async
def get_user(user_id):
    User = get_user_model()
    try:
        return User.objects.get(id=user_id)
    except User.DoesNotExist:
        return 'AnonymousUser'


class TokenAuthMiddleware:

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

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

        token = parse_qs(scope["query_string"].decode("utf8"))["token"][0]
        print(token)
        try:
            # This will automatically validate the token and raise an error if token is invalid
            is_valid = UntypedToken(token)
        except (InvalidToken, TokenError) as e:
            # Token is invalid
            print(e)
            return None
        else:
            #  Then token is valid, decode it
            decoded_data = jwt_decode(token, settings.SECRET_KEY, algorithms=["HS256"])
            print(decoded_data)

            scope['user'] = await get_user(int(decoded_data.get('user_id', None)))

            # Return the inner application directly and let it run everything else

        return await self.app(scope, receive, send)

阿吉喜欢这样

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from django.urls import path

from channelsAPI.routing import websocket_urlpatterns
from channelsAPI.token_auth import TokenAuthMiddleware

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'VirtualCurruncy.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": TokenAuthMiddleware(
        URLRouter([
            path("virtualcoin/", websocket_urlpatterns),
        ])
    ),
})
于 2021-05-05T11:40:41.237 回答