#!/usr/bin/env python3 """fjcheck - a dependency-free x402 client for the FlapJax Agent/Wallet Checker. Usage: python fjcheck.py 0xADDRESS # full run: challenge -> sign -> retry python fjcheck.py 0xADDRESS --dry-run # stop before sending the payment python fjcheck.py --selftest # offline EIP-712 + ecrecover check What it actually does, in order: 1. GET /check?address=0x.. -> 200 (cached/credited) or 402 2. on 402: base64url-decode the PAYMENT-REQUIRED response header (x402 v2) 3. pick the offer whose network is eip155:8453 and asset is USDC on Base 4. build an EIP-712 TransferWithAuthorization (EIP-3009) over the offer's payTo / amount / maxTimeoutSeconds, using the token's own domain (name, version) taken from the offer's `extra` block 5. sign it with secp256k1, low-s normalised, recoverable v 6. retry the same URL with an X-PAYMENT header carrying the signed payload 7. print whatever came back, including a settlement failure No web3, no eth-account, no requests. stdlib only. """ import json, os, sys, time, base64, secrets, urllib.request, urllib.error HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) from wallet import keccak256, point_mul, point_add, priv_to_address # noqa: E402 BASE = "https://flapjax-check.run402.com" P = 2 ** 256 - 2 ** 32 - 977 N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 G = (0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8) # ---------- secp256k1 ECDSA ---------- def sign(digest: bytes, priv: int) -> tuple: z = int.from_bytes(digest, "big") % N while True: k = secrets.randbelow(N - 1) + 1 R = point_mul(k, G) r = R[0] % N if r == 0: continue s = (pow(k, -1, N) * (z + r * priv)) % N if s == 0: continue v = 27 + (R[1] & 1) + (2 if R[0] >= N else 0) if s > N // 2: # EIP-2 low-s s = N - s v ^= 1 return r, s, v def ecrecover(digest: bytes, r: int, s: int, v: int): x = r + (N if v >= 31 else 0) alpha = (pow(x, 3, P) + 7) % P beta = pow(alpha, (P + 1) // 4, P) y = beta if (beta & 1) == ((v - 27) & 1) else P - beta R = (x, y) z = int.from_bytes(digest, "big") % N rinv = pow(r, -1, N) Q = point_add(point_mul((s * rinv) % N, R), point_mul((-z * rinv) % N, G)) return "0x" + keccak256(Q[0].to_bytes(32, "big") + Q[1].to_bytes(32, "big"))[-20:].hex() # ---------- EIP-712 ---------- def eip712_digest(domain: dict, types: str, values: list) -> bytes: """domain: name/version/chainId/verifyingContract; types: canonical type string.""" dth = keccak256(b"EIP712Domain(string name,string version,uint256 chainId,address verifyingContract)") enc = b"".join([ dth, keccak256(domain["name"].encode()), keccak256(domain["version"].encode()), domain["chainId"].to_bytes(32, "big"), bytes(12) + bytes.fromhex(domain["verifyingContract"][2:].lower()), ]) sep = keccak256(enc) th = keccak256(types.encode()) body = [th] for v in values: body.append(v if isinstance(v, bytes) else int(v).to_bytes(32, "big")) struct = keccak256(b"".join(body)) return keccak256(b"\x19\x01" + sep + struct) AUTH_TYPES = ("TransferWithAuthorization(address from,address to,uint256 value," "uint256 validAfter,uint256 validBefore,bytes32 nonce)") def load_priv() -> int: d = json.load(open(os.path.join(HERE, "wallet_evm.json"))) return int(d["private_key_hex"], 16) def selftest() -> bool: priv = load_priv() addr = priv_to_address(priv)[0] domain = {"name": "USD Coin", "version": "2", "chainId": 8453, "verifyingContract": "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"} d = eip712_digest(domain, AUTH_TYPES, [int(addr, 16), int(addr, 16), 20000, 0, int(time.time()) + 300, secrets.token_bytes(32)]) r, s, v = sign(d, priv) back = ecrecover(d, r, s, v) print("address :", addr) print("digest :", "0x" + d.hex()) print("sig : 0x%064x%064x%02x" % (r, s, v)) print("recovered :", back) print("SELFTEST", "OK" if back.lower() == addr.lower() else "FAIL") return back.lower() == addr.lower() # ---------- HTTP ---------- def http(url: str, headers: dict = None): r = urllib.request.Request(url, headers=headers or {"User-Agent": "fjcheck/1.0"}) try: with urllib.request.urlopen(r, timeout=45) as x: return x.status, dict(x.headers), x.read().decode() except urllib.error.HTTPError as e: return e.code, dict(e.headers), e.read().decode() def decode_challenge(headers: dict): raw = None for k, v in headers.items(): if k.lower() == "payment-required": raw = v break if not raw: return None raw = raw.replace("-", "+").replace("_", "/") return json.loads(base64.b64decode(raw + "=" * (-len(raw) % 4)).decode()) def main() -> int: args = [a for a in sys.argv[1:] if not a.startswith("--")] if "--selftest" in sys.argv: return 0 if selftest() else 1 if not args: print(__doc__) return 2 addr = args[0].lower() dry = "--dry-run" in sys.argv url = f"{BASE}/check?address={addr}" st, hdr, body = http(url) print(f"[1] GET {url}\n -> {st}") if st == 200: print(json.dumps(json.loads(body), indent=1)[:4000]) return 0 ch = decode_challenge(hdr) if not ch: print(" no PAYMENT-REQUIRED header; body was:", body[:400]) return 1 print(f"[2] x402Version={ch['x402Version']} offers={len(ch['accepts'])}") off = next((a for a in ch["accepts"] if a.get("network") == "eip155:8453" and a.get("asset", "").lower() == "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913"), ch["accepts"][0]) print(" scheme=%s network=%s amount=%s (%s USDC) payTo=%s timeout=%ss" % (off["scheme"], off["network"], off["amount"], int(off["amount"]) / 1e6, off["payTo"], off.get("maxTimeoutSeconds"))) priv = load_priv() me = priv_to_address(priv)[0] now = int(time.time()) auth = {"from": me, "to": off["payTo"], "value": int(off["amount"]), "validAfter": 0, "validBefore": now + int(off.get("maxTimeoutSeconds", 300)), "nonce": secrets.token_bytes(32)} domain = {"name": off["extra"]["name"], "version": off["extra"]["version"], "chainId": int(off["network"].split(":")[1]) % (2 ** 256), "verifyingContract": off["asset"]} d = eip712_digest(domain, AUTH_TYPES, [int(auth["from"], 16), int(auth["to"], 16), auth["value"], auth["validAfter"], auth["validBefore"], auth["nonce"]]) r, s, v = sign(d, priv) sig = "0x%064x%064x%02x" % (r, s, v) print(f"[3] domain name={domain['name']!r} version={domain['version']!r} chainId={domain['chainId']}") print(f" digest=0x{d.hex()}") print(f" payer={me} validBefore={auth['validBefore']} (+{auth['validBefore']-now}s)") print(f"[4] recovered={ecrecover(d, r, s, v)} sig={sig[:20]}...") assert ecrecover(d, r, s, v).lower() == me.lower() payload = {"x402Version": 2, "scheme": off["scheme"], "network": off["network"], "payload": {"signature": sig, "authorization": {"from": auth["from"], "to": auth["to"], "value": str(auth["value"]), "validAfter": str(auth["validAfter"]), "validBefore": str(auth["validBefore"]), "nonce": "0x" + auth["nonce"].hex()}}} enc = base64.b64encode(json.dumps(payload).encode()).decode() print(f"[5] X-PAYMENT payload {len(enc)} b64 chars") path = os.path.join(HERE, "x402_payment.json") json.dump(payload, open(path, "w"), indent=1) print(" saved", path) if dry: print("[6] --dry-run: not sending") return 0 st2, h2, b2 = http(url, {"User-Agent": "fjcheck/1.0", "X-PAYMENT": enc}) print(f"[6] retry with X-PAYMENT -> {st2}") print(" ", b2[:900]) return 0 if st2 == 200 else 1 if __name__ == "__main__": sys.exit(main())