"""Crypto-signed, tamper-evident audit log for governance and compliance. Each entry is signed with Ed25519 and chained via previous-hash to form a tamper-evident log suitable for regulatory reviews. """ from __future__ import annotations import hashlib import json import time from dataclasses import dataclass from typing import Optional from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) from cryptography.hazmat.primitives import serialization @dataclass class AuditEntry: """A single signed audit log entry.""" sequence: int timestamp_ns: int event_id: str action: str details: dict prev_hash: str entry_hash: str signature: bytes def to_dict(self) -> dict: return { "sequence": self.sequence, "timestamp_ns": self.timestamp_ns, "event_id": self.event_id, "action": self.action, "details": self.details, "prev_hash": self.prev_hash, "entry_hash": self.entry_hash, "signature": self.signature.hex(), } class AuditLog: """Append-only, crypto-signed audit log with hash chaining.""" def __init__(self, private_key: Ed25519PrivateKey) -> None: self._private_key = private_key self._public_key = private_key.public_key() self._entries: list[AuditEntry] = [] self._prev_hash = "0" * 64 # Genesis hash def append(self, event_id: str, action: str, details: Optional[dict] = None) -> AuditEntry: details = details or {} sequence = len(self._entries) timestamp_ns = time.time_ns() # Compute hash of this entry's content content = json.dumps({ "sequence": sequence, "timestamp_ns": timestamp_ns, "event_id": event_id, "action": action, "details": details, "prev_hash": self._prev_hash, }, sort_keys=True) entry_hash = hashlib.sha256(content.encode()).hexdigest() signature = self._private_key.sign(entry_hash.encode()) entry = AuditEntry( sequence=sequence, timestamp_ns=timestamp_ns, event_id=event_id, action=action, details=details, prev_hash=self._prev_hash, entry_hash=entry_hash, signature=signature, ) self._entries.append(entry) self._prev_hash = entry_hash return entry def verify(self, public_key: Optional[Ed25519PublicKey] = None) -> bool: """Verify the entire chain: hash linkage + signatures.""" pk = public_key or self._public_key prev_hash = "0" * 64 for entry in self._entries: # Verify chain linkage if entry.prev_hash != prev_hash: return False # Recompute hash content = json.dumps({ "sequence": entry.sequence, "timestamp_ns": entry.timestamp_ns, "event_id": entry.event_id, "action": entry.action, "details": entry.details, "prev_hash": entry.prev_hash, }, sort_keys=True) expected_hash = hashlib.sha256(content.encode()).hexdigest() if entry.entry_hash != expected_hash: return False # Verify signature try: pk.verify(entry.signature, entry.entry_hash.encode()) except Exception: return False prev_hash = entry.entry_hash return True @property def entries(self) -> list[AuditEntry]: return list(self._entries) def to_dict(self) -> dict: return {"entries": [e.to_dict() for e in self._entries]}