#!/usr/bin/env python3 """Bounded 1NFER x402 client. Local keys only; no hidden retries or auto-funding.""" import argparse, base64, json, os, re, secrets, stat, sys, time from decimal import Decimal from pathlib import Path from urllib.error import HTTPError from urllib.request import Request, build_opener, ProxyHandler, HTTPSHandler, HTTPRedirectHandler ORIGIN = 'https://1nfer.net' PATH = '/v1/account/topups/x402' TOKEN = '0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913' NETWORK = 'eip155:8453' class Error(Exception): pass class NoRedirect(HTTPRedirectHandler): def redirect_request(self, *a, **kw): raise Error('Redirect refused') def read(path): fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(fd) as f: st = os.fstat(f.fileno()) if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or stat.S_IMODE(st.st_mode) & 0o077: raise Error('Private files require owner-only permissions') raw = f.read(32769) if len(raw) > 32768: raise Error('Private file too large') return json.loads(raw) def save(path, value, new=False): temp = path if new else path.with_name(path.name + '.tmp') fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, 'w') as f: json.dump(value, f); f.flush(); os.fsync(f.fileno()) if not new: os.replace(temp, path) fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) os.fsync(fd); os.close(fd) def money(raw): value = Decimal(raw) if not value.is_finite() or not 1 <= value <= 2000 or value.as_tuple().exponent < -2: raise Error('Use $1–2000, at most two decimal places') return format(value, '.2f') def request(path, body=None, key=None, proof=None): if not (path in (PATH, '/v1/agents/register', '/v1/account', '/v1/payments/x402') or re.fullmatch('/v1/invoices/inv_[A-Za-z0-9_-]{32}', path)): raise Error('Unsupported fixed-origin route') headers = {'Accept':'application/json', 'User-Agent':'1NFER-x402-client/1.0'} if key: headers['Authorization'] = 'Bearer ' + key if proof: headers['PAYMENT-SIGNATURE'] = base64.b64encode(json.dumps(proof).encode()).decode() data = None if body is None else json.dumps(body).encode() if data is not None: headers['Content-Type'] = 'application/json' opener = build_opener(ProxyHandler({}), NoRedirect(), HTTPSHandler()) req = Request(ORIGIN+path, data=data, headers=headers) try: response = opener.open(req, timeout=40) except HTTPError as exc: response = exc with response: raw = response.read(262145) if len(raw) > 262144: raise Error('Response too large') value = json.loads(raw) if not isinstance(value, dict): raise Error('Invalid response') return response.code, {k.upper():v for k,v in response.headers.items()}, value def requirements(state, receiver, cap): challenge = state['challenge']; rows = challenge.get('accepts') if challenge.get('x402Version') != 2 or challenge.get('resource',{}).get('url') != ORIGIN+PATH or not isinstance(rows,list) or len(rows)!=1: raise Error('Unsupported challenge or resource') req = rows[0] expected = {'scheme':'exact','network':NETWORK,'asset':TOKEN,'amount':str(int(Decimal(state['body']['credit_usd'])*1000000)), 'payTo':receiver,'maxTimeoutSeconds':300,'extra':{'name':'USD Coin','version':'2'}} if req != expected or Decimal(state['body']['credit_usd']) > Decimal(money(cap)): raise Error('Wrong recipient/network/token/amount or authorized cap') if not re.fullmatch('0x[0-9a-fA-F]{40}',receiver) or int(receiver,16)==0: raise Error('Invalid receiver') if state['invoice'].get('expires_at',0) <= int(time.time())+60: raise Error('Quote near expiry; do not sign') return req def signed_payment(state, wallet, receiver, cap): from eth_account import Account from eth_account.messages import encode_typed_data req = requirements(state,receiver,cap) payer = Account.from_key(wallet['private_key']) if wallet.get('address',payer.address).lower()!=payer.address.lower(): raise Error('Wallet address mismatch') now = int(time.time()) auth = {'from':payer.address,'to':receiver,'value':req['amount'],'validAfter':str(now-1), 'validBefore':str(now+240),'nonce':'0x'+secrets.token_hex(32)} types = {'TransferWithAuthorization':[{'name':k,'type':t} for k,t in [('from','address'),('to','address'),('value','uint256'),('validAfter','uint256'),('validBefore','uint256'),('nonce','bytes32')]]} message = {**auth, **{k:int(auth[k]) for k in ('value','validAfter','validBefore')}} typed = encode_typed_data(domain_data={'name':'USD Coin','version':'2','chainId':8453,'verifyingContract':TOKEN}, message_types=types,message_data=message) signature = '0x'+payer.sign_message(typed).signature.hex() return {'x402Version':2,'resource':state['challenge']['resource'],'accepted':req, 'payload':{'authorization':auth,'signature':signature}} def public(value): # Only selected account/payment fields, never keys, signatures or error bodies. return {k:value[k] for k in ('account_id','invoice_id','status','credit_usd','authorization_state','transaction','payment_confirmed') if k in value} def run(a): directory = Path(a.directory).expanduser() if a.command=='register': directory.mkdir(mode=0o700,parents=True,exist_ok=True) st=directory.lstat() if not stat.S_ISDIR(st.st_mode) or st.st_uid!=os.getuid() or stat.S_IMODE(st.st_mode)&0o077: raise Error('Use a private directory outside Git (chmod 700)') # Serialize each directory, including all writes and network outcomes. import fcntl with open(directory/'client.lock','a') as lock: os.chmod(directory/'client.lock',0o600) fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB) account_path=directory/'account.private.json'; intent=directory/'payment.private.json' if a.command=='register': if not account_path.exists():save(account_path,{'api_key':'hs-live-'+secrets.token_hex(32)},True) account=read(account_path) code,_,value=request('/v1/agents/register',{'api_key':account['api_key'],'label':a.label},None) if code not in (200,201): raise Error('Registration HTTP '+str(code)+'; retain the SAME saved key') if account.get('account_id') and account['account_id']!=value.get('account_id'):raise Error('Account identity changed') account['account_id']=value['account_id'];save(account_path,account) return {'registered':True,'account_id':account['account_id'],'key_saved_locally':True} account=read(account_path) if a.command=='quote': code,_,terms=request('/v1/payments/x402') if code!=200 or terms.get('enabled') is not True:raise Error('x402 currently unavailable; native checkout remains separate') usd=money(a.usd) if not intent.exists():save(intent,{'body':{'credit_usd':usd,'idempotency_key':secrets.token_hex(24)}},True) state=read(intent) if state['body']['credit_usd']!=usd:raise Error('Keep the original amount; use a separate directory for another payment') # Unsigned same-intent readback is safe even after a lost submit response. code,headers,value=request(PATH,state['body'],account['api_key']) if code==402: challenge=json.loads(base64.b64decode(headers.get('PAYMENT-REQUIRED',headers.get('Payment-Required','')),validate=True)) state.update(challenge=challenge,invoice=value) elif code in (200,202):state['invoice']=value else:raise Error('Quote HTTP '+str(code)+'; keep intent, no new payment') save(intent,state) return {**public(value),'pay_to':state.get('challenge',{}).get('accepts',[{}])[0].get('payTo'),'network':NETWORK,'asset':'USDC'} state=read(intent) if a.command=='pay': if state.get('submit_intent'): return {'submitted_or_uncertain':True,'next_action':'Run quote with SAME amount to recover invoice, then status. Do not sign or send again.'} if not a.approve_payment:raise Error('Explicit --approve-payment is required') # Balance check before making a spend authorization; fixed public Base RPC. from eth_account import Account wallet=read(Path(a.wallet_file).expanduser());payer=Account.from_key(wallet['private_key']) rpc_req=Request('https://mainnet.base.org',data=json.dumps({'jsonrpc':'2.0','id':1,'method':'eth_call', 'params':[{'to':TOKEN,'data':'0x70a08231'+payer.address[2:].lower().zfill(64)},'latest']}).encode(), headers={'Content-Type':'application/json','User-Agent':'1NFER-x402-client/1.0','Accept':'application/json'}) with build_opener(ProxyHandler({}),NoRedirect(),HTTPSHandler()).open(rpc_req,timeout=20) as r:balance=json.loads(r.read(8192)) if 'result' not in balance or int(balance['result'],16)=deadline: result=public(value) if value.get('status')=='credited': c,_,current=request('/v1/account',key=account['api_key']) result['account_readback_http']=c return result time.sleep(min(20,max(0,deadline-time.monotonic()))) def main(): p=argparse.ArgumentParser(description=__doc__) p.add_argument('--directory',default='~/.local/share/1nfer-buyer') sub=p.add_subparsers(dest='command',required=True) register=sub.add_parser('register');register.add_argument('--label',default='x402-agent') q=sub.add_parser('quote');q.add_argument('--usd',default='1.00') pay=sub.add_parser('pay');pay.add_argument('--wallet-file',required=True);pay.add_argument('--pay-to',required=True) pay.add_argument('--max-usd',required=True);pay.add_argument('--approve-payment',action='store_true') status=sub.add_parser('status');status.add_argument('--wait',type=int,default=0,choices=range(0,1801),metavar='0..1800') args=p.parse_args() try:print(json.dumps(run(args))) except Exception as exc: print(json.dumps({'ok':False,'error_type':type(exc).__name__,'detail':str(exc) if isinstance(exc,Error) else 'Operation interrupted; retain local state. No automatic retry.'}));return 1 return 0 if __name__=='__main__':sys.exit(main())