#!/usr/bin/env python3 """btc_x402.py - paid Bitcoin node data: x402 per call + Lightning prepaid/subscription. Layers: Bitcoin node (data, via local bitcoin-core.cli) -> this server (x402 per-call rail + Lightning recurring rail) -> the UnyKorn chain (chain.unykorn.ai) registry for offers and payment-verification records. GET /health free GET /offer free, offer manifest GET /blockcount /feeestimate paid: X-PAYMENT (x402) or Authorization: Bearer POST /keys create API key POST /topup {key, sats} BOLT11 invoice GET /topup/ poll; credits sats once settled POST /subscribe {key, plan} monthly plan invoice Config: btc402.env (see btc402.env.example) or environment. Stdlib only. Run: python3 btc_x402.py """ import base64, json, os, shutil, subprocess, time, urllib.request def load_env(path=os.environ.get("BTC402_ENV", os.path.join(os.path.dirname(os.path.abspath(__file__)), "btc402.env"))): if os.path.isfile(path): for line in open(path): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) load_env() import lightning, registry # noqa: E402 (read env after loading) from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # noqa: E402 CLI = os.environ.get("BTC_CLI") or shutil.which("bitcoin-core.cli") or "/snap/bin/bitcoin-core.cli" FACILITATOR = os.environ.get("FACILITATOR_URL", "").rstrip("/") PAY_TO = os.environ.get("PAY_TO", "") NETWORK = os.environ.get("X402_NETWORK", "base") ASSET = os.environ.get("X402_ASSET", "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913") # USDC on Base PRICE_ATOMIC = os.environ.get("PRICE_ATOMIC", "1000") # 0.001 USDC PRICE_SATS = int(os.environ.get("PRICE_SATS", "1")) # per call on the API-key path HOST, PORT = os.environ.get("HOST", "127.0.0.1"), int(os.environ.get("PORT", "8402")) ROUTES = { "/blockcount": (["getblockcount"], "Current Bitcoin mainnet block height"), "/feeestimate": (["estimatesmartfee", "6"], "Fee estimate (BTC/kvB) for confirmation within 6 blocks"), } def cfg_summary(): return {"PRICE_ATOMIC": PRICE_ATOMIC, "PRICE_SATS": PRICE_SATS, "X402_NETWORK": NETWORK, "X402_ASSET": ASSET, "PAY_TO": PAY_TO, "BTC_PAY_TO": os.environ.get("BTC_PAY_TO", ""), "PLANS": lightning.PLANS} def node(args): """Whitelisted read-only RPC against the local node. No wallet calls.""" r = subprocess.run([CLI, *args], capture_output=True, text=True, timeout=15) if r.returncode != 0: raise RuntimeError(r.stderr.strip() or "bitcoin-cli failed") out = r.stdout.strip() try: return json.loads(out) except ValueError: return out def requirements(path, desc, host, scheme="https"): return {"scheme": "exact", "network": NETWORK, "maxAmountRequired": PRICE_ATOMIC, "resource": f"{scheme}://{host}{path}", "description": desc, "mimeType": "application/json", "payTo": PAY_TO, "maxTimeoutSeconds": 60, "asset": ASSET, "extra": {"name": "USD Coin", "version": "2"}} def _cdp_creds(): """CDP API key id/secret from CDP_KEY_FILE (KEY=VALUE file) or env. Never logged.""" kid, sec = os.environ.get("CDP_API_KEY_ID", ""), os.environ.get("CDP_API_KEY_SECRET", "") path = os.environ.get("CDP_KEY_FILE", "") if path and os.path.isfile(path): for line in open(path): if "=" in line: k, v = line.strip().split("=", 1) v = v.strip().strip('"').strip("'") if k.strip() == "CDP_API_KEY_ID": kid = v if k.strip() == "CDP_API_KEY_SECRET": sec = v return kid, sec.replace("\\n", "\n") def _b64u(b): return base64.urlsafe_b64encode(b).rstrip(b"=").decode() def cdp_jwt(method, url): """Coinbase CDP bearer JWT: EdDSA for Ed25519 keys, ES256 for PEM EC keys. Needs 'cryptography'.""" from cryptography.hazmat.primitives.asymmetric import ed25519, ec from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature import secrets as _s kid, sec = _cdp_creds() if not kid or not sec: raise RuntimeError("CDP credentials not configured") now = int(time.time()) claims = {"sub": kid, "iss": "cdp", "nbf": now, "exp": now + 120, "uris": [f"{method} {url.split('://', 1)[1]}"]} if "BEGIN" in sec: key, alg = serialization.load_pem_private_key(sec.encode(), None), "ES256" else: key, alg = ed25519.Ed25519PrivateKey.from_private_bytes(base64.b64decode(sec)[:32]), "EdDSA" hdr = {"alg": alg, "kid": kid, "typ": "JWT", "nonce": _s.token_hex(16)} msg = (_b64u(json.dumps(hdr).encode()) + "." + _b64u(json.dumps(claims).encode())).encode() if alg == "EdDSA": sig = key.sign(msg) else: r, s_ = decode_dss_signature(key.sign(msg, ec.ECDSA(hashes.SHA256()))) sig = r.to_bytes(32, "big") + s_.to_bytes(32, "big") return msg.decode() + "." + _b64u(sig) def facilitator(action, payload, reqs): body = json.dumps({"x402Version": 1, "paymentPayload": payload, "paymentRequirements": reqs}).encode() url = f"{FACILITATOR}/{action}" headers = {"Content-Type": "application/json", "User-Agent": "btc402/1"} auth = os.environ.get("FACILITATOR_AUTH", "") # "cdp" or "bearer" if auth == "cdp": headers["Authorization"] = "Bearer " + cdp_jwt("POST", url) elif auth == "bearer" and os.environ.get("FACILITATOR_TOKEN"): headers["Authorization"] = "Bearer " + os.environ["FACILITATOR_TOKEN"] req = urllib.request.Request(url, data=body, headers=headers) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read()) def on_ln_settle(payment_hash, sats, kind): registry.verification_record("lightning", payment_hash, sats, "sats", kind) def call_record(rail, ref, amount, unit, path): return {"X-UnyKorn-Record": registry.header(registry.verification_record(rail, ref, amount, unit, path))} class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *a): # never log Authorization headers print("%s %s" % (self.address_string(), fmt % a)) def send_json(self, code, obj, headers=None): data = json.dumps(obj).encode() self.send_response(code) self.send_header("Content-Type", "application/json") for k, v in (headers or {}).items(): self.send_header(k, v) self.send_header("Content-Length", str(len(data))) self.end_headers(); self.wfile.write(data) def body(self): n = int(self.headers.get("Content-Length") or 0) return json.loads(self.rfile.read(n) or b"{}") if n else {} def do_POST(self): path = self.path.split("?")[0] try: if path == "/keys": return self.send_json(201, lightning.create_key()) b = self.body() if path == "/topup": return self.send_json(201, lightning.topup(b.get("key"), b.get("sats", 0))) if path == "/subscribe": return self.send_json(201, lightning.subscribe(b.get("key"), b.get("plan"))) return self.send_json(404, {"error": "not found"}) except ValueError as e: return self.send_json(400, {"error": str(e)}) except Exception as e: return self.send_json(502, {"error": f"lightning backend: {e}"}) def do_GET(self): path = self.path.split("?")[0] try: if path == "/health": return self.send_json(200, {"ok": True, "blocks": node(["getblockcount"])}) if path == "/offer": return self.send_json(200, registry.offer_manifest(ROUTES, cfg_summary())) if path.startswith("/topup/"): return self.send_json(200, lightning.poll(path[7:], on_ln_settle)) except ValueError as e: return self.send_json(404, {"error": str(e)}) except Exception as e: return self.send_json(500, {"error": str(e)}) if path not in ROUTES: return self.send_json(404, {"error": "not found"}) args, desc = ROUTES[path] _host = self.headers.get("Host", f"{HOST}:{PORT}") _xfp = (self.headers.get("X-Forwarded-Proto") or "").split(",")[0].strip().lower() _local = _host.split(":")[0] in ("127.0.0.1", "localhost", "::1", "[::1]") _scheme = _xfp if _xfp in ("http", "https") else ("http" if _local else "https") reqs = requirements(path, desc, _host, _scheme) auth = self.headers.get("Authorization", "") if auth.startswith("Bearer "): # Lightning-funded API-key path ok, info = lightning.debit(auth[7:].strip(), PRICE_SATS, f"call:{path}") if not ok: return self.send_json(402, {"error": info, "price_sats": PRICE_SATS, "topup": "POST /topup"}) try: result = node(args) hdr = call_record("apikey", f"{auth[-8:]}:{time.time_ns()}", PRICE_SATS, "sats", path) return self.send_json(200, {"result": result, "charged_sats": PRICE_SATS, "account": info}, hdr) except Exception as e: return self.send_json(500, {"error": str(e)}) header = self.headers.get("X-PAYMENT") if not header: return self.send_json(402, {"x402Version": 1, "error": "X-PAYMENT header is required", "accepts": [reqs], "alternative": {"apiKey": "Authorization: Bearer ", "price_sats": PRICE_SATS}}) try: payload = json.loads(base64.b64decode(header)) v = facilitator("verify", payload, reqs) if not v.get("isValid"): return self.send_json(402, {"x402Version": 1, "error": v.get("invalidReason", "invalid payment"), "accepts": [reqs]}) result = node(args) s = facilitator("settle", payload, reqs) if not s.get("success"): return self.send_json(402, {"x402Version": 1, "error": s.get("errorReason", "settlement failed"), "accepts": [reqs]}) hdr = call_record("x402", s.get("transaction") or json.dumps(s, sort_keys=True), PRICE_ATOMIC, "usdc_atomic", path) hdr["X-PAYMENT-RESPONSE"] = base64.b64encode(json.dumps(s).encode()).decode() return self.send_json(200, {"result": result}, hdr) except Exception as e: return self.send_json(500, {"error": str(e)}) if __name__ == "__main__": if not FACILITATOR or not PAY_TO: print("WARNING: FACILITATOR_URL / PAY_TO not set: x402 calls will return 402 but cannot settle.") lightning.conn() print("offer registered:", registry.register_offer(registry.offer_manifest(ROUTES, cfg_summary()))) registry.start_retry() print(f"btc402 listening on http://{HOST}:{PORT}") ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()