diff --git a/deltatrace/dsl.py b/deltatrace/dsl.py index fef9abf..a7497eb 100644 --- a/deltatrace/dsl.py +++ b/deltatrace/dsl.py @@ -114,3 +114,62 @@ class TraceGraph: next_level.append(hashlib.sha256((left + right).encode()).hexdigest()) level = next_level return level[0] + + def merkle_proof(self, index: int) -> List[Dict[str, str]]: + """Compute a Merkle-like proof for the node at the given index. + + This returns a list of dicts, each describing the sibling hash at a + particular tree level required to recompute the Merkle root. Each entry + has: + - direction: 'left' or 'right' indicating the position of the sibling + - hash: the hexadecimal digest of the sibling node at that level + + Note: This is a lightweight, deterministic proof suitable for auditing + provenance in MVP deployments. It relies on a fixed leaf ordering (the + order of self.nodes) and a deterministic leaf hashing similar to merkle_root. + """ + import hashlib + import json + + def _canonicalize(obj: object) -> Dict[str, object]: + if hasattr(obj, "__dict__"): + data = getattr(obj, "__dict__") + return {k: v for k, v in data.items()} + return {"repr": str(obj)} + + if index < 0 or index >= len(self.nodes): + raise IndexError("index out of range for merkle_proof") + + # Build leaf hashes from canonicalized node representations + leaves: List[str] = [] + for n in self.nodes: + canon = _canonicalize(n) + serialized = json.dumps(canon, sort_keys=True, separators=(",", ":")) + leaves.append(hashlib.sha256(serialized.encode()).hexdigest()) + + proof: List[Dict[str, str]] = [] + idx = index + level = leaves + while len(level) > 1: + # Sibling hash for the current index + if idx % 2 == 0: + sib_index = idx + 1 + direction = "right" + else: + sib_index = idx - 1 + direction = "left" + if sib_index < len(level): + proof.append({"direction": direction, "hash": level[sib_index]}) + else: + # If no true sibling (odd last item), replicate the node itself + proof.append({"direction": direction, "hash": level[idx]}) + # Compute next level + next_level: List[str] = [] + 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).encode()).hexdigest()) + level = next_level + idx = idx // 2 + + return proof