"""w42_m7_e2e.py — bounty #136, M7: the *wallet* half, proven end to end. Runs the REAL patched wallet class from the fork (`cashu.wallet.wallet.Wallet`) against the REAL running mint that serves both rails (bolt11/sat and xmr), on a real regtest Monero chain. Nothing mocked, no raw HTTP shortcuts for the wallet steps: every wallet action below goes through the wallet's own public API (`load_mint`, `request_mint`, `get_mint_quote`, `mint`, `melt_quote`, `melt`) so the checks fail if the wallet cannot talk to the xmr rail. G0 Wallet(url, db, unit="xmr").load_mint() loads the XMR keyset and not the sat one G0b the pre-patch URL shape (/v1/mint/quote/bolt11 with unit=xmr) is REJECTED by the mint -> that is exactly the wall the unpatched wallet hits G1 request_mint() returns a Monero address (95 chars), not a bolt11 invoice; the stored quote is labelled method="xmr" (core/base.py fix) G2 get_mint_quote() through the wallet reports unpaid G3 an external wallet (w2) really pays that address (real tx hash) G4 the wallet's own get_mint_quote() flips to PAID G5 wallet.mint() issues the note; the wallet balance is the full amount G6 wallet.melt_quote(addr?amount=N) returns a quote, method="xmr", amount from the request (this is the call that used to die in bolt11.decode) G7 wallet.melt() pays out G8 the destination subaddress really received the coins (chain check, new incoming tx) G9 the spent proofs are gone from the wallet and read SPENT at the mint G10 /v1/info unchanged (no restart, no keyset rotation) """ import asyncio import json import os import shutil import sys import time import urllib.error import urllib.request sys.path.insert(0, "/home/choka/cashu34") from cashu.wallet.wallet import Wallet # noqa: E402 MINT = "http://127.0.0.1:3338" PW = "http://127.0.0.1:38103/json_rpc" PD = "http://127.0.0.1:38101/json_rpc" AMT = 1048576 # 2**20 piconero DBDIR = os.path.expanduser("~/w42walletdb") RESULTS = [] def check(name, ok, detail=""): RESULTS.append((name, bool(ok), detail)) print("CHECK %-70s %s %s" % (name, "PASS" if ok else "FAIL", detail), flush=True) def rpc(url, method, params=None, allow_error=False): body = json.dumps({"jsonrpc": "2.0", "id": "0", "method": method, "params": params or {}}).encode() req = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}) raw = json.loads(urllib.request.urlopen(req, timeout=180).read().decode()) if raw.get("error"): if allow_error: return raw["error"] raise RuntimeError("%s: %s" % (method, raw["error"])) return raw.get("result", {}) def api(method, path, body=None): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(MINT + path, data=data, method=method, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=90) as f: return f.status, json.loads(f.read().decode()) except urllib.error.HTTPError as e: return e.code, e.read().decode()[:200] def open_w(name): return rpc(PW, "open_wallet", {"filename": name, "password": ""}) def mine(n, addr): return rpc(PD, "generateblocks", {"amount_of_blocks": n, "wallet_address": addr}) def addr_of(name): open_w(name) return rpc(PW, "get_address", {"account_index": 0})["address"] def incoming(account=0): r = rpc(PW, "get_transfers", {"account_index": account, "in": True, "pool": True}) return r.get("in", []) + r.get("pool", []) async def main(): st, info = api("GET", "/v1/info") n5 = sorted((m["method"], m["unit"]) for m in info["nuts"]["5"]["methods"]) check("G0 mint advertises both rails in one process", n5 == [("bolt11", "sat"), ("xmr", "xmr")], str(n5)) st, ks = api("GET", "/v1/keysets") xks = [k for k in ks["keysets"] if k.get("unit") == "xmr"][0] kid = xks["id"] input_fee = -(-int(xks.get("input_fee_ppk", 0) or 0) // 1000) MELT = AMT - input_fee # --- G0b: the URL shape the unpatched wallet used is rejected by the mint ------ st_old, body_old = api("POST", "/v1/mint/quote/bolt11", {"unit": "xmr", "amount": AMT}) check("G0b the pre-patch path /v1/mint/quote/bolt11 with unit=xmr is rejected", st_old >= 400, "http=%s %s" % (st_old, str(body_old)[:80])) # --- the wallet --------------------------------------------------------------- if os.path.isdir(DBDIR): shutil.rmtree(DBDIR) os.makedirs(DBDIR) w = await Wallet.with_db(url=MINT, db=DBDIR, name="w42", unit="xmr") await w.load_mint() units = sorted({k.unit.name for k in w.keysets.values()}) check("G0 Wallet(unit='xmr') loaded the xmr keyset and no other unit", units == ["xmr"], "keysets units=%s ids=%s" % (units, [i[:12] for i in w.keysets])) quote = await w.request_mint(AMT) is_addr = len(str(quote.request)) == 95 and not str(quote.request).startswith("ln") check("G1 request_mint() returned a Monero address, not a bolt11 invoice", is_addr, "request=%s… len=%d" % (str(quote.request)[:18], len(str(quote.request)))) check("G1b the stored quote is labelled with the xmr method (core/base.py fix)", str(quote.method) == "xmr" and str(quote.unit) == "xmr", "method=%s unit=%s" % (quote.method, quote.unit)) q = await w.get_mint_quote(quote.quote) check("G2 the wallet reads its own quote back as UNPAID", str(q.state_val) == "unpaid", "state=%s" % q.state_val) # --- a real external payment -------------------------------------------------- open_w("w2") w2addr = rpc(PW, "get_address", {"account_index": 0})["address"] tr = rpc(PW, "transfer", {"destinations": [{"amount": AMT, "address": quote.request}], "account_index": 0, "priority": 0, "ring_size": 11}) print("### funding tx %s" % tr.get("tx_hash"), flush=True) check("G3 an external wallet really paid the address (real tx)", bool(tr.get("tx_hash")), str(tr.get("tx_hash"))[:20]) mine(1, w2addr) stq = None for _ in range(20): stq = await w.get_mint_quote(quote.quote) if str(stq.state_val) in ("paid", "issued"): break time.sleep(2) check("G4 the wallet's own get_mint_quote() flips to PAID", str(stq.state_val) in ("paid", "issued"), "state=%s" % stq.state_val) if str(stq.state_val) not in ("paid", "issued"): return summary() proofs = await w.mint(AMT, quote.quote) await w.load_proofs() check("G5 wallet.mint() issued the note through the xmr rail", sum(p.amount for p in proofs) == AMT, "proofs=%d" % len(proofs)) check("G5b the wallet balance is the full amount", w.available_balance.amount == AMT, "balance=%s (%s)" % (w.available_balance.amount, w.unit.str(w.available_balance.amount))) # --- melt: the call that used to die in bolt11.decode ------------------------- # How much can this wallet actually melt? amount requested from the address + # the mint's melt fee reserve + the input fee for as many proofs as we hold. ppk = int(xks.get("input_fee_ppk", 0) or 0) in_fee = -(-(len(proofs) * ppk) // 1000) dest_amt = AMT - in_fee print("### %d proofs, input_fee_ppk=%d -> input fee %d, melting %d" % (len(proofs), ppk, in_fee, dest_amt), flush=True) open_w("w2") dest = rpc(PW, "create_address", {"account_index": 0, "label": "m7-melt-dest"})["address"] before = len(incoming()) mq = await w.melt_quote("%s?amount=%d" % (dest, dest_amt)) fr = int(mq.fee_reserve or 0) if fr: dest_amt -= fr mq = await w.melt_quote("%s?amount=%d" % (dest, dest_amt)) check("G6 wallet.melt_quote(monero address) works — no bolt11 decoding off-rail", mq.amount == dest_amt and str(mq.method) == "xmr", "amount=%s method=%s fee_reserve=%s" % (mq.amount, mq.method, mq.fee_reserve)) if mq.amount != dest_amt: return summary() resp = await w.melt(proofs, "%s?amount=%d" % (dest, dest_amt), fr, mq.quote) check("G7 wallet.melt() paid out", str(resp.state) in ("PAID", "paid"), "state=%s" % resp.state) pre = str(getattr(resp, "payment_preimage", "") or "") check("G7b the melt carries the on-chain txid the mint broadcast", len(pre) == 64 and all(c in "0123456789abcdef" for c in pre), "payment_preimage=%s" % pre[:24]) # the mint pays out of its own wallet: wait for the tx to reach the chain hit = [] for _ in range(12): mine(1, w2addr) time.sleep(5) open_w("w2") inc = incoming() hit = [t for t in inc if t.get("address") == dest] if hit: break got = sum(int(t.get("amount", 0)) for t in hit) check("G8 the destination subaddress really received the coins (chain check)", len(hit) >= 1 and got >= dest_amt, "new_incoming=%d received=%d want>=%d" % (len(hit), got, dest_amt)) check("G8b that is more incoming transfers than before the melt", len(inc) > before, "incoming %d -> %d" % (before, len(inc))) await w.load_proofs() check("G9 the spent proofs are gone from the wallet", w.available_balance.amount == 0, "balance=%s (was %d)" % (w.available_balance.amount, AMT)) st, info2 = api("GET", "/v1/info") n52 = sorted((m["method"], m["unit"]) for m in info2["nuts"]["5"]["methods"]) check("G10 /v1/info unchanged at the end (no restart, no keyset rotation)", n52 == n5, str(n52)) open_w("w1") return summary() def summary(): npass = sum(1 for _, ok, _ in RESULTS if ok) print("\n=== %d/%d checks passed ===" % (npass, len(RESULTS)), flush=True) for name, ok, detail in RESULTS: if not ok: print("FAILED:", name, detail, flush=True) return 0 if npass == len(RESULTS) else 1 if __name__ == "__main__": sys.exit(asyncio.run(main()))