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