Coverage for backend/django/authentication/middleware.py: 98%

1from channels.middleware import BaseMiddleware 

2from django.contrib.auth.middleware import RemoteUserMiddleware 

3from CoreRoot import settings 

4from CoreRoot.helpers import get_asgi_header_value 

5from authentication.token_helpers import build_human_user_access_token 

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" 

26def dummy_auth_header_middleware(get_response): 

27 """ 

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

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

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

31 for use without requiring any additional config from developers. 

32 """ 

33 

34 def middleware(request): 

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

36 

37 if request.META.get("HTTP_DAPR_API_TOKEN"): 

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

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

40 return get_response(request) 

41 

42 if remote_user_header is None: 

43 request.META[settings.REMOTE_USER_HEADER] = REMOTE_USER_ID 

44 request.META[settings.REMOTE_USER_EMAIL_HEADER] = REMOTE_USER_EMAIL 

45 request.META[settings.REMOTE_USER_GROUPS_HEADER] = ( 

46 settings.PLATFORM_ADMINISTRATORS_GROUP 

47 ) 

48 request.META[settings.REMOTE_USER_ACCESS_TOKEN_HEADER] = ( 

49 build_human_user_access_token() 

50 ) 

51 

52 response = get_response(request) 

53 

54 return response 

55 

56 return middleware 

57 

58 

59class AsgiAuthHeaderMiddleware(BaseMiddleware): 

60 """ 

61 The ASGI compatible version of dummy_auth_header_middleware. 

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

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

64 upstream consumers of the ASGI scope. 

65 """ 

66 

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

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

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

70 remote_user_header_value = get_asgi_header_value( 

71 scope["headers"], settings.ASGI_REMOTE_USER_HEADER 

72 ) 

73 

74 if remote_user_header_value is None: 

75 scope["headers"].append( 

76 (settings.ASGI_REMOTE_USER_HEADER.encode(), REMOTE_USER_ID.encode()) 

77 ) 

78 scope["headers"].append( 

79 ( 

80 settings.ASGI_REMOTE_USER_EMAIL_HEADER.encode(), 

81 REMOTE_USER_EMAIL.encode(), 

82 ) 

83 ) 

84 scope["headers"].append( 

85 ( 

86 settings.ASGI_REMOTE_USER_GROUPS_HEADER.encode(), 

87 settings.PLATFORM_ADMINISTRATORS_GROUP.encode(), 

88 ) 

89 ) 

90 scope["headers"].append( 

91 ( 

92 settings.ASGI_REMOTE_USER_ACCESS_TOKEN_HEADER.encode(), 

93 build_human_user_access_token().encode(), 

94 ) 

95 ) 

96 

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