Django Channels Authentication Middleware

[John-Doherty01]

Websockets are great to have within an API service toolkit for full duplex communication. Django's most popular implementation of this is https://channels.readthedocs.io/en/stable/

Handling authentication for this django channel library can be handled by providing a authentication middleware class which can then be referenced by django channels, example here using a custom authentication class called TokenAuthMiddlewareStack:

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": TokenAuthMiddlewareStack(
        URLRouter(
            websocket_urlpatterns
        )
    ),
})

I think it would be great if django-oauth-toolkit provided an out the box implementation of an authentication middleware which is compatible with django channels. I've made a quick working PoC below but hasn't been tested security wise:

from channels.auth import AuthMiddlewareStack
from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware
from oauth2_provider.models import AccessToken
from django.contrib.auth.models import AnonymousUser
from urllib.parse import parse_qs

@database_sync_to_async
def get_user(scope):
    user = None

    query_string = scope.get('query_string').decode()
    params = parse_qs(query_string)
    token_key = params.get('token')[0]

    if token_key is not None and token_key != "":
        token = AccessToken.objects.get(token=token_key)
        token_valid = token.is_valid()
        if token_valid:
            user = token.user

    return user or AnonymousUser()

class TokenAuthMiddleware(BaseMiddleware):

    async def __call__(self, scope, receive, send):
        scope['user'] = await get_user(scope)
        return await super().__call__(scope, receive, send)

TokenAuthMiddlewareStack = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner))

I seen this has been mentioned in the past but thought it would be good to bring up again. I spent the time making this PoC but I'm not an active member of this repo, would be great to hear feedback of active contributors regarding this PoC.

[auvipy]

you are welcome to come with a contribution

[yurasavin]

Hi! I've also was looking for a way to authenticate user with request auth header. Here is my implementation if it's help to someone

from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware
from channels.security.websocket import WebsocketDenier
from django.contrib.auth import authenticate
from django.http import HttpRequest

class OAuth2Middleware(BaseMiddleware):
    @database_sync_to_async
    def _authenticate(self, token: str) -> User | None:
        http_request = HttpRequest()
        http_request.META["Authorization"] = token
        return authenticate(request=http_request)

    async def authenticate_user(self, scope) -> None:
        scope["user"] = None

        token = None
        for header_name, header_value in scope["headers"]:
            if header_name == b"authorization":
                token = header_value
                break

        if not token:
            return

        scope["user"] = await self._authenticate(token)

    async def __call__(self, scope, receive, send):
        await self.authenticate_user(scope)

        if not scope["user"]:
            denier = WebsocketDenier()
            return await denier(scope, receive, send)

        return await super().__call__(scope, receive, send)