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

1"""Optional read-audit middleware — logs GET requests (off by default).""" 

2 

3from __future__ import annotations 

4 

5from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint 

6from starlette.requests import Request 

7from starlette.responses import Response 

8from starlette.types import ASGIApp 

9 

10from lexigram.admin.audit.correlation import get_correlation_id 

11from lexigram.contracts.admin.audit_entry import AuditEntry 

12from lexigram.contracts.admin.audit_logger import AdminAuditLoggerProtocol 

13 

14 

15class AdminReadAuditMiddleware(BaseHTTPMiddleware): 

16 """Middleware that logs admin GET requests for compliance auditing. 

17 

18 Only active when ``read_audit_enabled`` is ``True``. When enabled, 

19 every GET request matching typical admin resource patterns 

20 (``/admin/...``) is recorded as a low-verbosity ``AuditEntry`` 

21 with ``outcome="success"`` and no before/after diff. 

22 """ 

23 

24 def __init__( 

25 self, 

26 app: ASGIApp, 

27 audit_logger: AdminAuditLoggerProtocol, 

28 read_audit_enabled: bool = False, 

29 admin_prefix: str = "/admin", 

30 ) -> None: 

31 super().__init__(app) 

32 self._audit_logger = audit_logger 

33 self._read_audit_enabled = read_audit_enabled 

34 self._admin_prefix = admin_prefix 

35 

36 async def dispatch( 

37 self, request: Request, call_next: RequestResponseEndpoint 

38 ) -> Response: 

39 response = await call_next(request) 

40 

41 if not self._read_audit_enabled: 

42 return response 

43 

44 if request.method.upper() != "GET": 

45 return response 

46 

47 if not request.url.path.startswith(self._admin_prefix): 

48 return response 

49 

50 if request.url.path in ( 

51 f"{self._admin_prefix}/login", 

52 f"{self._admin_prefix}/logout", 

53 f"{self._admin_prefix}/health", 

54 f"{self._admin_prefix}/static", 

55 ): 

56 return response 

57 

58 user_id = getattr(request.state, "user", None) 

59 user_id_str = ( 

60 str(getattr(user_id, "user_id", user_id)) if user_id else "anonymous" 

61 ) 

62 

63 entry = AuditEntry( 

64 admin_user_id=user_id_str, 

65 action="read.list" 

66 if "/" + request.url.path.split("/")[-1] == "" 

67 else "read.detail", 

68 resource_type="admin_page", 

69 resource_id=request.url.path, 

70 outcome="success", 

71 correlation_id=get_correlation_id(), 

72 request_id=request.headers.get("X-Request-ID"), 

73 request_ip=request.client.host if request.client else None, 

74 metadata={"method": "GET", "query": str(request.query_params)}, 

75 ) 

76 await self._audit_logger.write(entry) 

77 return response 

78 

79 

80__all__ = ["AdminReadAuditMiddleware"]