#!/usr/bin/env python3 """Second pass, after the first one graded 1,064 of 1,551 hosts BROKEN. That number was mostly me. Three bugs, all in the direction that flattered the finding: 1. I read 4,096 bytes of the 402 body and then called json.loads on the fragment. 142 servers sent perfectly good payment requirements and my truncation turned them into "unparseable". 2. x402 v2 puts the payment requirements in the PAYMENT-REQUIRED *header*. I only ever looked at the body, so 471 correct v2 servers came back as "402 without accepts". 3. Every listing declares its HTTP method in extensions.bazaar.info.input, and I sent GET to all of them. 232 replied 405 and 179 replied 404, which is what a POST-only route says to a GET. A 69% breakage rate was the most striking number I could have published today, and it was an artefact of the instrument. Same lesson as the pre-written conclusion this morning: the check has to be able to disagree with me. Grades: LIVE 402 and machine-readable terms somewhere an x402 client would look (body accepts / paymentRequirements, or the PAYMENT-REQUIRED header). OPAQUE 402, but nothing an agent could actually pay -- no terms anywhere. A wall with no keyhole. Human-readable "payment required" counts as opaque, because the customer here is not a human. OPEN 200 without payment. The paywall is not in front of the resource. MISMATCH terms served, but paying the named address is not paying the address the catalogue advertises for this resource. BROKEN answered, but not with 402 and not with the goods. DEAD no answer: DNS, refused, TLS, timeout. """ import json, socket, ssl, time, urllib.error, urllib.parse, urllib.request from concurrent.futures import ThreadPoolExecutor UA = ("Tenner/1.0 (autonomous agent; x402 Bazaar liveness survey; " "one unpaid request per host, method as declared in the listing; " "http://144-31-195-17.traefik.me/)") TIMEOUT = 15 MAXBODY = 262144 def declared(rec): """Method and example body the seller published in their own listing.""" try: inp = rec["extensions"]["bazaar"]["info"]["input"] except Exception: return "GET", None, None m = str(inp.get("method") or "GET").upper() return m, inp.get("body"), inp.get("queryParams") def payto_of(rec): for a in rec.get("accepts", []) or []: if a.get("payTo"): return str(a["payTo"]) return None def terms_from(body, headers): """Everywhere an x402 client is entitled to look for payment terms.""" out = [] for key in ("payment-required", "x-payment-required", "www-authenticate", "x-accept-payment"): v = headers.get(key) if v: out.append(("header:" + key, v)) if body: try: d = json.loads(body) except Exception: d = None if isinstance(d, dict): acc = d.get("accepts") or d.get("paymentRequirements") if isinstance(acc, dict): acc = [acc] if acc: out.append(("body", acc)) return out def served_addrs(terms): addrs = [] for src, val in terms: if src == "body": for a in val: if isinstance(a, dict) and a.get("payTo"): addrs.append(str(a["payTo"])) else: # v2 header is a base64 or structured blob; pull anything that # looks like an address rather than pretending to parse it fully. import re addrs += re.findall(r"0x[a-fA-F0-9]{40}", str(val)) return addrs def probe(rec): url = rec.get("resource", "") method, body, qp = declared(rec) out = {"resource": url, "host": urllib.parse.urlparse(url).netloc.lower(), "method": method, "listed_payTo": payto_of(rec)} data = None headers = {"User-Agent": UA, "Accept": "application/json,*/*"} if method in ("POST", "PUT", "PATCH"): data = json.dumps(body if isinstance(body, dict) else {}).encode() headers["Content-Type"] = "application/json" elif isinstance(qp, dict) and qp: sep = "&" if "?" in url else "?" url = url + sep + urllib.parse.urlencode( {k: v for k, v in qp.items() if isinstance(v, (str, int, float))}) req = urllib.request.Request(url, method=method, data=data, headers=headers) t0 = time.time() try: with urllib.request.urlopen(req, timeout=TIMEOUT) as r: out["code"] = r.status out["grade"] = "OPEN" out["bytes"] = len(r.read(MAXBODY)) except urllib.error.HTTPError as e: out["code"] = e.code try: raw = e.read(MAXBODY).decode("utf-8", "replace") except Exception: raw = "" hdrs = {k.lower(): v for k, v in (e.headers or {}).items()} if e.code == 402: terms = terms_from(raw, hdrs) if not terms: out["grade"] = "OPAQUE" out["body"] = raw[:300] else: out["src"] = terms[0][0] addrs = served_addrs(terms) out["served_payTo"] = addrs[0] if addrs else None if out["listed_payTo"] and addrs and out["listed_payTo"] not in addrs: out["grade"] = "MISMATCH" else: out["grade"] = "LIVE" else: out["grade"] = "BROKEN" out["why"] = f"HTTP {e.code}" out["body"] = raw[:200] except (urllib.error.URLError, socket.timeout, ssl.SSLError, ConnectionError, OSError) as e: reason = getattr(e, "reason", e) out["grade"] = "DEAD" out["why"] = f"{type(reason).__name__}: {str(reason)[:90]}" except Exception as e: out["grade"] = "DEAD" out["why"] = f"{type(e).__name__}: {str(e)[:90]}" out["ms"] = int((time.time() - t0) * 1000) return out def main(): recs = json.load(open("/home/agent/x402survey/raw.json")) by_host = {} for r in recs: h = urllib.parse.urlparse(r.get("resource", "")).netloc.lower() if h and h not in by_host: by_host[h] = r targets = list(by_host.values()) print(f"probing {len(targets)} hosts, method as declared", flush=True) res = [] with ThreadPoolExecutor(max_workers=24) as ex: for i, o in enumerate(ex.map(probe, targets), 1): res.append(o) if i % 200 == 0: print(f" {i}/{len(targets)}", flush=True) json.dump(res, open("/home/agent/x402survey/probe2_hosts.json", "w")) tally = {} for o in res: tally[o["grade"]] = tally.get(o["grade"], 0) + 1 print("RESULT", json.dumps(tally, sort_keys=True)) if __name__ == "__main__": main()