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

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 typing import TYPE_CHECKING 

10 

11from starlette.requests import Request 

12from starlette.requests import Request as StarletteRequest 

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

14 

15from lexigram.admin.auth.models import GUEST_USER 

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

17from lexigram.contracts import AuthenticatedUserProtocol 

18from lexigram.di.decorators import inject 

19from lexigram.logging import get_logger 

20 

21if TYPE_CHECKING: 

22 from lexigram.admin.auth.protocols import AdminSessionServiceProtocol 

23 

24logger = get_logger(__name__) 

25 

26 

27@inject 

28class AdminAuthMiddleware: 

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

30 

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

32 

33 This middleware: 

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

35 2. Loads AdminUser from session or creates guest user 

36 3. Injects AdminUser into DI request scope 

37 4. Optionally redirects unauthenticated users to login 

38 

39 The injected AdminUser is then available to all controllers 

40 via dependency injection. 

41 """ 

42 

43 def __init__( 

44 self, 

45 app: ASGIApp, 

46 user_store: AbstractAdminUserStore | None = None, 

47 session_service: AdminSessionServiceProtocol | None = None, 

48 require_auth: bool = False, 

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

50 ): 

51 """Initialize auth middleware. 

52 

53 Args: 

54 app: ASGI application 

55 user_store: AbstractAdminUserStore for loading users 

56 session_service: AdminSessionServiceProtocol for TTL enforcement 

57 require_auth: If True, redirect unauthenticated users 

58 excluded_paths: Paths that don't require authentication 

59 """ 

60 self.app = app 

61 self.user_store = user_store 

62 self._session_service = session_service 

63 self.require_auth = require_auth 

64 self.excluded_paths = excluded_paths or [] 

65 

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

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

68 

69 Args: 

70 path: Request path to check 

71 

72 Returns: 

73 True if path is excluded, False otherwise 

74 """ 

75 for pattern in self.excluded_paths: 

76 if pattern.endswith("*"): 

77 # Wildcard matching 

78 prefix = pattern[:-1] 

79 if path.startswith(prefix): 

80 return True 

81 elif path == pattern: 

82 return True 

83 return False 

84 

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

86 """Pure ASGI middleware implementation.""" 

87 

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

89 # Pass through non-HTTP requests 

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

91 return 

92 

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

94 

95 # Check if path is excluded 

96 if self._is_path_excluded(path): 

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

98 return 

99 

100 # Build request for store access 

101 request = StarletteRequest(scope, receive) 

102 

103 # Resolve tenant 

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

105 pass 

106 

107 # Load user from session 

108 user = await self._load_user(request) 

109 

110 # Check if authentication is required 

111 if self.require_auth and ( 

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

113 ): 

114 from starlette.exceptions import HTTPException 

115 

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

117 

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

119 scope["user"] = user 

120 

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

122 if "state" not in scope: 

123 scope["state"] = {} 

124 scope["state"]["user"] = user 

125 

126 # Continue with request 

127 try: 

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

129 finally: 

130 pass 

131 

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

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

134 

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

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

137 to direct user_store lookup for backward compatibility with sessions 

138 created before the session_id was stored. 

139 

140 Args: 

141 request: The incoming request 

142 

143 Returns: 

144 AdminUser instance (or GUEST_USER if not authenticated) 

145 """ 

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

147 # SessionMiddleware + AuthController. 

148 try: 

149 if "session" not in request.scope: 

150 return GUEST_USER 

151 

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

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

154 if session_id and self._session_service is not None: 

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

156 if session_data is None: 

157 # Session expired or revoked — clear cookie 

158 request.session.clear() 

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

160 return GUEST_USER 

161 

162 admin_id = session_data.get("admin_id") 

163 if admin_id is None: 

164 request.session.clear() 

165 return GUEST_USER 

166 

167 user = ( 

168 await self.user_store.get_by_id(admin_id) 

169 if self.user_store 

170 else None 

171 ) 

172 if user is None or not user.is_active: 

173 await self._session_service.revoke_session(session_id) 

174 request.session.clear() 

175 return GUEST_USER 

176 

177 logger.debug( 

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

179 user.user_id, 

180 ) 

181 return user 

182 

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

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

185 if user_id: 

186 user_store = self.user_store 

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

188 if user and user.is_active: 

189 logger.debug( 

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

191 user.user_id, 

192 ) 

193 return user 

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

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

196 

197 return GUEST_USER 

198 

199 

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

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

202 

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

204 when DI injection is not available. 

205 

206 Args: 

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

208 

209 Returns: 

210 The current User or GUEST_USER 

211 

212 Example: 

213 ```python 

214 @get("/admin/profile") 

215 async def profile(request: Request): 

216 user = current_user(request) 

217 return {"username": user.name} 

218 ``` 

219 """ 

220 # Try to get from request state first 

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

222 return request.state.user 

223 

224 return GUEST_USER