130 lines
3.7 KiB
Python
130 lines
3.7 KiB
Python
"""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]}
|