build(agent): weasel-1#856f80 iteration

cosmic_ledger/delta.py

@@ -78,6 +78,42 @@ def verify_delta_root(delta_entries: List[dict]) -> bool:
             return False
     return True
 
+
+def verify_inclusion(leaf_digest: str, merkle_path: List[str], root: str, index: int) -> bool:
+    """Verify that a leaf digest is included in a Merkle tree with given root.
+
+    leaf_digest: hex string of the leaf digest
+    merkle_path: list of sibling digests (hex) from leaf up to, but not including, the root
+    root: expected Merkle root (hex)
+    index: the zero-based index of the leaf within the original leaves array
+
+    This mirrors the path generation in merkle_path_for_index: when a sibling is
+    missing the leaf is duplicated. We reconstruct the hash up to the root and
+    compare with the provided root.
+    """
+    if not leaf_digest:
+        return False
+    try:
+        node = bytes.fromhex(leaf_digest)
+    except Exception:
+        return False
+
+    idx = index
+    for sib_hex in merkle_path:
+        try:
+            sib = bytes.fromhex(sib_hex)
+        except Exception:
+            return False
+        if idx % 2 == 0:
+            # current node is left child
+            node = hashlib.sha256(node + sib).digest()
+        else:
+            # current node is right child
+            node = hashlib.sha256(sib + node).digest()
+        idx = idx // 2
+
+    return node.hex() == (root or "")
+
 class DeltaLog:
     def __init__(self):
         self.entries = []  # each entry is a dict with digest and payload

test.sh

@@ -40,3 +40,11 @@ print('Delta replication test passed')
 PY
 
 echo "Tests completed."
+
+# Run unit tests in tests/ (if pytest is available in the environment)
+if command -v pytest >/dev/null 2>&1; then
+    echo "Running pytest..."
+    pytest -q tests || { echo "pytest failed"; exit 1; }
+else
+    echo "pytest not found; skipping pytest run"
+fi

tests/test_basic.py

@@ -1,3 +1,9 @@
+import sys
+import pathlib
+
+# Ensure repository root is on sys.path so tests can import the package
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
+
 from cosmic_ledger.ledger import LocalLedger
 
 def test_basic_replication():

tests/test_delta_inclusion.py

@@ -0,0 +1,33 @@
+import sys
+import pathlib
+import hashlib
+
+# Ensure repository root is on sys.path so tests can import the package
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
+
+from cosmic_ledger.delta import DeltaLog, merkle_path_for_index, verify_inclusion
+
+
+def test_merkle_inclusion_valid_and_tampered():
+    dl = DeltaLog()
+    # add three simple payloads
+    entries = [ {"v": i} for i in range(3) ]
+    for e in entries:
+        dl.add_entry(e)
+
+    leaves = [en["digest"] for en in dl.entries]
+    # for each leaf, compute path and verify inclusion
+    for i, digest in enumerate(leaves):
+        path_info = merkle_path_for_index(leaves, i)
+        root = path_info["root"]
+        path = path_info["path"]
+        assert verify_inclusion(digest, path, root, i)
+
+    # Tamper with a sibling digest and ensure verification fails
+    tampered_path = path.copy()
+    # flip one byte in the first sibling (if present) to simulate corruption
+    if tampered_path:
+        first = tampered_path[0]
+        # replace first hex char with something else (safe mutation)
+        tampered_path[0] = ("0" if first[0] != "0" else "1") + first[1:]
+        assert not verify_inclusion(leaves[0], tampered_path, root, 0)