Лучшие практики для аутентификации каналов 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)
Вернуться на верх