Django Channels Auth Middleware for rest_framework_simplejwt

django_channels_routing.py

import logging
from urllib.parse import parse_qs
from django.conf.urls import url
from channels.routing import ProtocolTypeRouter, URLRouter
from rest_framework_simplejwt.authentication import JWTAuthentication

logger = logging.getLogger(__name__)

class JwtTokenAuthMiddleware:
    """
    JWT token authorization middleware for Django Channels 2
    Assumes Websocket urls in the format:
      ws://domain.com/ws/objects/5?access_token=your_access_token_here_ABCDEFT123456
    """

    def __init__(self, inner):
        self.inner = inner

    def __call__(self, scope):
        close_old_connections()
        try:
            queries = parse_qs(scope['query_string'].strip().decode())
            raw_token = queries['access_token'][0]
            auth = JWTAuthentication()
            validated_token = auth.get_validated_token(raw_token)
            user = auth.get_user(validated_token)
            scope['user'] = user
        except Exception as e:
            logger.error(e)
        return self.inner(scope)


application = ProtocolTypeRouter({
    "websocket": JwtTokenAuthMiddleware(
        URLRouter([
            url(r'^ws/objects/(?P<object_id>[^/]+)/$', ObjectConsumer),
        ])
    ),
})