authentik.root.asgi

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

 1"""
 2ASGI config for authentik project.
 3
 4It exposes the ASGI callable as a module-level variable named ``application``.
 5
 6For more information on this file, see
 7https://docs.djangoproject.com/en/3.0/howto/deployment/asgi/
 8"""
 9
10import django
11from channels.routing import ProtocolTypeRouter, URLRouter
12from django.core.asgi import get_asgi_application
13from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
14
15from authentik.root.setup import setup
16
17# DJANGO_SETTINGS_MODULE is set in gunicorn.conf.py
18
19setup()
20django.setup()
21
22
23from authentik.root import websocket  # noqa
24
25
26class LifespanApp:
27    """
28    temporary shim for https://github.com/django/channels/issues/1216
29    needed so that hypercorn doesn't display an error.
30    this uses ASGI 2.0 format, not the newer 3.0 single callable
31    """
32
33    def __init__(self, scope):
34        self.scope = scope
35
36    async def __call__(self, receive, send):
37        if self.scope["type"] == "lifespan":
38            while True:
39                message = await receive()
40                if message["type"] == "lifespan.startup":
41                    await send({"type": "lifespan.startup.complete"})
42                elif message["type"] == "lifespan.shutdown":
43                    await send({"type": "lifespan.shutdown.complete"})
44                    return
45
46
47class RouteNotFoundMiddleware:
48    """Middleware to ignore 404s for websocket requests
49    taken from https://github.com/django/daphne/issues/165#issuecomment-808284950"""
50
51    def __init__(self, app):
52        self.app = app
53
54    async def __call__(self, scope, receive, send):
55        try:
56            return await self.app(scope, receive, send)
57        except ValueError as exc:
58            if "No route found for path" in str(exc) and scope["type"] == "websocket":
59                await send({"type": "websocket.close"})
60            else:
61                raise exc
62
63
64class AuthentikAsgi(SentryAsgiMiddleware):
65    """Root ASGI App wrapper"""
66
67    def call_startup(self):
68        from authentik.root.signals import post_startup, pre_startup, startup
69
70        pre_startup.send(sender=self)
71        startup.send(sender=self)
72        post_startup.send(sender=self)
73
74
75application = AuthentikAsgi(
76    ProtocolTypeRouter(
77        {
78            "http": get_asgi_application(),
79            "websocket": RouteNotFoundMiddleware(URLRouter(websocket.websocket_urlpatterns)),
80            "lifespan": LifespanApp,
81        }
82    )
83)
class LifespanApp: 
27class LifespanApp:
28    """
29    temporary shim for https://github.com/django/channels/issues/1216
30    needed so that hypercorn doesn't display an error.
31    this uses ASGI 2.0 format, not the newer 3.0 single callable
32    """
33
34    def __init__(self, scope):
35        self.scope = scope
36
37    async def __call__(self, receive, send):
38        if self.scope["type"] == "lifespan":
39            while True:
40                message = await receive()
41                if message["type"] == "lifespan.startup":
42                    await send({"type": "lifespan.startup.complete"})
43                elif message["type"] == "lifespan.shutdown":
44                    await send({"type": "lifespan.shutdown.complete"})
45                    return

temporary shim for https://github.com/django/channels/issues/1216 needed so that hypercorn doesn't display an error. this uses ASGI 2.0 format, not the newer 3.0 single callable

LifespanApp(scope) 
34    def __init__(self, scope):
35        self.scope = scope
scope
class RouteNotFoundMiddleware: 
48class RouteNotFoundMiddleware:
49    """Middleware to ignore 404s for websocket requests
50    taken from https://github.com/django/daphne/issues/165#issuecomment-808284950"""
51
52    def __init__(self, app):
53        self.app = app
54
55    async def __call__(self, scope, receive, send):
56        try:
57            return await self.app(scope, receive, send)
58        except ValueError as exc:
59            if "No route found for path" in str(exc) and scope["type"] == "websocket":
60                await send({"type": "websocket.close"})
61            else:
62                raise exc

Middleware to ignore 404s for websocket requests taken from https://github.com/django/daphne/issues/165#issuecomment-808284950

RouteNotFoundMiddleware(app) 
52    def __init__(self, app):
53        self.app = app
app
class AuthentikAsgi(sentry_sdk.integrations.asgi.SentryAsgiMiddleware): 
65class AuthentikAsgi(SentryAsgiMiddleware):
66    """Root ASGI App wrapper"""
67
68    def call_startup(self):
69        from authentik.root.signals import post_startup, pre_startup, startup
70
71        pre_startup.send(sender=self)
72        startup.send(sender=self)
73        post_startup.send(sender=self)

Root ASGI App wrapper

def call_startup(self): 
68    def call_startup(self):
69        from authentik.root.signals import post_startup, pre_startup, startup
70
71        pre_startup.send(sender=self)
72        startup.send(sender=self)
73        post_startup.send(sender=self)
application = <AuthentikAsgi object>