diff --git a/conformance/.clusterfuzzlite/fuzz_receipt.py b/conformance/.clusterfuzzlite/fuzz_receipt.py index 5040eecd..763df568 100755 --- a/conformance/.clusterfuzzlite/fuzz_receipt.py +++ b/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,10 +104,37 @@ 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: @@ -115,16 +142,19 @@ def TestOneInput(data: bytes) -> None: 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}" diff --git a/conformance/tests/test_fuzz_receipt.py b/conformance/tests/test_fuzz_receipt.py index be010b29..e5f2cc7e 100644 --- a/conformance/tests/test_fuzz_receipt.py +++ b/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 == [[]]