Coverage for src/lexigram/admin/middleware/authorization.py: 0%

1"""Request-entry RBAC middleware (AUTH-09, AUTH-18). 

2 

3Checks every non-public request against an authorizer before dispatching 

4to the handler. Returns 401 for anonymous users, 403 for authorization 

5denials, and is HTMX-aware. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

11 

12from starlette.middleware.base import BaseHTTPMiddleware 

13from starlette.requests import Request 

14from starlette.responses import JSONResponse, RedirectResponse, Response 

15 

16from lexigram.admin.observability.admin_metrics import AdminMetrics 

17from lexigram.logging import get_logger 

18 

19if TYPE_CHECKING: 

20 from starlette.types import ASGIApp 

21 

22logger = get_logger(__name__) 

23 

24_PUBLIC_PATHS: tuple[str, ...] = ( 

25 "/admin/login", 

26 "/admin/setup", 

27 "/admin/static", 

28 "/admin/health", 

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

30 "/admin/login/2fa", 

31 "/admin/verify-email", 

32 "/admin/password-reset", 

33 "/admin/register", 

34) 

35 

36 

37class DefaultRequestAuthorizer: 

38 """Default request-entry authorizer — authenticated users pass (fail-closed on identity).""" 

39 

40 async def authorize_request(self, user: object, request: Request) -> bool: 

41 del request # unused 

42 return getattr(user, "user_id", None) is not None 

43 

44 

45@runtime_checkable 

46class RequestAuthorizerProtocol(Protocol): 

47 """Protocol for request-level authorization. 

48 

49 Concrete implementations (e.g. PiccolinaAdminAuthPolicy) implement 

50 ``authorize_request`` alongside the union ``AuthorizerProtocol`` methods. 

51 """ 

52 

53 async def authorize_request(self, user: object, request: Request) -> bool: 

54 """Return True if the user is authorized to access the request.""" 

55 ... 

56 

57 

58class AdminAuthorizationMiddleware(BaseHTTPMiddleware): 

59 """Middleware that enforces request-entry authorization.""" 

60 

61 def __init__( 

62 self, 

63 app: ASGIApp, 

64 authorizer: RequestAuthorizerProtocol, 

65 metrics: AdminMetrics | None = None, 

66 ) -> None: 

67 super().__init__(app) 

68 self._authorizer = authorizer 

69 self._metrics = metrics or AdminMetrics(None) 

70 

71 async def dispatch(self, request: Request, call_next: Any) -> Any: 

72 """Check authorization before dispatching to the next handler.""" 

73 path = request.url.path 

74 if any(path.startswith(p) for p in _PUBLIC_PATHS): 

75 return await call_next(request) 

76 

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

78 if user is None: 

79 logger.info( 

80 "admin_authz.unauthenticated", 

81 path=path, 

82 ) 

83 return self._unauthenticated(request) 

84 

85 if not await self._authorizer.authorize_request(user, request): 

86 logger.info( 

87 "admin_authz.denied", 

88 user_id=getattr(user, "user_id", "unknown"), 

89 path=path, 

90 ) 

91 resource = path.split("/")[2] if len(path.split("/")) > 2 else "unknown" 

92 self._metrics.record_authz_denied(resource=resource) 

93 return self._forbidden(request) 

94 

95 return await call_next(request) 

96 

97 @staticmethod 

98 def _unauthenticated( 

99 request: Request, 

100 ) -> JSONResponse | RedirectResponse | Response: 

101 """Redirect to login, with HX-Redirect for HTMX requests. 

102 

103 HTMX swaps responses into the current page, so a plain redirect 

104 would render the login page inside the target component. The 

105 HX-Redirect header forces a full browser navigation instead. 

106 """ 

107 login_url = f"/admin/login?next={request.url.path}" 

108 if request.headers.get("HX-Request") == "true": 

109 response = Response(status_code=200) 

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

111 return response 

112 return RedirectResponse(url=login_url, status_code=302) 

113 

114 @staticmethod 

115 def _forbidden(request: Request) -> JSONResponse: 

116 """Return 403 with user context.""" 

117 return JSONResponse( 

118 {"error": "forbidden", "path": request.url.path}, 

119 status_code=403, 

120 ) 

121 

122 

123__all__ = [ 

124 "AdminAuthorizationMiddleware", 

125 "DefaultRequestAuthorizer", 

126 "RequestAuthorizerProtocol", 

127]