Django 通道 JWT 身份验证
Posted
技术标签:
【中文标题】Django 通道 JWT 身份验证【英文标题】:Django Channels JWT Authentication 【发布时间】:2021-03-25 13:44:23 【问题描述】:我正在尝试访问 Consumer.py 范围内的用户,并从数据库中获取一些与用户相关的模型。但是,我用于验证所有 websocket 连接的 AuthMiddlewareStack
似乎无法正常工作。
当我在我的 REST 框架中使用 JWT 令牌 django-rest-framework-simplejwt
进行身份验证时,能够在 django 通道中验证 websocket 连接的最佳/安全方法是什么?
【问题讨论】:
【参考方案1】:好吧,我也遇到了同样的问题。首先,您不能使用 django 频道进行 JWT 身份验证,因为您可以通过频道发送的唯一内容是 query string
并且您不能设置标头参数或诸如 http
协议之类的东西(尤其是如果您使用 @ 987654324@ 作为您的客户端)。出于安全目的,我不想将我的令牌作为查询字符串发送(因为每个人都可以看到它)。所以我在这里解释我的解决方案,也许它也可以解决你的问题。我创建了一个用于在我的套接字中注册的 API,在该 API 中,我返回了一个票据(uuid 类型)作为响应,并且在同一个 API 中,我根据用户缓存了这个票据:
class RegisterFilterAPIView(APIView):
"""
get:
API view for retrieving ticket uuid.
"""
authentication_classes = (JWTAuthentication,)
permission_classes = (IsAuthenticatedOrReadOnly,)
def get(self, request, *args, **kwargs):
ticket_uuid = str(uuid4())
if request.user.is_anonymous:
cache.set(ticket_uuid, False, TICKET_EXPIRE_TIME)
else:
# You can set any condition based on logged in user here
cache.set(ticket_uuid, some_conditions, TICKET_EXPIRE_TIME)
return Response('ticket_uuid': ticket_uuid)
在这部分之后,我将此票作为查询字符串发送到我的套接字,例如:
var endpoint = 'ws://your/socket/endpoint/?ticket_uuid=some_ticket';
var newSocket = new WebSocket(endpoint);
newSocket.onmessage = function (e)
console.log("message", e)
;
newSocket.onopen = function (e)
console.log("open", e);
;
newSocket.onerror = function (e)
console.log("error", e)
;
newSocket.onclose = function (e)
console.log("close", e)
;
请注意,上面的代码是用JS
写的,所以你应该根据你的要求把它改成别的东西。最后,在我的消费者中,我处理了这张在我的注册 API 中创建的票:
from urllib.parse import parse_qsl
from django.core.cache import cache
from channels.generic.websocket import AsyncJsonWebsocketConsumer
class FilterConsumer(AsyncJsonWebsocketConsumer):
async def websocket_connect(self, event):
try:
query_string = self.scope['query_string'].decode('utf-8')
query_params = dict(parse_qsl(query_string))
ticket_uuid = query_params.get('ticket_uuid')
self.scope['has_ticket'] = cache.get(ticket_uuid)
if not cache.delete(ticket_uuid): # I destroyed ticket for performance and security purposes
raise Exception('ticket not found')
except:
await self.close()
return
await self.accept()
所以现在您有了一个安全的注册 API(如获取令牌 API),您可以根据您的 JWT 令牌生成令牌,但要确保您的服务器支持缓存后端服务。您还可以根据您的票值在您的 websocket 连接方法中设置self.scope['user']
。我希望这可以解决您的问题。
【讨论】:
留给未来的读者:关于在查询参数中使用令牌时的安全风险:检查此answer 的 cmets 中的讨论。【参考方案2】:频道 3 身份验证与频道 2 不同,您必须为此创建自己的身份验证中间件,首先创建文件 channelsmiddleware.py
"""General web socket middlewares
"""
from channels.db import database_sync_to_async
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from rest_framework_simplejwt.tokens import UntypedToken
from rest_framework_simplejwt.authentication import JWTTokenUserAuthentication
from rest_framework_simplejwt.state import User
from channels.middleware import BaseMiddleware
from channels.auth import AuthMiddlewareStack
from django.db import close_old_connections
from urllib.parse import parse_qs
from jwt import decode as jwt_decode
from django.conf import settings
@database_sync_to_async
def get_user(validated_token):
try:
user = get_user_model().objects.get(id=validated_token["user_id"])
# return get_user_model().objects.get(id=toke_id)
print(f"user")
return user
except User.DoesNotExist:
return AnonymousUser()
class JwtAuthMiddleware(BaseMiddleware):
def __init__(self, inner):
self.inner = inner
async def __call__(self, scope, receive, send):
# Close old database connections to prevent usage of timed out connections
close_old_connections()
# Get the token
token = parse_qs(scope["query_string"].decode("utf8"))["token"][0]
# Try to authenticate the user
try:
# This will automatically validate the token and raise an error if token is invalid
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)
# Will return a dictionary like -
#
# "token_type": "access",
# "exp": 1568770772,
# "jti": "5c15e80d65b04c20ad34d77b6703251b",
# "user_id": 6
#
# Get the user using ID
scope["user"] = await get_user(validated_token=decoded_data)
return await super().__call__(scope, receive, send)
def JwtAuthMiddlewareStack(inner):
return JwtAuthMiddleware(AuthMiddlewareStack(inner))
你可以像这样将它导入到你的消费者的 routing.py 或 asgi.py 文件中
"""
ASGI config for config project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/
"""
import os
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.core.asgi import get_asgi_application
from channels.security.websocket import AllowedHostsOriginValidator
from chat.consumers import ChatConsumer
from django.urls import path, re_path
from .channelsmiddleware import JwtAuthMiddlewareStack
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")
application = ProtocolTypeRouter(
"http": get_asgi_application(),
"websocket": AllowedHostsOriginValidator(
JwtAuthMiddlewareStack(
URLRouter(
[
#path(),your routes here
]
)
),
),
)
【讨论】:
以上是关于Django 通道 JWT 身份验证的主要内容,如果未能解决你的问题,请参考以下文章
JWT 身份验证:使用 UI 令牌对 Graphene/Django (GraphQL) 查询进行身份验证?
如何在 django rest 框架中验证 jwt 身份验证中的令牌
使用 Django Rest 框架进行 JWT 令牌身份验证