#!/usr/bin/env python3 """Re-derive a CTF settlement coverage manifest from any Polygon RPC endpoint. WHAT THIS IS ------------ A standalone checker for a published coverage manifest. You hand it the manifest JSON and the URL of a Polygon JSON-RPC endpoint you chose yourself. It reads the Gnosis ConditionalTokens ``ConditionResolution`` logs for each declared block range, canonicalises them, hashes them, and compares the hashes to the ones the manifest publishes. If they match, the publisher's store holds exactly the settlement events that endpoint reports over that span, in the same order, with the same decoded values. It is deliberately dependency-free: Python 3.9+ standard library only, no third-party packages, no database, no account, no credential beyond the RPC URL, and no reference to the publisher's own infrastructure. The point of the exercise is that you can reproduce the numbers WITHOUT trusting, or even contacting, the party that published them. A checker that needed the publisher's code would be the publisher asserting completeness in a second voice. Run it from the repository root. The checker and the manifest live in different trees, so a command written relative to either one alone cannot find the other, which is exactly what the first published version of the usage line did. python3 services/api/scripts/evidence/verify_ctf_resolution_coverage.py \ --manifest docs/evidence/ctf-resolution-coverage/2026-08-25-manifest.json \ --rpc-url https://polygon.gateway.tenderly.co \ --from-block 92600000 --to-block 92639309 The endpoint above is named because it is measured to serve the whole span; any Polygon endpoint you trust more is the better choice. See the directory README for what other free endpoints do, and for how to read a red result: MISMATCH is a finding about the manifest, UNVERIFIED is a finding about your endpoint. WHAT A GREEN RUN PROVES ----------------------- For every range it actually read: the publisher's set of ``ConditionResolution`` events over that block span is byte-identical, under the canonical encoding in this file, to what your endpoint serves. That covers presence (nothing missing), absence (nothing extra, including nothing invented), ordering, and - through the payload digest - that the publisher decoded conditionId, oracle, questionId, winning outcome index and the payout vector from those logs the same way this script does. WHAT A GREEN RUN DOES NOT PROVE, and this list is not decoration: * Nothing about blocks OUTSIDE the manifest's declared span. In particular the manifest's floor is an OPERATOR-DECLARED first block. If a resolution exists below it, this script will never look there and will still print PASS. The floor is a separate claim needing its own separate evidence. * Nothing about events on any other contract. This reads exactly one address and exactly one topic0. Conditions resolved through some other mechanism are outside the frame entirely. * Nothing about correctness of the settlements themselves. A ``ConditionResolution`` says an address reported a payout vector. It does not say the report was right, undisputed, or final, and this script cannot know. * Nothing about the publisher's SERVING path. This checks a manifest against a chain. Whether an API hands you the same rows is a different question. * Nothing your endpoint is wrong about. If your provider silently truncates a log response instead of erroring, you will see a mismatch, not a lie - but the mismatch will look like the publisher's fault. Re-run against a second, unrelated provider before concluding anything. The script prints the endpoint it used and the endpoint's chain id and head block for exactly this reason. FAIL-CLOSED ----------- A range this script could not READ is reported ``unverified``, never ``pass``, and makes the whole run exit non-zero. An RPC error is never counted as "zero logs found": that conflation is the specific defect this whole artifact exists to rule out, because a provider that refuses a query by RESPONSE SIZE returns an ERROR where a naive reader sees an empty result and records an empty range. Errors and empties are separated explicitly at the one place it matters, in ``_get_logs_range``. EXIT CODES ---------- 0 every selected range verified and matched 1 at least one selected range mismatched or could not be read 2 the manifest is internally inconsistent (bad tiling, bad root, bad shape) 3 usage or endpoint problem (unreachable, wrong chain, behind the span) """ from __future__ import annotations import argparse import datetime import hashlib import json import os import sys import time import urllib.error import urllib.request from collections.abc import Iterable from typing import Any # ---- the frame being checked --------------------------------------------- #: Gnosis ConditionalTokens on Polygon. Every Polymarket market settles here, #: and so does anything else that uses the same framework, which is why the #: manifest is a CTF index and not a venue index. CONDITIONAL_TOKENS = "0x4d97dcd97ec945f40cf65f87097ace5ea0476045" #: keccak256("ConditionResolution(bytes32,address,bytes32,uint256,uint256[])"). #: topics = [topic0, conditionId, oracle, questionId]; #: data = outcomeSlotCount (word 0) then ABI-encoded uint256[] payoutNumerators. TOPIC_CONDITION_RESOLUTION = ( "0xb44d84d3289691f71497564b85d4233648d9dbae8cbdbb4329f301c3a0185894" ) POLYGON_CHAIN_ID = 137 #: Bumping this changes every digest. It is written into the manifest and checked #: on load, so a manifest produced under a different encoding is refused loudly #: instead of failing as a mismatch on every range. CANONICAL_SPEC = "ctf-resolution-coverage/1" #: sha256 of zero bytes. The digest an EMPTY range must carry. Named because an #: empty range is a positive claim ("we looked here and there was nothing"), and #: it has to be checkable like any other. EMPTY_DIGEST = hashlib.sha256(b"").hexdigest() #: ``datetime.UTC`` is Python 3.11+ and this file's whole premise is that it runs #: on a stock interpreter someone already has. It is verified green under #: /usr/bin/python3 3.9.6, the one that ships with macOS, so the modern alias is #: refused here on purpose rather than because nobody noticed the lint. _UTC = datetime.timezone.utc # noqa: UP017 DEFAULT_CHUNK_BLOCKS = 10_000 MIN_CHUNK_BLOCKS = 1 DEFAULT_TIMEOUT_S = 60.0 DEFAULT_RETRIES = 4 class VerifyError(RuntimeError): """A condition the operator has to resolve; never silently absorbed.""" class RpcRangeRefusedError(RuntimeError): """The endpoint refused this width. Halve and retry; do NOT record an empty.""" class RpcUnavailableError(RuntimeError): """The endpoint could not answer. The range is unverified, never empty.""" # ---- canonical encoding --------------------------------------------------- # # One implementation, used by this checker AND by the publisher's manifest # builder, so the two cannot fork into two encodings under one spec name. Keep # it boring: fixed field order, explicit separators, lowercase hex, decimal # integers, one trailing newline per event. Anything clever here (JSON, # whitespace, locale-dependent formatting) is a future mismatch nobody can debug. def _hexint(value: Any) -> int: """A JSON-RPC quantity to int. Accepts 0x-hex or a plain integer.""" if isinstance(value, int): return value if isinstance(value, str) and value.strip(): return int(value, 16) raise VerifyError(f"cannot read quantity {value!r}") def _bytes32(topic: Any) -> str: """An indexed topic to 0x + 64 lowercase hex, or raise.""" if not isinstance(topic, str): raise VerifyError(f"topic is not a string: {topic!r}") body = topic[2:] if topic.startswith(("0x", "0X")) else topic if len(body) != 64: raise VerifyError(f"topic is not 32 bytes: {topic!r}") int(body, 16) # raises on non-hex return "0x" + body.lower() def _address_from_topic(topic: Any) -> str: """An indexed address topic to 0x + 40 lowercase hex (the low 20 bytes).""" return "0x" + _bytes32(topic)[-40:] def _data_words(data: Any) -> list[str]: """0x-prefixed ABI data split into 32-byte words.""" if not isinstance(data, str): raise VerifyError(f"log data is not a string: {data!r}") body = data[2:] if data.startswith(("0x", "0X")) else data return [body[i : i + 64] for i in range(0, len(body), 64)] def decode_resolution_log(log: dict[str, Any]) -> dict[str, Any]: """One raw ``eth_getLogs`` entry to the fields the manifest commits to. Raises: VerifyError: the log is not a well-formed ``ConditionResolution``. It is raised rather than swallowed because a log this script cannot decode is not a log it may quietly drop: dropping it would shrink the re-derived set and turn a decoder gap into a phantom "publisher has extra rows" finding. """ topics = log.get("topics") or [] if len(topics) != 4: raise VerifyError(f"expected 4 topics, got {len(topics)}") if _bytes32(topics[0]) != TOPIC_CONDITION_RESOLUTION: raise VerifyError(f"unexpected topic0 {topics[0]!r}") words = _data_words(log.get("data") or "0x") numerators: list[int] = [] if len(words) >= 3: # word0 = outcomeSlotCount; word1 = offset to the array; word2 = length; # words[3:3+length] = the numerators themselves. arr_len = int(words[2], 16) for i in range(arr_len): idx = 3 + i if idx < len(words): numerators.append(int(words[idx], 16)) winning_index = ( numerators.index(max(numerators)) if numerators and any(numerators) else None ) tx_hash = log.get("transactionHash") if not isinstance(tx_hash, str) or len(tx_hash) != 66: raise VerifyError(f"bad transactionHash {tx_hash!r}") return { "block_number": _hexint(log.get("blockNumber")), "tx_hash": tx_hash.lower(), "log_index": _hexint(log.get("logIndex")), "condition_id": _bytes32(topics[1]), "oracle": _address_from_topic(topics[2]), "question_id": _bytes32(topics[3]), "winning_index": winning_index, "payout_numerators": numerators, } def sort_key(event: dict[str, Any]) -> tuple[int, int, str]: """The total order the digests are taken in. ``(block_number, log_index)`` is already unique on an EVM chain, since logIndex is per-block and spans every transaction in the block. ``tx_hash`` is appended anyway so the order stays total even on a malformed input where that uniqueness does not hold - a set with a duplicated coordinate would otherwise hash differently depending on which copy the sorter happened to put first, which would make a genuine mismatch look like flakiness. """ return (event["block_number"], event["log_index"], event["tx_hash"]) def identity_line(event: dict[str, Any]) -> bytes: """block | txHash | logIndex | conditionId The tuple that answers "is it the same set of settlement events". Deliberately excludes every decoded value, so a decoder disagreement cannot be mistaken for a coverage gap. """ return ( f"{event['block_number']}|{event['tx_hash']}|" f"{event['log_index']}|{event['condition_id']}\n" ).encode() def payload_line(event: dict[str, Any]) -> bytes: """identity fields, then oracle | questionId | winningIndex | numerators ``winningIndex`` is argmax over the payout vector and empty when no numerator is positive. It is derived, so committing to it is redundant as data and valuable as a test: it is the one place the publisher's decode of the payout vector becomes checkable rather than merely restated. ``outcomeSlotCount`` (data word 0) is NOT in the line. The publisher does not persist it, and committing to a field the publisher would have to re-derive at manifest time would make the digest a statement about the manifest builder rather than about the store. """ nums = ",".join(str(n) for n in event["payout_numerators"]) win = "" if event["winning_index"] is None else str(event["winning_index"]) return ( f"{event['block_number']}|{event['tx_hash']}|" f"{event['log_index']}|{event['condition_id']}|" f"{event['oracle']}|{event['question_id']}|{win}|{nums}\n" ).encode() def range_digests(events: Iterable[dict[str, Any]]) -> tuple[str, str, int, Any, Any]: """(identity_digest, payload_digest, n, first_block, last_block). The caller must pass events ALREADY sorted by ``sort_key``; this does not sort, so that the sort is one decision made in one place at the call sites. """ ident = hashlib.sha256() payload = hashlib.sha256() n = 0 first_block: Any = None last_block: Any = None for ev in events: ident.update(identity_line(ev)) payload.update(payload_line(ev)) n += 1 if first_block is None: first_block = ev["block_number"] last_block = ev["block_number"] return ident.hexdigest(), payload.hexdigest(), n, first_block, last_block def root_line(entry: dict[str, Any]) -> bytes: """One manifest range, as it is committed to the root hash. Every published field of a range is in here. A field a reader can see but the root does not cover is a field an intermediary can edit without detection, which is the whole failure mode a root exists to close. """ first = "" if entry.get("first_block") is None else str(entry["first_block"]) last = "" if entry.get("last_block") is None else str(entry["last_block"]) return ( f"{entry['from_block']}:{entry['to_block']}:{entry['n']}:" f"{first}:{last}:{entry['identity_digest']}:{entry['payload_digest']}\n" ).encode() def manifest_root(ranges: list[dict[str, Any]]) -> str: """sha256 over every range's root line, in published order.""" h = hashlib.sha256() for entry in ranges: h.update(root_line(entry)) return h.hexdigest() # ---- manifest self-consistency ------------------------------------------- def check_manifest(manifest: dict[str, Any]) -> list[str]: """Everything checkable about a manifest WITHOUT touching a chain. Run before any RPC call. A manifest that does not tile its own declared span, or whose root does not recompute, is not worth spending an hour of somebody else's bandwidth on, and its failure mode should not be reported as "the chain disagrees". Returns: A list of problems, empty when the manifest is internally sound. """ problems: list[str] = [] spec = manifest.get("canonical_spec") if spec != CANONICAL_SPEC: problems.append( f"canonical_spec is {spec!r}, this checker implements {CANONICAL_SPEC!r}" ) if manifest.get("contract_address", "").lower() != CONDITIONAL_TOKENS: problems.append( f"contract_address {manifest.get('contract_address')!r} is not the CTF " f"address this checker reads ({CONDITIONAL_TOKENS})" ) if manifest.get("event_topic0", "").lower() != TOPIC_CONDITION_RESOLUTION: problems.append( f"event_topic0 {manifest.get('event_topic0')!r} is not ConditionResolution" ) if manifest.get("chain_id") != POLYGON_CHAIN_ID: problems.append(f"chain_id {manifest.get('chain_id')!r} is not {POLYGON_CHAIN_ID}") ranges = manifest.get("ranges") if not isinstance(ranges, list) or not ranges: problems.append("ranges is missing or empty") return problems floor = manifest.get("floor_block") through = manifest.get("swept_through_block") if not isinstance(floor, int) or not isinstance(through, int) or floor > through: problems.append(f"declared span [{floor}, {through}] is not a valid interval") return problems # Tiling: no gap, no overlap, exact cover of [floor, through]. A gap is a # block nobody claims to have read while the header claims the whole span. cursor = floor total = 0 for i, entry in enumerate(ranges): for key in ( "from_block", "to_block", "n", "identity_digest", "payload_digest", ): if key not in entry: problems.append(f"range {i} is missing {key}") return problems if entry["from_block"] != cursor: problems.append( f"range {i} starts at {entry['from_block']}, expected {cursor} " "(gap or overlap in the tiling)" ) return problems if entry["to_block"] < entry["from_block"]: problems.append(f"range {i} is inverted: {entry}") return problems if entry["n"] == 0: if entry["identity_digest"] != EMPTY_DIGEST: problems.append( f"range {i} declares n=0 but its identity_digest is not the " "sha256 of the empty string" ) if entry.get("first_block") is not None or entry.get("last_block") is not None: problems.append(f"range {i} declares n=0 but carries block bounds") else: fb, lb = entry.get("first_block"), entry.get("last_block") if not isinstance(fb, int) or not isinstance(lb, int): problems.append(f"range {i} has n>0 but no first/last block") elif not (entry["from_block"] <= fb <= lb <= entry["to_block"]): problems.append( f"range {i} block bounds [{fb}, {lb}] fall outside the range" ) total += entry["n"] cursor = entry["to_block"] + 1 if cursor != through + 1: problems.append( f"ranges cover up to {cursor - 1}, declared swept_through_block is {through}" ) declared_total = manifest.get("total_events") if declared_total is not None and declared_total != total: problems.append( f"total_events says {declared_total}, ranges sum to {total}" ) declared_root = manifest.get("manifest_root") computed_root = manifest_root(ranges) if declared_root != computed_root: problems.append( f"manifest_root {declared_root!r} does not recompute; got {computed_root!r}" ) return problems # ---- JSON-RPC ------------------------------------------------------------- def _post(url: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]: body = json.dumps(payload).encode() req = urllib.request.Request( url, data=body, headers={ "Content-Type": "application/json", "Accept": "application/json", # Named so an endpoint operator seeing this traffic can tell what it # is. An anonymous flood of getLogs is what gets a shared public # endpoint to start refusing. "User-Agent": "ctf-resolution-coverage-verifier/1", }, ) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read()) _WIDTH_REFUSAL_MARKERS = ( "response size exceeded", "query returned more than", "log response size", "range is too large", "block range is too wide", "exceed maximum block range", "limit exceeded", "too many results", "query timeout exceeded", "range too large", # QuickNode's public Polygon endpoint, measured 2026-08-25: "eth_getLogs is # limited to a 10,000 range". It arrives as HTTP 413 too, which this module # already treats as a refusal, but relying on the status code alone would # leave the classifier blind to any provider that words it this way over a # plain 400. "is limited to a", # dRPC's free Polygon plan, measured 2026-08-25 from this machine: HTTP 400, # `{"message": "ranges over 10000 blocks are not supported on free plan", # "code": 35}`. It is a genuine WIDTH refusal and not a blanket rejection: # the same query was refused at widths 10000, 5000, 1000, 500, 300, 200 and # 128 and SERVED at 100 and 10. So the message's own "10000" is not the # limit that is actually enforced, and a reader who trusts the wording will # conclude the endpoint is broken. Classifying it correctly is still right, # because it is what it says it is; whether following it is PRACTICAL is a # different question and the README answers it (at ~100 blocks per call this # manifest needs on the order of a million calls, so dRPC free is not a # usable endpoint for a full pass even though the classifier now handles it). "ranges over", "-32005", ) def _looks_like_width_refusal(message: str) -> bool: """Is this the endpoint saying "that window is too big", or a real failure? The distinction is the whole point of the module docstring's fail-closed paragraph. Providers signal width refusal through an ERROR, and every one of them words it differently, so this is a substring list rather than a code check. When in doubt the answer must be False: treating a real failure as a width refusal makes the script halve its way down to single blocks and then report a range unverified, which is noisy but safe, while the reverse (treating a width refusal as fatal) merely stops early. Neither direction can produce a false PASS, because neither produces an empty result. """ low = message.lower() return any(marker in low for marker in _WIDTH_REFUSAL_MARKERS) def rpc( url: str, method: str, params: list[Any], *, timeout: float = DEFAULT_TIMEOUT_S, retries: int = DEFAULT_RETRIES, sleep_ms: int = 0, ) -> Any: """One JSON-RPC call, with backoff on transport and rate-limit failures. Raises: RpcRangeRefusedError: the endpoint refused the query's width. RpcUnavailableError: the endpoint could not answer after ``retries`` attempts. """ last: str = "no attempt made" for attempt in range(retries): if sleep_ms: time.sleep(sleep_ms / 1000.0) try: out = _post(url, {"jsonrpc": "2.0", "id": 1, "method": method, "params": params}, timeout) except urllib.error.HTTPError as exc: # noqa: PERF203 detail = "" try: detail = exc.read().decode("utf-8", "replace")[:400] except Exception: # noqa: BLE001 pass last = f"HTTP {exc.code}: {detail or exc.reason}" if _looks_like_width_refusal(last) or exc.code == 413: raise RpcRangeRefusedError(last) from exc if exc.code not in (429, 500, 502, 503, 504): raise RpcUnavailableError(last) from exc except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as exc: last = f"{type(exc).__name__}: {exc}" else: if isinstance(out, dict) and out.get("error"): message = json.dumps(out["error"]) if _looks_like_width_refusal(message): raise RpcRangeRefusedError(message) raise RpcUnavailableError(f"rpc error: {message}") if not isinstance(out, dict) or "result" not in out: raise RpcUnavailableError(f"malformed response: {str(out)[:200]}") return out["result"] time.sleep(min(2**attempt, 8)) raise RpcUnavailableError(f"{method} failed after {retries} attempts: {last}") def _get_logs_range( url: str, from_block: int, to_block: int, *, chunk_blocks: int, sleep_ms: int, timeout: float, call_counter: list[int], ) -> list[dict[str, Any]]: """Every ConditionResolution log in [from_block, to_block], inclusive. Chunks the span and halves the chunk on a width refusal, exactly like a production sweep has to. The invariant that matters: this function returns a list ONLY when every block in the span was actually read. Any window that could not be read raises. There is no code path here where a failure becomes an empty list, because an empty list is itself an assertion - "there is nothing here" - and it must only ever be made by an endpoint that answered. """ out: list[dict[str, Any]] = [] start = from_block width = max(MIN_CHUNK_BLOCKS, chunk_blocks) while start <= to_block: end = min(start + width - 1, to_block) try: call_counter[0] += 1 logs = rpc( url, "eth_getLogs", [ { "address": CONDITIONAL_TOKENS, "topics": [TOPIC_CONDITION_RESOLUTION], "fromBlock": hex(start), "toBlock": hex(end), } ], timeout=timeout, sleep_ms=sleep_ms, ) except RpcRangeRefusedError: if width <= MIN_CHUNK_BLOCKS: raise width = max(MIN_CHUNK_BLOCKS, width // 2) continue if not isinstance(logs, list): raise RpcUnavailableError(f"eth_getLogs returned {type(logs).__name__}, not a list") out.extend(logs) start = end + 1 # Re-widen slowly after a refusal so one dense window does not force the # whole remaining span through at the narrow width. if width < chunk_blocks: width = min(chunk_blocks, width * 2) return out # ---- verification --------------------------------------------------------- def verify_range( url: str, entry: dict[str, Any], *, chunk_blocks: int, sleep_ms: int, timeout: float, call_counter: list[int], ) -> dict[str, Any]: """Re-derive one manifest range and compare. Never raises on a mismatch.""" result: dict[str, Any] = { "from_block": entry["from_block"], "to_block": entry["to_block"], "manifest_n": entry["n"], "status": "unverified", } try: raw = _get_logs_range( url, entry["from_block"], entry["to_block"], chunk_blocks=chunk_blocks, sleep_ms=sleep_ms, timeout=timeout, call_counter=call_counter, ) except (RpcRangeRefusedError, RpcUnavailableError) as exc: result["reason"] = f"endpoint could not serve this range: {exc}" return result removed = sum(1 for lg in raw if lg.get("removed") is True) try: events = [decode_resolution_log(lg) for lg in raw if lg.get("removed") is not True] except VerifyError as exc: result["reason"] = f"undecodable log in this range: {exc}" return result events.sort(key=sort_key) keys = [sort_key(ev) for ev in events] duplicates = len(keys) - len(set(keys)) ident, payload, n, first_block, last_block = range_digests(events) result.update( { "observed_n": n, "removed_logs_skipped": removed, "duplicate_coordinates": duplicates, "observed_identity_digest": ident, "observed_payload_digest": payload, } ) identity_ok = ident == entry["identity_digest"] payload_ok = payload == entry["payload_digest"] bounds_ok = first_block == entry.get("first_block") and last_block == entry.get("last_block") if identity_ok and payload_ok and bounds_ok and n == entry["n"]: result["status"] = "pass" return result result["status"] = "mismatch" detail = [] if n != entry["n"]: detail.append(f"count: manifest {entry['n']}, endpoint {n}") if not identity_ok: detail.append("identity digest differs") if not payload_ok: detail.append( "payload digest differs" + (" (identity matches, so this is a DECODE disagreement, not a coverage gap)" if identity_ok else "") ) if not bounds_ok: detail.append( f"block bounds: manifest [{entry.get('first_block')}, {entry.get('last_block')}], " f"endpoint [{first_block}, {last_block}]" ) result["reason"] = "; ".join(detail) return result def file_provenance(path: str) -> dict[str, Any]: """Identify one file exactly: its bytes, its size, and when it last changed. WHY THE REPORT CARRIES THIS. Five verification reports were published before this existed, and a reviewer was able to prove that four of them came from a revision of this checker that no longer exists, purely because they lack a field the current code always writes. That is the right conclusion from the wrong kind of evidence: it worked by accident, and only in the direction of doubt. A report that names the bytes that produced it needs no archaeology. A hash is recorded even though a comment-only edit churns it. The objection to hashing SOURCE is about pinning it in a document as a thing readers must match; recording it inside a run's own report is the opposite, since nothing ever has to equal it and its only job is to say which generation ran. Fail-loud rather than fail-absent: an unreadable file records the error under the same key, because a MISSING key is exactly the ambiguity this closes. The FULL PATH is deliberately not recorded, only the basename. These reports are published for strangers, and an absolute path publishes the operator's home directory and working-tree layout to everyone who opens the artifact while adding nothing a reader can use: the sha256 is what identifies a generation, and anyone matching it hashes their own copy. """ try: with open(path, "rb") as fh: blob = fh.read() except OSError as exc: # strerror, not str(exc): an OSError stringifies WITH the filename, which # would put the absolute path back into a published report through the # error branch. The test that pins this caught exactly that. return { "name": os.path.basename(path), "error": f"{type(exc).__name__}: {exc.strerror or 'unreadable'}", } return { "name": os.path.basename(path), "sha256": hashlib.sha256(blob).hexdigest(), "size_bytes": len(blob), "mtime_utc": datetime.datetime.fromtimestamp( os.path.getmtime(path), _UTC ).strftime("%Y-%m-%dT%H:%M:%SZ"), } def verdict_note(passed: int, mismatched: int, unverified: int) -> str: """The sentence printed after the counts, distinguishing the two red states. The two are not the same finding and must never read the same way: * MISMATCH means the endpoint and the manifest DISAGREE about what is on chain. That is a claim about the publisher, and it is the finding this artifact exists to make possible. * UNVERIFIED means the endpoint did not answer. That is a claim about the endpoint, and it says nothing whatsoever about the publisher. A run that is entirely unverified is the common first-attempt outcome on a free endpoint that prunes history or caps eth_getLogs hard, and the default reading of a red result is "the publisher is wrong". Saying which one happened, in the output rather than only in a README, is the difference between a third party retrying on another provider and a third party concluding the receipt is broken. """ if mismatched: return ( f"{mismatched} range(s) MISMATCHED: this endpoint and the manifest disagree " "about what is on chain. That is a finding about the manifest. Re-run against " "a second, unrelated provider before concluding, then report it." ) if unverified and passed == 0: return ( "NOTHING was verified and NOTHING mismatched: every selected range was " "unverified, so this endpoint could not serve the query at all (pruned " "history, a hard eth_getLogs cap, or rate limiting). This says nothing about " "the manifest. Try another endpoint, or a smaller --chunk-blocks." ) if unverified: return ( f"{unverified} range(s) UNVERIFIED and 0 mismatched: this endpoint could not " "serve part of the span. Nothing here contradicts the manifest; the unread " "ranges are simply unread. Re-run those ranges elsewhere." ) return "every selected range was read and matched." def select_ranges( ranges: list[dict[str, Any]], from_block: int | None, to_block: int | None, stride: int = 1, ) -> list[dict[str, Any]]: """The ranges a --from-block / --to-block / --stride selection asks for. Whole ranges only. A partially-selected range cannot be checked, because the digest is over the range's full contents, so a half-read range would report a mismatch that means nothing. Selecting nothing is an error rather than a vacuous pass. ``stride`` takes every Nth selected range, which is how you get systematic coverage across the whole span for a small fraction of the calls: 1-in-40 over this manifest is 22 ranges and a few hundred calls rather than several thousand. It is deliberately a SAMPLE and the run reports itself as partial, because a strided pass cannot find a problem in a range it did not read. Use it to gain confidence cheaply, not to conclude. """ lo = from_block if from_block is not None else -1 hi = to_block if to_block is not None else (1 << 62) picked = [r for r in ranges if r["from_block"] >= lo and r["to_block"] <= hi] if stride > 1: picked = picked[::stride] return picked def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser( description="Re-derive a CTF settlement coverage manifest from a Polygon RPC endpoint.", ) ap.add_argument("--manifest", required=True, help="path to the manifest JSON") ap.add_argument( "--rpc-url", required=True, help="any Polygon JSON-RPC endpoint that serves eth_getLogs over the span", ) ap.add_argument("--from-block", type=int, default=None) ap.add_argument("--to-block", type=int, default=None) ap.add_argument( "--stride", type=int, default=1, help="verify every Nth range in the window (a sample; the run reports partial)", ) ap.add_argument("--chunk-blocks", type=int, default=DEFAULT_CHUNK_BLOCKS) ap.add_argument( "--sleep-ms", type=int, default=0, help="pause before each RPC call; raise this on a shared public endpoint", ) ap.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT_S) ap.add_argument("--report", default=None, help="write the full JSON report here") ap.add_argument( "--skip-endpoint-checks", action="store_true", help="do not ask the endpoint for chain id and head block first", ) args = ap.parse_args(argv) try: with open(args.manifest, "rb") as fh: manifest_blob = fh.read() manifest = json.loads(manifest_blob) except (OSError, json.JSONDecodeError) as exc: print(f"FAIL: cannot read manifest: {exc}", file=sys.stderr) return 3 # Hashed from the bytes actually read, not re-read later, so the report # cannot end up naming a file that changed underneath the run. manifest_sha256 = hashlib.sha256(manifest_blob).hexdigest() problems = check_manifest(manifest) if problems: print("FAIL: the manifest is internally inconsistent, before any chain read:") for p in problems: print(f" - {p}") return 2 print( f"manifest OK: {len(manifest['ranges'])} ranges tiling " f"[{manifest['floor_block']}, {manifest['swept_through_block']}], " f"{manifest.get('total_events')} events, root {manifest['manifest_root']}" ) call_counter = [0] if not args.skip_endpoint_checks: try: chain_id = _hexint(rpc(args.rpc_url, "eth_chainId", [], timeout=args.timeout)) head = _hexint(rpc(args.rpc_url, "eth_blockNumber", [], timeout=args.timeout)) except (RpcUnavailableError, RpcRangeRefusedError, VerifyError) as exc: print(f"FAIL: endpoint unusable: {exc}", file=sys.stderr) return 3 call_counter[0] += 2 print(f"endpoint: {args.rpc_url} chain_id={chain_id} head={head}") if chain_id != POLYGON_CHAIN_ID: print(f"FAIL: endpoint is chain {chain_id}, not Polygon ({POLYGON_CHAIN_ID})") return 3 if head < manifest["swept_through_block"]: print( f"FAIL: endpoint head {head} is behind the manifest span end " f"{manifest['swept_through_block']}; it cannot serve the whole claim" ) return 3 if args.stride < 1: print("FAIL: --stride must be 1 or more", file=sys.stderr) return 3 selected = select_ranges( manifest["ranges"], args.from_block, args.to_block, args.stride ) if not selected: print("FAIL: the --from-block/--to-block window selects no whole range", file=sys.stderr) return 3 partial = len(selected) != len(manifest["ranges"]) print( f"verifying {len(selected)} of {len(manifest['ranges'])} ranges " f"[{selected[0]['from_block']}, {selected[-1]['to_block']}]" ) results: list[dict[str, Any]] = [] started = time.time() passed = mismatched = unverified = 0 for i, entry in enumerate(selected, 1): res = verify_range( args.rpc_url, entry, chunk_blocks=args.chunk_blocks, sleep_ms=args.sleep_ms, timeout=args.timeout, call_counter=call_counter, ) results.append(res) if res["status"] == "pass": passed += 1 elif res["status"] == "mismatch": mismatched += 1 print( f" MISMATCH [{res['from_block']}, {res['to_block']}]: {res.get('reason')}" ) else: unverified += 1 print( f" UNVERIFIED [{res['from_block']}, {res['to_block']}]: {res.get('reason')}" ) if i % 25 == 0 or i == len(selected): print( f" ... {i}/{len(selected)} ranges, {passed} pass, {mismatched} mismatch, " f"{unverified} unverified, {call_counter[0]} rpc calls, " f"{time.time() - started:.0f}s" ) report = { "checker": CANONICAL_SPEC, "generated_at": datetime.datetime.now(_UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), # WHAT RAN, on WHAT BYTES, with WHICH ARGUMENTS. Every one of these was # absent from the first five reports published from this script, and the # gap was load-bearing: a reader could not tell whether a report came # from the committed checker, nor which --chunk-blocks produced its call # count, nor whether the manifest on disk today is the one that was read. "invocation": { "from_block": args.from_block, "to_block": args.to_block, "stride": args.stride, "chunk_blocks": args.chunk_blocks, "sleep_ms": args.sleep_ms, "timeout_s": args.timeout, "skip_endpoint_checks": bool(args.skip_endpoint_checks), }, "checker_source": file_provenance(__file__), "manifest_sha256": manifest_sha256, "rpc_url": args.rpc_url, "manifest_root": manifest["manifest_root"], "manifest_path": args.manifest, "ranges_selected": len(selected), "ranges_in_manifest": len(manifest["ranges"]), "partial_selection": partial, "stride": args.stride, "passed": passed, "mismatched": mismatched, "unverified": unverified, "rpc_calls": call_counter[0], "elapsed_s": round(time.time() - started, 1), "results": results, } if args.report: with open(args.report, "w", encoding="utf-8") as fh: json.dump(report, fh, indent=2, sort_keys=True) print(f"report written to {args.report}") print( f"\n{passed} passed, {mismatched} mismatched, {unverified} unverified " f"in {call_counter[0]} RPC calls over {report['elapsed_s']}s" ) print(verdict_note(passed, mismatched, unverified)) if mismatched or unverified: print("RESULT: FAIL") return 1 if partial: print( "RESULT: PASS for the selected window only. The manifest's remaining " f"{len(manifest['ranges']) - len(selected)} ranges were not read and this " "run says nothing about them." ) else: print("RESULT: PASS for every range in the manifest.") return 0 if __name__ == "__main__": # pragma: no cover sys.exit(main())