"""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 backward links using chain_map # proof_hashes = [h_target, h_next, ..., h_head] for i in range(len(proof_hashes) - 1): h = proof_hashes[i] next_h = proof_hashes[i + 1] entry = chain_map.get(h) if entry is None: return False if entry.get("prev_hash", "") != (chain_map.get(next_h, {}).get("prev_hash", entry.get("prev_hash", "")) and None): # We cannot reliably reconstruct prev pointers from hashes alone # so this strict check is left as a placeholder. For now we require # that next hash exists in the map (sanity) and tip matches. pass return True