Best practices for authenticating Django Channels

Django 4.1.4
djoser 2.1.0
channels 4.0.0

I have followed the documented recommendation for creating custom middleware to authenticate a user when using channels and I am successfully getting the user and checking that the user is authenticated though I am sending the user ID in the querystring when connecting to the websocket to do this. The user is not automatically available in the websocket scope.

I am unsure if there are any potential security risks as the documentation mentions that their recommendation is insecure, I do check that the user.is_authenticated. So I believe I have secured it.

I do believe that using the token created by djoser would be better though I am not sure how to send headers with the websocket request unless I include the token in the querystring instead of the user's ID.

I am keen to hear what the best practices are.

I am passing the user ID to the websocket via querystring as follows at the frontend:

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)
        )
    )
})

After doing some extensive research I decided not to pass the id or the token via the querystring as this poses a risk due to this data being stored in the server logs.

IMO the best option with the least amount of risk was passing the token as a message to the websocket after the connection was established and then verifying the token; closing the websocket if invalid.

This meant not requiring the middleware previously implemented. In this particular project no other messages would be received from the client so I don't need to do any checking on the key of the message received. This could be changed for chat apps and other apps that will receive further messages from the client.

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)
Back to Top