Coverage for src/lexigram/admin/middleware/auth.py: 20%

1"""Authentication middleware for Lexigram Admin. 

2 

3This middleware integrates with Lexigram's DI container to provide 

4request-scoped user authentication and authorization. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import datetime 

10from typing import TYPE_CHECKING 

11 

12from starlette.requests import Request 

13from starlette.requests import Request as StarletteRequest 

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

15 

16from lexigram.admin.auth.models import GUEST_USER 

17from lexigram.admin.auth.store.base import AbstractAdminUserStore 

18from lexigram.contracts import AuthenticatedUserProtocol 

19from lexigram.di.decorators import inject 

20from lexigram.logging import get_logger 

21from lexigram.primitives import clock 

22 

23if TYPE_CHECKING: 

24 from lexigram.admin.auth.protocols import AdminSessionServiceProtocol 

25 

26logger = get_logger(__name__) 

27 

28 

29@inject 

30class AdminAuthMiddleware: 

31 """Pure ASGI middleware for admin authentication and authorization. 

32 

33 10x faster than BaseHTTPMiddleware - no task creation overhead. 

34 

35 This middleware: 

36 1. Checks if user is authenticated (via session/JWT) 

37 2. Loads AdminUser from session or creates guest user 

38 3. Injects AdminUser into DI request scope 

39 4. Optionally redirects unauthenticated users to login 

40 

41 The injected AdminUser is then available to all controllers 

42 via dependency injection. 

43 """ 

44 

45 def __init__( 

46 self, 

47 app: ASGIApp, 

48 user_store: AbstractAdminUserStore | None = None, 

49 session_service: AdminSessionServiceProtocol | None = None, 

50 require_auth: bool = False, 

51 excluded_paths: list[str] | None = None, 

52 ): 

53 """Initialize auth middleware. 

54 

55 Args: 

56 app: ASGI application 

57 user_store: AbstractAdminUserStore for loading users 

58 session_service: AdminSessionServiceProtocol for TTL enforcement 

59 require_auth: If True, redirect unauthenticated users 

60 excluded_paths: Paths that don't require authentication 

61 """ 

62 self.app = app 

63 self.user_store = user_store 

64 self._session_service = session_service 

65 self.require_auth = require_auth 

66 self.excluded_paths = excluded_paths or [] 

67 

68 def _is_path_excluded(self, path: str) -> bool: 

69 """Check if path is excluded from auth requirements. 

70 

71 Args: 

72 path: Request path to check 

73 

74 Returns: 

75 True if path is excluded, False otherwise 

76 """ 

77 for pattern in self.excluded_paths: 

78 if pattern.endswith("*"): 

79 # Wildcard matching 

80 prefix = pattern[:-1] 

81 if path.startswith(prefix): 

82 return True 

83 elif path == pattern: 

84 return True 

85 return False 

86 

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

88 """Pure ASGI middleware implementation.""" 

89 

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

91 # Pass through non-HTTP requests 

92 await self.app(scope, receive, send) 

93 return 

94 

95 path = scope.get("path", "") 

96 

97 # Check if path is excluded 

98 if self._is_path_excluded(path): 

99 await self.app(scope, receive, send) 

100 return 

101 

102 # Build request for store access 

103 request = StarletteRequest(scope, receive) 

104 

105 # Resolve tenant 

106 if hasattr(request.state, "tenant_id"): 

107 pass 

108 

109 # Load user from session 

110 user = await self._load_user(request) 

111 

112 # Check if authentication is required 

113 if self.require_auth and ( 

114 user is None or getattr(user, "user_id", "guest") == "guest" 

115 ): 

116 from starlette.exceptions import HTTPException 

117 

118 raise HTTPException(status_code=401, detail="Unauthorized") 

119 

120 # Store user in scope for access by other middleware/controllers 

121 scope["user"] = user 

122 

123 # Also store in scope state (becomes request.state) 

124 if "state" not in scope: 

125 scope["state"] = {} 

126 scope["state"]["user"] = user 

127 

128 # Continue with request 

129 try: 

130 await self.app(scope, receive, send) 

131 finally: 

132 pass 

133 

134 async def _load_user(self, request: Request) -> AuthenticatedUserProtocol | None: 

135 """Load user from request session or return guest user. 

136 

137 Validates via AdminSessionService (which enforces idle + absolute TTL) 

138 when both a session_id and session_service are available. Falls back 

139 to direct user_store lookup for backward compatibility with sessions 

140 created before the session_id was stored. 

141 

142 Args: 

143 request: The incoming request 

144 

145 Returns: 

146 AdminUser instance (or GUEST_USER if not authenticated) 

147 """ 

148 # Canonical admin auth path: signed Starlette session cookie managed by 

149 # SessionMiddleware + AuthController. 

150 try: 

151 if "session" not in request.scope: 

152 return GUEST_USER 

153 

154 # ── Session-service path (enforces TTL) ──────────────────── 

155 session_id = request.session.get("session_id") 

156 if self._session_service is not None: 

157 if not session_id: 

158 # Service-bound deployment without a service-managed 

159 # session: never consult the legacy fallback. 

160 return GUEST_USER 

161 session_data = await self._session_service.get_session(session_id) 

162 if session_data is None: 

163 # Session expired or revoked — clear cookie 

164 request.session.clear() 

165 logger.debug("session.expired_or_revoked", session_id=session_id) 

166 return GUEST_USER 

167 

168 admin_id = session_data.get("admin_id") 

169 if admin_id is None: 

170 request.session.clear() 

171 return GUEST_USER 

172 

173 user = ( 

174 await self.user_store.get_by_id(admin_id) 

175 if self.user_store 

176 else None 

177 ) 

178 if user is None or not user.is_active: 

179 await self._session_service.revoke_session(session_id) 

180 request.session.clear() 

181 return GUEST_USER 

182 

183 logger.debug( 

184 "Successfully loaded user %s from session (TTL-validated)", 

185 user.user_id, 

186 ) 

187 return user 

188 

189 # ── Legacy fallback (no session_service / no session_id) ─── 

190 user_id = request.session.get("admin_user_id") 

191 if user_id: 

192 expires_at_raw = request.session.get("admin_session_expires_at") 

193 if expires_at_raw is None: 

194 request.session.clear() 

195 return GUEST_USER 

196 try: 

197 expires_at = datetime.fromisoformat(expires_at_raw) 

198 except ValueError: 

199 request.session.clear() 

200 return GUEST_USER 

201 if expires_at.tzinfo is None or clock.now() >= expires_at: 

202 request.session.clear() 

203 return GUEST_USER 

204 user_store = self.user_store 

205 user = await user_store.get_by_id(user_id) if user_store else None 

206 if user and user.is_active: 

207 logger.debug( 

208 "Successfully loaded user %s from request.session", 

209 user.user_id, 

210 ) 

211 return user 

212 except (RuntimeError, ValueError, OSError, AssertionError) as e: 

213 logger.debug("Failed to load user from request.session: %s", e) 

214 

215 return GUEST_USER 

216 

217 

218def current_user(request: Request | None = None) -> AuthenticatedUserProtocol | None: 

219 """Get the current authenticated user from request scope. 

220 

221 This is a helper function that can be used in controllers 

222 when DI injection is not available. 

223 

224 Args: 

225 request: Request object (optional, will try to get from DI scope) 

226 

227 Returns: 

228 The current User or GUEST_USER 

229 

230 Example: 

231 ```python 

232 @get("/admin/profile") 

233 async def profile(request: Request): 

234 user = current_user(request) 

235 return {"username": user.name} 

236 ``` 

237 """ 

238 # Try to get from request state first 

239 if request and hasattr(request.state, "user"): 

240 return request.state.user 

241 

242 return GUEST_USER