"""Reproduce transaction lookup with Litecoin Core 0.21.5.8 on fresh regtest.""" import argparse import base64 from datetime import datetime, timezone import hashlib import http.client import json from pathlib import Path import shutil import socket import subprocess import tempfile import time PIN = "7430c6d0d4cb59d355361a0b90ffbfc4ac725609483e04c242bf95372cc9b8fe" VERSION = "/LitecoinCore:0.21.5.8/" def sha256(path): return hashlib.sha256(path.read_bytes()).hexdigest() class RPCError(RuntimeError): def __init__(self, error): self.error = error super().__init__(f'RPC {error["code"]}: {error["message"]}') class Lab: def __init__(self, directory, executable): self.directory = directory self.executable = executable self.trace = [] self.process = None def start(self, indexed): self.indexed = indexed with socket.socket() as reservation: reservation.bind(("127.0.0.1", 0)) self.port = reservation.getsockname()[1] datadir = self.directory / "data" conf = self.directory / "empty.conf" self.cookie = datadir / "regtest" / ".cookie" argv = [str(self.executable), "-regtest", f"-datadir={datadir}", f"-conf={conf}", "-server=1", "-rpcbind=127.0.0.1", "-rpcallowip=127.0.0.1", f"-rpcport={self.port}", "-listen=0", "-connect=0", "-dnsseed=0", "-discover=0", "-upnp=0", "-listenonion=0", "-prune=0", f"-txindex={int(indexed)}", "-fallbackfee=0.001", f"-disablewallet={int(indexed)}"] with (self.directory / f"daemon-{int(indexed)}.log").open("wb") as output: self.process = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=output, stderr=subprocess.STDOUT, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) deadline = time.monotonic() + 40 while time.monotonic() < deadline: if self.process.poll() is not None: raise RuntimeError("Core exited before RPC startup") try: info = self.rpc("getblockchaininfo", record=False) assert info["chain"] == "regtest" and info["pruned"] is False assert self.rpc("getnetworkinfo", record=False)["subversion"] == VERSION assert self.rpc("getconnectioncount") == 0 return except (OSError, http.client.HTTPException, RPCError): time.sleep(0.2) raise RuntimeError("Core RPC did not become ready") def rpc(self, method, *params, wallet=None, record=True): token = base64.b64encode(self.cookie.read_bytes().strip()).decode("ascii") conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=12) try: payload = json.dumps({"jsonrpc": "1.0", "id": "lookup-test", "method": method, "params": params}) conn.request("POST", "/" if wallet is None else f"/wallet/{wallet}", payload, {"Authorization": f"Basic {token}", "Content-Type": "application/json"}) response = conn.getresponse() packet = json.loads(response.read()) finally: conn.close() if record: self.trace.append({"txindex_enabled": self.indexed, "wallet": wallet, "method": method, "params": params, "result": packet["result"], "error": packet["error"]}) if packet["error"] is not None: raise RPCError(packet["error"]) return packet["result"] def stop(self): if self.process is None or self.process.poll() is not None: return try: self.rpc("stop", record=False) self.process.wait(timeout=25) finally: if self.process.poll() is None: self.process.kill() self.process.wait(timeout=10) def experiment(lab): cases = [] def check(label, actual, expected): if actual != expected: raise AssertionError(f"{label}: expected {expected!r}, got {actual!r}") cases.append({"case": label, "observed": actual}) def check_error(label, expected_code, expected_text, method, *params, wallet=None): try: lab.rpc(method, *params, wallet=wallet) except RPCError as exc: if exc.error["code"] != expected_code or expected_text not in exc.error["message"]: raise AssertionError(f"{label}: unexpected error") from exc cases.append({"case": label, "observed": exc.error}) return raise AssertionError(f"{label}: RPC unexpectedly succeeded") lab.start(indexed=False) check("no-index-status", lab.rpc("getindexinfo", "txindex"), {}) lab.rpc("createwallet", "maker") lab.rpc("createwallet", "observer") mining = lab.rpc("getnewaddress", "", "legacy", wallet="maker") lab.rpc("generatetoaddress", 101, mining) receiving = lab.rpc("getnewaddress", "", "legacy", wallet="maker") txid = lab.rpc("sendtoaddress", receiving, 1, wallet="maker") pending = lab.rpc("getrawtransaction", txid, True) check("txid-before-mining", pending["txid"], txid) check("mempool-before-mining", txid in lab.rpc("getrawmempool"), True) signed = bytes.fromhex(pending["hex"]) check("raw-bytes-hash-to-txid", hashlib.sha256(hashlib.sha256(signed).digest()).digest()[::-1].hex(), txid) output = next(out for out in pending["vout"] if out["scriptPubKey"]["addresses"] == [receiving] and out["value"] == 1) vout = output["n"] payment_block = lab.rpc("generatetoaddress", 1, mining)[0] check("mempool-after-mining", txid in lab.rpc("getrawmempool"), False) check_error("confirmed-bare-lookup-without-index", -5, "No such mempool transaction", "getrawtransaction", txid, True) explicit = lab.rpc("getrawtransaction", txid, True, payment_block) check("explicit-block-lookup", {"txid": explicit["txid"], "blockhash": explicit["blockhash"], "in_active_chain": explicit["in_active_chain"], "confirmations": explicit["confirmations"]}, {"txid": txid, "blockhash": payment_block, "in_active_chain": True, "confirmations": 1}) check("same-bytes-after-mining", explicit["hex"], pending["hex"]) header = lab.rpc("getblockheader", payment_block) check("payment-block-height-and-depth", [header["height"], header["confirmations"]], [102, 1]) containing_block = lab.rpc("getblock", payment_block, 1) check("payment-in-block-body", txid in containing_block["tx"], True) check_error("wrong-known-block", -5, "No such transaction found in the provided block", "getrawtransaction", txid, True, header["previousblockhash"]) check_error("unknown-block", -5, "Block hash not found", "getrawtransaction", txid, True, "0" * 64) coin = lab.rpc("gettxout", txid, vout) check("utxo-despite-bare-lookup-error", [coin["value"], coin["confirmations"]], [1, 1]) maker = lab.rpc("gettransaction", txid, wallet="maker") check("involved-wallet-knows-tx", [maker["txid"], maker["confirmations"]], [txid, 1]) check_error("unrelated-wallet", -5, "Invalid or non-wallet transaction id", "gettransaction", txid, wallet="observer") lab.stop() lab.start(indexed=True) deadline = time.monotonic() + 40 while time.monotonic() < deadline: indexes = lab.rpc("getindexinfo", "txindex") if indexes.get("txindex", {}).get("synced") is True: break time.sleep(0.2) else: raise RuntimeError("Transaction index did not catch up") check("index-height", indexes["txindex"]["best_block_height"], 102) check("same-tip-after-restart", lab.rpc("getbestblockhash"), payment_block) indexed = lab.rpc("getrawtransaction", txid, True) check("bare-indexed-lookup", [indexed["txid"], indexed["blockhash"], indexed["confirmations"]], [txid, payment_block, 1]) check("same-bytes-after-indexing", indexed["hex"], pending["hex"]) check("bare-lookup-omits-in-active-chain", "in_active_chain" in indexed, False) return {"version": VERSION, "network": "regtest", "daemon_sha256": PIN, "txid": txid, "vout": vout, "blockhash": payment_block, "block_height": header["height"], "cases": cases, "rpc_trace": lab.trace} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("daemon", type=Path, help="Verified Litecoin Core 0.21.5.8 Windows x64 litecoind.exe") parser.add_argument("output", type=Path, help="New JSON results file") args = parser.parse_args() if args.output.exists(): parser.error("Output already exists") if sha256(args.daemon) != PIN: parser.error("Daemon does not match the pinned Windows x64 SHA-256") started = datetime.now(timezone.utc).isoformat() with tempfile.TemporaryDirectory(prefix="lw-tx-lookup-") as tmp: directory = Path(tmp) executable = directory / "litecoind.exe" shutil.copyfile(args.daemon, executable) (directory / "data").mkdir() (directory / "empty.conf").touch() lab = Lab(directory, executable) try: results = experiment(lab) finally: lab.stop() results["started_at_utc"] = started results["finished_at_utc"] = datetime.now(timezone.utc).isoformat() results["runner_sha256"] = sha256(Path(__file__)) with args.output.open("x", encoding="utf-8", newline="\n") as destination: json.dump(results, destination, indent=2) destination.write("\n") print(f'{len(results["cases"])} checks passed; {len(results["rpc_trace"])} RPC records.') if __name__ == "__main__": main()