"django channels jwt auth" Code Answer's

django channels jwt auth

#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
#authmiddleware.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 cant then import it into your consumer's routing.py or asgi.py file like this
#asgi.py
    """
    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 
                        ]
                    )
                ),
            ),
        }
    )

Add Comment