你如何在 django 通道上使用令牌身份验证来验证 websocket？

Question

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

Answer 1

此答案对频道 1 有效。

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

我将在这里总结讨论。

将此 mixin 复制到您的项目中： https://gist.github.com/leonardoo/9574251b3c7eefccd84fc38905110ce4
将装饰器应用于 ws_connect

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

使用 mixin 并在升级/连接请求中使用正确的令牌后，消息将具有如下例所示的用户。如您所见，我们在 User 模型上实现了 has_permission()，因此它可以只检查其实例。如果没有token或者token无效，则消息上没有用户。


    #  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 2

我相信即使在 HTTPS 协议中，在查询字符串中发送令牌也会暴露令牌。为了解决这个问题，我使用了以下步骤：

创建一个基于令牌的 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})

在通道参数

session_key

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

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

Answer 3

对于Django-Channels2你可以写自定义的认证中间件 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 4

关于渠道 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 会，但是，不会创建一个频道 session 它只是存储用户在 ws_connect 中的消息 object 中。由于令牌不会在 ws_receive 中再次发送并且消息 object 也不可用，为了在 ws_receive 和 ws_disconnect 中也获得用户，您需要自己将其存储在 session 中。这是一种简单的方法：

@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

Answer 5

下面Django-Channels2个中间件验证生成的JWTs 通过 djangorestframework-jwt .

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

settings.py

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

routing.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 6

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

middleware.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)

routing.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 7

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),
        ])
    ),
})

Answer 8

ovveride 自定义 AuthMiddleware

from urllib.parse import parse_qs

from channels.auth import AuthMiddleware
from channels.db import database_sync_to_async
from channels.sessions import CookieMiddleware, SessionMiddleware

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


@database_sync_to_async
def get_user(scope):
    query_string = parse_qs(scope['query_string'].decode())
    token = query_string.get('token')
    if not token:
        return AnonymousUser()
    try:
        user = Token.objects.get(key=token[0]).user

    except Exception as exception:
        return AnonymousUser()
    if not user.is_active:
        return AnonymousUser()
    return user


class TokenAuthMiddleware(AuthMiddleware):
    async def resolve_scope(self, scope):
        scope['user']._wrapped = await get_user(scope)


def TokenAuthMiddlewareStack(inner):
    return CookieMiddleware(SessionMiddleware(TokenAuthMiddleware(inner)))

在asgi.py

中导入TokenAuthMiddlewareStack中间件

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application

from chat.api.router_ws import urlpatterns_websocket
from .middleware import TokenAuthMiddlewareStack

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
# Initialize Django ASGI application early to ensure the AppRegistry
# is populated before importing code that may import ORM models.

application = ProtocolTypeRouter({
  "http": get_asgi_application(),
  "websocket": AllowedHostsOriginValidator(
        TokenAuthMiddlewareStack(
            URLRouter(urlpatterns_websocket)
        )
    ),
})

在前端:new WebSocket(ws://8000/{your_path}?token=${localStorage.getItem('token')})

在 Consumer 中：您可以作为 self.scope["user"]

访问请求的用户

你如何在 django 通道上使用令牌身份验证来验证 websocket？

How do you authenticate a websocket with token authentication on django channels?

django-rest-framework

django-channels

auth-token

阿斯吉喜欢这样