diff --git a/src/willow/grovedb/__init__.py b/src/willow/grovedb/__init__.py index 1421b93..eacb258 100644 --- a/src/willow/grovedb/__init__.py +++ b/src/willow/grovedb/__init__.py @@ -132,6 +132,7 @@ from .verifier import ( verify_grovedb_proof, verify_proof_against_root, + check_envelope, quick_verify, VerifyOptions, ) @@ -227,6 +228,7 @@ # Main verifier functions "verify_grovedb_proof", "verify_proof_against_root", + "check_envelope", "quick_verify", "VerifyOptions", ] diff --git a/src/willow/grovedb/verifier.py b/src/willow/grovedb/verifier.py index 6bad0fc..1b1de8a 100644 --- a/src/willow/grovedb/verifier.py +++ b/src/willow/grovedb/verifier.py @@ -26,6 +26,44 @@ class VerifyOptions: """Options for proof verification.""" limit: Optional[int] = None deserialize_elements: bool = True + # The GroveDB path the query was made at. When set, the proof envelope + # must carry a lower layer for every segment (see check_envelope); without + # it an empty result set at that path cannot be told apart from a proof + # whose subtree was silently dropped. + expected_path: Optional[List[bytes]] = None + + +def check_envelope(proof: GroveDBProof, expected_path: List[bytes]) -> None: + """ + The check grovedb's own verifier does not make. The verifier finds the + next layer by the envelope's `lower_layers` map KEY, which is not + hash-bound: a prover who renames or drops the entry for a subtree on the + query path gets the same root hash with that subtree's results silently + gone, so a proof of "K = V" verifies as "K is absent". Requiring a layer + for every path segment closes it (once a layer is present its root is + hash-bound to the parent). `prove_options` is prover-chosen bytes that + steer limit accounting, so it is pinned to the default the chain's prover + uses. + """ + if proof.version != 0: + raise GroveDBVerificationError(f"Unsupported proof version: {proof.version}") + _check_prove_options(proof) + layer = proof.proof.root_layer + for i, seg in enumerate(expected_path): + nxt = layer.lower_layers.get(bytes_to_hex(seg)) + if nxt is None: + raise GroveDBVerificationError( + f"envelope: no lower layer for path segment {i} ({seg!r}); " + "the proof does not descend to the query path" + ) + layer = nxt + if layer.lower_layers: + raise GroveDBVerificationError("envelope: unexpected lower layers below the query path") + + +def _check_prove_options(proof: GroveDBProof) -> None: + if not proof.proof.prove_options.decrease_limit_on_empty_sub_query_result: + raise GroveDBVerificationError("envelope: non-default prove_options") def verify_grovedb_proof( @@ -56,6 +94,9 @@ def _verify_proof( """Verify a decoded GroveDB proof.""" if proof.version != 0: raise GroveDBVerificationError(f"Unsupported proof version: {proof.version}") + _check_prove_options(proof) + if options.expected_path is not None: + check_envelope(proof, options.expected_path) results: List[Dict[str, Any]] = [] limit = options.limit diff --git a/src/willow/proof.py b/src/willow/proof.py index 0b45c7b..e99f0b6 100644 --- a/src/willow/proof.py +++ b/src/willow/proof.py @@ -103,7 +103,7 @@ async def verify_query_proof( error="Empty proof provided" ) - return self._verify(proof_bytes) + return self._verify(proof_bytes, self._expected_path(path_query)) except ValueError as e: return ProofVerificationResult( @@ -140,7 +140,19 @@ async def verify_item_proof( return await self.verify_query_proof(proof_hex, documents, path_query) - def _verify(self, proof_bytes: bytes) -> ProofVerificationResult: + @staticmethod + def _expected_path(path_query: Optional[PathQueryData]) -> Optional[List[bytes]]: + # A non-empty path is required to descend; [] keeps the legacy + # behaviour for root-level items. + if path_query is None or not path_query.path: + return None + return [seg.encode() for seg in path_query.path] + + def _verify( + self, + proof_bytes: bytes, + expected_path: Optional[List[bytes]] = None, + ) -> ProofVerificationResult: """ Verify proof bytes and return result. @@ -153,7 +165,8 @@ def _verify(self, proof_bytes: bytes) -> ProofVerificationResult: try: grovedb_options = GroveDBVerifyOptions( limit=self.options.limit, - deserialize_elements=self.options.deserialize_elements + deserialize_elements=self.options.deserialize_elements, + expected_path=expected_path, ) if self.options.expected_root_hash: diff --git a/tests/test_grovedb_envelope.py b/tests/test_grovedb_envelope.py new file mode 100644 index 0000000..07822bc --- /dev/null +++ b/tests/test_grovedb_envelope.py @@ -0,0 +1,96 @@ +"""Regression tests for the GroveDB envelope-descent guard (check_envelope). + +The gap (grovedb 3.1.0): the verifier finds the next layer by the proof +envelope's `lower_layers` map key, which is not hash-bound. A prover who +renames or drops the entry for a subtree on the query path gets the SAME root +hash with that subtree's results silently gone, so a proof of "K = V" verifies +as "K is absent". check_envelope closes it by requiring a lower layer for every +path segment and pinning prove_options to the chain default. + +These tests construct the decoded envelope directly (dataclasses), so they +exercise the guard independently of the byte decoder — see the PR note about +the decoder format. +""" +import pytest + +from willow.grovedb import GroveDBVerificationError +from willow.grovedb.types import GroveDBProof, GroveDBProofV0, LayerProof, ProveOptions +from willow.grovedb.hash import bytes_to_hex +from willow.grovedb.verifier import check_envelope + +PATH = [b"subgroves", b"aave-v3-lending", b"indexed", b"Supply"] + + +def leaf(*keys): + return LayerProof(merk_proof=b"", lower_layers={bytes_to_hex(k): leaf() for k in keys}) + + +def honest_envelope(path=PATH, opts=True): + # A chain of single-child layers down the path, empty at the leaf. + layer = LayerProof(merk_proof=b"", lower_layers={}) + for seg in reversed(path): + layer = LayerProof(merk_proof=b"", lower_layers={bytes_to_hex(seg): layer}) + return GroveDBProof( + version=0, + proof=GroveDBProofV0( + root_layer=layer, + prove_options=ProveOptions(decrease_limit_on_empty_sub_query_result=opts), + ), + ) + + +def test_honest_envelope_descends_to_the_path(): + check_envelope(honest_envelope(), PATH) + + +def test_renamed_lower_layer_is_rejected(): + p = honest_envelope() + # Rename the root layer's only key: the descent for segment 0 now misses. + root = p.proof.root_layer + (only_key, sub), = list(root.lower_layers.items()) + root.lower_layers = {bytes_to_hex(b"forged"): sub} + with pytest.raises(GroveDBVerificationError, match="does not descend"): + check_envelope(p, PATH) + + +def test_dropped_lower_layer_is_rejected(): + p = honest_envelope() + # Walk to the 'indexed' layer and drop its 'Supply' child. + layer = p.proof.root_layer + for seg in PATH[:3]: + layer = layer.lower_layers[bytes_to_hex(seg)] + layer.lower_layers = {} + with pytest.raises(GroveDBVerificationError, match="does not descend"): + check_envelope(p, PATH) + + +def test_extra_lower_layers_below_the_path_are_rejected(): + p = honest_envelope(path=PATH[:3]) # descends only to 'indexed' + with pytest.raises(GroveDBVerificationError, match="unexpected lower layers"): + check_envelope(p, PATH[:2]) # ask it to stop at 'aave-v3-lending' + + +def test_non_default_prove_options_is_rejected(): + with pytest.raises(GroveDBVerificationError, match="prove_options"): + check_envelope(honest_envelope(opts=False), PATH) + + +def test_verify_options_expected_path_invokes_the_guard(monkeypatch): + # verify_grovedb_proof must run check_envelope when expected_path is set. + from willow.grovedb import verifier + + called = {} + + def fake_decode(_data): + return honest_envelope() + + def fake_layer(*_a, **_k): + return "root" + + monkeypatch.setattr(verifier, "decode_grovedb_proof", fake_decode) + monkeypatch.setattr(verifier, "_verify_layer_proof", lambda *a, **k: b"\x00" * 32) + real_check = verifier.check_envelope + monkeypatch.setattr(verifier, "check_envelope", lambda p, path: called.setdefault("path", path) or real_check(p, path)) + + verifier.verify_grovedb_proof(b"x", verifier.VerifyOptions(expected_path=PATH)) + assert called["path"] == PATH