1"""

2 CMVK Audit Trail Module

3

4 Provides immutable audit logging for verification operations.

5 All verifications can be logged with timestamps, inputs, and results

6 for compliance and forensic analysis.

7"""

8

9 from __future__ import annotations

10

11 import hashlib

12 import json

13 import threading

14 import uuid

15 from dataclasses import asdict, dataclass, field

16 from datetime import UTC, datetime

17 from pathlib import Path

18 from typing import Any

19

20 import numpy as np

21

22

23@dataclass(frozen=True)

24 class AuditEntry:

25 """

26 Immutable audit record for a single verification.

27

28 All fields are frozen to ensure auditability.

29

30 Attributes:

31 id: Unique identifier for this verification

32 timestamp: UTC timestamp of verification

33 operation: Type of verification operation

34 inputs_hash: SHA-256 hash of input data (for privacy)

35 result_summary: Summary of verification result

36 drift_score: The drift score from verification

37 confidence: The confidence score

38 metric_used: Distance metric used

39 profile_used: Threshold profile applied (if any)

40 passed: Whether verification passed thresholds

41 metadata: Additional context (user, session, etc.)

42 checksum: Integrity checksum for this entry

43 """

44

45 id: str

46 timestamp: str

47 operation: str

48 inputs_hash: str

49 result_summary: dict

50 drift_score: float

51 confidence: float

52 metric_used: str

53 profile_used: str | None

54 passed: bool

55 metadata: dict

56 checksum: str

57

58 def to_dict(self) -> dict:

59 """Convert to dictionary for serialization."""

60 return asdict(self)

61

62 def verify_integrity(self) -> bool:

63 """Verify the entry's checksum is valid."""

64 computed = _compute_checksum(

65 self.id,

66 self.timestamp,

67 self.operation,

68 self.inputs_hash,

69 self.drift_score,

70 self.confidence,

71 )

72 return computed == self.checksum

73

74

75@dataclass

76 class AuditTrail:

77 """

78 Audit trail manager for verification operations.

79

80 Maintains an immutable log of all verifications with optional

81 persistence to file.

82

83 Thread-safe for concurrent verifications.

84 """

85

86 entries: list[AuditEntry] = field(default_factory=list)

87 persist_path: Path | None = None

88 auto_persist: bool = False

89 _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

90

91 def __post_init__(self) -> None:

92 """Load existing entries if persist path exists."""

93 if self.persist_path and self.persist_path.exists():

94 self._load_from_file()

95

96 def log(

97 self,

98 operation: str,

99 inputs: dict[str, Any],

100 drift_score: float,

101 confidence: float,

102 metric_used: str = "cosine",

103 profile_used: str | None = None,

104 passed: bool = True,

105 result_details: dict | None = None,

106 metadata: dict | None = None,

107 ) -> AuditEntry:

108 """

109 Log a verification operation.

110

111 Args:

112 operation: Type of operation (e.g., "verify_embeddings")

113 inputs: Input data (will be hashed, not stored directly)

114 drift_score: Drift score from verification

115 confidence: Confidence score

116 metric_used: Distance metric used

117 profile_used: Threshold profile applied

118 passed: Whether verification passed

119 result_details: Additional result information

120 metadata: Additional context (user, session, etc.)

121

122 Returns:

123 The created AuditEntry

124 """

125 entry_id = str(uuid.uuid4())

126 timestamp = datetime.now(UTC).isoformat()

127 inputs_hash = _hash_inputs(inputs)

128

129 checksum = _compute_checksum(

130 entry_id,

131 timestamp,

132 operation,

133 inputs_hash,

134 drift_score,

135 confidence,

136 )

137

138 entry = AuditEntry(

139 id=entry_id,

140 timestamp=timestamp,

141 operation=operation,

142 inputs_hash=inputs_hash,

143 result_summary=result_details or {},

144 drift_score=drift_score,

145 confidence=confidence,

146 metric_used=metric_used,

147 profile_used=profile_used,

148 passed=passed,

149 metadata=metadata or {},

150 checksum=checksum,

151 )

152

153 with self._lock:

154 self.entries.append(entry)

155 if self.auto_persist and self.persist_path:

156 self._persist_entry(entry)

157

158 return entry

159

160 def get_entries(

161 self,

162 start_time: datetime | None = None,

163 end_time: datetime | None = None,

164 operation: str | None = None,

165 passed_only: bool | None = None,

166 ) -> list[AuditEntry]:

167 """

168 Query audit entries with optional filters.

169

170 Args:

171 start_time: Filter entries after this time

172 end_time: Filter entries before this time

173 operation: Filter by operation type

174 passed_only: If True, only passed; if False, only failed; if None, all

175

176 Returns:

177 List of matching AuditEntry records

178 """

179 with self._lock:

180 results = list(self.entries)

181

182 if start_time:

183 start_iso = start_time.isoformat()

184 results = [e for e in results if e.timestamp >= start_iso]

185

186 if end_time:

187 end_iso = end_time.isoformat()

188 results = [e for e in results if e.timestamp <= end_iso]

189

190 if operation:

191 results = [e for e in results if e.operation == operation]

192

193 if passed_only is not None:

194 results = [e for e in results if e.passed == passed_only]

195

196 return results

197

198 def get_statistics(self) -> dict:

199 """

200 Calculate summary statistics for the audit trail.

201

202 Returns:

203 Dictionary with counts, pass rates, and drift statistics

204 """

205 with self._lock:

206 entries = list(self.entries)

207

208 if not entries:

209 return {"total_entries": 0}

210

211 drift_scores = [e.drift_score for e in entries]

212 passed_count = sum(1 for e in entries if e.passed)

213

214 # Group by operation

215 by_operation: dict[str, list[AuditEntry]] = {}

216 for e in entries:

217 by_operation.setdefault(e.operation, []).append(e)

218

219 operation_stats = {

220 op: {

221 "count": len(ops),

222 "pass_rate": sum(1 for e in ops if e.passed) / len(ops),

223 "mean_drift": float(np.mean([e.drift_score for e in ops])),

224 }

225 for op, ops in by_operation.items()

226 }

227

228 return {

229 "total_entries": len(entries),

230 "passed_count": passed_count,

231 "failed_count": len(entries) - passed_count,

232 "pass_rate": passed_count / len(entries),

233 "mean_drift": float(np.mean(drift_scores)),

234 "std_drift": float(np.std(drift_scores)),

235 "min_drift": float(np.min(drift_scores)),

236 "max_drift": float(np.max(drift_scores)),

237 "by_operation": operation_stats,

238 }

239

240 def verify_integrity(self) -> tuple[bool, list[str]]:

241 """

242 Verify integrity of all audit entries.

243

244 Returns:

245 Tuple of (all_valid, list of invalid entry IDs)

246 """

247 invalid_ids = []

248 with self._lock:

249 for entry in self.entries:

250 if not entry.verify_integrity():

251 invalid_ids.append(entry.id)

252

253 return len(invalid_ids) == 0, invalid_ids

254

255 def export_json(self, path: Path | str) -> None:

256 """Export audit trail to JSON file."""

257 path = Path(path)

258 with self._lock:

259 data = {

260 "exported_at": datetime.now(UTC).isoformat(),

261 "entry_count": len(self.entries),

262 "entries": [e.to_dict() for e in self.entries],

263 }

264

265 path.write_text(json.dumps(data, indent=2))

266

267 def export_csv(self, path: Path | str) -> None:

268 """Export audit trail to CSV file."""

269 import csv

270

271 path = Path(path)

272 with self._lock:

273 entries = list(self.entries)

274

275 if not entries:

276 path.write_text("")

277 return

278

279 fieldnames = [

280 "id",

281 "timestamp",

282 "operation",

283 "drift_score",

284 "confidence",

285 "metric_used",

286 "profile_used",

287 "passed",

288 "inputs_hash",

289 ]

290

291 with path.open("w", newline="") as f:

292 writer = csv.DictWriter(f, fieldnames=fieldnames)

293 writer.writeheader()

294 for entry in entries:

295 writer.writerow({k: getattr(entry, k) for k in fieldnames})

296

297 def _load_from_file(self) -> None:

298 """Load entries from persist file."""

299 if not self.persist_path or not self.persist_path.exists():

300 return

301

302 try:

303 data = json.loads(self.persist_path.read_text())

304 for entry_dict in data.get("entries", []):

305 self.entries.append(AuditEntry(**entry_dict))

306 except (json.JSONDecodeError, TypeError) as e:

307 # Log warning but don't fail

308 print(f"Warning: Could not load audit trail from {self.persist_path}: {e}")

309

310 def _persist_entry(self, entry: AuditEntry) -> None:

311 """Append entry to persist file."""

312 if not self.persist_path:

313 return

314

315 # Read existing or create new

316 if self.persist_path.exists():

317 try:

318 data = json.loads(self.persist_path.read_text())

319 except json.JSONDecodeError:

320 data = {"entries": []}

321 else:

322 data = {"entries": []}

323

324 data["entries"].append(entry.to_dict())

325 data["updated_at"] = datetime.now(UTC).isoformat()

326 data["entry_count"] = len(data["entries"])

327

328 self.persist_path.write_text(json.dumps(data, indent=2))

329

330 def clear(self) -> None:

331 """Clear all entries (use with caution)."""

332 with self._lock:

333 self.entries.clear()

334

335

336# ============================================================================

337# Helper Functions

338# ============================================================================

339

340

341 def _hash_inputs(inputs: dict[str, Any]) -> str:

342 """Create SHA-256 hash of inputs for privacy-preserving audit."""

343

344 def serialize(obj: Any) -> Any:

345 """Convert objects to JSON-serializable format."""

346 if isinstance(obj, np.ndarray) or hasattr(obj, "tolist"):

347 return obj.tolist()

348 elif hasattr(obj, "__dict__"):

349 return str(type(obj).__name__)

350 return obj

351

352 # Create deterministic JSON representation

353 serialized = json.dumps(

354 {k: serialize(v) for k, v in sorted(inputs.items())},

355 sort_keys=True,

356 )

357 return hashlib.sha256(serialized.encode()).hexdigest()

358

359

360 def _compute_checksum(*values: Any) -> str:

361 """Compute checksum for audit entry integrity."""

362 content = "|".join(str(v) for v in values)

363 return hashlib.sha256(content.encode()).hexdigest()[:16]

364

365

366# ============================================================================

367# Global Audit Trail Instance

368# ============================================================================

369

370_global_audit_trail: AuditTrail | None = None

371

372

373 def get_audit_trail() -> AuditTrail:

374 """Get or create the global audit trail instance."""

375 global _global_audit_trail

376 if _global_audit_trail is None:

377 _global_audit_trail = AuditTrail()

378 return _global_audit_trail

379

380

381 def configure_audit_trail(

382 persist_path: Path | str | None = None,

383 auto_persist: bool = False,

384) -> AuditTrail:

385 """

386 Configure the global audit trail.

387

388 Args:

389 persist_path: Path to persist audit entries

390 auto_persist: Whether to automatically persist each entry

391

392 Returns:

393 The configured AuditTrail instance

394 """

395 global _global_audit_trail

396 _global_audit_trail = AuditTrail(

397 persist_path=Path(persist_path) if persist_path else None,

398 auto_persist=auto_persist,

399 )

400 return _global_audit_trail