57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
"""Delta-Sync ledger: tamper-evident log with replay capability.
|
|
|
|
The ledger is a lightweight, chain-logged store where each entry records the
|
|
payload hash and a chain hash of the previous entry to provide auditability and
|
|
replay verification in offline mode.
|
|
"""
|
|
from __future__ import annotations
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass, asdict
|
|
from typing import List, Dict, Any
|
|
|
|
|
|
@dataclass
|
|
class LedgerEntry:
|
|
seq: int
|
|
payload: Dict[str, Any]
|
|
prev_hash: str
|
|
hash: str
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class DeltaSyncLedger:
|
|
def __init__(self) -> None:
|
|
self.entries: List[LedgerEntry] = []
|
|
self._last_hash = "" # Genesis chain
|
|
|
|
@staticmethod
|
|
def _hash(payload: Dict[str, Any], prev_hash: str) -> str:
|
|
h = hashlib.sha256()
|
|
h.update(json.dumps(payload, sort_keys=True).encode("utf-8"))
|
|
h.update(prev_hash.encode("utf-8"))
|
|
return h.hexdigest()
|
|
|
|
def append(self, payload: Dict[str, Any]) -> LedgerEntry:
|
|
seq = len(self.entries) + 1
|
|
entry_hash = self._hash(payload, self._last_hash)
|
|
entry = LedgerEntry(seq=seq, payload=payload, prev_hash=self._last_hash, hash=entry_hash)
|
|
self.entries.append(entry)
|
|
self._last_hash = entry_hash
|
|
return entry
|
|
|
|
def verify(self) -> bool:
|
|
prev = ""
|
|
for e in self.entries:
|
|
if e.prev_hash != prev:
|
|
return False
|
|
if e.hash != self._hash(e.payload, e.prev_hash):
|
|
return False
|
|
prev = e.hash
|
|
return True
|
|
|
|
def replay(self) -> List[Dict[str, Any]]:
|
|
return [e.to_dict() for e in self.entries]
|