[Fixed]-Django Channels JWT Authentication

Channels 3 auth is different from channels 2 you will have to create your own auth middleware for that start by creating a file channelsmiddleware.py

"""General web socket middlewares
"""

from channels.db import database_sync_to_async
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from rest_framework_simplejwt.tokens import UntypedToken
from rest_framework_simplejwt.authentication import JWTTokenUserAuthentication
from rest_framework_simplejwt.state import User
from channels.middleware import BaseMiddleware
from channels.auth import AuthMiddlewareStack
from django.db import close_old_connections
from urllib.parse import parse_qs
from jwt import decode as jwt_decode
from django.conf import settings


@database_sync_to_async
def get_user(validated_token):
    try:
        user = get_user_model().objects.get(id=validated_token["user_id"])
        # return get_user_model().objects.get(id=toke_id)
        print(f"{user}")
        return user
   
    except User.DoesNotExist:
        return AnonymousUser()



class JwtAuthMiddleware(BaseMiddleware):
    def __init__(self, inner):
        self.inner = inner

    async def __call__(self, scope, receive, send):
       # Close old database connections to prevent usage of timed out connections
        close_old_connections()

        # Get the token
        token = parse_qs(scope["query_string"].decode("utf8"))["token"][0]

        # Try to authenticate the user
        try:
            # This will automatically validate the token and raise an error if token is invalid
            UntypedToken(token)
        except (InvalidToken, TokenError) as e:
            # Token is invalid
            print(e)
            return None
        else:
            #  Then token is valid, decode it
            decoded_data = jwt_decode(token, settings.SECRET_KEY, algorithms=["HS256"])
            print(decoded_data)
            # Will return a dictionary like -
            # {
            #     "token_type": "access",
            #     "exp": 1568770772,
            #     "jti": "5c15e80d65b04c20ad34d77b6703251b",
            #     "user_id": 6
            # }

            # Get the user using ID
            scope["user"] = await get_user(validated_token=decoded_data)
        return await super().__call__(scope, receive, send)


def JwtAuthMiddlewareStack(inner):
    return JwtAuthMiddleware(AuthMiddlewareStack(inner))

you can import it into your consumer’s routing.py or asgi.py file like this

"""
ASGI config for config 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.1/howto/deployment/asgi/
"""

import os
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.core.asgi import get_asgi_application
from channels.security.websocket import AllowedHostsOriginValidator
from chat.consumers import ChatConsumer
from django.urls import path, re_path
from .channelsmiddleware import JwtAuthMiddlewareStack

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")

application = ProtocolTypeRouter(
    {
        "http": get_asgi_application(),
        "websocket": AllowedHostsOriginValidator(
            JwtAuthMiddlewareStack(
                URLRouter(
                    [
                        #path(),your routes here 
                    ]
                )
            ),
        ),
    }
)

Well I had the same problem. First of all you can’t do JWT authentication with django channels because the only thing that you can send through your channels is query string and you can’t set header parameters or such thing like http protocol (especially if your using JavaScript as your client side). I didn’t want to send my token as query string because of security purpose (because every one can see it). So I explain my solution here and maybe it can solve your problem too. I created an API for registering in my socket and in that API I returned a ticket (uuid type) as a response and in the same API I cached this ticket based on a user:

class RegisterFilterAPIView(APIView):
    """
        get:
            API view for retrieving ticket uuid.
    """
    authentication_classes = (JWTAuthentication,)
    permission_classes = (IsAuthenticatedOrReadOnly,)

    def get(self, request, *args, **kwargs):
        ticket_uuid = str(uuid4())

        if request.user.is_anonymous:
            cache.set(ticket_uuid, False, TICKET_EXPIRE_TIME)
        else:
            # You can set any condition based on logged in user here
            cache.set(ticket_uuid, some_conditions, TICKET_EXPIRE_TIME)

        return Response({'ticket_uuid': ticket_uuid})

After this part I sent this ticket as a query string to my socket like:

var endpoint = 'ws://your/socket/endpoint/?ticket_uuid=some_ticket';
var newSocket = new WebSocket(endpoint);

newSocket.onmessage = function (e) {
    console.log("message", e)
};
newSocket.onopen = function (e) {
    console.log("open", e);
};
newSocket.onerror = function (e) {
    console.log("error", e)
};
newSocket.onclose = function (e) {
    console.log("close", e)
};

Note that the above codes are written in JS so you should change it to something else based on your requirements. And finally in my consumer I handled this ticket which is created in my register API:

from urllib.parse import parse_qsl
from django.core.cache import cache
from channels.generic.websocket import AsyncJsonWebsocketConsumer


class FilterConsumer(AsyncJsonWebsocketConsumer):

    async def websocket_connect(self, event):
        try:
            query_string = self.scope['query_string'].decode('utf-8')
            query_params = dict(parse_qsl(query_string))
            ticket_uuid = query_params.get('ticket_uuid')
            self.scope['has_ticket'] = cache.get(ticket_uuid)
            if not cache.delete(ticket_uuid): # I destroyed ticket for performance and security purposes
                raise Exception('ticket not found')
        except:
            await self.close()
            return

        await self.accept()

So now you have a register API (like obtain token API) which is safe and you can generate a token based on your JWT token but make sure your server supports a cache backend service. You can also set self.scope['user'] in your websocket connect method based on your ticket value. I hope this can solve your problem.