78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime
|
|
from typing import List
|
|
from typing import Tuple
|
|
import json
|
|
|
|
from .dsl import PoeticCert, ReplayBundle
|
|
|
|
|
|
class GovernanceLedger:
|
|
"""A simple tamper-evident governance ledger with per-entry hashes.
|
|
|
|
Logs are append-only and anchored via a rolling hash to the previous entry,
|
|
producing a chain of custody suitable for auditing even on intermittent
|
|
connectivity.
|
|
"""
|
|
|
|
def __init__(self, device_id: str = "unknown"):
|
|
self.device_id = device_id
|
|
self._entries: List[dict] = []
|
|
self._head_hash = ""
|
|
|
|
def log(self, entry: dict) -> None:
|
|
timestamp = datetime.utcnow().isoformat() + "Z"
|
|
payload = {
|
|
"device_id": self.device_id,
|
|
"timestamp": timestamp,
|
|
"entry": entry,
|
|
"parent": self._head_hash,
|
|
}
|
|
self._head_hash = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
|
|
self._entries.append(payload)
|
|
|
|
def emit_poetic_cert(self, invariants: dict, summary: str, issuer: str) -> PoeticCert:
|
|
"""Emit a compact PoeticCert for quick triage and machine checks.
|
|
|
|
This is intentionally minimal and human-readable while remaining
|
|
machine-checkable.
|
|
"""
|
|
timestamp = datetime.utcnow().isoformat() + "Z"
|
|
cert = PoeticCert(invariants=invariants, summary=summary, issued_by=issuer, timestamp=timestamp)
|
|
# Also log the issuance to the ledger for auditability
|
|
self.log({"type": "poetic_cert_issued", "cert_summary": summary, "issued_by": issuer})
|
|
return cert
|
|
|
|
def create_replay_bundle(self, entries: List[dict], signer: str, nonce: str = "") -> ReplayBundle:
|
|
"""Construct a ReplayBundle with a compact merkle-style root.
|
|
|
|
For the MVP we compute a deterministic root by hashing each entry and
|
|
then hashing the concatenation of sorted entry hashes. This produces a
|
|
compact fingerprint suitable for off-device anchoring.
|
|
"""
|
|
# Normalize and hash entries deterministically
|
|
entry_hashes = []
|
|
for e in entries:
|
|
h = hashlib.sha256(json.dumps(e, sort_keys=True).encode()).hexdigest()
|
|
entry_hashes.append(h)
|
|
|
|
# deterministic ordering
|
|
entry_hashes.sort()
|
|
concat = "".join(entry_hashes).encode()
|
|
merkle_root = hashlib.sha256(concat).hexdigest() if concat else ""
|
|
|
|
bundle = ReplayBundle(entries=list(entries), signer=signer, merkle_root=merkle_root, nonce=nonce)
|
|
# log bundle creation
|
|
self.log({"type": "replay_bundle_created", "signer": signer, "merkle_root": merkle_root})
|
|
return bundle
|
|
|
|
@property
|
|
def entries(self) -> List[dict]:
|
|
return self._entries[:]
|
|
|
|
def latest_hash(self) -> str:
|
|
return self._head_hash
|