c3lf-system-3/core/core/asgi.py

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

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application
from notify_sessions.routing import websocket_urlpatterns

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings')

django_asgi_app = get_asgi_application()


class TokenAuthMiddleware:
    """
    Token authorization middleware for Django Channels 2
    """

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

    def __call__(self, scope):
        import base64
        headers = dict(scope['headers'])
        if b'authorization' in headers:
            try:
                token_name, token_key = headers[b'authorization'].decode().split()
                if token_name == 'Basic':
                    b64 = base64.b64decode(token_key)
                    user = b64.decode().split(':')[0]
                    password = b64.decode().split(':')[1]
                    print(user, password)
                else:
                    print("Token name is not Basic")
                    scope['user'] = None
            except:
                print("Token is not valid")
                scope['user'] = None
        else:
            print("Token is not in headers")
            scope['user'] = None


TokenAuthMiddlewareStack = lambda inner: TokenAuthMiddleware(AuthMiddlewareStack(inner))

websocket_asgi_app = AllowedHostsOriginValidator(
    AuthMiddlewareStack(
        URLRouter(
            websocket_urlpatterns
        )
    )
)

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": websocket_asgi_app,
})