WebSocket_JWT_authentication_middleware.py

"""
Jwt middleware for django channels
"""

from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from jwt import decode as jwt_decode
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from rest_framework_simplejwt.tokens import UntypedToken


class JwtMiddleware:
    """
    Custom middleware that return user ID from token.
    """

    def __init__(self, app):
        # Store the ASGI application we were passed
        self.app = app

    async def __call__(self, scope, receive, send):
        """
            look up for token in headers , token must insert in headers like :
            "token" : "kwer32442341234234234.dsf23sdfsdfsdfqqw.213132123asdf"
     
        """
        token = scope["headers"][1][1]
        try:
            valid = UntypedToken(token)
        except (InvalidToken, TokenError) as e:
            scope['user'] = AnonymousUser()
        else:
            decoded_data = jwt_decode(token, settings.SECRET_KEY, algorithms=[settings.SIMPLE_JWT["ALGORITHM"]])

            scope['user'] = decoded_data['user_id']

        return await self.app(scope, receive, send)