88 lines
2.6 KiB
Python
88 lines
2.6 KiB
Python
"""Append-only governance ledger for BeVault."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
def _stable_hash(payload: Dict[str, Any]) -> str:
|
|
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
@dataclass
|
|
class LedgerEntry:
|
|
index: int
|
|
event_type: str
|
|
payload: Dict[str, Any]
|
|
timestamp: float
|
|
previous_hash: str
|
|
entry_hash: str
|
|
|
|
|
|
@dataclass
|
|
class GovernanceLedger:
|
|
"""Tamper-evident event ledger with hash chaining."""
|
|
|
|
entries: List[LedgerEntry] = field(default_factory=list)
|
|
|
|
def append(self, event_type: str, payload: Optional[Dict[str, Any]] = None, timestamp: Optional[float] = None) -> LedgerEntry:
|
|
timestamp = time.time() if timestamp is None else timestamp
|
|
payload = payload or {}
|
|
previous_hash = self.entries[-1].entry_hash if self.entries else ""
|
|
entry = LedgerEntry(
|
|
index=len(self.entries),
|
|
event_type=event_type,
|
|
payload=payload,
|
|
timestamp=timestamp,
|
|
previous_hash=previous_hash,
|
|
entry_hash="",
|
|
)
|
|
entry.entry_hash = _stable_hash(
|
|
{
|
|
"index": entry.index,
|
|
"event_type": entry.event_type,
|
|
"payload": entry.payload,
|
|
"timestamp": entry.timestamp,
|
|
"previous_hash": entry.previous_hash,
|
|
}
|
|
)
|
|
self.entries.append(entry)
|
|
return entry
|
|
|
|
def verify(self) -> bool:
|
|
previous_hash = ""
|
|
for index, entry in enumerate(self.entries):
|
|
if entry.index != index:
|
|
return False
|
|
if entry.previous_hash != previous_hash:
|
|
return False
|
|
expected_hash = _stable_hash(
|
|
{
|
|
"index": entry.index,
|
|
"event_type": entry.event_type,
|
|
"payload": entry.payload,
|
|
"timestamp": entry.timestamp,
|
|
"previous_hash": entry.previous_hash,
|
|
}
|
|
)
|
|
if entry.entry_hash != expected_hash:
|
|
return False
|
|
previous_hash = entry.entry_hash
|
|
return True
|
|
|
|
def anchor(self) -> str:
|
|
return self.entries[-1].entry_hash if self.entries else ""
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"entries": [entry.__dict__ for entry in self.entries],
|
|
"anchor": self.anchor(),
|
|
}
|
|
|
|
|
|
__all__ = ["GovernanceLedger", "LedgerEntry"]
|