Coverage for src/lexigram/admin/auth/guards.py: 0%

1"""Authentication and authorization guards for lexigram-admin. 

2 

3Provides middleware and guard utilities for protecting routes. 

4Integrates with lexigram-auth session management. 

5""" 

6 

7from __future__ import annotations 

8 

9import base64 

10from dataclasses import dataclass 

11from functools import wraps 

12import hmac 

13from typing import TYPE_CHECKING, Any 

14 

15from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint 

16from starlette.requests import Request 

17from starlette.responses import RedirectResponse, Response 

18 

19from lexigram.admin.auth.permissions import PermissionSet, get_user_permissions 

20from lexigram.admin.exceptions import ErrorCode, PermissionDeniedError 

21from lexigram.contracts import ( 

22 AuthorizerProtocol, 

23 AuthProviderProtocol, 

24) 

25from lexigram.contracts.web import RequestProtocol, ResponseProtocol 

26from lexigram.di.decorators import inject 

27from lexigram.logging import get_logger 

28from lexigram.result import Err, Ok, Result 

29from lexigram.serialization.backends import json as json_backend 

30 

31if TYPE_CHECKING: 

32 from collections.abc import Awaitable, Callable 

33 

34logger = get_logger(__name__) 

35 

36 

37@dataclass 

38class GuardConfig: 

39 """Configuration for authentication guards.""" 

40 

41 login_url: str = "/admin/login" 

42 logout_url: str = "/admin/logout" 

43 exempt_paths: tuple[str, ...] = ( 

44 "/admin/login", 

45 "/admin/static", 

46 "/admin/health", 

47 # Standalone pre-session flows (own CSRF + guest handling): 

48 "/admin/setup", 

49 "/admin/verify-email", 

50 "/admin/password-reset", 

51 ) 

52 # Whether to accept Authorization: Bearer <token> for admin APIs. 

53 # Default: False to enforce strict cookie-based admin sessions. 

54 allow_bearer_tokens: bool = False 

55 htmx_redirect_header: str = "HX-Redirect" 

56 

57 

58@inject 

59class AuthGuardMiddleware(BaseHTTPMiddleware): 

60 """Middleware that enforces authentication on admin routes. 

61 

62 Checks for valid session and loads user into request.state. 

63 Redirects unauthenticated requests to login page. 

64 

65 For HTMX requests, returns HX-Redirect header instead of 302. 

66 """ 

67 

68 def __init__( 

69 self, 

70 app: Any, 

71 auth_provider: AuthProviderProtocol | None = None, 

72 config: GuardConfig | None = None, 

73 authorizer: AuthorizerProtocol | None = None, 

74 ) -> None: 

75 super().__init__(app) 

76 self.auth_provider = auth_provider 

77 self.config = config or GuardConfig() 

78 self.authorizer = authorizer 

79 

80 async def dispatch( # type: ignore[override] 

81 self, 

82 request: RequestProtocol, # type: ignore[override] 

83 call_next: RequestResponseEndpoint, 

84 ) -> ResponseProtocol: 

85 # Skip auth for exempt paths 

86 if self._is_exempt(request.url.path): 

87 return await call_next(request) # type: ignore[return-value, arg-type] 

88 

89 # Check if user is already loaded by AdminAuthMiddleware 

90 user = getattr(request.state, "user", None) 

91 

92 # If not, try to load it (fallback/standalone usage) 

93 if user is None: 

94 user = await self._get_authenticated_user(request) 

95 

96 if not self._is_authenticated(user): 

97 # If request has Authorization: Bearer, return 401 instead of redirect 

98 auth_header = request.headers.get("Authorization", "") 

99 if auth_header.startswith("Bearer "): 

100 from starlette.responses import JSONResponse 

101 

102 return JSONResponse( # type: ignore[return-value] 

103 {"authenticated": False, "detail": "Invalid or missing token"}, 

104 status_code=401, 

105 ) 

106 return self._redirect_to_login(request) 

107 

108 # Ensure user and permissions are in state 

109 request.state.user = user 

110 if not hasattr(request.state, "permissions"): 

111 # Use injected authorizer if available 

112 if self.authorizer: 

113 try: 

114 request.state.permissions = get_user_permissions( 

115 user, 

116 self.authorizer, 

117 ) 

118 except ( 

119 ConnectionError, 

120 RuntimeError, 

121 ValueError, 

122 TypeError, 

123 AttributeError, 

124 ): 

125 # Authorization failed, skip permissions 

126 logger.debug( 

127 "Could not compute user permissions", 

128 exc_info=True, 

129 ) 

130 request.state.permissions = None 

131 else: 

132 # No authorizer available, skip permissions 

133 request.state.permissions = None 

134 

135 return await call_next(request) # type: ignore[return-value, arg-type] 

136 

137 def _is_authenticated(self, user: Any) -> bool: 

138 """Check if user is traditionally authenticated (not guest).""" 

139 if user is None: 

140 return False 

141 

142 # Check common identity fields 

143 user_id = getattr(user, "user_id", None) or getattr(user, "id", None) 

144 if not user_id or user_id == "guest": 

145 return False 

146 

147 # Check activity 

148 return getattr(user, "is_active", True) 

149 

150 def _is_exempt(self, path: str) -> bool: 

151 """Check if path is exempt from auth.""" 

152 return any(path.startswith(exempt) for exempt in self.config.exempt_paths) 

153 

154 async def _get_authenticated_user(self, request: RequestProtocol) -> Any | None: 

155 """Get authenticated user from signed request session.""" 

156 if "session" in request.scope: # type: ignore[attr-defined] 

157 user_id = request.session.get("admin_user_id") # type: ignore[attr-defined] 

158 if user_id: 

159 try: 

160 if hasattr(self.auth_provider, "user_store"): 

161 return await self.auth_provider.user_store.get_user_by_id( # type: ignore[union-attr] 

162 user_id, 

163 ) 

164 return None 

165 except ( 

166 ConnectionError, 

167 RuntimeError, 

168 ValueError, 

169 TypeError, 

170 AttributeError, 

171 ): 

172 logger.debug("Session user resolution failed", exc_info=True) 

173 

174 # Try Authorization header (for API calls) 

175 # By default, admin routes do NOT accept bearer tokens to keep a strict 

176 # separation between admin sessions (cookie-based) and application 

177 # JWTs. This can be enabled explicitly via GuardConfig.allow_bearer_tokens. 

178 if self.config.allow_bearer_tokens: 

179 auth_header = request.headers.get("Authorization", "") 

180 if auth_header.startswith("Bearer "): 

181 token = auth_header[7:] 

182 try: 

183 # lexigram-auth uses authenticate_user or verify_token but AuthGuard usually checks tokens too 

184 if hasattr(self.auth_provider, "verify_token"): 

185 token_result = self.auth_provider.verify_token(token) # type: ignore[union-attr] 

186 if hasattr(token_result, "__await__"): 

187 token_result = await token_result 

188 # Handle Result[VerifiedToken, ...] (new API) 

189 if hasattr(token_result, "is_ok"): 

190 if token_result.is_ok(): 

191 verified = token_result.unwrap() # type: ignore[union-attr] 

192 return ( 

193 await self.auth_provider.user_store.get_user_by_id( # type: ignore[union-attr] 

194 verified.user_id, 

195 ) 

196 ) 

197 # Fallback: legacy dict payload (older providers) 

198 elif token_result and "sub" in token_result: # type: ignore[operator] 

199 return await self.auth_provider.user_store.get_user_by_id( # type: ignore[union-attr] 

200 token_result["sub"], # type: ignore[index] 

201 ) 

202 elif self.auth_provider is not None and hasattr( 

203 self.auth_provider, "validate_token" 

204 ): 

205 payload = self.auth_provider.validate_token(token) 

206 if hasattr(payload, "__await__"): 

207 payload = await payload 

208 if ( 

209 payload 

210 and "sub" in payload 

211 and self.auth_provider is not None 

212 ): 

213 user_store = getattr(self.auth_provider, "user_store", None) 

214 if user_store is not None: 

215 return await user_store.get_user_by_id( 

216 payload["sub"], 

217 ) 

218 return payload 

219 except ( 

220 ConnectionError, 

221 RuntimeError, 

222 ValueError, 

223 TypeError, 

224 AttributeError, 

225 ) as e: 

226 # Provide richer diagnostic logging so we can see token header issues (e.g., unexpected 'alg') 

227 try: 

228 header = None 

229 try: 

230 segment = token.split(".", 1)[0] 

231 padded = segment + "=" * (-len(segment) % 4) 

232 header = json_backend.loads( 

233 base64.urlsafe_b64decode(padded) 

234 ) 

235 except (ValueError, TypeError): 

236 header = None 

237 logger.warning( 

238 "Token validation failed: %s - header=%s", 

239 str(e), 

240 header, 

241 exc_info=True, 

242 ) 

243 except (OSError, ValueError, TypeError) as e: 

244 logger.warning("Token validation failed", exc_info=True) 

245 return None 

246 

247 def _redirect_to_login(self, request: RequestProtocol) -> ResponseProtocol: 

248 """Create redirect response to login page.""" 

249 # Build redirect URL with return path 

250 return_to = request.url.path 

251 if request.url.query: 

252 return_to = f"{return_to}?{request.url.query}" 

253 

254 login_url = f"{self.config.login_url}?next={return_to}" 

255 

256 # For HTMX requests, use HX-Redirect header 

257 if request.headers.get("HX-Request"): 

258 response = Response(status_code=200) 

259 response.headers[self.config.htmx_redirect_header] = login_url 

260 return response # type: ignore[return-value] 

261 

262 return RedirectResponse(url=login_url, status_code=302) # type: ignore[return-value] 

263 

264 

265class PermissionGuard: 

266 """GuardProtocol that checks permissions on specific routes. 

267 

268 Usage with @use_guards decorator: 

269 class UserController(Controller): 

270 @get("/admin/users") 

271 @use_guards(PermissionGuard("users.list")) 

272 async def list_users(self, request: Request) -> ...: ... 

273 

274 Usage as standalone callable: 

275 guard = PermissionGuard("users.delete") 

276 result = await guard(request) 

277 if result.is_err(): 

278 raise result.unwrap_err() 

279 """ 

280 

281 def __init__( 

282 self, 

283 *permissions: str, 

284 require_all: bool = False, 

285 message: str | None = None, 

286 authorizer: AuthorizerProtocol | None = None, 

287 ): 

288 self.permissions = permissions 

289 self.require_all = require_all 

290 self.message = message 

291 self._authorizer = authorizer 

292 

293 async def __call__( 

294 self, request: RequestProtocol 

295 ) -> Result[None, PermissionDeniedError]: 

296 """Check permissions. Returns Ok(None) on success, Err(PermissionDeniedError) on denial.""" 

297 user = getattr(request.state, "user", None) 

298 

299 if user is None: 

300 return Err(PermissionDeniedError(message="Authentication required")) 

301 

302 user_perms: PermissionSet = getattr(request.state, "permissions", None) # type: ignore[assignment] 

303 if user_perms is None: 

304 # Use injected authorizer or fallback to request.state.permissions if already set 

305 authorizer = self._authorizer 

306 if authorizer is None: 

307 # Permissions should have been set by middleware 

308 return Err( 

309 PermissionDeniedError( 

310 message="Authorization service unavailable", 

311 ) 

312 ) 

313 user_perms = get_user_permissions(user, authorizer) 

314 

315 if self.require_all: 

316 if not user_perms.has_all(*self.permissions): 

317 missing = list( 

318 filter(lambda p: not user_perms.has(p), self.permissions), 

319 ) 

320 return Err( 

321 PermissionDeniedError( 

322 message=self.message 

323 or f"Missing permissions: {', '.join(missing)}", 

324 required_permission=str(self.permissions), 

325 ) 

326 ) 

327 elif not user_perms.has_any(*self.permissions): 

328 return Err( 

329 PermissionDeniedError( 

330 message=self.message 

331 or f"Requires permission: {' or '.join(self.permissions)}", 

332 required_permission=str(self.permissions), 

333 ) 

334 ) 

335 return Ok(None) 

336 

337 def __matmul__(self, func: Callable) -> Callable: 

338 """Allow usage as @guard decorator via @ operator.""" 

339 return self.wrap(func) 

340 

341 def wrap( 

342 self, 

343 func: Callable[..., Awaitable[Any]], 

344 ) -> Callable[..., Awaitable[Any]]: 

345 """Wrap a function with permission check. 

346 

347 Raises PermissionDeniedError if the guard check fails. 

348 """ 

349 

350 @wraps(func) 

351 async def wrapper(request: Request, *args, **kwargs) -> Any: 

352 result = await self(request) # type: ignore[arg-type] 

353 if result.is_err(): 

354 raise result.unwrap_err() 

355 return await func(request, *args, **kwargs) 

356 

357 return wrapper 

358 

359 

360class RoleGuard: 

361 """GuardProtocol that checks roles on specific routes.""" 

362 

363 def __init__( 

364 self, 

365 *roles: str, 

366 require_all: bool = False, 

367 message: str | None = None, 

368 ): 

369 self.roles = roles 

370 self.require_all = require_all 

371 self.message = message 

372 

373 async def __call__( 

374 self, request: RequestProtocol 

375 ) -> Result[None, PermissionDeniedError]: 

376 """Check roles. Returns Ok(None) on success, Err(PermissionDeniedError) on denial.""" 

377 user = getattr(request.state, "user", None) 

378 

379 if user is None: 

380 return Err(PermissionDeniedError(message="Authentication required")) 

381 

382 user_roles = set(getattr(user, "roles", []) or []) 

383 

384 if self.require_all: 

385 if not all(r in user_roles for r in self.roles): 

386 missing = list(filter(lambda r: r not in user_roles, self.roles)) 

387 return Err( 

388 PermissionDeniedError( 

389 message=self.message 

390 or f"Requires all roles: {', '.join(missing)}", 

391 ) 

392 ) 

393 elif not user_roles.intersection(self.roles): 

394 return Err( 

395 PermissionDeniedError( 

396 message=self.message or f"Requires role: {' or '.join(self.roles)}", 

397 ) 

398 ) 

399 return Ok(None) 

400 

401 

402def require_auth(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]: 

403 """Simple decorator to require authentication. 

404 

405 Just checks that user exists in request.state. 

406 """ 

407 

408 @wraps(func) 

409 async def wrapper(request: Request, *args, **kwargs) -> Any: 

410 user = getattr(request.state, "user", None) 

411 if user is None: 

412 raise PermissionDeniedError(message="Authentication required") 

413 return await func(request, *args, **kwargs) 

414 

415 return wrapper 

416 

417 

418def csrf_protect(func: Callable[..., Awaitable[Any]]) -> Callable[..., Awaitable[Any]]: 

419 """Decorator to require valid CSRF token for state-changing operations. 

420 

421 Checks for CSRF token in: 

422 1. X-CSRF-Token header 

423 2. csrf_token form field 

424 

425 HTMX requests automatically include the token via hx-headers. 

426 """ 

427 

428 @wraps(func) 

429 async def wrapper(request: Request, *args, **kwargs) -> Any: 

430 # Skip for safe methods 

431 if request.method in ("GET", "HEAD", "OPTIONS"): 

432 return await func(request, *args, **kwargs) 

433 

434 # Get expected token from session 

435 session = getattr(request.state, "session", None) 

436 expected_token = getattr(session, "csrf_token", None) if session else None 

437 

438 if not expected_token: 

439 # Fail closed: a session without CSRF state cannot authorize 

440 # state-changing requests. 

441 logger.warning("csrf_rejected_no_session_token") 

442 raise PermissionDeniedError( 

443 message="Missing CSRF session state", 

444 code=ErrorCode.AUTH_INVALID_TOKEN, 

445 ) 

446 

447 # Get submitted token 

448 submitted_token = request.headers.get("X-CSRF-Token") 

449 

450 if not submitted_token: 

451 # Try form data 

452 try: 

453 form = request.scope.get("admin_form_data") 

454 if form is None: 

455 form = await request.form() 

456 submitted_token = form.get("csrf_token") # type: ignore[assignment] 

457 except ( 

458 ConnectionError, 

459 RuntimeError, 

460 ValueError, 

461 TypeError, 

462 AttributeError, 

463 ): 

464 pass 

465 

466 if not submitted_token or not hmac.compare_digest( 

467 submitted_token, expected_token 

468 ): 

469 raise PermissionDeniedError( 

470 message="Invalid or missing CSRF token", 

471 code=ErrorCode.AUTH_INVALID_TOKEN, 

472 ) 

473 

474 return await func(request, *args, **kwargs) 

475 

476 return wrapper 

477 

478 

479class CompositeGuard: 

480 """Combine multiple guards with AND/OR logic. 

481 

482 Usage: 

483 guard = CompositeGuard( 

484 PermissionGuard("users.list"), 

485 RoleGuard("admin"), 

486 logic="or" # User needs permission OR role 

487 ) 

488 """ 

489 

490 def __init__( 

491 self, 

492 *guards: PermissionGuard | RoleGuard, 

493 logic: str = "and", # "and" or "or" 

494 ): 

495 self.guards = guards 

496 self.logic = logic 

497 

498 async def __call__( 

499 self, request: RequestProtocol 

500 ) -> Result[None, PermissionDeniedError]: 

501 """Execute guards based on logic. 

502 

503 Returns Ok(None) when guard(s) pass. Returns Err(PermissionDeniedError) 

504 on denial. For "and" logic, the first failure short-circuits. For "or" 

505 logic, the last failure is returned if all guards deny. 

506 """ 

507 if self.logic == "and": 

508 # All guards must pass — short-circuit on first failure 

509 for guard in self.guards: 

510 result = await guard(request) 

511 if result.is_err(): 

512 return result 

513 return Ok(None) 

514 # At least one guard must pass 

515 last_failure: Result[None, PermissionDeniedError] = Err( 

516 PermissionDeniedError(message="All guards denied access") 

517 ) 

518 for guard in self.guards: 

519 result = await guard(request) 

520 if result.is_ok(): 

521 return Ok(None) 

522 last_failure = result 

523 return last_failure