Coverage for backend/authentication/middleware.py: 88%

33 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-11-06 23:27 +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 

6 

7 

8class AhuoraRemoteUserMiddleware(RemoteUserMiddleware): 

9 """Mirror Django's remote-user middleware while binding to our header names. 

10 

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 """ 

15 header = settings.REMOTE_USER_HEADER 

16 force_logout_if_no_header = False 

17 

18#### 

19REMOTE_USER_ID = "0195c52c-7843-7e16-8790-3f4f24bc22e3" 

20REMOTE_USER_EMAIL = "debug@example.com" 

21REMOTE_USER_ACCESS_TOKEN = jwt.encode({ 

22 "exp": 1756268489, 

23 "iat": 1756268189, 

24 "auth_time": 1756268189, 

25 "jti": "f7cf0ea7-6e22-4f37-be16-32399127e500", 

26 "iss": "http://issuer.com", 

27 "aud": ["platform"], 

28 "sub": "a60d7a1e-7d65-4b8d-92a9-1c0dcc1fe44f", 

29 "typ": "Bearer", 

30 "azp": "platform", 

31 "sid": "70d7f166-facb-4c09-ad27-bc3f6ee897ce", 

32 "acr": "1", 

33 "allowed-origins": ["*"], 

34 "scope": "openid profile email", 

35 "email_verified": True, 

36 "name": "John Smith", 

37 "preferred_username": "example@email.com", 

38 "given_name": "John", 

39 "family_name": "Smith", 

40 "email": "example@email.com" 

41}, "secret", algorithm="HS256") 

42 

43 

44def dummy_auth_header_middleware(get_response): 

45 """ 

46 Intended for use when settings.DEBUG is True. If the header defined 

47 by the above middleware is not set, then it will be set to a default 

48 username, allowing for a "remote" test account to be created and made available 

49 for use without requiring any additional config from developers. 

50 """ 

51 

52 def middleware(request): 

53 remote_user_header = request.META.get(settings.REMOTE_USER_HEADER) 

54 

55 if request.META.get('HTTP_DAPR_API_TOKEN'): 

56 # If the Dapr API token is set, we assume this is a Dapr sidecar request 

57 # and we do not set the remote user header. 

58 return get_response(request) 

59 

60 if remote_user_header is None: 

61 request.META[settings.REMOTE_USER_HEADER] = REMOTE_USER_ID 

62 request.META[settings.REMOTE_USER_EMAIL_HEADER] = REMOTE_USER_EMAIL 

63 request.META[settings.REMOTE_USER_GROUPS_HEADER] = settings.PLATFORM_ADMINISTRATORS_GROUP 

64 request.META[settings.REMOTE_USER_ACCESS_TOKEN_HEADER] = REMOTE_USER_ACCESS_TOKEN 

65 

66 response = get_response(request) 

67 

68 return response 

69 

70 return middleware 

71 

72class AsgiAuthHeaderMiddleware(BaseMiddleware): 

73 """ 

74 The ASGI compatible version of dummy_auth_header_middleware. 

75 Intended for use when settings.DEBUG is True. If the remote user header 

76 is not set, then it will be set to a default username, which is made available to 

77 upstream consumers of the ASGI scope. 

78 """ 

79 

80 async def __call__(self, scope, receive, send): 

81 """Inject default remote-user headers when absent in debug environments.""" 

82 if "headers" in scope: 82 ↛ 90line 82 didn't jump to line 90 because the condition on line 82 was always true

83 remote_user_header_value = get_asgi_header_value(scope["headers"], settings.ASGI_REMOTE_USER_HEADER) 

84 

85 if remote_user_header_value is None: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true

86 scope["headers"].append((settings.ASGI_REMOTE_USER_HEADER.encode(), REMOTE_USER_ID.encode())) 

87 scope["headers"].append((settings.ASGI_REMOTE_USER_EMAIL_HEADER.encode(), REMOTE_USER_EMAIL.encode())) 

88 scope["headers"].append((settings.ASGI_REMOTE_USER_GROUPS_HEADER.encode(), settings.PLATFORM_ADMINISTRATORS_GROUP.encode())) 

89 

90 return await super().__call__(scope, receive, send)