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

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 "/admin/events", 

30) 

31 

32 

33@runtime_checkable 

34class RequestAuthorizerProtocol(Protocol): 

35 """Protocol for request-level authorization. 

36 

37 Concrete implementations (e.g. PiccolinaAdminAuthPolicy) also satisfy 

38 ``AdminAuthorizerProtocol`` from ``lexigram-contracts`` by implementing 

39 the CRUD methods alongside this one. 

40 """ 

41 

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

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

44 ... 

45 

46 

47class AdminAuthorizationMiddleware(BaseHTTPMiddleware): 

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

49 

50 def __init__( 

51 self, 

52 app: ASGIApp, 

53 authorizer: RequestAuthorizerProtocol, 

54 metrics: AdminMetrics | None = None, 

55 ) -> None: 

56 super().__init__(app) 

57 self._authorizer = authorizer 

58 self._metrics = metrics or AdminMetrics(None) 

59 

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

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

62 path = request.url.path 

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

64 return await call_next(request) 

65 

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

67 if user is None: 

68 logger.info( 

69 "admin_authz.unauthenticated", 

70 path=path, 

71 ) 

72 return self._unauthenticated(request) 

73 

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

75 logger.info( 

76 "admin_authz.denied", 

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

78 path=path, 

79 ) 

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

81 self._metrics.record_authz_denied(resource=resource) 

82 return self._forbidden(request) 

83 

84 return await call_next(request) 

85 

86 @staticmethod 

87 def _unauthenticated( 

88 request: Request, 

89 ) -> JSONResponse | RedirectResponse | Response: 

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

91 

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

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

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

95 """ 

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

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

98 response = Response(status_code=200) 

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

100 return response 

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

102 

103 @staticmethod 

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

105 """Return 403 with user context.""" 

106 return JSONResponse( 

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

108 status_code=403, 

109 ) 

110 

111 

112__all__ = [ 

113 "AdminAuthorizationMiddleware", 

114 "RequestAuthorizerProtocol", 

115]