test(conformance): fuzz registry-entry inputs alongside receipts

conformance/.clusterfuzzlite/fuzz_receipt.py

@@ -10,9 +10,9 @@
   root of the supplied entry and the claim is the one that root includes. Forging either takes a
   SHA-256 collision, so a PASS on anything else is a defect in the replay.
 
-Half the inputs are raw receipt bytes; the other half start from a genuine
-five-leaf receipt and mutate one field, which reaches the replay loop far more
-often than random JSON would.
+Half the inputs vary the receipt and half vary the registry entry. Each lane
+uses either raw JSON or a mutation of its genuine fixture. Entry parsing is
+exercised directly even when the entry cannot reach the inclusion replay.
 """
 import contextlib
 import hashlib
@@ -104,27 +104,57 @@ def _mutated(fdp: atheris.FuzzedDataProvider) -> object:
     return receipt
 
 
+def _mutated_entry(fdp: atheris.FuzzedDataProvider) -> object:
+    entry = dict(_ENTRY)
+    field = fdp.PickValueInList(sorted(entry))
+    kind = fdp.ConsumeIntInRange(0, 4)
+    if kind == 0:
+        del entry[field]
+    elif kind == 1:
+        entry[field] = fdp.ConsumeUnicode(80)
+    elif kind == 2:
+        entry[field] = fdp.ConsumeIntInRange(-2, 2**64)
+    elif kind == 3:
+        entry[field] = None
+    else:
+        entry[field] = fdp.ConsumeBool()
+    return entry
+
+
 def TestOneInput(data: bytes) -> None:
     fdp = atheris.FuzzedDataProvider(data)
     claim = _CLAIMS[fdp.ConsumeIntInRange(0, len(_CLAIMS) - 1)]
+    entry = _ENTRY
     if fdp.ConsumeBool():
+        receipt = _RECEIPT
+        if fdp.ConsumeBool():
+            entry = _mutated_entry(fdp)
+        else:
+            try:
+                entry = loads_strict(fdp.ConsumeBytes(fdp.remaining_bytes()))
+            except LoadError:
+                return
+    elif fdp.ConsumeBool():
         receipt = _mutated(fdp)
     else:
         try:
             receipt = loads_strict(fdp.ConsumeBytes(fdp.remaining_bytes()))
         except LoadError:
             return
 
+    with contextlib.suppress(InclusionError):
+        parse_entry(entry)
+
     with contextlib.suppress(InclusionError):
         verify_inclusion(claim, *parse_receipt(receipt))
 
     if not isinstance(receipt, dict):
         return
-    findings = tr_anc.check(claim, receipt=receipt, entry=_ENTRY)
+    findings = tr_anc.check(claim, receipt=receipt, entry=entry)
     if any(f.code == "TR-ANC-002" and f.passed() for f in findings):
         assert claim is _CLAIMS[_TARGET], f"inclusion proven for the wrong claim: {receipt!r}"
         index, path, _, _ = parse_proof(receipt)
-        root, count, _, _ = parse_entry(_ENTRY)
+        root, count, _, _ = parse_entry(entry)
         assert root == _ROOT, f"inclusion proven under another root: {receipt!r}"
         assert verify_inclusion(claim, index, path, count, root), (
             f"inclusion not reproduced against the supplied entry: {receipt!r}"

conformance/tests/test_fuzz_receipt.py

@@ -56,3 +56,125 @@ def test_pass_oracle_independently_replays_proof(harness, monkeypatch):
     monkeypatch.setattr(harness["tr_anc"], "check", lambda *a, **kw: [fake_pass])
     with pytest.raises(AssertionError, match="not reproduced against the supplied entry"):
         harness["TestOneInput"](json.dumps(receipt).encode())
+
+
+def _entry_lane(harness, monkeypatch, entry):
+    provider = harness["atheris"].FuzzedDataProvider
+
+    class EntryProvider(provider):
+        def __init__(self, data):
+            super().__init__(data)
+            self.choices = iter((True, False))
+
+        def ConsumeBool(self):
+            return next(self.choices)
+
+    monkeypatch.setattr(harness["atheris"], "FuzzedDataProvider", EntryProvider)
+    harness["TestOneInput"](json.dumps(entry).encode())
+
+
+def test_entry_lane_reaches_parser_and_checker(harness, monkeypatch):
+    entry = {**harness["_ENTRY"], "producer": "other/1.0.0"}
+    seen = []
+    parse = harness["parse_entry"]
+    check = harness["tr_anc"].check
+    globals_ = harness["TestOneInput"].__globals__
+
+    def observe_parse(value):
+        seen.append(("parse", value))
+        return parse(value)
+
+    def observe_check(*args, **kwargs):
+        seen.append(("check", kwargs["entry"]))
+        return check(*args, **kwargs)
+
+    monkeypatch.setitem(globals_, "parse_entry", observe_parse)
+    monkeypatch.setattr(harness["tr_anc"], "check", observe_check)
+    _entry_lane(harness, monkeypatch, entry)
+    assert ("parse", entry) in seen
+    assert ("check", entry) in seen
+
+
+@pytest.mark.parametrize("missing", ["ts", "merkle_root", "leaf_count", "producer", "batch_id"])
+def test_entry_missing_fields_refuse(harness, monkeypatch, missing):
+    entry = dict(harness["_ENTRY"])
+    del entry[missing]
+    with pytest.raises(harness["InclusionError"], match="missing required"):
+        harness["parse_entry"](entry)
+    _entry_lane(harness, monkeypatch, entry)
+
+
+@pytest.mark.parametrize("field", ["ts", "producer", "batch_id"])
+@pytest.mark.parametrize("value", [None, True, 7, [], {}])
+def test_entry_nonstring_fields_refuse(harness, monkeypatch, field, value):
+    entry = {**harness["_ENTRY"], field: value}
+    with pytest.raises(harness["InclusionError"], match="non-empty string"):
+        harness["parse_entry"](entry)
+    _entry_lane(harness, monkeypatch, entry)
+
+
+def test_entry_lane_valid_twin_and_no_fixture_mutation(harness, monkeypatch):
+    before = json.dumps(harness["_ENTRY"], sort_keys=True)
+    _entry_lane(harness, monkeypatch, dict(harness["_ENTRY"]))
+    assert json.dumps(harness["_ENTRY"], sort_keys=True) == before
+
+
+def test_entry_lane_false_pass_is_caught(harness, monkeypatch):
+    entry = {**harness["_ENTRY"], "merkle_root": "sha256:" + "00" * 32}
+    fake_pass = types.SimpleNamespace(code="TR-ANC-002", passed=lambda: True)
+    monkeypatch.setattr(harness["tr_anc"], "check", lambda *a, **kw: [fake_pass])
+    with pytest.raises(AssertionError, match="another root"):
+        _entry_lane(harness, monkeypatch, entry)
+
+
+@pytest.mark.parametrize("kind", range(5))
+def test_structured_entry_lane_mutates_selected_field(harness, monkeypatch, kind):
+    seen = []
+    check = harness["tr_anc"].check
+    provider = harness["atheris"].FuzzedDataProvider
+
+    class EntryProvider(provider):
+        def __init__(self, data):
+            super().__init__(data)
+            self.ints = iter((harness["_TARGET"], kind, 7))
+            self.bools = iter((True, True, False))
+
+        def ConsumeIntInRange(self, lower, upper):
+            return next(self.ints)
+
+        def ConsumeBool(self):
+            return next(self.bools)
+
+        def PickValueInList(self, values):
+            return "producer"
+
+        def ConsumeUnicode(self, length):
+            return "changed/1.0.0"
+
+    def observe(*args, **kwargs):
+        seen.append(kwargs["entry"])
+        return check(*args, **kwargs)
+
+    monkeypatch.setattr(harness["atheris"], "FuzzedDataProvider", EntryProvider)
+    monkeypatch.setattr(harness["tr_anc"], "check", observe)
+    harness["TestOneInput"](b"")
+    expected = dict(harness["_ENTRY"])
+    if kind == 0:
+        del expected["producer"]
+    else:
+        expected["producer"] = {1: "changed/1.0.0", 2: 7, 3: None, 4: False}[kind]
+    assert seen == [expected]
+    assert harness["_ENTRY"]["producer"] == "fuzz/1.0.0"
+
+
+def test_nonobject_entry_reaches_direct_parser(harness, monkeypatch):
+    seen = []
+    parse = harness["parse_entry"]
+
+    def observe(value):
+        seen.append(value)
+        return parse(value)
+
+    monkeypatch.setitem(harness["TestOneInput"].__globals__, "parse_entry", observe)
+    _entry_lane(harness, monkeypatch, [])
+    assert seen == [[]]