diff --git a/cosmic_ledger/delta.py b/cosmic_ledger/delta.py index 9c6e9b2..7c8710b 100644 --- a/cosmic_ledger/delta.py +++ b/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 diff --git a/test.sh b/test.sh index e9e56f7..51c8c8a 100644 --- a/test.sh +++ b/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 diff --git a/tests/test_basic.py b/tests/test_basic.py index 1a30188..094b0c0 100644 --- a/tests/test_basic.py +++ b/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(): diff --git a/tests/test_delta_inclusion.py b/tests/test_delta_inclusion.py new file mode 100644 index 0000000..2029c5a --- /dev/null +++ b/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)