#!/usr/bin/env python3 """price.py: what a 402 actually costs, resolved from the asset rather than read off the server's own description of itself. Written at wake dc7d9337 (2026-09-23), hunting the fifth member of a family this toolbox has now been bitten by four times: a check that reads the cheap half of an answer and reports the whole. A 200 is not a date (fresh.py, #131). A status line is not the only way to be alive (#132). A 200 is not a page (soft404.py, #135). One handshake is not access (drift.py, #138). And the one this instrument is for: A 402 IS NOT A PRICE. `payments.py`, my own instrument, ends every payable row with "OK 402 live". It reaches that verdict from a status code and a list of protocol names, and it prints `scheme`, `network`, `amount` and `asset` as its evidence. Two things are wrong with that evidence and both are invisible to its reader: 1. It never reads `payTo`. A challenge with no destination is not payable by anyone, and nothing in the report would have said so. 2. `amount` is an integer in the ATOMIC UNITS of `asset`, and the challenge does not carry the asset's decimals. "2000" is $0.002 if the asset has 6 decimals and $0.000000000000002 if it has 18. Every dollar figure in this registry's payment notes is a conversion somebody did in their head from a contract address they recognised. So the price of a call rests on a fact no party to the exchange states: how many decimals the token at that address has. The server asserts a friendly `extra.name` ("USD Coin"); that is the server describing itself, on exactly the axis a dishonest or mistaken server would describe itself wrongly. The chain will answer the same question for free. WHAT THIS DOES. For every `payments` block claiming `machinePayable`, it asks each GET-able endpoint the registry names, parses the x402 challenge, and for every `accepts` entry resolves the asset against the network's own public RPC (`decimals()` and `symbol()` for EVM, `getTokenSupply` for Solana). Then it states the price in the asset's own units, and compares that with the dollar figures the entry's own notes claim, as a CANDIDATE and never a finding: this test reads numbers out of prose, and prose can name a figure in order to deny it (the defect drift.py shipped and had to fix a day later). python3 tools/price.py --clone /tmp/reg python3 tools/price.py --selftest # no network, no clone VERDICTS, one per challenge: PRICED at least one accept carries scheme, network, amount, asset AND payTo, and the asset resolved on its own chain. A machine has everything it needs. UNPRICED a challenge document exists and no accept is both complete and resolvable. The missing field or the unresolved network is named. STATUS a 402 with no challenge document at all: money-shaped, and there is nothing in it to pay. Firecrawl's out-of-credits 402 is this shape, and its entry correctly says machinePayable: false. THE CONTROLS, both of which must pass before a single price is printed: - example.com must not answer a payment challenge (the network is sane); - a known NON-TOKEN address must fail to resolve as a token. An EVM node answers `eth_call` on an address with no code with an empty result, and an instrument that reads `0x` as zero prints DECIMALS 0 -- which would turn 30000 atomic units into 30000 dollars. That is the soft404 defect exactly (a failure to measure read as a measurement), so the resolver refuses any result that is not 32 bytes, and the control proves it refuses. DISCLOSED LIMITS. Only GET-able endpoints are asked; a bare POST to an MCP server is a different request from the one the record made. Only the networks in RPCS resolve; anything else is UNRESOLVED and says so rather than being dropped. A resolved symbol is what the CONTRACT says it is, which is better evidence than what the server says it is and is still not proof the token is the real USDC: that is a claim about which address the market treats as canonical, and no single RPC call settles it. """ import argparse import base64 import json import os import re import sys import urllib.error import urllib.request from decimal import Decimal sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import corpus # noqa: E402 UA = {"User-Agent": "lintel-price (Public Agents registry, cto agent)"} CONTROL_URL = "https://example.com/" # Public, unauthenticated JSON-RPC, one per CAIP-2 network seen in this # registry's live challenges. Measured reachable from this container at # 2026-09-23T10:07Z, except eip155:137, whose public endpoint answers 401. RPCS = { "eip155:8453": ("evm", "https://mainnet.base.org"), "eip155:42161": ("evm", "https://arb1.arbitrum.io/rpc"), "eip155:196": ("evm", "https://rpc.xlayer.tech"), "eip155:137": ("evm", "https://polygon-bor-rpc.publicnode.com"), "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp": ("svm", "https://api.mainnet-beta.solana.com"), } DECIMALS_SEL = "0x313ce567" # decimals() SYMBOL_SEL = "0x95d89b41" # symbol() SETTLEMENT_FIELDS = ("scheme", "network", "amount", "asset", "payTo") # A dollar figure as an entry's notes write one: $0.002, $0.01, $1. MONEY_RE = re.compile(r"\$([0-9]+(?:\.[0-9]+)?)") def http(url, method="GET", body=None, headers=None, timeout=20): req = urllib.request.Request(url, data=body, method=method, headers={**UA, **(headers or {})}) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, dict(r.headers), r.read(20000) except urllib.error.HTTPError as e: return e.code, dict(e.headers), e.read(20000) except Exception as e: return None, {"error": str(e)[:150]}, b"" def rpc(url, payload, timeout=20): st, _, body = http(url, "POST", json.dumps(payload).encode(), {"Content-Type": "application/json"}, timeout) if st != 200: return None try: return json.loads(body) except Exception: return None def decode_word(hexstr): """One 32-byte ABI word as an int, or None. THE RULE THIS INSTRUMENT EXISTS TO GET RIGHT: an address with no contract answers `0x`, and a node under load answers an error object. Neither is a zero. Anything that is not exactly 32 bytes of hex is NOT A READING. """ if not isinstance(hexstr, str) or not hexstr.startswith("0x"): return None h = hexstr[2:] if len(h) != 64: return None try: return int(h, 16) except ValueError: return None def decode_string(hexstr): """An ABI-encoded dynamic string (offset, length, bytes), or None.""" if not isinstance(hexstr, str) or not hexstr.startswith("0x"): return None h = hexstr[2:] if len(h) < 128: # Some old tokens return a bytes32 symbol rather than a string. if len(h) == 64: try: return bytes.fromhex(h).rstrip(b"\x00").decode("utf-8", "replace") or None except ValueError: return None return None try: length = int(h[64:128], 16) raw = bytes.fromhex(h[128:128 + length * 2]) return raw.decode("utf-8", "replace") except (ValueError, IndexError): return None def resolve_asset(network, asset, cache=None): """(decimals, symbol, note) for an asset on a named chain, or Nones. Never guesses. An unknown network, an unreachable node or a reply that is not a 32-byte word all come back as `decimals=None` with the reason in the note, because a price computed from a guessed decimals is worse than no price at all. """ if cache is not None and (network, asset) in cache: return cache[(network, asset)] out = (None, None, "") kind_url = RPCS.get(network) if not kind_url: out = (None, None, "no public RPC configured for %s" % network) else: kind, url = kind_url if kind == "evm": if not re.fullmatch(r"0x[0-9a-fA-F]{40}", asset or ""): out = (None, None, "asset %r is not an EVM address" % (asset,)) else: d = rpc(url, {"jsonrpc": "2.0", "id": 1, "method": "eth_call", "params": [{"to": asset, "data": DECIMALS_SEL}, "latest"]}) dec = decode_word((d or {}).get("result")) if dec is None or dec > 36: out = (None, None, "no 32-byte decimals() answer from %s (not a token here)" % url) else: s = rpc(url, {"jsonrpc": "2.0", "id": 1, "method": "eth_call", "params": [{"to": asset, "data": SYMBOL_SEL}, "latest"]}) out = (dec, decode_string((s or {}).get("result")), "") else: d = rpc(url, {"jsonrpc": "2.0", "id": 1, "method": "getTokenSupply", "params": [asset]}) val = ((d or {}).get("result") or {}).get("value") or {} dec = val.get("decimals") if isinstance(dec, int) and 0 <= dec <= 36: out = (dec, None, "symbol not read on this chain (no metadata call)") else: out = (None, None, "getTokenSupply gave no decimals (not a mint here)") if cache is not None: cache[(network, asset)] = out return out def human(amount, decimals): """Atomic units to the asset's own units. Refuses anything unreadable.""" if decimals is None: return None try: n = Decimal(str(amount)) except Exception: return None if n != n.to_integral_value(): return None return n / (Decimal(10) ** decimals) def parse_challenge(status, headers, body): """The x402 document a 402 carries, from the header or the body, or None.""" if status != 402: return None doc = None pr = headers.get("payment-required") or headers.get("Payment-Required") if pr: try: doc = json.loads(base64.b64decode(pr + "=" * (-len(pr) % 4))) except Exception: doc = None if not isinstance(doc, dict): try: doc = json.loads(body.decode("utf-8", "replace")) except Exception: doc = None if not isinstance(doc, dict): return {"accepts": [], "raw": True} accepts = doc.get("accepts") return {"accepts": accepts if isinstance(accepts, list) else [], "doc": doc} def grade(challenge, cache=None, resolver=resolve_asset): """PRICED / UNPRICED / STATUS for one challenge, with the reason. `resolver` is injected so --selftest can falsify every rule here with no network at all. """ if challenge is None: return None accepts = [a for a in challenge.get("accepts") or [] if isinstance(a, dict)] if not accepts: return {"verdict": "STATUS", "quotes": [], "why": "a 402 with no accepts document: it says money and names nothing to pay"} quotes, problems = [], [] for a in accepts: amount = a.get("amount", a.get("maxAmountRequired")) missing = [f for f in SETTLEMENT_FIELDS if not (a.get(f) if f != "amount" else amount)] if missing: problems.append("an accept on %s is missing %s" % (a.get("network"), ",".join(missing))) continue dec, sym, note = resolver(a["network"], a["asset"], cache) value = human(amount, dec) if value is None: problems.append("an accept on %s names %s and %s" % (a.get("network"), amount, note or "the amount is not an integer")) continue quotes.append({"network": a["network"], "asset": a["asset"], "payTo": a["payTo"], "scheme": a.get("scheme"), "atomic": str(amount), "decimals": dec, "symbol": sym, "value": str(value), "note": note}) if quotes: return {"verdict": "PRICED", "quotes": quotes, "why": "", "problems": problems} return {"verdict": "UNPRICED", "quotes": [], "why": "; ".join(problems) or "no accept could be priced", "problems": problems} def candidates(clone, tool_id): """GET-able endpoints the registry itself names, from probe records.""" out = [] pdir = os.path.join(clone, "registry", "evidence", "probes") for name in sorted(os.listdir(pdir)) if os.path.isdir(pdir) else []: if not name.endswith(".json"): continue e = json.load(open(os.path.join(pdir, name), encoding="utf-8")) if (e.get("subject") or {}).get("id") != tool_id: continue if ((e.get("request") or {}).get("method", "GET")).upper() != "GET": continue if e.get("surface"): out.append({"url": e["surface"], "record": e.get("id"), "question": e.get("question"), "entry": e}) return out def discloses_empty_accepts(record): """Does the registry's own record for this surface already say the 402 carries nothing payable? drift.py's rule, applied here: a disagreement the record already states is DISCLOSED, not drift. A re-find that cannot say "since when" is worth less than one that can, and this sweep must not report as news something a filed probe said first. It reads prose, so it is a reason to look and never a proof: every sentence it matched is printed under the row. """ if not record: return [] blob = record.get("observed") or {} said = [] for key, val in (blob.get("decoded") or {}).items(): if isinstance(val, str) and re.search(r"empty\s+accepts|accepts.{0,20}empty|nothing .{0,20}to pay", val, re.I): said.append("%s: %s" % (key, val)) fin = record.get("finding") or "" if re.search(r"empty\s+accepts|nothing payable|offers nothing payable", fin, re.I): said.append("finding: " + fin) return said def quote_key(g): """What a client would cache from a challenge: where to pay and how much.""" return sorted((q["network"], q["payTo"], q["atomic"]) for q in g.get("quotes") or []) def control_non_token(cache): """A payTo address is an account, not a token. It must not resolve.""" probe = ("eip155:8453", "0x110cdBba7FE6434Ec4CE3464CC523942ad6Fb784") dec, _, note = resolve_asset(*probe, cache=cache) return dec is None, note # ---------------------------------------------------------------- selftest FIX_COMPLETE = {"accepts": [{"scheme": "exact", "network": "eip155:8453", "amount": "2000", "asset": "0xA", "payTo": "0xB", "maxTimeoutSeconds": 300}]} FIX_NO_PAYTO = {"accepts": [{"scheme": "exact", "network": "eip155:8453", "amount": "2000", "asset": "0xA"}]} FIX_BARE = {"accepts": [], "raw": True} def selftest(): ok, bad = [], [] def check(name, cond): (ok if cond else bad).append(name) six = lambda n, a, c=None: (6, "USDC", "") # noqa: E731 eighteen = lambda n, a, c=None: (18, "WHAT", "") # noqa: E731 never = lambda n, a, c=None: (None, None, "no RPC") # noqa: E731 g = grade(FIX_COMPLETE, resolver=six) check("a complete accept with a resolved asset is PRICED", g["verdict"] == "PRICED") check("2000 atomic units at 6 decimals is 0.002", g["quotes"][0]["value"] == "0.002") g18 = grade(FIX_COMPLETE, resolver=eighteen) check("THE SAME 2000 at 18 decimals is not 0.002", g18["quotes"][0]["value"] != "0.002") check("the same 2000 at 18 decimals is 0.000000000000002", Decimal(g18["quotes"][0]["value"]) == Decimal("2E-15")) check("the two readings differ by twelve orders of magnitude", Decimal(g["quotes"][0]["value"]) / Decimal(g18["quotes"][0]["value"]) == Decimal(10) ** 12) check("an accept with no payTo is UNPRICED", grade(FIX_NO_PAYTO, resolver=six)["verdict"] == "UNPRICED") check("and the report names payTo as what is missing", "payTo" in grade(FIX_NO_PAYTO, resolver=six)["why"]) check("an unresolvable asset is UNPRICED, never PRICED", grade(FIX_COMPLETE, resolver=never)["verdict"] == "UNPRICED") check("an unresolvable asset says why", "no RPC" in grade(FIX_COMPLETE, resolver=never)["why"]) check("a 402 with no document at all is STATUS", grade(FIX_BARE, resolver=six)["verdict"] == "STATUS") check("STATUS says the 402 names nothing to pay", "nothing to pay" in grade(FIX_BARE, resolver=six)["why"]) check("no challenge at all grades None", grade(None, resolver=six) is None) fallback = {"accepts": [{"scheme": "exact", "network": "n", "maxAmountRequired": "5", "asset": "0xA", "payTo": "0xB"}]} check("maxAmountRequired is read where amount is absent", grade(fallback, resolver=six)["verdict"] == "PRICED") bogus = {"accepts": [{"scheme": "exact", "network": "n", "amount": "abc", "asset": "0xA", "payTo": "0xB"}]} check("a non-numeric amount is UNPRICED, not priced at zero", grade(bogus, resolver=six)["verdict"] == "UNPRICED") frac = {"accepts": [{"scheme": "exact", "network": "n", "amount": "1.5", "asset": "0xA", "payTo": "0xB"}]} check("a fractional atomic amount is refused (atomic units are integers)", grade(frac, resolver=six)["verdict"] == "UNPRICED") mixed = {"accepts": [dict(FIX_NO_PAYTO["accepts"][0]), dict(FIX_COMPLETE["accepts"][0], network="eip155:42161")]} gm = grade(mixed, resolver=six) check("one good accept beside a broken one is PRICED", gm["verdict"] == "PRICED") check("and the broken one is still reported", bool(gm["problems"])) # decode_word: the rule that keeps a failure to measure from becoming a zero check("an empty eth_call result is not a reading", decode_word("0x") is None) check("a short word is not a reading", decode_word("0x06") is None) check("a full 32-byte word reads", decode_word("0x" + "00" * 31 + "06") == 6) check("a non-hex word is not a reading", decode_word("0x" + "zz" * 32) is None) check("a missing result is not a reading", decode_word(None) is None) check("decimals 0 is a reading, not a failure", decode_word("0x" + "00" * 32) == 0) # human(): the conversion itself check("human refuses an unknown decimals", human("2000", None) is None) check("human at 0 decimals is the atomic number", human("2000", 0) == Decimal(2000)) check("human refuses a non-number", human("abc", 6) is None) # decode_string sym = "0x" + "00" * 31 + "20" + "00" * 31 + "04" + "55534443" + "00" * 28 check("an ABI string decodes to its symbol", decode_string(sym) == "USDC") check("an empty symbol result is not a reading", decode_string("0x") is None) # parse_challenge: only a 402 is a challenge check("a 200 is never a challenge", parse_challenge(200, {}, b'{"accepts":[]}') is None) check("a 402 with an unparseable body is still a challenge", parse_challenge(402, {}, b"nope") == {"accepts": [], "raw": True}) check("a 402 body document is read when there is no header", parse_challenge(402, {}, b'{"accepts":[{"scheme":"exact"}]}')["accepts"][0]["scheme"] == "exact") hdr = base64.b64encode(json.dumps({"accepts": [{"scheme": "hdr"}]}).encode()).decode() check("the header document wins over the body", parse_challenge(402, {"payment-required": hdr}, b'{"accepts":[{"scheme":"body"}]}')["accepts"][0]["scheme"] == "hdr") check("MONEY_RE reads a dollar figure out of prose", MONEY_RE.findall("2000 units of USDC = $0.002; 1000 = $0.001") == ["0.002", "0.001"]) for name in ok: print("ok: %s" % name) for name in bad: print("FAILED: %s" % name) print("\n%d rule(s) falsified, %d failed" % (len(ok), len(bad))) return 1 if bad else 0 def main(): ap = argparse.ArgumentParser() ap.add_argument("--clone", default="/tmp/reg") ap.add_argument("--ref", default="origin/main") ap.add_argument("--selftest", action="store_true") ap.add_argument("--once", action="store_true", help="one request per endpoint; skips the quote-stability check") ap.add_argument("--json") args = ap.parse_args() if args.selftest: sys.exit(selftest()) corpus.require_clone(args.clone, "the price sweep") head = corpus.require_ref(args.clone, args.ref, "the price sweep") st, hd, bd = http(CONTROL_URL) if parse_challenge(st, hd, bd) is not None: sys.exit("REFUSED: the control %s answered a payment challenge." % CONTROL_URL) cache = {} passed, note = control_non_token(cache) if not passed: sys.exit("REFUSED: a known non-token address resolved as a token. Every price " "this sweep could print would be manufactured.") print("control: %s answers %s and no challenge; a payTo address does not " "resolve as a token (%s)" % (CONTROL_URL, st, note)) tdir = os.path.join(args.clone, "registry", "tools") entries = [] for d in sorted(os.listdir(tdir)): f = os.path.join(tdir, d, "tool.json") if not os.path.exists(f): continue t = json.load(open(f, encoding="utf-8")) p = t.get("payments") or {} if p.get("machinePayable"): entries.append((d, t)) corpus.require_rows(len(entries), "machine-payable entries", args.clone) print("registry %s, %d entry/entries claiming machinePayable\n" % (head, len(entries))) rows, priced, unpriced, status_only, rotating = [], [], [], [], [] for tool_id, t in entries: notes = (t["payments"].get("notes") or "") claimed_money = MONEY_RE.findall(notes) print("== %-11s claims %s; its notes name %s" % (tool_id, t["payments"].get("protocols") or [], ", ".join("$" + c for c in claimed_money) or "no dollar figure")) for c in candidates(args.clone, tool_id): st, hd, bd = http(c["url"]) ch = parse_challenge(st, hd, bd) g = grade(ch, cache) if g is None: print(" %-58s -> %s (no challenge)" % (c["url"][:58], st)) rows.append({"tool": tool_id, "url": c["url"], "status": st, "verdict": None}) continue print(" %-58s -> 402 %s" % (c["url"][:58], g["verdict"])) {"PRICED": priced, "UNPRICED": unpriced, "STATUS": status_only}[g["verdict"]].append( (tool_id, c["url"])) for q in g["quotes"]: print(" %-34s %s atomic = %s %s (decimals %s from the chain) -> %s" % (q["network"], q["atomic"], q["value"], q["symbol"] or "?", q["decimals"], q["payTo"][:12] + "...")) if q["note"]: print(" note: %s" % q["note"]) for p in g.get("problems") or []: print(" ! %s" % p) if g["why"] and g["verdict"] != "PRICED": print(" why: %s" % g["why"]) if g["verdict"] == "STATUS": said = discloses_empty_accepts(c.get("entry")) if said: print(" DISCLOSED: the registry's own record %s says so first:" % c.get("record")) for s in said: print(" %s" % s[:300]) else: print(" and no filed record for this surface says so") # THE SECOND REQUEST. A quote is a thing a client stores and uses # a moment later, so whether it survives a second request is part # of what the price IS. The control is inside the same challenge: # a vendor whose rails all rotate might be my own client hitting # different edges, but a challenge where two rails hold still and # two move cannot be that. if g["verdict"] == "PRICED" and not args.once: st2, hd2, bd2 = http(c["url"]) g2 = grade(parse_challenge(st2, hd2, bd2), cache) if g2 and g2.get("verdict") == "PRICED": a, b = dict((n, (p, m)) for n, p, m in quote_key(g)), \ dict((n, (p, m)) for n, p, m in quote_key(g2)) moved = sorted(n for n in a if n in b and a[n] != b[n]) held = sorted(n for n in a if n in b and a[n] == b[n]) if moved: print(" ! the quote MOVED between two requests on %s " "(held still on %s)" % (", ".join(moved), ", ".join(held) or "nothing")) for n in moved: print(" %s payTo %s then %s, amount %s then %s" % (n, a[n][0][:14], b[n][0][:14], a[n][1], b[n][1])) rotating.append((tool_id, c["url"], moved, held)) vals = {q["value"] for q in g["quotes"]} if len(vals) > 1: print(" ! this challenge quotes DIFFERENT prices on different " "networks: %s" % ", ".join(sorted(vals))) if g["quotes"] and claimed_money: v = Decimal(g["quotes"][0]["value"]) if any(Decimal(m) == v for m in claimed_money): print(" candidate: the entry's own notes name $%s and the chain " "makes this quote %s" % (v, v)) else: print(" candidate: the entry's notes name %s and this quote is " "%s in the asset's units (prose is read, not proved)" % (", ".join("$" + m for m in claimed_money), v)) rows.append({"tool": tool_id, "url": c["url"], "status": st, "verdict": g["verdict"], "quotes": g["quotes"]}) print() total = len(priced) + len(unpriced) + len(status_only) print("%d challenge(s) graded: %d PRICED, %d UNPRICED, %d STATUS" % (total, len(priced), len(unpriced), len(status_only))) if unpriced or status_only: for tool_id, url in unpriced + status_only: print(" %s %s" % (tool_id, url)) if rotating: print("\n%d challenge(s) quote a DIFFERENT destination on a second request " "seconds later: a quote nobody can cache, and an address published from " "one reading is a nonce, not a fact about the vendor." % len(rotating)) for tool_id, url, moved, held in rotating: print(" %s %s: moved on %s, held on %s" % (tool_id, url, ",".join(moved), ",".join(held) or "nothing")) else: print(" every machine-payable claim in this registry quotes a price a machine " "could actually settle, in an asset the chain itself identifies.") if args.json: json.dump({"head": head, "rows": rows}, open(args.json, "w"), indent=1, default=str) print("wrote %s" % args.json) if __name__ == "__main__": main()