Coverage for backend/django/authentication/middleware.py: 98%
35 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-03-26 20:57 +0000
« prev ^ index » next coverage.py v7.10.7, created at 2026-03-26 20:57 +0000
1import jwt
2from channels.middleware import BaseMiddleware
3from django.contrib.auth.middleware import RemoteUserMiddleware
4from CoreRoot import settings
5from CoreRoot.helpers import get_asgi_header_value
8class AhuoraRemoteUserMiddleware(RemoteUserMiddleware):
9 """Mirror Django's remote-user middleware while binding to our header names.
11 The middleware trusts `settings.REMOTE_USER_HEADER` for identity and skips the
12 default forced-logout behaviour so an absent header does not terminate the
13 session during multi-hop requests handled by infrastructure components.
14 """
16 header = settings.REMOTE_USER_HEADER
17 force_logout_if_no_header = False
19 def process_request(self, request):
20 super().process_request(request)
23####
24REMOTE_USER_ID = "0195c52c-7843-7e16-8790-3f4f24bc22e3"
25REMOTE_USER_EMAIL = "debug@example.com"
26REMOTE_USER_ACCESS_TOKEN = jwt.encode(
27 {
28 "exp": 1756268489,
29 "iat": 1756268189,
30 "auth_time": 1756268189,
31 "jti": "f7cf0ea7-6e22-4f37-be16-32399127e500",
32 "iss": "http://issuer.com",
33 "aud": ["platform"],
34 "sub": "a60d7a1e-7d65-4b8d-92a9-1c0dcc1fe44f",
35 "typ": "Bearer",
36 "azp": "platform",
37 "sid": "70d7f166-facb-4c09-ad27-bc3f6ee897ce",
38 "acr": "1",
39 "allowed-origins": ["*"],
40 "scope": "openid profile email",
41 "email_verified": True,
42 "name": "John Smith",
43 "preferred_username": "example@email.com",
44 "given_name": "John",
45 "family_name": "Smith",
46 "email": "example@email.com",
47 },
48 "secret",
49 algorithm="HS256",
50)
53def dummy_auth_header_middleware(get_response):
54 """
55 Intended for use when settings.DEBUG is True. If the header defined
56 by the above middleware is not set, then it will be set to a default
57 username, allowing for a "remote" test account to be created and made available
58 for use without requiring any additional config from developers.
59 """
61 def middleware(request):
62 remote_user_header = request.META.get(settings.REMOTE_USER_HEADER)
64 if request.META.get("HTTP_DAPR_API_TOKEN"):
65 # If the Dapr API token is set, we assume this is a Dapr sidecar request
66 # and we do not set the remote user header.
67 return get_response(request)
69 if remote_user_header is None:
70 request.META[settings.REMOTE_USER_HEADER] = REMOTE_USER_ID
71 request.META[settings.REMOTE_USER_EMAIL_HEADER] = REMOTE_USER_EMAIL
72 request.META[settings.REMOTE_USER_GROUPS_HEADER] = (
73 settings.PLATFORM_ADMINISTRATORS_GROUP
74 )
75 request.META[settings.REMOTE_USER_ACCESS_TOKEN_HEADER] = (
76 REMOTE_USER_ACCESS_TOKEN
77 )
79 response = get_response(request)
81 return response
83 return middleware
86class AsgiAuthHeaderMiddleware(BaseMiddleware):
87 """
88 The ASGI compatible version of dummy_auth_header_middleware.
89 Intended for use when settings.DEBUG is True. If the remote user header
90 is not set, then it will be set to a default username, which is made available to
91 upstream consumers of the ASGI scope.
92 """
94 async def __call__(self, scope, receive, send):
95 """Inject default remote-user headers when absent in debug environments."""
96 if "headers" in scope: 96 ↛ 118line 96 didn't jump to line 118 because the condition on line 96 was always true
97 remote_user_header_value = get_asgi_header_value(
98 scope["headers"], settings.ASGI_REMOTE_USER_HEADER
99 )
101 if remote_user_header_value is None:
102 scope["headers"].append(
103 (settings.ASGI_REMOTE_USER_HEADER.encode(), REMOTE_USER_ID.encode())
104 )
105 scope["headers"].append(
106 (
107 settings.ASGI_REMOTE_USER_EMAIL_HEADER.encode(),
108 REMOTE_USER_EMAIL.encode(),
109 )
110 )
111 scope["headers"].append(
112 (
113 settings.ASGI_REMOTE_USER_GROUPS_HEADER.encode(),
114 settings.PLATFORM_ADMINISTRATORS_GROUP.encode(),
115 )
116 )
118 return await super().__call__(scope, receive, send)