Coverage for src/mathenjeu/apps/staticapp/authmount.py: 45%

1""" 

2@file 

3@brief Custom Router to check for authentification. 

4""" 

5from starlette.requests import Request 

6from starlette.responses import RedirectResponse 

7from starlette.routing import Mount 

8from starlette.staticfiles import StaticFiles 

9from starlette.types import ASGIApp, Scope, Receive, Send 

10 

11 

12class _CommonMethods: 

13 

14 def get_app_session(self, scope): 

15 """ 

16 Retrieves the :epkg:`starlette` application and 

17 the session. 

18 

19 @param scope request 

20 @return application, session 

21 """ 

22 app = scope.get('app', None) 

23 if app is None: 

24 return None, None 

25 request = Request(scope) # , receive=receive) 

26 session = app._get_session(request) 

27 return app, session 

28 

29 

30class AuthMount(_CommonMethods, Mount): 

31 """ 

32 The router checks for authentification by looking for a cookie 

33 which contains an alias. This alias can only be set if the user 

34 was able to authentify himself. 

35 """ 

36 

37 def __init__(self, path: str, app: ASGIApp, name: str = None) -> None: 

38 """ 

39 @param path application mapped to this path 

40 @param app application 

41 @param name name 

42 """ 

43 Mount.__init__(self, path=path, app=app, name=name) 

44 

45 async def __call__(self, scope: Scope, receive: Receive, send: Send): 

46 """ 

47 Checks the user is authenticated, falls back in 

48 the previous behavior, otherwise redirect to the 

49 authentification page (``/login``). 

50 """ 

51 redirect = True 

52 app, session = self.get_app_session(scope) 

53 if session is not None: 

54 if app is not None: 

55 app._log_event("staticpage", scope, session=session) 

56 alias = session.get('alias', None) 

57 if alias is not None: 

58 await Mount.__call__(self, scope, receive, send) 

59 redirect = False 

60 if redirect: 

61 # Requires authentification. 

62 path = scope.get('root_path', '') + '%2F' + scope.get('path', '') 

63 path = path.replace('/', "%2F") 

64 resp = RedirectResponse(url='/login?returnto=' + path) 

65 await resp(scope, receive, send) 

66 

67 

68class AuthStaticFiles(_CommonMethods, StaticFiles): 

69 """ 

70 Overloads *StaticFiles* to check authentification. 

71 """ 

72 

73 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

74 """ 

75 The ASGI entry point. 

76 """ 

77 redirect = True 

78 app, session = self.get_app_session(scope) 

79 if session is not None: 

80 if app is not None: 

81 app._log_event("staticpage", scope, session=session) 

82 alias = session.get('alias', None) 

83 if alias is not None: 

84 assert scope["type"] == "http" 

85 

86 if not self.config_checked: 

87 await self.check_config() 

88 self.config_checked = True 

89 

90 path = self.get_path(scope) 

91 response = await self.get_response(path, scope) 

92 await response(scope, receive, send) 

93 redirect = False 

94 if redirect: 

95 # Requires authentification. 

96 path = scope.get('root_path', '') + '%2F' + scope.get('path', '') 

97 path = path.replace('/', "%2F") 

98 resp = RedirectResponse(url='/login?returnto=' + path) 

99 await resp(scope, receive, send)