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

cosmic_ledger/delta.py

@@ -1,6 +1,7 @@
 import hashlib
 import json
 from typing import List
+from .crypto import Signer
 
 def _hash_bytes(data: bytes) -> str:
     return hashlib.sha256(data).hexdigest()
@@ -166,6 +167,52 @@ class DeltaLog:
         digests = [e["digest"] for e in self.entries]
         return merkle_root(digests)
 
+    def export_delta_block(self, since_index: int = 0, signer_key: bytes = None) -> dict:
+        """Export a compact delta block suitable for DTN-friendly transfer.
+
+        The block contains:
+          - entries: full LedgerEntry dicts augmented with digest, delta_root, merkle_path
+          - digests: list of leaf digests included in the block (in order)
+          - root: merkle root for the block's leaves
+          - anchor: optional anchor recorded on this log
+          - aggregated_sig: HMAC-SHA256 of the root if signer_key is provided (hex)
+
+        This is a lightweight, backwards-compatible representation that
+        recipients can verify using verify_delta_root and (optional) aggregated_sig.
+        """
+        leaves = [e["digest"] for e in self.entries]
+        # entries since index
+        entries = []
+        for i in range(since_index, len(self.entries)):
+            entry = self.entries[i]
+            path_info = merkle_path_for_index(leaves, i)
+            full_entry = entry["payload"].copy()
+            full_entry.update({
+                "digest": entry["digest"],
+                "delta_root": path_info["root"],
+                "merkle_path": path_info["path"],
+                "index": i,
+            })
+            entries.append(full_entry)
+
+        block_digests = [e["digest"] for e in self.entries[since_index:]]
+        block_root = merkle_root(block_digests)
+
+        aggregated_sig = None
+        if signer_key is not None:
+            signer = Signer(signer_key)
+            # sign the block root bytes
+            aggregated_sig = signer.sign(block_root.encode("utf-8")).hex()
+
+        return {
+            "entries": entries,
+            "digests": block_digests,
+            "root": block_root,
+            "anchor": self.anchor,
+            "aggregated_sig": aggregated_sig,
+            "since_index": since_index,
+        }
+
     def checkpoint(self, prune_before_index: int = 0) -> dict:
         """Create a checkpoint for entries up to prune_before_index (exclusive) and optionally prune them.
 
@@ -203,3 +250,39 @@ class DeltaLog:
             if digest in cp.get("digests", []):
                 return True
         return False
+
+
+def verify_delta_block(block: dict, signer_key: bytes = None) -> bool:
+    """Verify the integrity of a delta block.
+
+    Checks performed:
+      - recompute merkle root from provided digests and compare to block['root']
+      - ensure each entry's digest matches the corresponding digest
+      - if signer_key is provided, verify aggregated_sig matches HMAC of root
+    """
+    if not block:
+        return False
+    digests = block.get("digests", [])
+    computed_root = merkle_root(digests)
+    if computed_root != (block.get("root") or ""):
+        return False
+
+    # ensure entries line up with digests
+    entries = block.get("entries", [])
+    for i, entry in enumerate(entries):
+        if entry.get("digest") != digests[i]:
+            return False
+
+    # verify optional aggregated signature
+    if signer_key is not None:
+        sig_hex = block.get("aggregated_sig")
+        if sig_hex is None:
+            return False
+        signer = Signer(signer_key)
+        expected = signer.sign((computed_root or "").encode("utf-8"))
+        try:
+            return expected.hex() == sig_hex
+        except Exception:
+            return False
+
+    return True

tests/test_delta_block.py

@@ -0,0 +1,40 @@
+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.delta import DeltaLog, verify_delta_block
+
+
+def test_export_and_verify_delta_block_with_signature():
+    dl = DeltaLog()
+    # add a few payloads
+    for i in range(4):
+        dl.add_entry({"v": i})
+
+    key = b'unique-test-key-xyz'
+    block = dl.export_delta_block(since_index=0, signer_key=key)
+
+    assert "root" in block
+    assert "digests" in block
+    assert "aggregated_sig" in block
+    assert block["aggregated_sig"] is not None
+
+    # verify should succeed with correct key
+    assert verify_delta_block(block, signer_key=key)
+
+
+def test_verify_delta_block_tamper_detection():
+    dl = DeltaLog()
+    for i in range(3):
+        dl.add_entry({"v": i})
+
+    key = b'another-key-123'
+    block = dl.export_delta_block(since_index=0, signer_key=key)
+
+    # tamper with a digest
+    if block["digests"]:
+        block["digests"][0] = ("0" if block["digests"][0][0] != "0" else "1") + block["digests"][0][1:]
+
+    assert not verify_delta_block(block, signer_key=key)