diff --git a/script/smoke/chain.py b/script/smoke/chain.py index 550e7f1..8551ec1 100644 --- a/script/smoke/chain.py +++ b/script/smoke/chain.py @@ -18,6 +18,7 @@ import time import urllib.error import urllib.request +from collections.abc import Callable from eth_account import Account from eth_account.signers.local import LocalAccount @@ -40,6 +41,46 @@ def log(msg: str) -> None: print(f"[smoke] {msg}", file=sys.stderr) +def run_collected( + checks: "list[tuple[str, Callable[[], None]]]", + *, + label: str, + informational: "frozenset[str] | set[str]" = frozenset(), + step_prefix: str = "", +) -> None: + """Run every check, recording failures instead of aborting at the first. + + `die()` raises SystemExit, so a plain sequence stops at the first failure and the remaining + checks never execute. Each `(name, thunk)` is isolated and its failure recorded. Names in + `informational` are reported but do not fail the run. Raises SystemExit at the end if any + required check failed. + """ + findings: list[tuple[str, str, bool]] = [] + for i, (name, thunk) in enumerate(checks, 1): + step(f"{step_prefix}{i}", name) + required = name not in informational + try: + thunk() + except SystemExit as exc: + detail = str(exc.code or "").replace("[smoke] ERROR: ", "") + print(f" ✗ {'FINDING' if required else 'info'}: {detail}", file=sys.stderr) + findings.append((name, detail, required)) + except Exception as exc: # noqa: BLE001 - harness/RPC error, surface as a finding + detail = f"{type(exc).__name__}: {exc}" + print(f" ✗ ERROR: {detail}", file=sys.stderr) + findings.append((name, detail, required)) + + required_fail = [f for f in findings if f[2]] + info_only = [f for f in findings if not f[2]] + log(f"{label}: {len(checks) - len(findings)}/{len(checks)} checks held") + for name, detail, _ in findings: + log(f" ✗ {name} — {detail}") + if info_only: + log(f"({len(info_only)} accepted divergence(s) reported as informational)") + if required_fail: + raise SystemExit(f"[smoke] {label}: {len(required_fail)} finding(s) need triage") + + def step(n: object, desc: str) -> None: print(f" \u2192 [{n}] {desc}", file=sys.stderr) @@ -560,12 +601,27 @@ def create_policy_with_accounts(self, admin: ChecksumAddress, ptype: int, accoun receipt = self.send(self.policy.functions.createPolicyWithAccounts(admin, ptype, accounts), self.deployer) return self._policy_id_from(receipt) + def create_composite_policy(self, admin: ChecksumAddress, ptype: int, child_ids: list[int]) -> int: + """Create a UNION/INTERSECT composite over existing simple policies; return the new id.""" + receipt = self.send(self.policy.functions.createCompositePolicy(admin, ptype, child_ids), self.deployer) + return self._policy_id_from(receipt) + def _policy_id_from(self, receipt: TxReceipt) -> int: events = self.policy.events.PolicyCreated().process_receipt(receipt, errors=DISCARD) if not events: die("PolicyCreated event not found in receipt") return int(events[0]["args"]["policyId"]) + def composite_children_from(self, receipt: TxReceipt, policy_id: int | None = None) -> list[int]: + """Decode `CompositePolicyUpdated` from a SPECIFIC receipt and return its `childPolicyIds`.""" + events = self.policy.events.CompositePolicyUpdated().process_receipt(receipt, errors=DISCARD) + if policy_id is not None: + events = [e for e in events if int(e["args"]["policyId"]) == policy_id] + if not events: + suffix = f" for policyId={policy_id}" if policy_id is not None else "" + die(f"CompositePolicyUpdated event not found in receipt{suffix}") + return [int(child) for child in events[0]["args"]["childPolicyIds"]] + def _norm(v: object) -> object: """Normalize for comparison: lowercase hex/addresses so checksums match.""" diff --git a/script/smoke/config.py b/script/smoke/config.py index 33a7b33..95f7952 100644 --- a/script/smoke/config.py +++ b/script/smoke/config.py @@ -36,9 +36,18 @@ def amt(whole: int, decimals: int) -> int: VARIANT_ASSET = 0 VARIANT_STABLECOIN = 1 -# PolicyType enum (IPolicyRegistry). +# PolicyType enum (IPolicyRegistry). BLOCKLIST/ALLOWLIST are "simple" policies (decide from an +# address set); UNION/INTERSECT are "composite" gates over 2..4 simple children. POLICY_TYPE_BLOCKLIST = 0 POLICY_TYPE_ALLOWLIST = 1 +POLICY_TYPE_UNION = 2 +POLICY_TYPE_INTERSECT = 3 + +# Composite child-set bounds. Outside this range the registry reverts +# ChildPoliciesOutsideOfRange(MIN_CHILD_POLICIES, MAX_CHILD_POLICIES). Distinct from the +# 64-account membership batch limit (BatchSizeTooLarge). +MIN_CHILD_POLICIES = 2 +MAX_CHILD_POLICIES = 4 # Built-in policy IDs: ALWAYS_ALLOW = 0, ALWAYS_BLOCK = (uint64(ALLOWLIST) << 56) | 1. ALWAYS_ALLOW_ID = 0 diff --git a/script/smoke/journeys/policy_registry.py b/script/smoke/journeys/policy_registry.py index 9b809c3..b19cf99 100644 --- a/script/smoke/journeys/policy_registry.py +++ b/script/smoke/journeys/policy_registry.py @@ -4,6 +4,11 @@ admin lifecycle, and — the part that matters most — a token actually enforcing a policy (PolicyForbids on transfer + mint). Edges cover the registry's reverts and the token-side write-time validation, then a flow-level event check. + +`_composite` covers the UNION/INTERSECT composite gates: construction, live +gate evaluation over child membership, full-replacement `updateComposite` +semantics, the child-set validation reverts, and token enforcement through a +composite. """ from __future__ import annotations @@ -100,14 +105,203 @@ def _edges(c: Chain, tok, pid_r: int, pid_b: int) -> None: c.expect_revert("PolicyNotFound", tok.functions.updatePolicy(config.TRANSFER_SENDER_POLICY, 999999), c.DEPLOYER) +def _composite(c: Chain, pid_b: int) -> None: + """Composite (UNION / INTERSECT) policies: construction, evaluation, update, edges, enforcement.""" + carol = c.cfg.new_addr("carol") + dave = c.cfg.new_addr("dave") + erin = c.cfg.new_addr("erin") + + step(17, "seed two simple children, then create a UNION and an INTERSECT over them") + # childX allows {alice, dave}; childY allows {bob, dave}. dave is the only account in BOTH. + pid_x = c.create_policy_with_accounts(c.DEPLOYER, config.POLICY_TYPE_ALLOWLIST, [c.ALICE, dave]) + pid_y = c.create_policy_with_accounts(c.DEPLOYER, config.POLICY_TYPE_ALLOWLIST, [c.BOB, dave]) + pid_union = c.create_composite_policy(c.DEPLOYER, config.POLICY_TYPE_UNION, [pid_x, pid_y]) + pid_inter = c.create_composite_policy(c.DEPLOYER, config.POLICY_TYPE_INTERSECT, [pid_x, pid_y]) + c.assert_eq(c.policy.functions.policyExists(pid_union).call(), True, "UNION composite exists") + c.assert_eq(c.policy.functions.policyAdmin(pid_union).call(), c.DEPLOYER, "UNION admin == deployer") + c.assert_eq(c.policy.functions.policyExists(pid_inter).call(), True, "INTERSECT composite exists") + c.assert_eq(c.policy.functions.policyAdmin(pid_inter).call(), c.DEPLOYER, "INTERSECT admin == deployer") + + step(18, "UNION authorizes an account in ANY child; denies one in none") + c.assert_eq(c.policy.functions.isAuthorized(pid_union, c.ALICE).call(), True, "UNION: alice (childX only) allowed") + c.assert_eq(c.policy.functions.isAuthorized(pid_union, c.BOB).call(), True, "UNION: bob (childY only) allowed") + c.assert_eq(c.policy.functions.isAuthorized(pid_union, dave).call(), True, "UNION: dave (both children) allowed") + c.assert_eq(c.policy.functions.isAuthorized(pid_union, carol).call(), False, "UNION: carol (no child) denied") + + step(19, "INTERSECT authorizes only an account in EVERY child") + c.assert_eq(c.policy.functions.isAuthorized(pid_inter, dave).call(), True, "INTERSECT: dave (every child) allowed") + c.assert_eq(c.policy.functions.isAuthorized(pid_inter, c.ALICE).call(), False, "INTERSECT: alice missing childY -> denied") + c.assert_eq(c.policy.functions.isAuthorized(pid_inter, c.BOB).call(), False, "INTERSECT: bob missing childX -> denied") + c.assert_eq(c.policy.functions.isAuthorized(pid_inter, carol).call(), False, "INTERSECT: carol (no child) denied") + + step(20, "live evaluation: mutate a CHILD's membership, composite verdict flips (no call on the composite)") + c.send(c.policy.functions.updateAllowlist(pid_x, True, [carol]), c.deployer) + c.assert_eq(c.policy.functions.isAuthorized(pid_union, carol).call(), True, "UNION: carol allowed after childX add") + c.assert_eq(c.policy.functions.isAuthorized(pid_inter, carol).call(), False, "INTERSECT: carol still denied (childY missing)") + c.send(c.policy.functions.updateAllowlist(pid_y, True, [carol]), c.deployer) + c.assert_eq(c.policy.functions.isAuthorized(pid_inter, carol).call(), True, "INTERSECT: carol allowed once in every child") + c.send(c.policy.functions.updateAllowlist(pid_x, False, [carol]), c.deployer) + c.assert_eq(c.policy.functions.isAuthorized(pid_inter, carol).call(), False, "INTERSECT: carol denied again after childX removal") + + step(21, "updateComposite REPLACES the child set (no merge); event carries the exact new set") + pid_c1 = c.create_policy_with_accounts(c.DEPLOYER, config.POLICY_TYPE_ALLOWLIST, [erin]) + pid_c2 = c.create_policy_with_accounts(c.DEPLOYER, config.POLICY_TYPE_ALLOWLIST, [erin]) + c.assert_eq(c.policy.functions.isAuthorized(pid_union, c.ALICE).call(), True, "pre-update: alice allowed via childX") + receipt = c.send(c.policy.functions.updateComposite(pid_union, [pid_c1, pid_c2]), c.deployer) + c.assert_eq(c.policy.functions.isAuthorized(pid_union, c.ALICE).call(), False, "post-update: alice denied (old children dropped, not merged)") + c.assert_eq(c.policy.functions.isAuthorized(pid_union, c.BOB).call(), False, "post-update: bob denied (old children dropped)") + c.assert_eq(c.policy.functions.isAuthorized(pid_union, erin).call(), True, "post-update: erin allowed via the new children") + # Payload-level check: assert_events_emitted is presence-only, so decode this specific receipt. + c.assert_eq( + c.composite_children_from(receipt, pid_union), + [pid_c1, pid_c2], + "CompositePolicyUpdated payload == exactly the new child ids", + ) + + step(22, "child count outside [2,4] -> ChildPoliciesOutsideOfRange") + too_few = [pid_x] + too_many = [pid_x, pid_y, pid_c1, pid_c2, pid_b] + c.assert_eq(len(too_few) < config.MIN_CHILD_POLICIES, True, "fixture: 1 child is below MIN_CHILD_POLICIES") + c.assert_eq(len(too_many) > config.MAX_CHILD_POLICIES, True, "fixture: 5 children is above MAX_CHILD_POLICIES") + c.expect_revert( + "ChildPoliciesOutsideOfRange", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_UNION, too_few), + c.DEPLOYER, + ) + c.expect_revert( + "ChildPoliciesOutsideOfRange", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_UNION, too_many), + c.DEPLOYER, + ) + c.expect_revert("ChildPoliciesOutsideOfRange", c.policy.functions.updateComposite(pid_union, too_few), c.DEPLOYER) + c.expect_revert("ChildPoliciesOutsideOfRange", c.policy.functions.updateComposite(pid_union, too_many), c.DEPLOYER) + + step(23, "a built-in sentinel as a child -> InvalidChildPolicy") + c.expect_revert( + "InvalidChildPolicy", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_UNION, [config.ALWAYS_ALLOW_ID, pid_x]), + c.DEPLOYER, + ) + c.expect_revert( + "InvalidChildPolicy", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_UNION, [config.ALWAYS_BLOCK_ID, pid_x]), + c.DEPLOYER, + ) + c.expect_revert( + "InvalidChildPolicy", c.policy.functions.updateComposite(pid_union, [config.ALWAYS_ALLOW_ID, pid_x]), c.DEPLOYER + ) + + step(24, "a composite as a child -> InvalidChildPolicy") + c.expect_revert( + "InvalidChildPolicy", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_INTERSECT, [pid_inter, pid_x]), + c.DEPLOYER, + ) + c.expect_revert("InvalidChildPolicy", c.policy.functions.updateComposite(pid_union, [pid_inter, pid_x]), c.DEPLOYER) + + step(25, "non-admin updateComposite -> Unauthorized") + c.expect_revert("Unauthorized", c.policy.functions.updateComposite(pid_union, [pid_x, pid_y]), c.USER2) + + step(26, "composite creator input guards: ZeroAddress, IncompatiblePolicyType") + # ZeroAddress outranks every later check, so the child set here is deliberately valid. + c.expect_revert( + "ZeroAddress", + c.policy.functions.createCompositePolicy(config.ZERO, config.POLICY_TYPE_UNION, [pid_x, pid_y]), + c.DEPLOYER, + ) + # A composite must be UNION or INTERSECT; a simple gate is rejected by the composite creator. + for simple in (config.POLICY_TYPE_ALLOWLIST, config.POLICY_TYPE_BLOCKLIST): + c.expect_revert( + "IncompatiblePolicyType", + c.policy.functions.createCompositePolicy(c.DEPLOYER, simple, [pid_x, pid_y]), + c.DEPLOYER, + ) + # Mirror: updateComposite must reject a SIMPLE target, and the membership mutators must reject a + # COMPOSITE target. Together with the conformance checks at the end of the journey this closes the + # type-guard matrix in both directions, so a red conformance check reads as "the two creator-side + # bugs" rather than "the type byte is ignored everywhere". + c.expect_revert("IncompatiblePolicyType", c.policy.functions.updateComposite(pid_x, [pid_x, pid_y]), c.DEPLOYER) + c.expect_revert( + "IncompatiblePolicyType", c.policy.functions.updateAllowlist(pid_union, True, [c.BOB]), c.DEPLOYER + ) + c.expect_revert( + "IncompatiblePolicyType", c.policy.functions.updateBlocklist(pid_inter, True, [c.BOB]), c.DEPLOYER + ) + + step(27, "non-existent child -> PolicyNotFound, and it outranks InvalidChildPolicy") + # A well-formed but never-created id. Type byte 0 (BLOCKLIST) — note this is exactly the shape whose + # read-side semantics are permissive (an empty blocklist authorizes everyone), so child-existence + # validation is the only thing standing between a permissionless caller and a composite that + # authorizes every account while looking like a legitimate admin-owned gate. + ghost = 999999 + c.assert_eq(c.policy.functions.policyExists(ghost).call(), False, "fixture: ghost child id does not exist") + c.expect_revert( + "PolicyNotFound", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_UNION, [pid_x, ghost]), + c.DEPLOYER, + ) + # Ghost in FIRST position too: an impl that only validates childPolicyIds[0] passes the case above. + c.expect_revert( + "PolicyNotFound", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_UNION, [ghost, pid_x]), + c.DEPLOYER, + ) + c.expect_revert("PolicyNotFound", c.policy.functions.updateComposite(pid_union, [pid_x, ghost]), c.DEPLOYER) + # Precedence: existence is checked across the WHOLE set before validity. The invalid child is + # placed FIRST and the ghost LAST, so a per-element validator would answer InvalidChildPolicy. + c.expect_revert( + "PolicyNotFound", + c.policy.functions.createCompositePolicy(c.DEPLOYER, config.POLICY_TYPE_UNION, [pid_inter, ghost]), + c.DEPLOYER, + ) + + step(28, "create ASSET token wired to a UNION composite on TRANSFER_RECEIVER + MINT_RECEIVER") + # Dedicated children so the token's gate is unaffected by the mutations above. + pid_t1 = c.create_policy_with_accounts(c.DEPLOYER, config.POLICY_TYPE_ALLOWLIST, [c.ALICE]) + pid_t2 = c.create_policy_with_accounts(c.DEPLOYER, config.POLICY_TYPE_ALLOWLIST, [c.DEPLOYER]) + pid_tok = c.create_composite_policy(c.DEPLOYER, config.POLICY_TYPE_UNION, [pid_t1, pid_t2]) + c.assert_eq(c.policy.functions.isAuthorized(pid_tok, c.ALICE).call(), True, "gate: alice authorized (child 1)") + c.assert_eq(c.policy.functions.isAuthorized(pid_tok, c.DEPLOYER).call(), True, "gate: deployer authorized (child 2)") + c.assert_eq(c.policy.functions.isAuthorized(pid_tok, c.BOB).call(), False, "gate: bob denied (neither child)") + + salt = c.cfg.salt_for("composite-enforce") + params = AssetCreateParams("Composite Gated Asset", "CGATE", c.DEPLOYER, config.ASSET_DECIMALS).encode() + init_calls = [ + init_call(c.asset_abi, "updatePolicy", config.TRANSFER_RECEIVER_POLICY, pid_tok), + init_call(c.asset_abi, "updatePolicy", config.MINT_RECEIVER_POLICY, pid_tok), + init_call(c.asset_abi, "grantRole", config.MINT_ROLE, c.DEPLOYER), + ] + tok_addr = c.predict_b20(config.VARIANT_ASSET, salt) + c.create_b20(config.VARIANT_ASSET, salt, params, init_calls) + ctok = c.asset_at(tok_addr) + c.assert_eq(ctok.functions.policyId(config.MINT_RECEIVER_POLICY).call(), pid_tok, "MINT_RECEIVER_POLICY == composite") + c.assert_eq( + ctok.functions.policyId(config.TRANSFER_RECEIVER_POLICY).call(), pid_tok, "TRANSFER_RECEIVER_POLICY == composite" + ) + + step(29, "composite-authorized receivers: mint + transfer succeed") + c.send(ctok.functions.mint(c.ALICE, config.amt(100, 18)), c.deployer) + c.assert_eq(ctok.functions.balanceOf(c.ALICE).call(), config.amt(100, 18), "alice minted (authorized via child 1)") + c.send(ctok.functions.mint(c.DEPLOYER, config.amt(100, 18)), c.deployer) + c.send(ctok.functions.transfer(c.ALICE, config.amt(1, 18)), c.deployer) + c.assert_eq(ctok.functions.balanceOf(c.ALICE).call(), config.amt(101, 18), "transfer to composite-authorized receiver") + + step(30, "composite-denied receiver on transfer -> PolicyForbids") + c.expect_revert("PolicyForbids", ctok.functions.transfer(c.BOB, config.amt(1, 18)), c.DEPLOYER) + + step(31, "composite-denied receiver on mint -> PolicyForbids") + c.expect_revert("PolicyForbids", ctok.functions.mint(c.BOB, config.amt(1, 18)), c.DEPLOYER) + + def _events(c: Chain) -> None: - step(17, "expected events emitted across the flow") + step(32, "expected events emitted across the flow") c.assert_events_emitted( "policy events", "PolicyCreated(uint64,address,uint8)", "AllowlistUpdated(uint64,address,bool,address[])", "PolicyAdminStaged(uint64,address,address)", "PolicyAdminUpdated(uint64,address,address)", + "CompositePolicyUpdated(uint64,address,uint64[])", "B20Created(address,uint8,string,string,uint8,bytes)", "PolicyUpdated(bytes32,uint64,uint64)", "Transfer(address,address,uint256)", @@ -120,5 +314,6 @@ def run(c: Chain) -> None: pid_b = _journey(c) tok, pid_r = _enforcement(c) _edges(c, tok, pid_r, pid_b) + _composite(c, pid_b) _events(c) log("policy-registry: OK") diff --git a/script/smoke/journeys/precompile_invariants.py b/script/smoke/journeys/precompile_invariants.py index 3570cef..39dd154 100644 --- a/script/smoke/journeys/precompile_invariants.py +++ b/script/smoke/journeys/precompile_invariants.py @@ -19,20 +19,21 @@ from __future__ import annotations -import sys from collections.abc import Callable +from functools import partial from hexbytes import HexBytes from web3 import Web3 from .. import config from ..abis import forcefeeder_artifact, probe_artifact -from ..chain import Chain, log, ok, step +from ..chain import Chain, log, ok, run_collected, step from ..codec import AssetCreateParams, init_call FACTORY = config.B20_FACTORY POLICY = config.POLICY_REGISTRY ALLOWLIST = config.POLICY_TYPE_ALLOWLIST +UNION = config.POLICY_TYPE_UNION # Check names that are known/accepted divergences: reported, but do not fail the run. INFORMATIONAL: set[str] = set() @@ -52,9 +53,28 @@ def _addr_word(address: str) -> bytes: # ── raw calldata edges (no helper contract; web3 emits bytes Solidity wouldn't) ──────────────────── +def _composite_write_calldata(c: Chain) -> list[tuple[str, bytes]]: + """Canonical calldata for the two composite write selectors. + + `msg.value != 0` is the first check in dispatch, ahead of version resolution and selector + matching, so arg values are immaterial and no pre-Cobalt gating is needed. + """ + children = [(ALLOWLIST << 56) | 2, (ALLOWLIST << 56) | 3] + return [ + ("createCompositePolicy", _clean(c.policy, "createCompositePolicy", c.DEPLOYER, UNION, children)), + ("updateComposite", _clean(c.policy, "updateComposite", (UNION << 56) | 4, children)), + ] + + def _payable_rejected(c: Chain, _probe) -> None: create_data = _clean(c.policy, "createPolicy", c.DEPLOYER, ALLOWLIST) c.expect_raw_revert("value on createPolicy", POLICY, create_data, value=1) + # Every policy-registry selector is nonpayable (IPolicyRegistry: the check runs at the top of + # dispatch), but only createPolicy was ever covered. `error_name` is pinned for these two because + # `expect_raw_revert` treats ANY non-ContractLogicError exception as a pass — without it, a + # node-level rejection (insufficient funds, bad gas estimate) would masquerade as NonPayable. + for name, data in _composite_write_calldata(c): + c.expect_raw_revert(f"value on {name}", POLICY, data, value=1, error_name="NonPayable") def _unknown_selector_reverts(c: Chain, _probe) -> None: @@ -93,11 +113,17 @@ def _staticcall_read_only(c: Chain, probe) -> None: def _value_forwarding_rejected(c: Chain, probe) -> None: - create_data = _clean(c.policy, "createPolicy", c.DEPLOYER, ALLOWLIST) overrides = {"from": c.DEPLOYER, "value": 1} + create_data = _clean(c.policy, "createPolicy", c.DEPLOYER, ALLOWLIST) fn = probe.functions.probeCall(POLICY, create_data) res = fn.call(overrides) c.assert_eq(res[0], False, "createPolicy rejects forwarded value", repro_fn=fn, repro_overrides=overrides) + # Same guard, reached from a real contract frame, for the two composite write selectors. + for name, data in _composite_write_calldata(c): + fn = probe.functions.probeCall(POLICY, data) + c.assert_eq( + fn.call(overrides)[0], False, f"{name} rejects forwarded value", repro_fn=fn, repro_overrides=overrides + ) def _returndata_fidelity(c: Chain, probe) -> None: @@ -170,7 +196,7 @@ def _create_gas_independent_of_prefunded_balance(c: Chain, _probe) -> None: # Ordered audit checklist: (name, fn). `name` doubles as the INFORMATIONAL downgrade key. CHECKS: list[tuple[str, Callable[[Chain, object], None]]] = [ - ("payable rejected (value on non-payable createPolicy)", _payable_rejected), + ("payable rejected (value on non-payable createPolicy + composite writes)", _payable_rejected), ("unknown selector reverts (no silent fallthrough)", _unknown_selector_reverts), ("empty calldata reverts (no implicit receive/fallback)", _empty_calldata_reverts), ("truncated args revert (strict ABI decode)", _truncated_args_revert), @@ -185,10 +211,6 @@ def _create_gas_independent_of_prefunded_balance(c: Chain, _probe) -> None: ] -def _detail(exc: SystemExit) -> str: - return str(exc.code or "").replace("[smoke] ERROR: ", "") - - def _setup(c: Chain): step("setup", "deploy PrecompileProbe helper") abi, bytecode = probe_artifact() @@ -200,30 +222,9 @@ def _setup(c: Chain): def run(c: Chain) -> None: log("precompile-invariants: starting (collect-all; findings reported at the end)") probe = _setup(c) - - findings: list[tuple[str, str, bool]] = [] # (name, detail, required) - for i, (name, fn) in enumerate(CHECKS, 1): - step(i, name) - required = name not in INFORMATIONAL - try: - fn(c, probe) - except SystemExit as exc: # assertion/expectation failed inside a check - detail = _detail(exc) - tag = "FINDING" if required else "info" - print(f" \u2717 {tag}: {detail}", file=sys.stderr) - findings.append((name, detail, required)) - except Exception as exc: # noqa: BLE001 - harness/RPC error, surface as a finding - detail = f"{type(exc).__name__}: {exc}" - print(f" \u2717 ERROR: {detail}", file=sys.stderr) - findings.append((name, detail, required)) - - required_fail = [f for f in findings if f[2]] - info_only = [f for f in findings if not f[2]] - log(f"precompile-invariants: {len(CHECKS) - len(findings)}/{len(CHECKS)} invariants held") - for name, detail, _ in findings: - log(f" \u2717 {name} \u2014 {detail}") - if info_only: - log(f"({len(info_only)} accepted divergence(s) reported as informational)") - if required_fail: - raise SystemExit(f"[smoke] precompile-invariants: {len(required_fail)} finding(s) need triage") + run_collected( + [(name, partial(fn, c, probe)) for name, fn in CHECKS], + label="precompile-invariants", + informational=INFORMATIONAL, + ) log("precompile-invariants: OK")