我们想为我们的 websocket 使用 django-channels，但我们也需要进行身份验证。我们有一个使用 django-rest-framework 运行的 rest api，我们在那里使用令牌来验证用户，但相同的功能似乎没有内置到 django-channels 中。
【问题讨论】：
标签： django-rest-framework django-channels auth-token
此答案对频道 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 以了解如何使用它。
User 模型上实现了has_permission()，因此它可以检查其实例。如果没有令牌或令牌无效，则消息上不会有用户。感谢 github 用户 leonardoo 分享他的 mixin。
【讨论】：
get_group 函数在做什么？如果有帮助，您能否展示您的模型样本。谢谢
get_id 还没有在那里，所以我也添加了它。它只是从组名返回组 id。您是否能够使用这些相同的令牌以其他方式进行身份验证？ （不是通过 websockets）你的具体问题是什么？
我相信即使在 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})
在channels参数的查询参数中使用这个session_key
我知道有一个额外的 API 调用，但我相信它比在 URL 字符串中发送令牌更安全。
编辑：这只是解决此问题的另一种方法，如 cmets 中所述，get 参数仅在 http 协议的 url 中公开，无论如何都应该避免。
【讨论】：
对于 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),
]),
),
})
【讨论】：
关于频道 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
【讨论】：
以下 Django-Channels 2 中间件对生成的 JWT 进行身份验证 由djangorestframework-jwt .
token 可以通过 djangorestframework-jwt http API 设置，它也将被发送给 WebSocket 连接如果JWT_AUTH_COOKIE 被定义。
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))
【讨论】：
self.scope['user'] 仍然返回AnonymousUser 任何帮助将不胜感激。
如果您使用的是 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()),
]
)
)
)
})
【讨论】：
AllowedHostsOriginValidator。 Django3.1 和 channel3。
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),
])
),
})
【讨论】：