66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
from dataclasses import dataclass
|
|
from typing import List, Optional, Dict, Any
|
|
from .delta import verify_delta_block, verify_inclusion, merkle_path_for_index
|
|
|
|
|
|
@dataclass
|
|
class DeltaBlock:
|
|
"""A small wrapper representing a compact delta-block used by adapters.
|
|
|
|
Fields mirror the structure produced by DeltaLog.export_delta_block to
|
|
provide a minimal, well-typed API surface for adapters and demos.
|
|
"""
|
|
entries: List[Dict[str, Any]]
|
|
digests: List[str]
|
|
root: str
|
|
anchor: Optional[str]
|
|
aggregated_sig: Optional[str]
|
|
since_index: int = 0
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"entries": self.entries,
|
|
"digests": self.digests,
|
|
"root": self.root,
|
|
"anchor": self.anchor,
|
|
"aggregated_sig": self.aggregated_sig,
|
|
"since_index": self.since_index,
|
|
}
|
|
|
|
@staticmethod
|
|
def from_dict(d: Dict[str, Any]) -> "DeltaBlock":
|
|
return DeltaBlock(
|
|
entries=d.get("entries", []),
|
|
digests=d.get("digests", []),
|
|
root=d.get("root", ""),
|
|
anchor=d.get("anchor"),
|
|
aggregated_sig=d.get("aggregated_sig"),
|
|
since_index=d.get("since_index", 0),
|
|
)
|
|
|
|
|
|
def verify_block(block: DeltaBlock, signer_key: Optional[bytes] = None) -> bool:
|
|
"""Convenience wrapper around verify_delta_block for DeltaBlock instances."""
|
|
return verify_delta_block(block.to_dict(), signer_key=signer_key)
|
|
|
|
|
|
def verify_entry_inclusion(entry_digest: str, block: DeltaBlock, index: int) -> bool:
|
|
"""Verify that a given entry digest is included in the block's Merkle root.
|
|
|
|
This function reconstructs the merkle path information from the block's
|
|
entries for the requested index and calls the library-level verify_inclusion
|
|
helper. This is useful for adapters that need to prove a single entry's
|
|
inclusion without re-computing the whole tree.
|
|
"""
|
|
if index < 0 or index >= len(block.digests):
|
|
return False
|
|
# find the merkle_path attached to the corresponding entry if present
|
|
try:
|
|
entry = block.entries[index]
|
|
except Exception:
|
|
return False
|
|
|
|
merkle_path = entry.get("merkle_path") if isinstance(entry, dict) else None
|
|
root = block.root
|
|
return verify_inclusion(entry_digest, merkle_path or [], root, index)
|