66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
"""
|
|
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,
|
|
})
|