如何在 django 通道上使用令牌认证来认证 websocket？
Posted
技术标签:
【中文标题】如何在 django 通道上使用令牌认证来认证 websocket？【英文标题】：How do you authenticate a websocket with token authentication on django channels? 【发布时间】：2017-09-09 14:42:22 【问题描述】：我们想为我们的 websocket 使用 django-channels，但我们也需要进行身份验证。我们有一个使用 django-rest-framework 运行的 rest api，我们在那里使用令牌来验证用户，但相同的功能似乎没有内置到 django-channels 中。
【问题讨论】：
github.com/jaquan1227/django-channel-jwt-auth 勾选此项，您可以将 jwt 放入查询中，它将获取具有该用户 ID 的用户。 【参考方案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()，因此它可以检查其实例。如果没有令牌或令牌无效，则消息上不会有用户。
感谢 github 用户 leonardoo 分享他的 mixin。
【讨论】：
get_group 函数在做什么？如果有帮助，您能否展示您的模型样本。谢谢
就是这样，我把例子做得更完整了。这只是一些基本的字符串操作。
我在这方面已经有很长一段时间了，我这辈子都无法让它发挥作用。你成功了吗？
是的，它就像示例一样工作。我注意到get_id 还没有在那里，所以我也添加了它。它只是从组名返回组 id。您是否能够使用这些相同的令牌以其他方式进行身份验证？ （不是通过 websockets）你的具体问题是什么？
只需强调 leonardoo 的 mixin 的第 47 行的重要性：request.META["HTTP_AUTHORIZATION"] = "token ".format(auth_token) 在我的例子中，token auth header 是 "Bearer" ，而不是“令牌”，所以我只需将其更改为： request.META["HTTP_AUTHORIZATION"] = "Bearer ".format(auth_token)【参考方案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)
在channels参数的查询参数中使用这个session_key
我知道有一个额外的 API 调用，但我相信它比在 URL 字符串中发送令牌更安全。
编辑：这只是解决此问题的另一种方法，如 cmets 中所述，get 参数仅在 http 协议的 url 中公开，无论如何都应该避免。
【讨论】：
这不是真的。我投错了，但我不能再改变它了。唯一的安全风险是用户浏览器历史记录，但问题中没有提到浏览器，因为这个案例是关于一个移动应用程序，即使它在浏览器中，你也不会访问 websocket url，它会在 js 中生成和请求，因此它甚至可能不会出现在历史记录中。 ***.com/questions/499591/are-https-urls-encrypted 这些 url 将出现在中间人攻击或代理服务器场景中。令牌将显示在此类代理服务器的访问日志中 接收服务器当然需要URL，如果中间有人，所有数据都会被泄露。但是我们使用 ssl； https和wss。因此，在代理的情况下，将有一条通过代理的安全隧道，与与代理本身的连接无关。主机名当然是可见的，但我们不在乎。 是的，感谢您更新我。它不是安全问题。最初删除了这个解决方案，但现在保留它作为另一种方式。【参考方案3】：对于 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),
]),
),
)
【讨论】：
我也在使用 django 通道版本 2。连接到 websockets 时如何连接传递授权标头令牌？ 我们也搬到了频道 2，我使用这个答案来帮助实施，所以我现在接受了它而不是我自己的。 我必须在客户端/javascript 端做什么？ 任何人都可以帮助我如何通过 websockets 传递授权标头？ 我很困惑人们如何让它发挥作用。根据这个 SO answer，websocket 连接无法自定义标头 - ***.com/questions/4361173/…【参考方案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 确实，但是，不 创建一个频道会话，它只是将用户存储在 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
【讨论】：
【参考方案5】：以下 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 任何帮助将不胜感激。【参考方案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()),
]
)
)
)
)
【讨论】：
就我而言，我必须省略AllowedHostsOriginValidator。 Django3.1 和 channel3。
握手时只会运行一次吗？还是在每个“消息”上？【参考方案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)
这样的Asgi
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),
])
),
)
【讨论】：
以上是关于如何在 django 通道上使用令牌认证来认证 websocket？的主要内容，如果未能解决你的问题，请参考以下文章