Coverage for src/lexigram/admin/middleware/auth_guard.py: 87%

1"""Session-based auth guard middleware for the Lexigram Admin panel. 

2 

3Redirects unauthenticated requests to the login page. Bypass paths 

4(login, setup, static assets, health) are always passed through so 

5authentication pages remain accessible before a session exists. 

6""" 

7 

8from __future__ import annotations 

9 

10from urllib.parse import quote 

11 

12from starlette.responses import RedirectResponse, Response 

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

14 

15from lexigram.logging import get_logger 

16 

17logger = get_logger(__name__) 

18 

19# Paths (or path suffixes) that are always accessible without a session. 

20_BYPASS_SUFFIXES: frozenset[str] = frozenset( 

21 { 

22 "/login", 

23 "/login/", 

24 "/logout", 

25 "/logout/", 

26 "/setup", 

27 "/setup/", 

28 "/health", 

29 "/health/", 

30 "/login/2fa", 

31 "/login/2fa/", 

32 "/verify-email", 

33 "/verify-email/", 

34 "/password-reset", 

35 "/password-reset/", 

36 "/register", 

37 "/register/", 

38 } 

39) 

40 

41_BYPASS_PREFIXES: tuple[str, ...] = ( 

42 "/static/", 

43 "/admin/static/", 

44) 

45 

46# Token-bearing sub-paths that must remain reachable without a session 

47# (e.g. the email verification and password-reset links emailed to admins). 

48_BYPASS_TOKEN_PREFIXES: tuple[str, ...] = ( 

49 "/admin/verify-email/", 

50 "/admin/password-reset/", 

51) 

52 

53# Exact public routes derived from _BYPASS_SUFFIXES — full-path membership 

54# only, never suffix matching, so resource names like "login", "register", 

55# "health", or "setup" cannot shadow protected admin routes. 

56_BYPASS_ROUTES: frozenset[str] = frozenset( 

57 f"/admin{s.rstrip('/')}" for s in _BYPASS_SUFFIXES 

58) 

59 

60 

61class AdminAuthGuardMiddleware: 

62 """Pure ASGI middleware that enforces session-based authentication. 

63 

64 Any request whose path is not in the bypass list must carry a 

65 Starlette session with ``admin_user_id`` set. If the session is 

66 missing or empty the client is redirected to ``/admin/login``. 

67 

68 This middleware is intentionally lightweight — it does not touch the 

69 database. It relies solely on the signed session cookie that 

70 ``AuthController`` writes on successful login. 

71 """ 

72 

73 def __init__(self, app: ASGIApp) -> None: 

74 """Initialise the middleware. 

75 

76 Args: 

77 app: The next ASGI application in the stack. 

78 """ 

79 self._app = app 

80 

81 # ------------------------------------------------------------------ 

82 # ASGI callable 

83 # ------------------------------------------------------------------ 

84 

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

86 """Process an incoming HTTP request. 

87 

88 Args: 

89 scope: ASGI connection scope. 

90 receive: ASGI receive callable. 

91 send: ASGI send callable. 

92 """ 

93 if scope["type"] != "http": 

94 await self._app(scope, receive, send) 

95 return 

96 

97 path: str = scope.get("path", "") 

98 

99 if self._is_bypass_path(path): 

100 await self._app(scope, receive, send) 

101 return 

102 

103 # Inspect the Starlette session (populated by SessionMiddleware). 

104 if "session" in scope: 

105 user_id = scope["session"].get("admin_user_id") 

106 if user_id: 

107 await self._app(scope, receive, send) 

108 return 

109 

110 # No valid session — redirect to login preserving the original URL. 

111 # HTMX requests get HX-Redirect so the browser performs a full page 

112 # navigation; a plain 307 would make htmx swap the login page into 

113 # the current component (e.g. a widget container). 

114 logger.debug("auth_guard.unauthenticated path=%s", path) 

115 next_url = quote(path, safe="/") 

116 login_url = f"/admin/login?next={next_url}" 

117 if self._is_htmx(scope): 

118 response = Response(status_code=200) 

119 response.headers["HX-Redirect"] = login_url 

120 else: 

121 response = RedirectResponse(url=login_url, status_code=307) 

122 await response(scope, receive, send) 

123 

124 # ------------------------------------------------------------------ 

125 # Helpers 

126 # ------------------------------------------------------------------ 

127 

128 @staticmethod 

129 def _is_htmx(scope: Scope) -> bool: 

130 """Return True when the request carries the htmx HX-Request header. 

131 

132 Args: 

133 scope: ASGI connection scope. 

134 

135 Returns: 

136 True for htmx fragment requests. 

137 """ 

138 headers = dict(scope.get("headers") or ()) 

139 return headers.get(b"hx-request") == b"true" 

140 

141 def _is_bypass_path(self, path: str) -> bool: 

142 """Return True if the path should bypass auth enforcement. 

143 

144 Args: 

145 path: The request path. 

146 

147 Returns: 

148 True when the path maps to a public endpoint or static asset. 

149 """ 

150 stripped = path.rstrip("/") or path 

151 if stripped in _BYPASS_ROUTES: 

152 return True 

153 

154 if any(path.startswith(prefix) for prefix in _BYPASS_TOKEN_PREFIXES): 

155 return True 

156 

157 return any(path.startswith(prefix) for prefix in _BYPASS_PREFIXES)