132 lines
4.9 KiB
Python
132 lines
4.9 KiB
Python
"""PlanDelta Compact Provenance (PCP) - tiny Merkle-DAG provenance.
|
|
|
|
This module provides a minimal, op-based Merkle-DAG encoding of PlanDelta
|
|
evolution. Each PCP entry is a small JSON-serializable dict containing:
|
|
|
|
- entry_id: short random id
|
|
- op: a small description of the delta-op (e.g., "add-commitment")
|
|
- author: who created this entry
|
|
- ts: timestamp
|
|
- prev_hash: hex of previous entry (empty for root)
|
|
- hash: sha256 hex of the canonical entry payload
|
|
|
|
The "chain" is simply a list of entries where each entry.prev_hash points to
|
|
the hash of the previous entry. Inclusion proofs are represented as the list of
|
|
hashes from a target entry up to the head. This is intentionally small and
|
|
deterministic for DTN-friendly admission controllers and unit tests.
|
|
"""
|
|
from typing import Dict, Any, List, Optional, Tuple
|
|
import time
|
|
import uuid
|
|
import json
|
|
import hashlib
|
|
|
|
|
|
def _canonical(entry: Dict[str, Any]) -> str:
|
|
# deterministic JSON over core fields (excluding computed 'hash')
|
|
payload = {
|
|
"entry_id": entry["entry_id"],
|
|
"op": entry["op"],
|
|
"author": entry.get("author", ""),
|
|
"ts": int(entry.get("ts", 0)),
|
|
"prev_hash": entry.get("prev_hash", ""),
|
|
}
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def _entry_hash(entry: Dict[str, Any]) -> str:
|
|
return hashlib.sha256(_canonical(entry).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def create_entry(op: str, author: str = "", prev_hash: str = "", ts: Optional[float] = None) -> Dict[str, Any]:
|
|
"""Create a PCP entry and compute its hash."""
|
|
eid = uuid.uuid4().hex[:12]
|
|
ts_val = time.time() if ts is None else float(ts)
|
|
entry = {
|
|
"entry_id": eid,
|
|
"op": op,
|
|
"author": author,
|
|
"ts": int(ts_val),
|
|
"prev_hash": prev_hash,
|
|
}
|
|
entry["hash"] = _entry_hash(entry)
|
|
return entry
|
|
|
|
|
|
def build_chain(ops: List[Tuple[str, str]]) -> List[Dict[str, Any]]:
|
|
"""Build a chain from a list of (op, author) tuples. Returns list of entries.
|
|
|
|
The first element will be the root (prev_hash="").
|
|
"""
|
|
chain: List[Dict[str, Any]] = []
|
|
prev = ""
|
|
for op, author in ops:
|
|
e = create_entry(op=op, author=author, prev_hash=prev)
|
|
chain.append(e)
|
|
prev = e["hash"]
|
|
return chain
|
|
|
|
|
|
def head_hash(chain: List[Dict[str, Any]]) -> str:
|
|
return chain[-1]["hash"] if chain else ""
|
|
|
|
|
|
def inclusion_proof(chain: List[Dict[str, Any]], target_entry_id: str) -> List[str]:
|
|
"""Return list of hashes from target entry up to the head (inclusive).
|
|
|
|
Raises ValueError if target not found.
|
|
"""
|
|
idx = next((i for i, e in enumerate(chain) if e["entry_id"] == target_entry_id), None)
|
|
if idx is None:
|
|
raise ValueError("target entry not in chain")
|
|
return [e["hash"] for e in chain[idx:]]
|
|
|
|
|
|
def verify_inclusion(chain_head_hash: str, proof_hashes: List[str], chain_map: Optional[Dict[str, Dict[str, Any]]] = None) -> bool:
|
|
"""Verify that the proof leads to the provided head hash.
|
|
|
|
The proof is expected to be the list of consecutive entry.hash values from
|
|
the target up to the head. We verify that each entry's prev_hash matches
|
|
the previous element's hash by optionally consulting chain_map (a mapping
|
|
of hash->entry). If chain_map is not provided, we only check that the last
|
|
hash equals chain_head_hash.
|
|
"""
|
|
if not proof_hashes:
|
|
return False
|
|
if proof_hashes[-1] != chain_head_hash:
|
|
return False
|
|
if chain_map is None:
|
|
# best-effort: we have only hashes; accept if tip matches
|
|
return True
|
|
# verify forward links using chain_map
|
|
# proof_hashes = [h_target, h_next, ..., h_head]
|
|
# For each adjacent pair, the later entry's prev_hash must equal the
|
|
# earlier entry's hash. We consult chain_map which maps hash->entry.
|
|
for i in range(len(proof_hashes) - 1):
|
|
h_curr = proof_hashes[i]
|
|
h_next = proof_hashes[i + 1]
|
|
# ensure the provided map contains the entry and that the entry's
|
|
# recorded hash matches the canonical hash of its payload. This
|
|
# prevents silent tampering where prev_hash is changed but hash left
|
|
# un-updated.
|
|
next_entry = chain_map.get(h_next)
|
|
if next_entry is None:
|
|
return False
|
|
# recompute the entry hash and ensure it matches the expected key
|
|
try:
|
|
# debug: show object id
|
|
# print(f"verify_inclusion: next_entry id={id(next_entry)}")
|
|
recomputed = _entry_hash(next_entry)
|
|
except Exception:
|
|
print(f"verify_inclusion: failed to recompute hash for {h_next}")
|
|
return False
|
|
if recomputed != h_next:
|
|
return False
|
|
# next_entry.prev_hash should point to current hash
|
|
prev = next_entry.get("prev_hash", "")
|
|
# debug print
|
|
# print(f"verify link: next={h_next} prev={prev} expected={h_curr}")
|
|
if prev != h_curr:
|
|
return False
|
|
return True
|