Лучшие практики для аутентификации каналов Django
Django 4.1.4
djoser 2.1.0
каналы 4.0.0
Я следовал документированным рекомендациям по созданию пользовательского промежуточного ПО для аутентификации пользователя при использовании каналов, и я успешно получаю пользователя и проверяю, что пользователь аутентифицирован, хотя я отправляю ID пользователя в строке запроса при подключении к websocket для этого. Пользователь не доступен автоматически в области видимости websocket.
Я не уверен, есть ли какие-либо потенциальные риски безопасности, так как в документации упоминается, что их рекомендация небезопасна, я проверяю, что user.is_authenticated. Так что я считаю, что я обеспечил безопасность.
Я считаю, что использование токена, созданного djoser, было бы лучше, хотя я не уверен, как отправить заголовки с запросом websocket, если я не включу токен в строку запроса вместо ID пользователя.
Мне интересно узнать, какова лучшая практика.
На фронтенде я передаю идентификатор пользователя в websocket через querystring следующим образом:
websocket.value = new WebSocket(`ws://127.0.0.1:8000/ws/marketwatch/? ${authStore.userId}`)
middleware.py
from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist
@database_sync_to_async
def get_user(user_id):
User = get_user_model()
try:
user = User.objects.get(id=user_id)
except ObjectDoesNotExist:
return AnonymousUser()
else:
if user.is_authenticated:
return user
else:
return AnonymousUser()
class QueryAuthMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
scope['user'] = await get_user(int(scope["query_string"].decode()))
return await self.app(scope, receive, send)
consumers.py
import os
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from channels.security.websocket import AllowedHostsOriginValidator
from api.middleware import QueryAuthMiddleware
from .routing import ws_urlpatterns
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'api.settings')
application = ProtocolTypeRouter({
'http':get_asgi_application(),
'websocket': AllowedHostsOriginValidator(
QueryAuthMiddleware(
URLRouter(ws_urlpatterns)
)
)
})
После проведения обширного исследования я решил не передавать id или токен через строку запроса, так как это представляет риск из-за того, что эти данные будут храниться в логах сервера.
IMO лучшим вариантом с наименьшим риском была передача токена в виде сообщения в вебсокет после установления соединения, а затем проверка токена; закрытие вебсокета в случае его недействительности.
Это означало отказ от ранее реализованного промежуточного программного обеспечения. В данном конкретном проекте от клиента не будет получено никаких других сообщений, поэтому мне не нужно делать никакой проверки ключа полученного сообщения. Это может быть изменено для чат-приложений и других приложений, которые будут получать дополнительные сообщения от клиента.
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async
import json
from rest_framework.authtoken.models import Token
class MarketWatchConsumer(AsyncWebsocketConsumer):
@database_sync_to_async
def verify_token(self, token_dict):
try:
token = Token.objects.get(key=token_dict['token'])
except Token.DoesNotExist:
return False
else:
if token.user.is_active:
return True
else:
return False
async def connect(self):
await self.channel_layer.group_add('group', self.channel_name)
await self.accept()
async def receive(self, text_data=None, bytes_data=None):
valid_token = await self.verify_token(json.loads(text_data))
if not valid_token:
await self.close()
async def disconnect(self, code):
await self.channel_layer.group_discard('group', self.channel_name)