52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
import hashlib
|
|
import json
|
|
from typing import List
|
|
|
|
def _hash_bytes(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
def merkle_root(digests: List[str]) -> str:
|
|
if not digests:
|
|
return ""
|
|
level = [bytes.fromhex(d) for d in digests]
|
|
while len(level) > 1:
|
|
next_level = []
|
|
for i in range(0, len(level), 2):
|
|
left = level[i]
|
|
right = level[i+1] if i+1 < len(level) else level[i]
|
|
next_level.append(hashlib.sha256(left + right).digest())
|
|
level = next_level
|
|
return level[0].hex()
|
|
|
|
class DeltaLog:
|
|
def __init__(self):
|
|
self.entries = [] # each entry is a dict with digest and payload
|
|
self.anchor = None # optional cloud/ground anchor for global verifiability
|
|
|
|
def add_entry(self, entry: dict) -> str:
|
|
# entry must be serializable, and we store a digest for Merkle
|
|
payload_bytes = json.dumps(entry, sort_keys=True).encode('utf-8')
|
|
digest = hashlib.sha256(payload_bytes).hexdigest()
|
|
self.entries.append({
|
|
"digest": digest,
|
|
"payload": entry,
|
|
})
|
|
return digest
|
|
|
|
def anchor_root(self, anchor: str) -> None:
|
|
"""Record an optional anchor for the current delta log.
|
|
|
|
This does not alter existing entries; it simply stores a reference
|
|
to a trusted anchor (e.g., ground control) to tie the local log to
|
|
an external verifiable state.
|
|
"""
|
|
self.anchor = anchor
|
|
|
|
def delta_from_index(self, index: int) -> List[dict]:
|
|
# return full payloads for simplicity (MVP). In a real system you'd return compact digests with proofs.
|
|
return [e["payload"] for e in self.entries[index:]]
|
|
|
|
def root(self) -> str:
|
|
digests = [e["digest"] for e in self.entries]
|
|
return merkle_root(digests)
|