Coverage for src / lexigram / contracts / audit / types.py: 2%

1"""Canonical audit-trail value types for security and compliance logging.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from enum import StrEnum, auto 

8from typing import Any 

9 

10__all__ = [ 

11 "AuditEntry", 

12 "AuditEventSeverity", 

13 "AuditMismatch", 

14 "AuditMismatchReason", 

15 "AuditQuery", 

16 "RetentionDecision", 

17 "RetentionPolicy", 

18] 

19 

20 

21class AuditEventSeverity(StrEnum): 

22 """Severity level for audit log entries.""" 

23 

24 LOW = auto() # Informational — routine operations 

25 MEDIUM = auto() # Notable — privilege changes, config updates 

26 HIGH = auto() # Security-sensitive — auth failures, data access 

27 CRITICAL = auto() # Compliance-critical — data deletion, impersonation 

28 

29 

30class AuditMismatchReason(StrEnum): 

31 """Why an audit entry failed checksum verification.""" 

32 

33 CHECKSUM_MISMATCH = auto() # Stored checksum does not match recomputed checksum 

34 NO_CHECKSUM_PRESENT = auto() # Entry has no stored checksum (pre-checksum row) 

35 

36 

37@dataclass(frozen=True) 

38class AuditEntry: 

39 """Immutable record of a security-relevant or compliance-relevant operation. 

40 

41 Canonical audit type used by all packages. Supersedes the former 

42 ``AuditEvent`` (simpler, no old/new values) and the ``AuditEntry`` 

43 from ``contracts/observability/audit``. 

44 

45 Attributes: 

46 action: Dot-notation action identifier (e.g. ``"user.update"``). 

47 actor_id: ID of the user or service that performed the action. 

48 resource_type: Kind of affected resource (e.g. ``"User"``). 

49 resource_id: ID of the affected resource. 

50 outcome: ``"success"`` or ``"failure"``. 

51 severity: Severity level (defaults to MEDIUM). 

52 occurred_at: UTC datetime of the event (defaults to now). 

53 metadata: Arbitrary additional context. 

54 old_values: Field values before the action (optional). 

55 new_values: Field values after the action (optional). 

56 source: Originating subsystem (e.g. ``"sql"``, ``"admin"``, ``"ai"``). 

57 tenant_id: Multi-tenant scoping (optional). 

58 correlation_id: Request correlation ID for tracing (optional). 

59 causation_id: Causing event ID for event-driven traces (optional). 

60 command_payload_hash: SHA-256 hash of the command payload (optional). 

61 payload_size_bytes: Size of the payload in bytes (optional). 

62 checksum: HMAC-SHA256 checksum of the persisted row, populated 

63 by stores on read-back (None for pre-checksum entries). 

64 """ 

65 

66 action: str 

67 actor_id: str 

68 resource_type: str = "" 

69 resource_id: str = "" 

70 outcome: str = "success" 

71 severity: AuditEventSeverity = AuditEventSeverity.MEDIUM 

72 occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

73 metadata: dict[str, Any] = field(default_factory=dict) 

74 old_values: dict[str, Any] | None = None 

75 new_values: dict[str, Any] | None = None 

76 source: str = "" 

77 tenant_id: str | None = None 

78 correlation_id: str | None = None 

79 causation_id: str | None = None 

80 command_payload_hash: bytes | None = None 

81 payload_size_bytes: int | None = None 

82 checksum: str | None = None 

83 

84 

85@dataclass(frozen=True) 

86class AuditQuery: 

87 """Composable filter criteria for querying the audit trail. 

88 

89 All fields are optional; omitted fields are not applied as filters. 

90 

91 Attributes: 

92 actor_id: Filter by actor identifier. 

93 action: Filter by action verb (exact match). 

94 resource_type: Filter by resource type. 

95 resource_id: Filter by resource identifier. 

96 source: Filter by originating subsystem. 

97 severity: Filter by severity level. 

98 outcome: Filter by outcome. 

99 tenant_id: Filter by tenant. 

100 correlation_id: Filter by request correlation ID. 

101 since: Only entries at or after this UTC datetime. 

102 until: Only entries at or before this UTC datetime. 

103 limit: Maximum number of entries (default 100). 

104 offset: Entries to skip for pagination. 

105 """ 

106 

107 actor_id: str | None = None 

108 action: str | None = None 

109 resource_type: str | None = None 

110 resource_id: str | None = None 

111 source: str | None = None 

112 severity: AuditEventSeverity | None = None 

113 outcome: str | None = None 

114 tenant_id: str | None = None 

115 correlation_id: str | None = None 

116 since: datetime | None = None 

117 until: datetime | None = None 

118 limit: int = 100 

119 offset: int = 0 

120 

121 

122@dataclass(frozen=True) 

123class AuditMismatch: 

124 """Record of a checksum verification failure. 

125 

126 Attributes: 

127 entry_id: Identifier of the audit entry with a bad checksum. 

128 expected_checksum: Checksum stored in the database. 

129 actual_checksum: Checksum recomputed from current data. 

130 reason: Whether the entry is tampered or simply has no stored checksum. 

131 """ 

132 

133 entry_id: str 

134 expected_checksum: str 

135 actual_checksum: str 

136 reason: AuditMismatchReason = AuditMismatchReason.CHECKSUM_MISMATCH 

137 

138 

139class RetentionDecision(StrEnum): 

140 """Outcome of a retention policy evaluation.""" 

141 

142 RETAIN = auto() # Keep indefinitely 

143 RETAIN_UNTIL = auto() # Keep until expiry 

144 ARCHIVE = auto() # Move to cold storage 

145 PURGE = auto() # Eligible for deletion 

146 

147 

148@dataclass(frozen=True) 

149class RetentionPolicy: 

150 """Configuration for audit retention behavior. 

151 

152 Attributes: 

153 name: Identifier for this policy. 

154 default_retention_days: Default days to retain entries (0 = indefinite). 

155 severity_overrides: Per-severity retention days (key = severity value). 

156 source_overrides: Per-source retention days (key = source string). 

157 """ 

158 

159 name: str 

160 default_retention_days: int = 365 

161 severity_overrides: dict[str, int] = field(default_factory=dict) 

162 source_overrides: dict[str, int] = field(default_factory=dict)