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

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 

16 header = settings.REMOTE_USER_HEADER 

17 force_logout_if_no_header = False 

18 

19 def process_request(self, request): 

20 super().process_request(request) 

21 

22 

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) 

51 

52 

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

60 

61 def middleware(request): 

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

63 

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) 

68 

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 ) 

78 

79 response = get_response(request) 

80 

81 return response 

82 

83 return middleware 

84 

85 

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

93 

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 ) 

100 

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 ) 

117 

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