Using Token as authentication for websocket | django == 4.0.3 | channels == 3.0.4

Using Token as authentication for websocket

Version:

        django == 4.0.3
        channels == 3.0.4
    

Important notice: must declare first before, AuthMiddlewareStack.

import os
import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')

django.setup()

from channels.auth import AuthMiddlewareStack
    

1.Add middleware.py in your django app.

Example token headers.
Authorization : Token [user token here]
    from channels.auth import AuthMiddlewareStack
    from channels.db import database_sync_to_async
    from django.contrib.auth.models import AnonymousUser
    
    from rest_framework.authtoken.models import Token
    
    
    @database_sync_to_async
    def get_user(headers):
        try:
            token_name, token_key = headers[b'authorization'].decode().split()
            if token_name == 'Token':
                token = Token.objects.get(key=token_key)
                return token.user
        except Token.DoesNotExist:
            return AnonymousUser()
    
    
    class TokenAuthMiddleware:
        def __init__(self, inner):
            self.inner = inner
    
        async def __call__(self, scope, receive, send):
            headers = dict(scope['headers'])
            if b'authorization' in headers:
                scope['user'] = await get_user(headers)
            return await self.inner(scope, receive, send)
    
    
    class TokenAuthMiddlewareInstance:
        """
        Yeah, this is black magic:
        https://github.com/django/channels/issues/1399
        """
    
        def __init__(self, scope, middleware):
            self.middleware = middleware
            self.scope = dict(scope)
            self.inner = self.middleware.inner
    
        async def __call__(self, receive, send):
            headers = dict(self.scope['headers'])
            if b'authorization' in headers:
                self.scope['user'] = await get_user(headers)
            inner = self.inner(self.scope)
            return await inner(receive, send)
    
    
    def TokenAuthMiddlewareStack(inner): return TokenAuthMiddleware(
        AuthMiddlewareStack(inner))    

2.Add TokenAuthMiddlewareStack in your asgi.py

    """
    ASGI config for myapp 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.2/howto/deployment/asgi/
    """
    import os
    import django
    
    from channels.auth import AuthMiddlewareStack
    from channels.routing import ProtocolTypeRouter, URLRouter
    from django.core.asgi import get_asgi_application
    from channels.security.websocket import AllowedHostsOriginValidator
    
    from myapp.middleware import TokenAuthMiddlewareStack # middleware custom Token auth
    import myapp.routing
    
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')
    
    django.setup()
    
    application = ProtocolTypeRouter({
        # "http": get_asgi_application,
        "websocket": TokenAuthMiddlewareStack(
            URLRouter(
                myapp.routing.websocket_urlpatterns
            )
        ),
    })
in your consumers.py, check user object if exist or not.
self.scope['user'] # user object if exist if not then anonymous

Add new comment