#!/usr/bin/env python3 """payments.py: every `payments` block in the registry, measured against the endpoints the registry itself names, in one pass. Written at wake ec0e3ab2 (2026-09-20), the night the seventh payments block merged (Intercom and Shopify UCP, both `machinePayable: false`). Issue #123, filed the wake before, said the cell's problem in one line: three entries claim an agent can pay without a human and only two of them carry a probe record. The third, Bitrefill, carries its measurement as dated prose in a profile, which no instrument reads and no checker can re-run. drift.py asks whether an entry's ACCESS claim is still true. This asks the same question of its PAYMENT claim, and the two are different questions: a surface that answers 401 to a stranger is gated on identity, one that answers 402 is gated on money, and an entry that confuses them sends a reader to the wrong door. The registry's `payments.machinePayable` is exactly the boolean "can a machine pay its way in here", so it is checkable, and the check is one unpaid request per endpoint. python3 tools/payments.py --clone /tmp/reg python3 tools/payments.py --clone /tmp/reg --json out.json WHERE THE ENDPOINTS COME FROM, and this is the point of the instrument. Not from `surfaces`, which on most entries holds documentation pages: a docs page never answers 402 and asking it would manufacture a false negative. The candidates are, in order: 1. every probe record filed about that subject, which carries a real endpoint and the METHOD it was measured with; 2. any URL that a profile sentence mentions within 300 characters of "402", "x402" or "machine payment", tagged `prose`, because that is where an unrecorded measurement hides (Bitrefill's is the reason this rule exists). A GET is sent to every candidate whose evidence says GET, and to every prose candidate. POST-only endpoints are REPORTED AND NOT CALLED: a bare POST to an MCP server is a different request from the one the record made, and a half-request is worse evidence than none. That is a disclosed limit, not a silence, and it means this sweep can confirm a `true` and can only ever WEAKLY support a `false`. THE CONTROL, which must pass before any row is worth reading: a URL that is certainly not payable (example.com) must not report a payment challenge, and every entry whose block says `machinePayable: true` must produce at least one 402 from its own named endpoints. If the second half fails, either the entry has drifted or this instrument is looking in the wrong place; both are findings, and the run says which it cannot tell apart. """ import argparse import json import os import re import sys import urllib.error import urllib.request sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import corpus # noqa: E402 UA = {"User-Agent": "lintel-payments (Public Agents registry, cto agent)"} CONTROL_URL = "https://example.com/" NEAR = 300 # characters around a 402 mention that make a URL a candidate PAY_WORDS = re.compile(r"\b(402|x402|mpp|machine[- ]payment|machine[- ]payable)\b", re.I) URL_RE = re.compile(r"https?://[^\s)\]\"'>,`]+") # A host or path with a wildcard in it, written the way a profile writes one: # `api.bitrefill.com/x402/*`. This is not a URL and cannot be called, and an # entry whose only named endpoint is one of these is the gap worth printing. GLOB_RE = re.compile(r"`([a-z0-9.-]+\.[a-z]{2,}/[^`\s]*\*[^`\s]*)`", re.I) def clean(url): """A URL as a profile writes it, minus the prose around it. Markdown line-wraps and trailing punctuation produced rows like `https://docs` and `.../docs/errors-and-r` on the first run; a host with no dot is a fragment, not an endpoint, and calling it teaches nothing. """ url = url.rstrip(".,;:`'\"") host = url.split("//", 1)[-1].split("/", 1)[0] return url if "." in host and len(host) > 3 else None def fetch(url, timeout=20): """One unpaid GET. A 402 is an answer, not an error, so it is caught.""" req = urllib.request.Request(url, headers=UA, method="GET") try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, dict(r.headers), r.read(4000) except urllib.error.HTTPError as e: return e.code, dict(e.headers), e.read(4000) except Exception as e: return None, {"error": str(e)[:120]}, b"" def challenge(status, headers, body): """What a response says about paying, in the words the response used. Three shapes are recognised, all of them seen on this registry's own entries: the x402 `payment-required` document, an MPP `www-authenticate` line, and a bare 402 with neither (which is still a 402 and still says money, so it is reported rather than dropped). """ if status != 402: return None found = {"protocols": [], "detail": {}} wa = headers.get("www-authenticate") or headers.get("WWW-Authenticate") or "" if "mpp" in wa.lower() or "method=" in wa.lower(): found["protocols"].append("mpp") found["detail"]["www-authenticate"] = wa[:200] pr = headers.get("payment-required") or headers.get("Payment-Required") or "" doc = None if pr: found["protocols"].append("x402") found["detail"]["payment-required"] = pr[:120] try: doc = json.loads(body.decode("utf-8", "replace")) except Exception: doc = None if isinstance(doc, dict): accepts = doc.get("accepts") if isinstance(accepts, list) and accepts: if "x402" not in found["protocols"]: found["protocols"].append("x402") # x402 v2 names the price `amount`; earlier drafts and some # servers say `maxAmountRequired`. Reading only one of them prints # a null price for a document that states one, which is the # failure mode this whole toolbox exists to refuse. found["detail"]["accepts"] = [ {"scheme": a.get("scheme"), "network": a.get("network"), "amount": a.get("amount", a.get("maxAmountRequired")), "asset": a.get("asset"), "maxTimeoutSeconds": a.get("maxTimeoutSeconds")} for a in accepts if isinstance(a, dict) ][:6] if not found["protocols"]: found["protocols"].append("402-no-named-protocol") return found def shape_notes(ch): """Things about a live challenge that a reader would want said out loud. The first one is measured, not guessed (wake ec0e3ab2): x402 v2 defines `maxTimeoutSeconds` as a number of SECONDS, and one vendor on this registry answers with a unix timestamp of issue-time plus sixty. Two requests a minute apart showed the value advancing with the clock, which is what makes it a deadline and not a long duration. A client that reads it per the spec believes it has decades; a client that reads it as a deadline has a minute. Both readings are defensible from the document alone, which is exactly why it belongs in a registry. """ notes = [] for a in ch.get("detail", {}).get("accepts", []) or []: t = a.get("maxTimeoutSeconds") if isinstance(t, int) and t > 10 ** 9: notes.append("maxTimeoutSeconds=%d on %s is a unix timestamp (%sZ), not a " "duration in seconds as x402 v2 defines the field" % (t, a.get("network"), __import__("datetime").datetime.utcfromtimestamp(t).isoformat())) return notes def candidates(clone, tool_id): """Endpoints the registry itself names for this subject, with provenance.""" 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 url = e.get("surface") if url: out.append({"url": url, "from": "probe:" + e.get("question", "?"), "record": e.get("id"), "method": (e.get("request") or {}).get("method", "GET")}) prof = os.path.join(clone, "registry", "tools", tool_id, "profile.md") if os.path.exists(prof): text = open(prof, encoding="utf-8").read() for m in PAY_WORDS.finditer(text): window = text[max(0, m.start() - NEAR): m.end() + NEAR] for u in URL_RE.finditer(window): url = clean(u.group(0)) if not url or any(c["url"] == url for c in out): continue if "public-agents" in url: # our own site, never the vendor's door continue out.append({"url": url, "from": "prose", "record": None, "method": "GET"}) for g in GLOB_RE.finditer(window): pat = g.group(1) if not any(c.get("pattern") == pat for c in out): out.append({"url": None, "pattern": pat, "from": "prose-pattern", "record": None, "method": "PATTERN"}) return out def main(): ap = argparse.ArgumentParser() ap.add_argument("--clone", default="/tmp/reg") ap.add_argument("--ref", default="origin/main", help="the tree this sweep claims to measure") ap.add_argument("--json") args = ap.parse_args() corpus.require_clone(args.clone, "the payments sweep") head = corpus.require_ref(args.clone, args.ref, "the payments sweep") st, hd, bd = fetch(CONTROL_URL) if challenge(st, hd, bd) is not None: sys.exit("REFUSED: the control %s reported a payment challenge (%s). " "This network is answering something other than the internet." % (CONTROL_URL, st)) print("control: %s answers %s and no payment challenge" % (CONTROL_URL, st)) 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")) if t.get("payments"): entries.append((d, t)) corpus.require_rows(len(entries), "payments blocks", args.clone) print("registry %s, %d payments block(s)\n" % (head, len(entries))) rows, disagree, unsupported = [], [], [] for tool_id, t in entries: p = t["payments"] claimed = bool(p.get("machinePayable")) print("== %-13s machinePayable=%-5s protocols=%s methods=%s humanBilling=%s" % (tool_id, claimed, p.get("protocols") or [], p.get("methods") or [], p.get("humanBilling"))) seen402, asked, patterns = [], 0, [] for c in candidates(args.clone, tool_id): if c["method"].upper() == "PATTERN": patterns.append(c["pattern"]) print(" %-4s %-58s %s (a wildcard, not an endpoint: nothing here can call it)" % ("----", c["pattern"][:58], c["from"])) continue if c["method"].upper() != "GET": print(" %-4s %-58s %s (not called: this instrument only GETs)" % ("POST", c["url"][:58], c["from"])) continue asked += 1 st, hd, bd = fetch(c["url"]) ch = challenge(st, hd, bd) mark = "402" if ch else str(st) print(" %-4s %-58s %s -> %s%s" % ("GET", c["url"][:58], c["from"], mark, " " + ",".join(ch["protocols"]) if ch else "")) if ch: seen402.append({"url": c["url"], "from": c["from"], **ch}) for k, v in ch["detail"].items(): print(" %s: %s" % (k, json.dumps(v)[:180])) for note in shape_notes(ch): print(" NOTE %s" % note) rows.append({"tool": tool_id, "url": c["url"], "from": c["from"], "status": st, "challenge": ch}) if claimed and not seen402: unsupported.append(tool_id) print(" !! claims machinePayable, and no endpoint this sweep could GET answered 402") if patterns: print(" the only payable address this entry names is a PATTERN (%s):" % ", ".join(patterns)) print(" the claim may be perfectly true and no checker can re-run it,") print(" because nothing in the entry is a URL. That is the gap, not the claim.") if claimed and seen402: live = sorted({q for s in seen402 for q in s["protocols"]}) want = sorted(p.get("protocols") or []) if want and set(want) - set(live): disagree.append((tool_id, want, live)) print(" !! entry names %s; this sweep saw %s" % (want, live)) else: print(" OK 402 live, protocols %s" % live) if not claimed and seen402: disagree.append((tool_id, [], sorted({q for s in seen402 for q in s["protocols"]}))) print(" !! says NOT machine-payable and an endpoint answered 402") if not claimed and not seen402: print(" ok nothing answered 402 (%d GET-able endpoint(s); POST-only ones not called)" % asked) print() payable = [d for d, t in entries if t["payments"].get("machinePayable")] print("%d of %d blocks claim machinePayable; %d of those reproduce a live 402" % (len(payable), len(entries), len(payable) - len(unsupported))) if unsupported: print(" UNSUPPORTED (drifted, or this sweep is looking in the wrong place; " "it cannot tell those apart on its own): %s" % ", ".join(unsupported)) if disagree: for tool_id, want, live in disagree: print(" DISAGREES %s: entry %s, measured %s" % (tool_id, want or "false", live)) if not unsupported and not disagree: print(" every payment claim in the registry answers for itself today.") withprobe = set() pdir = os.path.join(args.clone, "registry", "evidence", "probes") for name in sorted(os.listdir(pdir)): e = json.load(open(os.path.join(pdir, name), encoding="utf-8")) if e.get("question") == "payment": withprobe.add((e.get("subject") or {}).get("id")) missing = [d for d in payable if d not in withprobe] print("\ncoverage: %d of %d machine-payable entries carry a probe record asking `payment`" % (len(payable) - len(missing), len(payable))) if missing: print(" no payment probe: %s (issue #123)" % ", ".join(missing)) if args.json: json.dump({"head": head, "rows": rows}, open(args.json, "w"), indent=1) print("\nwrote %s" % args.json) if __name__ == "__main__": main()