#!/usr/bin/env python3 """ARION x402 sell-side shim (stdlib only). Serves deterministic paid endpoints: POST /v1/address-check {"chain": "nano|evm|solana", "address": "..."} -> validation verdict. POST /v1/prescan {"scanner": "mcp|timing|proto|ssrf|ws|injection|redos", "code": "", "filename": "optional.ext"} -> findings list from ARION's deterministic pre-review tools (pattern scan, NOT an audit - honest scope per PROFILE.md). POST /v1/settle-check {"chain": "evm|solana|nano", "tx": "", "min_conf"?: int, "payer"/"payee"/"amount_raw"?: nano narrowers} -> single-tx settlement recompute from public RPC (status, deltas, finality tier). Deterministic recompute, not an audit; counter-leg scan is depth-bounded on nano and not run on evm/solana. POST /v1/dep-check {"manifest": "", "filename": "package.json|requirements.txt|Cargo.toml|go.mod"} -> dependency supply-chain pre-scan findings (unpinned versions, non-registry sources, install-time code exec, typosquat proximity, path/replace overrides, missing-lockfile note). PRE-SCAN, not an audit, no CVE feed - honest scope per PROFILE.md. Three payment dialects, all real-money-settled straight to ARION's own payTo (facilitators hold nothing; we hold no keys): a) x402 v2 "exact" on eip155:8453 (USDC $0.001): unpaid -> 402 + PAYMENT-REQUIRED; client signs an EIP-3009 authorization, retries with PAYMENT-SIGNATURE; we settle through facilitator.payai.network (fallback facilitator.xpay.sh) and answer 200 + PAYMENT-RESPONSE. Replay key = authorization nonce. b) x402 v2 "exact" on nano:mainnet: same flow, settled through facilitator.pursekeeper.dev (/verify then /settle). c) X-Nano-Payment: client broadcasts its own send, retries with header X-Nano-Payment: ; we check it through pursekeeper.dev/v1/verify?hash&to&min_raw and consume the hash once. Replay protection: every consumed hash is persisted in consumed.json next to this script; a presented-again hash answers 402 payment_reused and an unverifiable one 402 payment_invalid (pursekeeper's three checks). Paid routes also accept GET with query params (same auth headers) so GET-only directory verifiers and x402 clients that prefer query-style requests still see a real 402 and can pay for a served answer. Free routes: GET /, /health, /v1/price, /v1/requirements. Self-locating; no /workspace paths hardcoded. """ import base64, hashlib, hmac, importlib.util, json, os, re, shutil, socket, sys import tempfile, threading, time import urllib.request, urllib.error from urllib.parse import urlencode, urlparse, parse_qs from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import addresscheck PAID_PATHS = ("/v1/address-check", "/v1/prescan", "/v1/dep-check", "/v1/settle-check") PORT = int(os.environ.get("SHIM_PORT", "8402")) PAY_TO = "nano_3m8cz87zwxb1y16ob4bzp1eyek78qaig8ktohk7d45b18sh6u9exbowbnekr" PRICE_RAW = "500000000000000000000000000" # 0.0005 XNO PRICE_XNO = "0.0005" EVM_PAY_TO = "0x6E9c17439Cf81247965f9543645cFc8E746c4588" USDC_BASE = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" PRICE_USDC = "1000" # 0.001 USDC (6dp) EVM_FACILITATORS = ["https://facilitator.payai.network", "https://facilitator.xpay.sh"] FACILITATOR = "https://facilitator.pursekeeper.dev" PK_VERIFY_BASE = "https://pursekeeper.dev/v1/verify" NANO_BLOCK_HASH_RE = re.compile(r"^[0-9A-Fa-f]{64}$") IDX_VERIFY_HASH = ("175e8a8c6a98adedda7e83a9a1bb4e86e0d90de6" "c67ac37b49c1d94a7ca013d1") CONSUMED_LOCK = threading.Lock() MAX_CONCURRENT_REQUESTS = 32 REQUEST_SEMAPHORE = threading.BoundedSemaphore(MAX_CONCURRENT_REQUESTS) REQUEST_TIMEOUT_SECONDS = 15 def valid_nano_block_hash(value): return bool(NANO_BLOCK_HASH_RE.fullmatch(value or "")) def pk_verify_url(h): return PK_VERIFY_BASE + "?" + urlencode( {"hash": h, "to": PAY_TO, "min_raw": PRICE_RAW}) STATE_FILE = os.environ.get( "SHIM_STATE_FILE", os.path.join(HERE, "consumed.json")) MUSEDIN_WH_PATH = "/webhook/musedin" MUSEDIN_WH_SECRET_FILE = os.environ.get( "MUSEDIN_WH_SECRET_FILE", os.path.join(HERE, "musedin_webhook_secret.txt")) MUSEDIN_WH_LOG = os.environ.get( "MUSEDIN_WH_LOG", os.path.join(HERE, "musedin_webhook_log.jsonl")) UA = {"User-Agent": "ARION-x402-shim/1.0", "Content-Type": "application/json"} # --- paid pre-review scanners (siblings under deliverables/) --------------- PRESCANNERS = { "mcp": "mcp-prescan/mcp_prescan.py", "timing": "timing-prescan/timing_prescan.py", "proto": "proto-prescan/proto_prescan.py", "ssrf": "ssrfsink-prescan/ssrfsink_prescan.py", "ws": "ws-prescan/ws_prescan.py", "injection": "injection-prescan/injection_prescan.py", "redos": "redos-prescan/redos_prescan.py", } _SCANNER_MODS = {} def _scanner(name): if name not in _SCANNER_MODS: path = os.path.normpath(os.path.join(HERE, "..", PRESCANNERS[name])) spec = importlib.util.spec_from_file_location("ps_" + name, path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) _SCANNER_MODS[name] = mod return _SCANNER_MODS[name] def run_prescan(req_body): """Validate-then-scan. Returns (status_code, obj).""" name = req_body.get("scanner") code = req_body.get("code") if name not in PRESCANNERS: return 400, {"error": "bad_scanner", "scanners": sorted(PRESCANNERS)} if not isinstance(code, str) or not code: return 400, {"error": "bad_code"} fname = os.path.basename(str(req_body.get("filename") or "snippet.txt")) ext = os.path.splitext(fname)[1] or ".txt" tmp = None try: fd, tmp = tempfile.mkstemp(suffix=ext) with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(code) mod = _scanner(name) if hasattr(mod, "scan_file"): res = mod.scan_file(tmp, fname) elif hasattr(mod, "scan"): res = mod.scan(tmp) res = res[0] if isinstance(res, tuple) else res else: res = mod.scan_lines(tmp, code.splitlines(keepends=True)) rows = [f.as_dict() for f in res] except Exception as e: return 500, {"error": "scan_failed", "detail": str(e)[:200]} finally: if tmp: try: os.unlink(tmp) except Exception: pass return 200, {"scanner": name, "findings": rows, "count": len(rows)} # --- dependency supply-chain pre-scan (dep-prescan sibling) ---------------- # Manifests dep_prescan.scan_target can scan a single file of (MANIFESTS # entries with a scan fn). Directory scans and pyproject/Pipfile/setup.py # note-only manifests are out of scope for this endpoint. DEP_MANIFESTS = ("package.json", "requirements.txt", "Cargo.toml", "go.mod") _DEP_MOD = None def _dep(): global _DEP_MOD if _DEP_MOD is None: path = os.path.normpath(os.path.join( HERE, "..", "dep-prescan", "dep_prescan.py")) spec = importlib.util.spec_from_file_location("ps_dep", path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) _DEP_MOD = mod return _DEP_MOD def dep_check_valid(req_body): """Pre-settle validation: never burn a payment on an unscannable request.""" m = req_body.get("manifest") fname = os.path.basename(str(req_body.get("filename") or "")) return (isinstance(m, str) and bool(m) and fname in DEP_MANIFESTS) def address_check_valid(req_body): """Pre-settle validation: never burn a payment on an unverifiable address request (audit E1 2026-10-09).""" c = req_body.get("chain") a = req_body.get("address") return (isinstance(c, str) and c.lower() in addresscheck.CHAINS and isinstance(a, str) and bool(a)) def run_depcheck(req_body): """Validate-then-scan. Returns (status_code, obj).""" if not dep_check_valid(req_body): return 400, {"error": "bad_dep_check_request", "manifests": sorted(DEP_MANIFESTS)} fname = os.path.basename(str(req_body.get("filename"))) tmpdir = None try: tmpdir = tempfile.mkdtemp(prefix="depcheck_") target = os.path.join(tmpdir, fname) with open(target, "w", encoding="utf-8") as f: f.write(req_body["manifest"]) findings, scanned, notes = _dep().scan_target(target) except Exception as e: return 500, {"error": "scan_failed", "detail": str(e)[:200]} finally: if tmpdir: shutil.rmtree(tmpdir, ignore_errors=True) return 200, {"manifest": fname, "findings": findings, "count": len(findings), "scanned": [b for b, _ in scanned], "notes": notes} # --- single-tx settlement recompute (settlement-verify sibling) ------------ SETTLE_CHAINS = ("evm", "base", "solana", "nano") EVM_TX_RE = re.compile(r"^0x[0-9a-fA-F]{64}$") SOL_SIG_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,128}$") _SETTLE = {} def _settle_mod(kind): if kind not in _SETTLE: path = os.path.normpath(os.path.join( HERE, "..", "settlement-verify", "receipt_verify.py" if kind == "evmsol" else "xno_receipt.py")) spec = importlib.util.spec_from_file_location("sv_" + kind, path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) _SETTLE[kind] = mod return _SETTLE[kind] def settle_check_valid(req_body): """Pre-settle validation: never burn a payment on an unverifiable tx.""" for key in ("min_conf", "amount_raw"): v = req_body.get(key) if v not in (None, ""): try: int(v) except (TypeError, ValueError): return False chain = str(req_body.get("chain") or "").lower() tx = str(req_body.get("tx") or "") if chain == "xno": chain = "nano" if chain == "base": chain = "evm" # endpoint's evm leg is Base-only (EVM_RPCS) if chain == "evm": return bool(EVM_TX_RE.match(tx)) if chain == "solana": return bool(SOL_SIG_RE.match(tx)) if chain == "nano": return bool(NANO_BLOCK_HASH_RE.match(tx)) return False def run_settlecheck(req_body): """Validate-then-verify. Returns (status_code, obj).""" if not settle_check_valid(req_body): return 400, {"error": "bad_settle_check_request", "chains": list(SETTLE_CHAINS)} chain = str(req_body.get("chain")).lower() if chain == "xno": chain = "nano" if chain == "base": chain = "evm" # endpoint's evm leg is Base-only (EVM_RPCS) tx = str(req_body.get("tx")) try: min_conf = max(0, min(50, int(req_body.get("min_conf") or 0))) except (TypeError, ValueError): return 400, {"error": "bad_min_conf"} try: if chain == "evm": mod = _settle_mod("evmsol") out = mod.evm_verify(tx, None, mod.EVM_RPCS, min_conf, None) elif chain == "solana": mod = _settle_mod("evmsol") out = mod.sol_verify(tx, mod.SOL_RPC, min_conf) else: mod = _settle_mod("xno") amt = req_body.get("amount_raw") try: amt = int(amt) if amt not in (None, "") else None except (TypeError, ValueError): return 400, {"error": "bad_amount_raw"} receipt, fails, warns = mod.verify_send( tx, req_body.get("payer") or None, req_body.get("payee") or None, amt, mod.RPCS, 60) if receipt is None: out = {"verdict": "OUT-OF-SCOPE", "fails": fails} else: receipt["warnings"] = warns receipt["fails"] = fails out = receipt except Exception as e: return 500, {"error": "verify_failed", "detail": str(e)[:200]} out["scope"] = ("single-tx recompute from public RPC, not an audit; " "counter-leg scan depth-bounded (nano) / not run " "(evm, solana)") return 200, out CARD = { "service": "address-check", "seller": "ARION - autonomous agent under a published 3-law constitution; " "writes and payments are host-mediated and human-approved", "homepage": "https://files.profullstack.com/~arion/public/index.html", "endpoint": "POST /v1/address-check {\"chain\",\"address\"}", "endpoints": [ "POST /v1/address-check {\"chain\",\"address\"}", "POST /v1/prescan {\"scanner\",\"code\",\"filename\"?}", "POST /v1/dep-check {\"manifest\",\"filename\"}", "POST /v1/settle-check {\"chain\",\"tx\",\"min_conf\"?}", "GET equivalents accepted: same path + query params, same " "payment headers"], "chains": ["nano", "evm", "solana"], "price": {"xno": PRICE_XNO, "raw": PRICE_RAW, "payTo": PAY_TO, "network": "nano:mainnet", "usdc_base": PRICE_USDC, "payTo_evm": EVM_PAY_TO, "network_evm": "eip155:8453"}, "pay": [ "x402 v2 scheme exact on eip155:8453 (USDC $0.001) - retry with " "PAYMENT-SIGNATURE; settled via facilitator.payai.network / xpay.sh", "x402 v2 scheme exact on nano:mainnet - retry with " "PAYMENT-SIGNATURE; settled via facilitator.pursekeeper.dev", "or broadcast a send to payTo and retry with " "X-Nano-Payment: "], "free": ["GET /", "GET /v1/card", "GET /health", "GET /v1/price", "GET /v1/requirements", "GET /openapi.json", "GET /.well-known/x402", "GET /robots.txt"], "scanners": sorted(PRESCANNERS), "dep_manifests": sorted(DEP_MANIFESTS), "note": "deterministic; answers computed locally, no keys held; " "prescan/dep-check are pattern pre-reviews, not audits; " "every consumed payment hash is recorded for replay protection" } def jpost(url, payload, timeout=25): req = urllib.request.Request(url, method="POST", data=json.dumps(payload).encode(), headers=UA) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, json.load(r) except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read().decode()[:4000]) except Exception: return e.code, {"ok": False, "error": "http_%s" % e.code} except Exception as e: return 0, {"ok": False, "error": "unreachable", "detail": str(e)[:200]} def jget(url, timeout=20): req = urllib.request.Request(url, headers=UA) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, json.load(r) except Exception as e: return 0, {"ok": False, "error": "unreachable", "detail": str(e)[:200]} # --- discovery (x402scan / bazaar / 402index) ------------------------------- ENDPOINT_META = { "/v1/address-check": { "schema": {"type": "object", "properties": { "chain": {"type": "string", "enum": ["nano", "evm", "solana"]}, "address": {"type": "string"}}, "required": ["chain", "address"]}, "example_in": {"chain": "nano", "address": "nano_3m8cz87zwxb1y16ob4bzp1eyek78qaig8k" "tohk7d45b18sh6u9exbowbnekr"}, "example_out": {"verdict": {"valid": True}, "seller": "ARION"}}, "/v1/prescan": { "schema": {"type": "object", "properties": { "scanner": {"type": "string", "enum": sorted(PRESCANNERS)}, "code": {"type": "string"}, "filename": {"type": "string"}}, "required": ["scanner", "code"]}, "example_in": {"scanner": "injection", "code": "eval(input())", "filename": "a.py"}, "example_out": {"scanner": "injection", "findings": [], "count": 0, "seller": "ARION"}}, "/v1/dep-check": { "schema": {"type": "object", "properties": { "manifest": {"type": "string", "description": "manifest file content"}, "filename": {"type": "string", "enum": sorted(DEP_MANIFESTS)}}, "required": ["manifest", "filename"]}, "example_in": {"filename": "requirements.txt", "manifest": "reqeusts>=2\nflask==3.0.0\n"}, "example_out": {"manifest": "requirements.txt", "findings": [], "count": 0, "scanned": ["requirements.txt"], "notes": [], "seller": "ARION"}}, "/v1/settle-check": { "schema": {"type": "object", "properties": { "chain": {"type": "string", "enum": ["evm", "base", "solana", "nano"], "description": "evm/base = Base L2 " "(endpoint RPCs are " "Base-only)"}, "tx": {"type": "string", "description": "evm tx hash / solana " "signature / nano send-block " "hash"}, "min_conf": {"type": "integer", "minimum": 0, "maximum": 50}, "payer": {"type": "string"}, "payee": {"type": "string"}, "amount_raw": {"type": "string"}}, "required": ["chain", "tx"]}, "example_in": {"chain": "evm", "tx": "0x467eaa3b57a3a44db26ae7e98f6b40c63f07ebfb" "24a09d63be8c6dbfd63f9e62"}, "example_out": {"verdict": "VALID", "seller": "ARION"}}, } def output_schema(path, method="POST"): m = ENDPOINT_META.get(path) if not m: return None return {"input": {"type": "http", "method": method, "discoverable": True, "bodyType": "json", "schema": m["schema"], "example": m["example_in"]}, "output": {"type": "json", "example": m["example_out"]}} def requirements(path=None, method="POST"): r = {"scheme": "exact", "network": "nano:mainnet", "asset": "XNO", "payTo": PAY_TO, "amount": PRICE_RAW, "maxTimeoutSeconds": 60, "extra": {"asset": "XNO", "work": "required", "workThreshold": "fffffff800000000"}} os_ = output_schema(path, method) if os_: r["outputSchema"] = os_ return r def requirements_evm(path=None, method="POST"): r = {"scheme": "exact", "network": "eip155:8453", "asset": USDC_BASE, "payTo": EVM_PAY_TO, "amount": PRICE_USDC, "maxTimeoutSeconds": 60, "extra": {"name": "USD Coin", "version": "2"}} os_ = output_schema(path, method) if os_: r["outputSchema"] = os_ return r def accepts(path=None, method="POST"): return [requirements_evm(path, method), requirements(path, method)] def public_base(): try: return open(os.path.join(HERE, "current_url.txt")).read().strip() except Exception: return "" ENDPOINT_DESCR = { "/v1/address-check": "Deterministic chain address-format validation " "(nano/evm/solana).", "/v1/prescan": "Deterministic pattern pre-review of a code snippet " "(not an audit).", "/v1/dep-check": "Dependency supply-chain pre-scan of a manifest " "(unpinned/non-registry/install-script/typosquat " "flags; not an audit, no CVE feed).", "/v1/settle-check": "Single-tx settlement recompute from public RPC " "(evm tx hash / solana signature / nano send " "block): status, deltas, finality tier. " "Deterministic recompute, not an audit.", } def bazaar_extension(path, method="POST"): """PayAI/x402 Bazaar discovery declaration (BodyDiscoveryExtension shape, matching @x402/extensions declareDiscoveryExtension).""" m = ENDPOINT_META.get(path) if not m: return None return {"info": {"input": {"type": "http", "method": method, "bodyType": "json", "body": m["example_in"]}, "output": {"type": "json", "example": m["example_out"]}}, "schema": { "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", "properties": { "input": {"type": "object", "properties": { "type": {"type": "string", "const": "http"}, "method": {"type": "string", "enum": ["GET", "POST", "PUT", "PATCH"]}, "bodyType": {"type": "string", "enum": ["json", "form-data", "text"]}, "body": m["schema"]}, "required": ["type", "method", "bodyType", "body"], "additionalProperties": False}, "output": {"type": "object", "properties": { "type": {"type": "string"}, "example": {"type": "object"}}, "required": ["type"]}}, "required": ["input"]}} def payment_required_payload(path=None, method="POST"): """Full x402 v2 PaymentRequired object: resource metadata, accepts, and the bazaar discovery extension (echoed by v2 clients to the facilitator so the resource can be catalogued).""" base = public_base() pr = {"x402Version": 2, "accepts": accepts(path, method)} if path: pr["resource"] = { "url": (base + path) if base else path, "description": ENDPOINT_DESCR.get( path, "ARION paid endpoint"), "mimeType": "application/json", "serviceName": "ARION x402 microservices", "tags": ["verification", "static-analysis", "agent", "deterministic"]} ext = bazaar_extension(path, method) if ext: pr["extensions"] = {"bazaar": ext} return pr ROUTE_NAMES = { "/v1/address-check": "Chain address format validation", "/v1/prescan": "Code pattern pre-review", "/v1/dep-check": "Dependency supply-chain pre-scan", "/v1/settle-check": "Single-tx settlement recompute", } def well_known(): """/.well-known/x402 manifest, built per-request so url fields follow the recorded tunnel URL (current_url.txt) across restarts. `payment` + `resourceDetails` follow the seller-manifest pattern indexers (agent402.tools) read for displayName/price/network.""" base = public_base() resources = ["POST /v1/address-check", "POST /v1/prescan", "POST /v1/dep-check", "POST /v1/settle-check", "GET /v1/address-check?chain&address", "GET /v1/prescan?scanner&code&filename", "GET /v1/dep-check?manifest&filename", "GET /v1/settle-check?chain&tx&min_conf"] details = [] for res in resources: method, route_q = res.split(" ", 1) route = route_q.split("?")[0] details.append( {"url": (base + route) if base else route, "route": res, "method": method, "name": ROUTE_NAMES[route], "description": ENDPOINT_DESCR.get( route, "ARION paid endpoint"), "price": 0.001, "priceUsd": 0.001, "currency": "USDC", "network": "eip155:8453"}) return {"version": 1, "name": "ARION", "url": base, "homepage": base, "documentation": ((base + "/openapi.json") if base else "/openapi.json"), "description": "Deterministic micro-verification APIs by ARION, " "an autonomous agent (human-supervised, " "three-law constitution): address-format " "validation, code prescan, dependency " "supply-chain pre-scan, single-tx settlement " "recompute. Pre-scans, not audits. USDC on " "Base or XNO per call via x402.", "instructions": "All endpoints answer 402 (GET parity). " "Free: GET /health /v1/price " "/v1/requirements /v1/card /openapi.json. " "AI authorship disclosed; deterministic " "recompute, never a full audit.", "resources": resources, "resourceDetails": details, "payment": {"x402": { "version": 2, "currency": "USDC", "networks": ["eip155:8453", "nano:mainnet"], "primaryNetwork": "eip155:8453", "priceRange": "$0.001", "payTo": EVM_PAY_TO, "payToByNetwork": {"eip155:8453": EVM_PAY_TO, "nano:mainnet": PAY_TO}, "nonCustodial": True}}, "ownershipProofs": []} OPENAPI = { "openapi": "3.0.1", "info": {"title": "ARION x402 microservices", "version": "1.0.0", "description": "Deterministic paid endpoints by ARION, an " "autonomous agent under a published 3-law " "constitution. prescan is a pattern pre-review, " "not an audit."}, "servers": [{"url": "/"}], "paths": {p: {"post": { "summary": ("validate a chain address" if "address" in p else "dependency supply-chain pre-scan (not an audit)" if "dep" in p else "single-tx settlement recompute (not an audit)" if "settle" in p else "deterministic pattern pre-review (not an audit)"), "x-payment-info": {"protocols": ["x402"], "pricing": {"usdc_base": PRICE_USDC, "xno_raw": PRICE_RAW}}, "requestBody": {"required": True, "content": {"application/json": {"schema": m["schema"], "example": m["example_in"]}}}, "responses": {"200": {"description": "paid result"}, "402": {"description": "payment required"}}}} for p, m in ENDPOINT_META.items()}} def load_consumed(): try: return set(json.load(open(STATE_FILE))) except Exception: return set() def save_consumed(s): tmp = STATE_FILE + ".tmp" json.dump(sorted(s), open(tmp, "w")) os.replace(tmp, STATE_FILE) CONSUMED = load_consumed() def load_musedin_secret(): try: return open(MUSEDIN_WH_SECRET_FILE).read().strip() except Exception: return "" def append_musedin_event(rec): try: with open(MUSEDIN_WH_LOG, "a") as f: f.write(json.dumps(rec, sort_keys=True) + "\n") except Exception: pass def musedin_delivery_id(body): # MusedIn pushes are at-least-once; dedup keys on the delivery id inside # the payload (field name not yet documented — probe the likely keys). if not isinstance(body, dict): return "" for k in ("delivery_id", "deliveryId", "delivery", "event_id", "id"): v = body.get(k) if isinstance(v, str) and v: return v if isinstance(v, dict): for kk in ("delivery_id", "id"): if isinstance(v.get(kk), str) and v[kk]: return v[kk] return "" def musedin_delivery_seen(did): # Scan the tail of the log instead of a second state file: the log is the # same source the brain reads, so dedup and audit can never disagree. if not did: return False try: with open(MUSEDIN_WH_LOG, "r") as f: lines = f.readlines()[-2000:] except Exception: return False for line in lines: try: rec = json.loads(line) except Exception: continue if not rec.get("sig_ok"): continue if musedin_delivery_id(rec.get("body")) == did: return True return False def pay_required_body(why=None, path=None, method="POST"): b = {"error": "payment_required", "price_raw": PRICE_RAW, "price_xno": PRICE_XNO, "pay_to": PAY_TO, "network": "nano:mainnet", "price_usdc": PRICE_USDC, "pay_to_evm": EVM_PAY_TO, "network_evm": "eip155:8453", "x402Version": 2, "accepts": accepts(path, method), "x402": payment_required_payload(path, method), "x_nano_payment": {"header": "X-Nano-Payment", "value": "send block hash to pay_to", "min_raw": PRICE_RAW}} if why: b["reason"] = why return b class H(BaseHTTPRequestHandler): server_version = "ARION-x402-shim/1.0" protocol_version = "HTTP/1.1" def setup(self): super().setup() self.connection.settimeout(REQUEST_TIMEOUT_SECONDS) def handle(self): if not REQUEST_SEMAPHORE.acquire(blocking=False): try: self.request.sendall( b"HTTP/1.1 503 Service Unavailable\r\n" b"Content-Type: application/json\r\n" b"Connection: close\r\n\r\n" b'{"error":"busy"}') except Exception: pass return try: super().handle() except (socket.timeout, TimeoutError): pass # slow/idle connection expired; release thread quietly finally: REQUEST_SEMAPHORE.release() def log_message(self, fmt, *a): sys.stderr.write("%s %s %s\n" % (time.strftime("%H:%M:%S"), self.address_string(), fmt % a)) def _send(self, code, obj, headers=None): body = json.dumps(obj).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Expose-Headers", "PAYMENT-REQUIRED, PAYMENT-RESPONSE") self.send_header("X-Robots-Tag", "all") for k, v in (headers or {}).items(): self.send_header(k, v) self.end_headers() self.wfile.write(body) def _free(self): p = self.path.split("?")[0] if p in ("/", "/v1/card"): return self._send(200, CARD) if p == "/health": return self._send(200, {"ok": True, "seller": "ARION", "consumed": len(CONSUMED)}) if p == "/v1/price": return self._send(200, {"price_raw": PRICE_RAW, "price_xno": PRICE_XNO, "pay_to": PAY_TO, "network": "nano:mainnet", "price_usdc": PRICE_USDC, "pay_to_evm": EVM_PAY_TO, "network_evm": "eip155:8453"}) if p == "/v1/requirements": return self._send(200, {"x402Version": 2, "accepts": accepts()}) if p == "/openapi.json": return self._send(200, OPENAPI) if p in ("/.well-known/x402", "/.well-known/x402.json"): return self._send(200, well_known()) if p == MUSEDIN_WH_PATH: return self._send(200, {"route": "musedin_webhook", "method": "POST", "configured": bool(load_musedin_secret())}) if p == "/robots.txt": body = b"User-agent: *\nAllow: /\n" self.send_response(200) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", str(len(body))) self.end_headers() return self.wfile.write(body) if p == "/.well-known/402index-verify.txt": body = (IDX_VERIFY_HASH + "\n").encode() self.send_response(200) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", str(len(body))) self.end_headers() return self.wfile.write(body) return self._send(404, {"error": "not_found"}) def _402(self, p): pr = base64.b64encode(json.dumps( payment_required_payload(p, self.command) ).encode()).decode() return self._send(402, pay_required_body(path=p, method=self.command), {"PAYMENT-REQUIRED": pr, "x402-price": "0.001", "x402-asset": "USDC", "x402-network": "eip155:8453", "x402-pay-to": EVM_PAY_TO}) def _err402(self, why, p): """Error-path 402: still a full v2 PaymentRequired (header + resource binding) so a retrying client keeps the requirement it needs to pay (audit S2 2026-10-09).""" pr = base64.b64encode(json.dumps( payment_required_payload(p, self.command) ).encode()).decode() return self._send(402, pay_required_body(why, p, self.command), {"PAYMENT-REQUIRED": pr}) def _has_payment(self): return bool(self.headers.get("PAYMENT-SIGNATURE") or self.headers.get("X-Nano-Payment")) def _has_v1_payment(self): # x402 v1 clients send X-PAYMENT; we only speak v2 — tell them # instead of looking identical to an unpaid request (audit S1). return bool(self.headers.get("X-PAYMENT")) def do_GET(self): p = self.path.split("?")[0] if p in PAID_PATHS: if not self._has_payment(): if self._has_v1_payment(): return self._err402( "x402_v1_unsupported: retry with " "PAYMENT-SIGNATURE (x402 v2)", p) return self._402(p) qs = parse_qs(urlparse(self.path).query) req_body = {k: v[0] for k, v in qs.items()} return self._paid(req_body, p) self._free() def do_HEAD(self): p = self.path.split("?")[0] if p in PAID_PATHS: pr = base64.b64encode(json.dumps( payment_required_payload(p, "GET")).encode()).decode() self.send_response(402) self.send_header("Content-Type", "application/json") self.send_header("PAYMENT-REQUIRED", pr) self.send_header("x402-price", "0.001") self.send_header("x402-asset", "USDC") self.send_header("x402-network", "eip155:8453") self.send_header("x402-pay-to", EVM_PAY_TO) self.end_headers() return known = ("/", "/v1/card", "/health", "/v1/price", "/v1/requirements", "/robots.txt", "/openapi.json", "/.well-known/x402", "/.well-known/x402.json", "/.well-known/402index-verify.txt") self.send_response(200 if p in known else 404) self.send_header("Content-Type", "text/plain" if p in ("/robots.txt", "/.well-known/402index-verify.txt") else "application/json") self.end_headers() def do_OPTIONS(self): self.send_response(204) self.send_header("Allow", "GET, HEAD, POST, OPTIONS") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, HEAD, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, PAYMENT-SIGNATURE, X-Nano-Payment, " "X-PAYMENT") self.send_header("Access-Control-Expose-Headers", "PAYMENT-REQUIRED, PAYMENT-RESPONSE") self.end_headers() def do_POST(self): p = self.path.split("?")[0] if p == MUSEDIN_WH_PATH: return self._musedin_webhook() if p not in PAID_PATHS: return self._send(404, {"error": "not_found"}) try: n = int(self.headers.get("Content-Length") or 0) except ValueError: return self._send(400, {"error": "bad_content_length"}) if n < 0: return self._send(400, {"error": "bad_content_length"}) if n > 262144: return self._send(413, {"error": "payload_too_large"}) try: raw = self.rfile.read(n) if n else b"" except (socket.timeout, TimeoutError): return self._send(408, {"error": "request_timeout"}) if not self._has_payment(): if self._has_v1_payment(): return self._err402( "x402_v1_unsupported: retry with " "PAYMENT-SIGNATURE (x402 v2)", p) return self._402(p) try: req_body = json.loads(raw.decode() or "{}") except Exception: return self._send(400, {"error": "bad_json"}) return self._paid(req_body, p) def _musedin_webhook(self): try: n = int(self.headers.get("Content-Length") or 0) except ValueError: return self._send(400, {"error": "bad_content_length"}) if n <= 0: return self._send(400, {"error": "bad_content_length"}) if n > 65536: return self._send(413, {"error": "payload_too_large"}) try: raw = self.rfile.read(n) except (socket.timeout, TimeoutError): return self._send(408, {"error": "request_timeout"}) secret = load_musedin_secret() try: got = bytes.fromhex( self.headers.get("X-MusedIn-Signature") or "") except ValueError: got = b"" ok = bool(secret) and hmac.compare_digest( hmac.new(secret.encode(), raw, hashlib.sha256).digest(), got) rec = {"received_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "sig_ok": ok} try: rec["body"] = json.loads(raw.decode() or "{}") except Exception: rec["parse_error"] = True did = musedin_delivery_id(rec.get("body")) if did: rec["delivery_id"] = did if ok and did and musedin_delivery_seen(did): rec["dedup"] = True append_musedin_event(rec) if not secret: return self._send(503, {"error": "webhook_secret_not_configured"}) if not ok: return self._send(401, {"error": "bad_signature"}) return self._send(200, {"ok": True, "dedup": bool(rec.get("dedup"))}) def _paid(self, req_body, p): sig = self.headers.get("PAYMENT-SIGNATURE") xhash = self.headers.get("X-Nano-Payment") if p == "/v1/prescan": # validate before settle so malformed scans never burn a payment name = req_body.get("scanner") if name not in PRESCANNERS or \ not isinstance(req_body.get("code"), str) or \ not req_body.get("code"): return self._send(400, {"error": "bad_prescan_request", "scanners": sorted(PRESCANNERS)}) if p == "/v1/address-check" and not address_check_valid(req_body): return self._send(400, {"error": "bad_address_check_request", "chains": sorted( set(addresscheck.CHAINS))}) if p == "/v1/dep-check" and not dep_check_valid(req_body): return self._send(400, {"error": "bad_dep_check_request", "manifests": sorted(DEP_MANIFESTS)}) if p == "/v1/settle-check" and not settle_check_valid(req_body): return self._send(400, {"error": "bad_settle_check_request", "chains": list(SETTLE_CHAINS)}) if xhash: return self._pay_by_hash(xhash.strip(), req_body, p) return self._pay_by_x402(sig, req_body, p) def _pay_by_hash(self, h, req_body, p): if not valid_nano_block_hash(h): return self._err402("payment_invalid:bad_hash_format", p) if h in CONSUMED: return self._err402("payment_reused", p) code, r = jget(pk_verify_url(h)) if r.get("ok") is not True: return self._err402( "payment_invalid:" + str(r.get("reason") or r.get("error") or "unreachable")[:120], p) with CONSUMED_LOCK: if h in CONSUMED: return self._err402("payment_reused", p) CONSUMED.add(h) try: save_consumed(CONSUMED) except Exception: CONSUMED.discard(h) return self._send(503, {"error": "state_persist_failed"}) return self._serve(req_body, {"paid_by": "x-nano-payment", "block": h}, path=p) def _pay_by_x402(self, sig, req_body, p): try: payload = json.loads(base64.b64decode(sig).decode()) except Exception: return self._err402("bad_payment_signature", p) acc = payload.get("accepted") or {} inner = payload.get("payload") or {} net = payload.get("network") or acc.get("network") or "" if net in ("eip155:8453", "base") or \ (not net and "authorization" in inner): return self._pay_by_x402_evm(payload, req_body, p) block = (payload.get("payload") or {}).get("block", "") bh = "" if isinstance(block, str): bh = block.upper() reqs = requirements() code, v = jpost(FACILITATOR + "/verify", {"x402Version": 2, "paymentPayload": payload, "paymentRequirements": reqs}) if not v.get("isValid"): return self._err402( "payment_invalid:" + str(v.get("invalidReason") or v.get("error")), p) if bh and bh in CONSUMED: return self._err402("payment_reused", p) code, s = jpost(FACILITATOR + "/settle", {"x402Version": 2, "paymentPayload": payload, "paymentRequirements": reqs}) if not s.get("success"): return self._err402( "settle_failed:" + str(s.get("errorReason") or s.get("error")), p) if bh: with CONSUMED_LOCK: if bh in CONSUMED: return self._err402("payment_reused", p) CONSUMED.add(bh) try: save_consumed(CONSUMED) except Exception: CONSUMED.discard(bh) return self._send(503, {"error": "state_persist_failed"}) pr = base64.b64encode(json.dumps( {"success": True, "transaction": s.get("transaction"), "network": "nano:mainnet", "payer": s.get("payer")} ).encode()).decode() return self._serve(req_body, {"paid_by": "x402-exact-nano", "tx": s.get("transaction")}, {"PAYMENT-RESPONSE": pr}, path=p) def _pay_by_x402_evm(self, payload, req_body, p): auth = (payload.get("payload") or {}).get("authorization") or {} nonce = auth.get("nonce") or "" key = "evm:%s" % nonce if nonce and key in CONSUMED: return self._err402("payment_reused", p) reqs = requirements_evm() v = s = None used = None for f in EVM_FACILITATORS: code, v = jpost(f + "/verify", {"x402Version": 2, "paymentPayload": payload, "paymentRequirements": reqs}) if code == 0: continue used = f if v.get("isValid"): code, s = jpost(f + "/settle", {"x402Version": 2, "paymentPayload": payload, "paymentRequirements": reqs}) break if used is None: return self._err402("payment_invalid:facilitators_unreachable", p) if not v.get("isValid"): return self._err402( "payment_invalid:" + str(v.get("invalidReason") or v.get("error")), p) if not s.get("success"): return self._err402( "settle_failed:" + str(s.get("errorReason") or s.get("error")), p) if nonce: with CONSUMED_LOCK: if key in CONSUMED: return self._err402("payment_reused", p) CONSUMED.add(key) try: save_consumed(CONSUMED) except Exception: CONSUMED.discard(key) return self._send(503, {"error": "state_persist_failed"}) pr = base64.b64encode(json.dumps( {"success": True, "transaction": s.get("transaction"), "network": "eip155:8453", "payer": s.get("payer")} ).encode()).decode() return self._serve(req_body, {"paid_by": "x402-exact-usdc-base", "tx": s.get("transaction"), "facilitator": used}, {"PAYMENT-RESPONSE": pr}, path=p) def _serve(self, req_body, receipt, headers=None, path="/v1/address-check"): if path in ("/v1/prescan", "/v1/dep-check", "/v1/settle-check"): run = {"/v1/prescan": run_prescan, "/v1/dep-check": run_depcheck, "/v1/settle-check": run_settlecheck}[path] code, out = run(req_body) if code != 200: # payment already settled; report honestly, keep receipt out["receipt"] = receipt out["refund_hint"] = PAY_TO return self._send(code, out, headers) out.setdefault("seller", "ARION") out["receipt"] = receipt out["served_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) if path != "/v1/settle-check": out["scope"] = ( "deterministic pattern pre-review, not an audit" if path == "/v1/prescan" else "dependency supply-chain pre-scan, not an " "audit, no CVE feed") return self._send(200, out, headers) verdict = addresscheck.check(req_body.get("chain"), req_body.get("address")) out = {"verdict": verdict, "seller": "ARION", "receipt": receipt, "served_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} return self._send(200, out, headers) if __name__ == "__main__": sys.stderr.write("shim on :%d payTo=%s price=%s raw\n" % (PORT, PAY_TO, PRICE_RAW)) ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()