"""Two fresh loopback-only Litecoin Core regtest nodes; no real funds. python offline-wallet-test.py.txt /path/litecoind /path/litecoin-cli output-directory Python 3 standard library. Generates new disposable wallets and directories. Never accepts an existing data directory, config, wallet, seed or RPC password. """ from datetime import datetime, timezone from decimal import Decimal from pathlib import Path import csv import hashlib import json import shutil import socket import subprocess import sys import tempfile import time def utc(): return datetime.now(timezone.utc).isoformat() def serial(value): if isinstance(value, Decimal): return format(value, "f") if isinstance(value, dict): return {k: serial(v) for k, v in value.items()} if isinstance(value, list): return [serial(v) for v in value] return value def write(path, value): path.write_text(json.dumps(serial(value), indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def main(daemon, cli, output): output.mkdir(parents=True, exist_ok=True) root = Path(tempfile.mkdtemp(prefix="lw-offline-wallet-regtest-")) reservations = [] for _ in range(4): sock = socket.socket() sock.bind(("127.0.0.1", 0)) reservations.append(sock) ports = [s.getsockname()[1] for s in reservations] assertions, trace, lifecycle, snapshots = [], [], [], [] started = utc() common = ["-regtest", "-connect=0", "-dnsseed=0", "-dns=0", "-discover=0", "-listenonion=0", "-upnp=0", "-onlynet=ipv4", "-maxconnections=12", "-dbcache=32", "-par=1", "-fallbackfee=0.00001", "-acceptnonstdtxn=0"] def check(name, condition, details=""): assertions.append({"name": name, "passed": bool(condition), "details": details}) if not condition: raise AssertionError(name + ": " + details) class Node: def __init__(self, label, offset): self.label, self.dir = label, root / label self.dir.mkdir() self.rpcport, self.p2pport = ports[offset:offset + 2] self.reservations = reservations[offset:offset + 2] self.process, self.log = None, None self.base = ["-regtest", "-datadir=" + str(self.dir), "-rpcport=" + str(self.rpcport)] def start(self): for s in self.reservations: s.close() self.log = (self.dir / "process.log").open("ab") flags = common + ["-datadir=" + str(self.dir), "-rpcport=" + str(self.rpcport), "-rpcbind=127.0.0.1", "-rpcallowip=127.0.0.1", "-port=" + str(self.p2pport), "-bind=127.0.0.1:" + str(self.p2pport)] self.process = subprocess.Popen([str(daemon), *flags], stdout=self.log, stderr=self.log) self.rpc("getblockchaininfo", wait=True) lifecycle.append({"time_utc": utc(), "node": self.label, "event": "started", "flags": common, "rpc_bound": "127.0.0.1:" + str(self.rpcport), "p2p_bound": "127.0.0.1:" + str(self.p2pport)}) def rpc(self, method, *args, wallet=None, wait=False, error_expected=False): command = [str(cli), *self.base] if wait: command.append("-rpcwait") if wallet: command.append("-rpcwallet=" + wallet) values = [json.dumps(a, separators=(",", ":")) if isinstance(a, (dict, list, bool)) else str(a) for a in args] command += [method, *values] response = subprocess.run(command, capture_output=True, text=True, timeout=40) if response.returncode: error = response.stderr.strip() trace.append({"time_utc": utc(), "node": self.label, "wallet": wallet, "method": method, "params": values, "expected_error": error_expected, "error": error}) if error_expected: return {"error": error} raise RuntimeError(method + ": " + error) try: value = json.loads(response.stdout, parse_float=Decimal) if response.stdout.strip() else None except json.JSONDecodeError: value = response.stdout.strip() # CLI prints string results without JSON quotes. trace.append({"time_utc": utc(), "node": self.label, "wallet": wallet, "method": method, "params": values, "result": serial(value)}) return value def stop(self): if self.process and self.process.poll() is None: self.rpc("stop") self.process.wait(timeout=30) lifecycle.append({"time_utc": utc(), "node": self.label, "event": "stopped", "exit_code": self.process.returncode}) if self.log: self.log.close() def peers(self): fields = ("addr", "addrbind", "inbound", "connection_type", "subver", "startingheight", "network") return [{k: p[k] for k in fields if k in p} for p in self.rpc("getpeerinfo")] sender, recipient = Node("sender", 0), Node("recipient", 2) def wait_for(predicate, label, timeout=35): deadline = time.monotonic() + timeout while time.monotonic() < deadline: if predicate(): return time.sleep(0.1) raise TimeoutError(label) def state(stage): item = {"stage": stage, "observed_at_utc": utc(), "sender_height": sender.rpc("getblockcount"), "recipient_height": recipient.rpc("getblockcount"), "recipient_balance_ltc": recipient.rpc("getbalance", wallet="recipient"), "recipient_balances": recipient.rpc("getbalances", wallet="recipient"), "sender_peers": sender.peers(), "recipient_peers": recipient.peers()} snapshots.append(item) return item try: check("Four distinct ephemeral ports", len(set(ports)) == 4) sender.start(); recipient.start() for node in (sender, recipient): check(node.label + " uses regtest", node.rpc("getblockchaininfo")["chain"] == "regtest") check(node.label + " initially has zero peers", node.rpc("getconnectioncount") == 0) version = node.rpc("getnetworkinfo")["subversion"] check(node.label + " version is 0.21.5.8", "0.21.5.8" in version, version) sender.rpc("createwallet", "sender") recipient.rpc("createwallet", "recipient") mining_address = sender.rpc("getnewaddress", "test-funding", "bech32", wallet="sender") addresses = [recipient.rpc("getnewaddress", "offline-receipt-" + str(i + 1), "bech32", wallet="recipient") for i in range(3)] for address in addresses: details = recipient.rpc("getaddressinfo", address, wallet="recipient") check("Recipient owns pre-generated address " + address, details["ismine"] is True) check("Native SegWit address " + address, details["iswitness"] is True and details["witness_version"] == 0) sender.rpc("generatetoaddress", 101, mining_address) recipient.rpc("addnode", "127.0.0.1:" + str(sender.p2pport), "onetry") wait_for(lambda: recipient.rpc("getblockcount") == 101, "initial recipient sync") before = state("before_shutdown") check("Initial recipient height 101", before["recipient_height"] == 101) check("Initial recipient balance zero", Decimal(before["recipient_balance_ltc"]) == 0) check("Initial tips equal", sender.rpc("getbestblockhash") == recipient.rpc("getbestblockhash")) check("Initial peers are loopback only", all(p["addr"].startswith("127.0.0.1:") for p in before["sender_peers"] + before["recipient_peers"])) recipient.stop() check("Recipient daemon exited cleanly", recipient.process.poll() == 0) wait_for(lambda: sender.rpc("getconnectioncount") == 0, "stopped peer disconnect") payments = [] for index, (address, amount) in enumerate(zip(addresses, ("0.125", "0.02", "0.03")), 1): check("Recipient remains stopped during payment " + str(index), recipient.process.poll() == 0) txid = sender.rpc("sendtoaddress", address, amount, wallet="sender") blocks = sender.rpc("generatetoaddress", 1, mining_address) wallet_tx = sender.rpc("gettransaction", txid, wallet="sender") decoded = sender.rpc("decoderawtransaction", wallet_tx["hex"]) matches = [o for o in decoded["vout"] if address in o["scriptPubKey"].get("addresses", [])] check("Exactly one intended recipient output " + str(index), len(matches) == 1) output_value = Decimal(matches[0]["value"]) check("Output equals full requested amount " + str(index), output_value == Decimal(amount)) check("Output is transparent native SegWit " + str(index), matches[0]["scriptPubKey"]["type"] == "witness_v0_keyhash") check("Sender confirms payment " + str(index), wallet_tx["confirmations"] == 1) payments.append({"payment_id": "P" + str(index), "txid": txid, "recipient_address": address, "output_n": matches[0]["n"], "output_value_ltc": output_value, "sender_confirmation_height": sender.rpc("getblockcount"), "block_hash": blocks[0], "recipient_daemon_stopped": True}) sender_height = sender.rpc("getblockcount") check("Sender reached 104 while recipient stopped", sender_height == 104) offline = {"stage": "sender_confirmed_recipient_stopped", "observed_at_utc": utc(), "sender_height": sender_height, "recipient_height": None, "recipient_balance_ltc": None, "recipient_last_known_height": 101, "recipient_last_known_balance_ltc": "0.00000000", "recipient_process_exit_code": recipient.process.poll(), "sender_peers": sender.peers(), "recipient_peers": []} snapshots.append(offline) check("Sender has zero peers during offline payments", offline["sender_peers"] == []) for payment in payments: confirmations = sender.rpc("gettransaction", payment["txid"], wallet="sender")["confirmations"] payment["sender_confirmations_at_height_104"] = confirmations check("Independent confirmation-depth arithmetic " + payment["payment_id"], confirmations == 104 - payment["sender_confirmation_height"] + 1) check("Output values total 0.175 LTC", sum((p["output_value_ltc"] for p in payments), Decimal(0)) == Decimal("0.175")) recipient.start() if "recipient" not in recipient.rpc("listwallets"): recipient.rpc("loadwallet", "recipient") disconnected = state("restarted_without_peers") check("Restarted recipient remains at height 101", disconnected["recipient_height"] == 101) check("Restarted recipient still shows zero", Decimal(disconnected["recipient_balance_ltc"]) == 0) check("Restarted recipient has no peers", disconnected["recipient_peers"] == []) for payment in payments: missing = recipient.rpc("gettransaction", payment["txid"], wallet="recipient", error_expected=True) check("Unknown locally before catch-up " + payment["payment_id"], "Invalid or non-wallet transaction id" in missing.get("error", "")) recipient.rpc("addnode", "127.0.0.1:" + str(sender.p2pport), "onetry") wait_for(lambda: recipient.rpc("getblockcount") == 104 and recipient.rpc("getbalance", wallet="recipient") == Decimal("0.175"), "recipient catch-up") after = state("after_loopback_catchup") check("Final recipient height 104", after["recipient_height"] == 104) check("Final recipient balance 0.175 LTC", Decimal(after["recipient_balance_ltc"]) == Decimal("0.175")) check("Final trusted balance 0.175 LTC", Decimal(after["recipient_balances"]["mine"]["trusted"]) == Decimal("0.175")) check("Final tips equal", sender.rpc("getbestblockhash") == recipient.rpc("getbestblockhash")) check("Final peers are loopback only", len(after["recipient_peers"]) == 1 and all(p["addr"].startswith("127.0.0.1:") for p in after["sender_peers"] + after["recipient_peers"])) for payment in payments: transaction = recipient.rpc("gettransaction", payment["txid"], wallet="recipient") payment["recipient_confirmations_after_catchup"] = transaction["confirmations"] payment["recipient_wallet_amount_ltc"] = transaction["amount"] check("Recipient value matches output " + payment["payment_id"], transaction["amount"] == payment["output_value_ltc"]) check("Recipient confirmation count matches sender " + payment["payment_id"], transaction["confirmations"] == payment["sender_confirmations_at_height_104"]) check("Pre-generated address receipt " + payment["payment_id"], recipient.rpc("getreceivedbyaddress", payment["recipient_address"], 1, wallet="recipient") == payment["output_value_ltc"]) received = recipient.rpc("listunspent", 1, 9999999, addresses, wallet="recipient") check("Exactly three transparent unspent receipts", len(received) == 3) check("Independent UTXO sum matches wallet", sum((o["amount"] for o in received), Decimal(0)) == Decimal("0.175")) for node in (recipient, sender): node.stop() check(node.label + " stopped at end", node.process.poll() == 0) result = {"version": "1.0.0", "started_at_utc": started, "completed_at_utc": utc(), "core_version": version, "network": "regtest", "platform": sys.platform, "binary_sha256": {"litecoind": hashlib.sha256(daemon.read_bytes()).hexdigest(), "litecoin-cli": hashlib.sha256(cli.read_bytes()).hexdigest()}, "scope": "Two new disposable nodes, loopback peers only, three ordinary native-SegWit receipts while recipient daemon is stopped. No mainnet timing, GUI/mobile wallet, service credit or MWEB test.", "node_options": common, "snapshots": snapshots, "payments": payments, "peer_lifecycle": lifecycle, "assertions_passed": len(assertions), "assertions": assertions} write(output / "results.json", result) write(output / "rpc-trace.json", {"scope": "Only generated test wallet RPC commands; no private-key export, seed or authentication cookie.", "calls": trace}) with (output / "payments.csv").open("w", encoding="utf-8", newline="") as handle: rows = serial(payments); writer = csv.DictWriter(handle, fieldnames=list(rows[0])); writer.writeheader(); writer.writerows(rows) with (output / "states.csv").open("w", encoding="utf-8", newline="") as handle: fields = ["stage", "observed_at_utc", "sender_height", "recipient_height", "recipient_balance_ltc"] writer = csv.DictWriter(handle, fieldnames=fields); writer.writeheader() for snapshot in serial(snapshots): writer.writerow({key: snapshot.get(key) for key in fields}) print(json.dumps({"success": True, "assertions_passed": len(assertions), "sender_final_height": 104, "recipient_final_height": 104, "recipient_final_ltc": "0.17500000", "output": str(output)})) except Exception: health = {} for node in (sender, recipient): if node.process and node.process.poll() is None: try: health[node.label] = {"peers": node.peers(), "chain": node.rpc("getblockchaininfo")} except Exception: pass write(output / "failed-network.json", health) write(output / "failed-assertions.json", assertions) write(output / "failed-rpc-trace.json", trace) raise finally: for node in (recipient, sender): try: node.stop() except Exception: if node.process and node.process.poll() is None: node.process.terminate() node.process.wait(timeout=10) for sock in reservations: sock.close() # Never delete wallets while one of these processes is still running. if all(n.process is None or n.process.poll() is not None for n in (sender, recipient)): shutil.rmtree(root) if __name__ == "__main__": if len(sys.argv) != 4: raise SystemExit("Usage: python offline-wallet-test.py.txt litecoind litecoin-cli output-directory") main(Path(sys.argv[1]).resolve(), Path(sys.argv[2]).resolve(), Path(sys.argv[3]).resolve())