Coverage for src/lumen/audit/events.py: 100%

1""" 

2This module builds FHIR AuditEvent resource records for logging REST operations. 

3It captures details about the action, actor identity, request source IP, and target resource. 

4""" 

5 

6from datetime import UTC, datetime 

7from typing import Any 

8 

9from ..core.context import current_claims, current_client_ip 

10 

11 

12def build_audit_event( 

13 action: str, 

14 resource_type: str, 

15 resource_id: str, 

16 outcome: str = "0", # 0 = Success, 4 = Minor failure, 8 = Serious failure, 12 = Major failure 

17) -> dict[str, Any]: 

18 """ 

19 Build a standard FHIR AuditEvent resource dictionary. 

20 

21 Args: 

22 action: "C" (Create), "R" (Read), "U" (Update), "D" (Delete), "E" (Execute) 

23 resource_type: The type of the target resource (e.g., "Patient") 

24 resource_id: The ID of the target resource. 

25 outcome: The outcome of the event. 

26 """ 

27 claims = current_claims.get() or {} 

28 client_ip = current_client_ip.get() or "unknown" 

29 

30 # Determine user identity from SMART-on-FHIR claims 

31 # Prefer 'sub' (subject), fallback to 'client_id', then 'anonymous' 

32 user_id = claims.get("sub") or claims.get("client_id") or "anonymous" 

33 

34 audit_event: dict[str, Any] = { 

35 "resourceType": "AuditEvent", 

36 "type": { 

37 "system": "http://terminology.hl7.org/CodeSystem/audit-event-type", 

38 "code": "rest", 

39 "display": "RESTful Operation", 

40 }, 

41 "action": action, 

42 "recorded": datetime.now(UTC).isoformat(), 

43 "outcome": outcome, 

44 "agent": [ 

45 { 

46 "requestor": True, 

47 "who": {"identifier": {"value": user_id}}, 

48 "network": { 

49 "address": client_ip, 

50 "type": "2", # IP address 

51 }, 

52 } 

53 ], 

54 "source": {"observer": {"identifier": {"value": "lumen-fhir-server"}}}, 

55 "entity": [ 

56 { 

57 "what": {"reference": f"{resource_type}/{resource_id}"}, 

58 "type": { 

59 "system": "http://terminology.hl7.org/CodeSystem/audit-entity-type", 

60 "code": "2", 

61 "display": "System Object", 

62 }, 

63 } 

64 ], 

65 } 

66 

67 return audit_event