Coverage for src / lexigram / contracts / audit / protocols.py: 0%

1"""Framework-wide audit logging protocols for security and compliance.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime 

6from typing import Protocol, runtime_checkable 

7 

8from lexigram.contracts.audit.types import ( 

9 AuditEntry, 

10 AuditMismatch, 

11 AuditQuery, 

12 RetentionDecision, 

13) 

14 

15__all__ = [ 

16 "AuditLoggerProtocol", 

17 "AuditStoreProtocol", 

18 "AuditVerifierProtocol", 

19 "RetentionPolicyProtocol", 

20] 

21 

22 

23@runtime_checkable 

24class AuditLoggerProtocol(Protocol): 

25 """Framework-wide protocol for recording and querying the audit trail. 

26 

27 All packages that perform security-sensitive or compliance-relevant 

28 operations accept this via constructor injection and call ``log()`` 

29 at key operation boundaries. 

30 

31 Implementations must not raise from ``log()`` — audit failure must 

32 never block the operation that triggered it. 

33 """ 

34 

35 async def log(self, entry: AuditEntry) -> None: 

36 """Persist an audit entry. Never raises. 

37 

38 Args: 

39 entry: The audit event to record. 

40 """ 

41 ... 

42 

43 async def query(self, query: AuditQuery) -> list[AuditEntry]: 

44 """Retrieve audit entries matching the given filters. 

45 

46 Args: 

47 query: Filter criteria encapsulated in an AuditQuery object. 

48 

49 Returns: 

50 List of matching entries, newest-first. 

51 """ 

52 ... 

53 

54 

55@runtime_checkable 

56class AuditStoreProtocol(Protocol): 

57 """Low-level persistence protocol for audit storage backends. 

58 

59 The ``AuditLogger`` delegates to this protocol for actual persistence. 

60 This protocol is append-only — purge operations happen at a higher 

61 level via ``AuditPurger``. 

62 """ 

63 

64 async def append(self, entry: AuditEntry) -> None: 

65 """Persist a single audit entry. 

66 

67 Args: 

68 entry: The audit event to store. 

69 """ 

70 ... 

71 

72 async def query(self, query: AuditQuery) -> list[AuditEntry]: 

73 """Retrieve entries matching filters, newest-first. 

74 

75 Args: 

76 query: Filter criteria. 

77 

78 Returns: 

79 List of matching entries. 

80 """ 

81 ... 

82 

83 async def count(self, query: AuditQuery) -> int: 

84 """Return count of entries matching filters. 

85 

86 Args: 

87 query: Filter criteria. 

88 

89 Returns: 

90 Number of matching entries. 

91 """ 

92 ... 

93 

94 async def delete_expired(self, cutoff: datetime) -> int: 

95 """Delete entries whose stored expiry precedes or equals the cutoff. 

96 

97 Entries are identified by the ``__expires_at`` metadata stamp 

98 written by ``AuditLogger.log()`` when a retention policy is 

99 configured. Entries without the stamp are never deleted. 

100 

101 Args: 

102 cutoff: UTC datetime; entries expiring at or before this 

103 instant are deleted. 

104 

105 Returns: 

106 Number of entries deleted. 

107 """ 

108 ... 

109 

110 

111@runtime_checkable 

112class AuditVerifierProtocol(Protocol): 

113 """Protocol for audit trail tamper detection via HMAC checksums.""" 

114 

115 async def verify_recent(self, *, limit: int = 100) -> list[AuditMismatch]: 

116 """Verify checksums for the most recent entries. 

117 

118 Args: 

119 limit: Number of recent entries to verify. 

120 

121 Returns: 

122 List of mismatches (empty list = all verified). 

123 """ 

124 ... 

125 

126 async def verify_entry(self, entry: AuditEntry) -> AuditMismatch | None: 

127 """Verify checksum for a single entry. 

128 

129 Args: 

130 entry: The audit entry to verify. 

131 

132 Returns: 

133 None when the entry verifies clean; an AuditMismatch whose 

134 reason is ``checksum_mismatch`` when tampered or 

135 ``no_checksum_present`` when the entry carries no stored 

136 checksum (a pre-checksum row) and cannot be verified. 

137 """ 

138 ... 

139 

140 

141@runtime_checkable 

142class RetentionPolicyProtocol(Protocol): 

143 """Protocol for audit retention policy evaluation.""" 

144 

145 async def evaluate(self, entry: AuditEntry) -> RetentionDecision: 

146 """Determine the retention decision for an entry. 

147 

148 Args: 

149 entry: The audit entry to evaluate. 

150 

151 Returns: 

152 A RetentionDecision value. 

153 """ 

154 ... 

155 

156 async def get_expiry(self, entry: AuditEntry) -> datetime | None: 

157 """Return the expiry datetime for an entry, or None for indefinite retention. 

158 

159 Args: 

160 entry: The audit entry to evaluate. 

161 

162 Returns: 

163 UTC datetime when the entry expires, or None. 

164 """ 

165 ...