From 6188665f46fab047f9db5759d62c92aa170c6122 Mon Sep 17 00:00:00 2001 From: Renato Britto Date: Fri, 27 Feb 2026 13:22:19 -0300 Subject: [PATCH] move fast break things --- backend/script/.env.template | 1 + backend/script/DONTRUN_mytest.py | 96 +++++++ backend/script/README.md | 286 ++++++++++++++++++- backend/script/bcli.py | 25 ++ backend/script/bitcoin_rpc.py | 73 ++--- backend/script/lib/bitcoin_rpc_regtest.py | 248 +++++++++++++++++ backend/script/lib/bitcoin_rpc_testnet.py | 127 +++++++++ backend/script/lib/clis.py | 167 ++++++++++++ backend/script/load.py | 29 ++ backend/script/reproduce.py | 317 +++++++++++++++------- 10 files changed, 1228 insertions(+), 141 deletions(-) create mode 100644 backend/script/.env.template create mode 100644 backend/script/DONTRUN_mytest.py create mode 100644 backend/script/bcli.py create mode 100644 backend/script/lib/bitcoin_rpc_regtest.py create mode 100644 backend/script/lib/bitcoin_rpc_testnet.py create mode 100644 backend/script/lib/clis.py create mode 100644 backend/script/load.py diff --git a/backend/script/.env.template b/backend/script/.env.template new file mode 100644 index 0000000..9cb34f7 --- /dev/null +++ b/backend/script/.env.template @@ -0,0 +1 @@ +SSHPASS= \ No newline at end of file diff --git a/backend/script/DONTRUN_mytest.py b/backend/script/DONTRUN_mytest.py new file mode 100644 index 0000000..91eaba9 --- /dev/null +++ b/backend/script/DONTRUN_mytest.py @@ -0,0 +1,96 @@ +""" +bitcoin_rpc.py — Thin wrapper around bitcoin-cli for Python tests. +Uses subprocess calls to bitcoin-cli -regtest. +""" + +import json +import subprocess +import time +import os +import requests + +CLI = "bitcoin-cli" +SIGNET_ARGS = [CLI] + +def cli_testnet(*args, wallet=None): + """Call Bitcoin RPC endpoint via HTTP and return parsed JSON or string.""" + + url = "https://bitcoin-testnet-rpc.publicnode.com" + args=args[0].split(' ') + print(f"args: {args}") + def parse_param(p): + try: + return json.loads(p) + except (json.JSONDecodeError, TypeError): + return p + + payload = { + "jsonrpc": "1.0", + "id": "cli", + "method": args[0] if args else "getblockcount", + "params": [parse_param(p) for p in args[1:]] if len(args) > 1 else [] + } + print(f"RPC request: {payload}") + + try: + response = requests.post(url, json=payload, timeout=60) + response.raise_for_status() + result = response.json() + + if "error" in result and result["error"]: + raise RuntimeError(f"bitcoin-cli error: {result['error']}") + + return result.get("result") + except requests.RequestException as e: + raise RuntimeError(f"RPC request failed: {str(e)}") + +def cli_mainnet(*args, wallet=None): + """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" + cmd = list(SIGNET_ARGS) + if wallet: + cmd.append(f"-rpcwallet={wallet}") + cmd.extend(str(a) for a in args) + cmd = ' '.join(cmd) + ssh_pass = os.environ.get("SSHPASS") or os.environ.get("SSH_PASS") + if not ssh_pass: + raise RuntimeError("Environment variable SSHPASS or SSH_PASS must be set") + thecli = f"sshpass -p {ssh_pass} ssh root@95.111.247.57 '{cmd}'" + result = subprocess.run(thecli, shell=True, capture_output=True, text=True, timeout=60) + if result.returncode != 0: + raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + + output = result.stdout.strip() + if not output: + return None + try: + return json.loads(output) + except json.JSONDecodeError: + return output + +def cli_regtest(*args, wallet=None): + """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" + cmd = [CLI, "-regtest"] + if wallet: + cmd.append(f"-rpcwallet={wallet}") + cmd.extend(str(a) for a in args) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + if result.returncode != 0: + raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + + output = result.stdout.strip() + if not output: + return None + try: + return json.loads(output) + except json.JSONDecodeError: + return output + +block_count = cli_testnet("getblockcount") +print("testnet:", block_count) +blockhash = cli_testnet(f"getblockhash {block_count}") +block = cli_testnet(f"getblock {blockhash}") +print("testnet:", block) + + +# print("mainnet:", cli_mainnet("getblockcount")) +# print("regtest:", cli_regtest("getblockcount")) diff --git a/backend/script/README.md b/backend/script/README.md index 7f3efbf..9a9e47d 100644 --- a/backend/script/README.md +++ b/backend/script/README.md @@ -26,4 +26,288 @@ kick jaguar paddle void -dinner \ No newline at end of file +dinner + +sshpass $SSHPASS ssh user@REMOTE_HOST 'bitcoin-cli getblockcount' + +electrum alice: +summer recipe phrase depth accident shuffle doctor trip hurdle jeans crop plate + +electrum bob: +curtain tiny reduce deer icon maid fresh lunch vivid inform woman squirrel + + +alice_t4: +- bulk cactus balance toward hawk glory regret blast cinnamon game confirm vintage +- tb1qllp2tyl03qskw49mg3wl04paegrrn7ll6rfef4 +bob_t4: +- paper pact now deal next unique option animal region dismiss detect pipe +- tb1qtxdyzss48eue5clr2v7cd76fjsv3krs77s5ma0 +carol_t4: +- regular hawk empty only caught train half upon chaos clap until guitar +- tb1qhkyuf6vkxefc8lpyuhxdutjhwc8m78hry8cthj +miner_t4: +- mutual east bench couple cage volume demand slush slab all swallow section +- tb1qa7wryry350swml5gnq92as5hxcgnewppqrsn6j +exchange_t4: +- awkward sick super sausage parrot apple bread loud kangaroo shop result issue +- tb1qa093eafvdz0wa2k6l274ve7asvm353mv45cl4w +risky_t4: +- town daring skill mechanic buzz head day quality inhale park cinnamon random +- tb1qhjd00gdjzne9khq6shlc9spcdag2gt8c468hfr + + + + + + +=========================================== + +bitcoin-cli -testnet4 unloadwallet alice +rm -rf ~/.bitcoin/testnet4/wallets/alice +python load.py "bulk cactus balance toward hawk glory regret blast cinnamon ga +me confirm vintage" +WIF: cTNftSdam7X3sdwVy8aQ8vshDjQBgosS76HWMg6XrGjqYDJyYHCD +Address: tb1qumm4e6h5pacudhn7np7lr4tdwgk8khulkaj0ny +{ + "descriptor": "wpkh(02439f42b07b13682a319c06cc209119d10f96f72db9ba5a0f1b843ef89ce7807c)#j50ja6ax", + "checksum": "p30q9gg3", + "isrange": false, + "issolvable": true, + "hasprivatekeys": true +} +bitcoin-cli -testnet4 createwallet "alice" + + + + + +bitcoin-cli -testnet4 getdescriptorinfo "wpkh(cTNftSdam7X3sdwVy8aQ8vshDjQBgosS76HWMg6XrGjqYDJyYHCD)" +wpkh(02439f42b07b13682a319c06cc209119d10f96f72db9ba5a0f1b843ef89ce7807c)#j50ja6ax +bitcoin-cli -testnet4 -rpcwallet=alice importdescriptors \ + '[{"desc":"wpkh(02439f42b07b13682a319c06cc209119d10f96f72db9ba5a0f1b843ef89ce7807c)#j50ja6ax","timestamp":"now"}]' + + + + + bitcoin-cli -testnet4 -rpcwallet=alice importdescriptors \ +'[{ + "desc":"wpkh(02439f42b07b13682a319c06cc209119d10f96f72db9ba5a0f1b843ef89ce7807c)#j50ja6ax", + "timestamp":"now", + "active":true +}]' + +================== + + +```bash +export user=alice +export seed="bulk cactus balance toward hawk glory regret blast cinnamon game confirm vintage" +bitcoin-cli -testnet4 unloadwallet "$user" +rm -rf ~/.bitcoin/testnet4/wallets/$user +bitcoin-cli -testnet4 createwallet "$user" true true "" false true + +export WIF="$(python3 - <<'PY' +from embit import bip32 +from embit.networks import NETWORKS +from embit.ec import PrivateKey +import hashlib, os +seed_phrase = os.environ['seed'] +seed_bytes = hashlib.pbkdf2_hmac('sha512', seed_phrase.encode(), b'electrum', iterations=2048) +root = bip32.HDKey.from_seed(seed_bytes, version=NETWORKS['test']['xprv']) +key = root.derive("m/84h/1h/0h/0/0") +privkey = PrivateKey(key.key.secret) +print(privkey.wif(NETWORKS['test'])) +PY +)" +export WIF + +# descriptor=$(bitcoin-cli -testnet4 getdescriptorinfo "wpkh($WIF)" | jq -r .descriptor) +# Get just the checksum +checksum=$(bitcoin-cli -testnet4 getdescriptorinfo "wpkh($WIF)" | jq -r .checksum) + +bitcoin-cli -testnet4 -rpcwallet=$user importdescriptors \ +"[{ + \"desc\": \"wpkh($WIF)#$checksum\", + \"timestamp\": \"now\" +}]" +``` + + + + +```bash +export user=alice +export seed="bulk cactus balance toward hawk glory regret blast cinnamon game confirm vintage" + +bitcoin-cli -testnet4 unloadwallet "$user" 2>/dev/null +rm -rf ~/.bitcoin/testnet4/wallets/$user +bitcoin-cli -testnet4 createwallet "$user" + +export WIF="$(python3 - <<'PY' +from embit import bip32 +from embit.networks import NETWORKS +from embit.ec import PrivateKey +import hashlib, os +seed_phrase = os.environ['seed'] +seed_bytes = hashlib.pbkdf2_hmac('sha512', seed_phrase.encode(), b'electrum', iterations=2048) +root = bip32.HDKey.from_seed(seed_bytes, version=NETWORKS['test']['xprv']) +key = root.derive("m/84h/1h/0h/0/0") +privkey = PrivateKey(key.key.secret) +print(privkey.wif(NETWORKS['test'])) +PY +)" + +checksum=$(bitcoin-cli -testnet4 getdescriptorinfo "wpkh($WIF)" | jq -r .checksum) + +bitcoin-cli -testnet4 -rpcwallet=$user importdescriptors \ +"[{ + \"desc\": \"wpkh($WIF)#$checksum\", + \"timestamp\": \"now\" +}]" + +# ensure balance is defined and compare using bc, fallback to 0 on error +result=$(echo "${balance:-0} > 0" | bc -l 2>/dev/null || echo 0) +if [ "$result" -eq 1 ]; then + echo "balance > 0" +else + echo "balance <= 0" +fi + +``` + + +```bash + +export user=alice +export seed="bulk cactus balance toward hawk glory regret blast cinnamon game confirm vintage" + +bitcoin-cli -testnet4 unloadwallet "$user" 2>/dev/null +rm -rf ~/.bitcoin/testnet4/wallets/$user +bitcoin-cli -testnet4 createwallet "$user" + +# Get the xprv root key +export XPRV="$(python3 - <<'PY' +from embit import bip32 +from embit.networks import NETWORKS +import hashlib, os + +seed_phrase = os.environ['seed'] +seed_bytes = hashlib.pbkdf2_hmac('sha512', seed_phrase.encode(), b'electrum', iterations=2048) +root = bip32.HDKey.from_seed(seed_bytes, version=NETWORKS['test']['xprv']) +print(root.to_base58()) +PY +)" + +# Import receiving addresses (m/0/*) +recv_check=$(bitcoin-cli -testnet4 getdescriptorinfo "wpkh($XPRV/0/*)" | jq -r .checksum) +# Import change addresses (m/1/*) +change_check=$(bitcoin-cli -testnet4 getdescriptorinfo "wpkh($XPRV/1/*)" | jq -r .checksum) + +bitcoin-cli -testnet4 -rpcwallet=$user importdescriptors \ +"[ + { + \"desc\": \"wpkh($XPRV/0/*)#$recv_check\", + \"timestamp\": 0, + \"range\": [0, 50], + \"internal\": false + }, + { + \"desc\": \"wpkh($XPRV/1/*)#$change_check\", + \"timestamp\": 0, + \"range\": [0, 50], + \"internal\": true + } +]" + + +bitcoin-cli -testnet4 -rpcwallet=$user getblockchaininfo + +# Check current block height +bitcoin-cli -testnet4 getblockcount + +# Rescan from a reasonable starting height (adjust as needed) +bitcoin-cli -testnet4 -rpcwallet=$user rescanblockchain 50000 +``` + + + + + + +alice_t4: +- bulk cactus balance toward hawk glory regret blast cinnamon game confirm vintage +- tb1qllp2tyl03qskw49mg3wl04paegrrn7ll6rfef4 +bob_t4: +- paper pact now deal next unique option animal region dismiss detect pipe +- tb1qtxdyzss48eue5clr2v7cd76fjsv3krs77s5ma0 +carol_t4: +- regular hawk empty only caught train half upon chaos clap until guitar +- tb1qhkyuf6vkxefc8lpyuhxdutjhwc8m78hry8cthj +miner_t4: +- mutual east bench couple cage volume demand slush slab all swallow section +- tb1qa7wryry350swml5gnq92as5hxcgnewppqrsn6j +exchange_t4: +- awkward sick super sausage parrot apple bread loud kangaroo shop result issue +- tb1qa093eafvdz0wa2k6l274ve7asvm353mv45cl4w +risky_t4: +- town daring skill mechanic buzz head day quality inhale park cinnamon random +- tb1qhjd00gdjzne9khq6shlc9spcdag2gt8c468hfr + + +```bash +export user=risky +export seed="town daring skill mechanic buzz head day quality inhale park cinnamon random" +python3 - <<'PY' +from embit import bip32 +from embit.networks import NETWORKS +from embit.script import p2wpkh +import hashlib + +seed_phrase = "town daring skill mechanic buzz head day quality inhale park cinnamon random" +seed_bytes = hashlib.pbkdf2_hmac("sha512", seed_phrase.encode("utf-8"), b"electrum", iterations=2048) +root = bip32.HDKey.from_seed(seed_bytes, version=NETWORKS["test"]["xprv"]) + +# Print first 5 receiving and first 3 change for m/0h path +print("=== Receiving (m/0h/0/*) ===") +for i in range(5): + child = root.derive(f"m/0h/0/{i}") + addr = p2wpkh(child.key).address(NETWORKS["test"]) + print(f" m/0h/0/{i} -> {addr}") + +print("=== Change (m/0h/1/*) ===") +for i in range(3): + child = root.derive(f"m/0h/1/{i}") + addr = p2wpkh(child.key).address(NETWORKS["test"]) + print(f" m/0h/1/{i} -> {addr}") +PY + +bitcoin-cli -testnet4 unloadwallet "$user" 2>/dev/null +rm -rf ~/.bitcoin/testnet4/wallets/$user +bitcoin-cli -testnet4 createwallet "$user" + +export XPRV="$(python3 - <<'PY' +from embit import bip32 +from embit.networks import NETWORKS +import hashlib +seed_phrase = "bulk cactus balance toward hawk glory regret blast cinnamon game confirm vintage" +seed_bytes = hashlib.pbkdf2_hmac("sha512", seed_phrase.encode("utf-8"), b"electrum", iterations=2048) +root = bip32.HDKey.from_seed(seed_bytes, version=NETWORKS["test"]["xprv"]) +account = root.derive("m/0h") +print(account.to_base58()) +PY +)" + +recv_check=$(bitcoin-cli -testnet4 getdescriptorinfo "wpkh($XPRV/0/*)" | jq -r .checksum) +change_check=$(bitcoin-cli -testnet4 getdescriptorinfo "wpkh($XPRV/1/*)" | jq -r .checksum) + +bitcoin-cli -testnet4 -rpcwallet=$user importdescriptors "[ + {\"desc\": \"wpkh($XPRV/0/*)#$recv_check\", \"timestamp\": \"now\", \"range\": [0, 50], \"internal\": false}, + {\"desc\": \"wpkh($XPRV/1/*)#$change_check\", \"timestamp\": \"now\", \"range\": [0, 50], \"internal\": true} +]" + +bitcoin-cli -testnet4 -rpcwallet=$user rescanblockchain 107570 +bitcoin-cli -testnet4 -rpcwallet=$user getbalance +``` \ No newline at end of file diff --git a/backend/script/bcli.py b/backend/script/bcli.py new file mode 100644 index 0000000..526eb5c --- /dev/null +++ b/backend/script/bcli.py @@ -0,0 +1,25 @@ +""" +bitcoin_rpc.py — Thin wrapper around bitcoin-cli for Python tests. +Uses subprocess calls to bitcoin-cli -regtest. +""" + +import json +import subprocess +import time +import os +import requests + +CLI = "bitcoin-cli" +SIGNET_ARGS = [CLI] + +from lib.bitcoin_rpc_testnet import cli +# from lib.bitcoin_rpc_testnet import cli + +block_count = cli("getblockcount") +print("testnet:", block_count) +blockhash = cli(f"getblockhash {block_count}") +block = cli(f"getblock {blockhash}") +print("testnet:", block) + +# print("mainnet:", cli_mainnet("getblockcount")) +# print("regtest:", cli_regtest("getblockcount")) diff --git a/backend/script/bitcoin_rpc.py b/backend/script/bitcoin_rpc.py index b35d345..c9a32d5 100644 --- a/backend/script/bitcoin_rpc.py +++ b/backend/script/bitcoin_rpc.py @@ -1,10 +1,15 @@ """ -bitcoin_rpc.py — Thin wrapper around bitcoin-cli for Python tests. -Connection settings are read from config.ini in the same directory. +bitcoin_rpc.py — Facade that selects regtest or testnet backend +based on config.ini [bitcoin] network setting. + +Exposes: + NETWORK – "regtest" or "testnet" + IS_REGTEST – True when running on regtest + cli(...) – RPC call routed to the correct backend + mine_blocks, get_tx, get_utxos, get_balance, send_raw, ... """ import json -import subprocess import time import os import configparser @@ -17,57 +22,31 @@ def _load_config(): cfg.read(config_path) return cfg["bitcoin"] if "bitcoin" in cfg else {} -def _build_base_args(section): - cli_bin = section.get("cli", "bitcoin-cli") - network = section.get("network", "regtest").strip().lower() - - args = [cli_bin] - - network_flags = { - "regtest": "-regtest", - "testnet": "-testnet", - "signet": "-signet", - } - if network in network_flags: - args.append(network_flags[network]) - - for key, flag in [("rpchost", "-rpcconnect"), ("rpcport", "-rpcport"), - ("rpcuser", "-rpcuser"), ("rpcpassword", "-rpcpassword")]: - value = section.get(key, "").strip() - if value: - args.append(f"{flag}={value}") - - return args - _cfg = _load_config() -_BASE_ARGS = _build_base_args(_cfg) +NETWORK = _cfg.get("network", "regtest").strip().lower() +IS_REGTEST = NETWORK == "regtest" + +# ── Select the right cli backend ───────────────────────────────────────────── + +if IS_REGTEST: + from lib.clis import cli_regtest as _cli_backend +else: + from lib.clis import cli_testnet as _cli_backend -# Keep these for any scripts that might reference them directly -CLI = _cfg.get("cli", "bitcoin-cli") -SIGNET_ARGS = _BASE_ARGS def cli(*args, wallet=None): - """Call bitcoin-cli [network] [wallet] and return parsed JSON or string.""" - cmd = list(_BASE_ARGS) - if wallet: - cmd.append(f"-rpcwallet={wallet}") - cmd.extend(str(a) for a in args) + """Route RPC calls to the active backend.""" + return _cli_backend(*args, wallet=wallet) - result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) - if result.returncode != 0: - raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") - - output = result.stdout.strip() - if not output: - return None - try: - return json.loads(output) - except json.JSONDecodeError: - return output +# ── Block mining (regtest only) ────────────────────────────────────────────── def mine_blocks(n=1): - """Mine n blocks on regtest using generatetoaddress.""" + """Mine n blocks. Only works on regtest; no-op on testnet.""" + if not IS_REGTEST: + # On testnet we cannot mine — just wait for propagation + time.sleep(2) + return get_block_count() miner_addr = cli("getnewaddress", "", "bech32", wallet="miner") cli("generatetoaddress", n, miner_addr) return int(cli("getblockcount")) @@ -160,10 +139,12 @@ def get_new_address(wallet_name, addr_type="bech32"): def send_to_address(wallet_name, address, amount): """Send BTC to an address.""" + print(f"Sending {amount:.8f} BTC from {wallet_name} to {address}...") return cli("sendtoaddress", address, f"{amount:.8f}", wallet=wallet_name) if __name__ == "__main__": + print(f"Network mode: {NETWORK} (IS_REGTEST={IS_REGTEST})") print("Testing RPC connection...") info = cli("getblockchaininfo") print(f" Chain: {info['chain']}") diff --git a/backend/script/lib/bitcoin_rpc_regtest.py b/backend/script/lib/bitcoin_rpc_regtest.py new file mode 100644 index 0000000..2a6ed14 --- /dev/null +++ b/backend/script/lib/bitcoin_rpc_regtest.py @@ -0,0 +1,248 @@ +""" +bitcoin_rpc.py — Thin wrapper around bitcoin-cli for Python tests. +Connection settings are read from config.ini in the same directory. +""" + +import json +import subprocess +import time +import os +import configparser + +# ── Load config ────────────────────────────────────────────────────────────── + +def _load_config(): + cfg = configparser.ConfigParser() + config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.ini") + cfg.read(config_path) + return cfg["bitcoin"] if "bitcoin" in cfg else {} + +def _build_base_args(section): + cli_bin = section.get("cli", "bitcoin-cli") + network = section.get("network", "regtest").strip().lower() + + args = [cli_bin] + + network_flags = { + "regtest": "-regtest", + "testnet": "-testnet", + "signet": "-signet", + } + if network in network_flags: + args.append(network_flags[network]) + + for key, flag in [("rpchost", "-rpcconnect"), ("rpcport", "-rpcport"), + ("rpcuser", "-rpcuser"), ("rpcpassword", "-rpcpassword")]: + value = section.get(key, "").strip() + if value: + args.append(f"{flag}={value}") + + return args + +_cfg = _load_config() +_BASE_ARGS = _build_base_args(_cfg) + +# Keep these for any scripts that might reference them directly +CLI = _cfg.get("cli", "bitcoin-cli") +SIGNET_ARGS = _BASE_ARGS + +# def cli(*args, wallet=None): +# """Call bitcoin-cli [network] [wallet] and return parsed JSON or string.""" +# cmd = list(_BASE_ARGS) + + + +from lib.clis import cli_regtest +def cli(*args, wallet=None): + return cli_regtest(*args, wallet=wallet) +# import requests + +# CLI = "bitcoin-cli" +# SIGNET_ARGS = [CLI] + +# def cli_testnet(*args, wallet=None): +# """Call Bitcoin RPC endpoint via HTTP and return parsed JSON or string.""" + +# url = "https://bitcoin-testnet-rpc.publicnode.com" +# args=args[0].split(' ') +# print(f"args: {args}") +# def parse_param(p): +# try: +# return json.loads(p) +# except (json.JSONDecodeError, TypeError): +# return p + +# payload = { +# "jsonrpc": "1.0", +# "id": "cli", +# "method": args[0] if args else "getblockcount", +# "params": [parse_param(p) for p in args[1:]] if len(args) > 1 else [] +# } +# print(f"RPC request: {payload}") + +# try: +# response = requests.post(url, json=payload, timeout=60) +# response.raise_for_status() +# result = response.json() + +# if "error" in result and result["error"]: +# raise RuntimeError(f"bitcoin-cli error: {result['error']}") + +# return result.get("result") +# except requests.RequestException as e: +# raise RuntimeError(f"RPC request failed: {str(e)}") + +# def cli_mainnet(*args, wallet=None): +# """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" +# cmd = list(SIGNET_ARGS) +# if wallet: +# cmd.append(f"-rpcwallet={wallet}") +# cmd.extend(str(a) for a in args) +# cmd = ' '.join(cmd) +# ssh_pass = os.environ.get("SSHPASS") or os.environ.get("SSH_PASS") +# if not ssh_pass: +# raise RuntimeError("Environment variable SSHPASS or SSH_PASS must be set") +# thecli = f"sshpass -p {ssh_pass} ssh root@95.111.247.57 '{cmd}'" +# result = subprocess.run(thecli, shell=True, capture_output=True, text=True, timeout=60) +# if result.returncode != 0: +# raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + +# output = result.stdout.strip() +# if not output: +# return None +# try: +# return json.loads(output) +# except json.JSONDecodeError: +# return output + +# def cli_regtest(*args, wallet=None): +# """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" +# cmd = [CLI, "-regtest"] +# if wallet: +# cmd.append(f"-rpcwallet={wallet}") +# cmd.extend(str(a) for a in args) +# result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) +# if result.returncode != 0: +# raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + +# output = result.stdout.strip() +# if not output: +# return None +# try: +# return json.loads(output) +# except json.JSONDecodeError: +# return output + +# def cli(*args, wallet=None): +# return cli_testnet(*args, wallet=wallet) +# # return cli_regtest(*args, wallet=wallet) + +def mine_blocks(n=1): + """Mine n blocks on regtest using generatetoaddress.""" + miner_addr = cli("getnewaddress", "", "bech32", wallet="miner") + cli("generatetoaddress", n, miner_addr) + return int(cli("getblockcount")) + + +def fund_wallet(wallet_name, amount=1.0, from_wallet="miner"): + """Send `amount` BTC from `from_wallet` to a new address in `wallet_name`.""" + addr = cli("getnewaddress", "", "bech32", wallet=wallet_name) + txid = cli("sendtoaddress", addr, f"{amount:.8f}", wallet=from_wallet) + return txid, addr + + +def wait_for_mempool_empty(timeout=60): + """Wait until mempool is empty (all txs mined).""" + for _ in range(timeout * 2): + info = cli("getmempoolinfo") + if info["size"] == 0: + return True + time.sleep(0.5) + return False + + +def get_tx(txid): + """Get decoded transaction.""" + return cli("getrawtransaction", txid, "true") + + +def get_utxos(wallet_name, min_conf=0): + """List unspent outputs for a wallet.""" + return cli("listunspent", min_conf, wallet=wallet_name) + + +def get_balance(wallet_name): + """Get wallet balance.""" + return float(cli("getbalance", wallet=wallet_name)) + + +def send_raw(hex_tx): + """Broadcast a raw transaction.""" + return cli("sendrawtransaction", hex_tx) + + +def create_funded_psbt(wallet_name, inputs, outputs, options=None): + """Create a funded PSBT.""" + args = ["walletcreatefundedpsbt", json.dumps(inputs), json.dumps(outputs), 0] + if options: + args.append(json.dumps(options)) + return cli(*args, wallet=wallet_name) + + +def process_psbt(wallet_name, psbt): + """Sign a PSBT.""" + return cli("walletprocesspsbt", psbt, wallet=wallet_name) + + +def finalize_psbt(psbt): + """Finalize a PSBT.""" + return cli("finalizepsbt", psbt) + + +def decode_psbt(psbt): + """Decode a PSBT.""" + return cli("decodepsbt", psbt) + + +def create_raw_tx(inputs, outputs): + """Create a raw transaction.""" + return cli("createrawtransaction", json.dumps(inputs), json.dumps(outputs)) + + +def sign_raw_tx(wallet_name, hex_tx): + """Sign a raw transaction.""" + return cli("signrawtransactionwithwallet", hex_tx, wallet=wallet_name) + + +def decode_raw_tx(hex_tx): + """Decode a raw transaction.""" + return cli("decoderawtransaction", hex_tx) + + +def get_block_count(): + """Get current block height.""" + return int(cli("getblockcount")) + + +def get_new_address(wallet_name, addr_type="bech32"): + """Get a new address.""" + return cli("getnewaddress", "", addr_type, wallet=wallet_name) + + +def send_to_address(wallet_name, address, amount): + """Send BTC to an address.""" + return cli("sendtoaddress", address, f"{amount:.8f}", wallet=wallet_name) + + +if __name__ == "__main__": + print("Testing RPC connection...") + info = cli("getblockchaininfo") + print(f" Chain: {info['chain']}") + print(f" Blocks: {info['blocks']}") + + for w in ["miner", "alice", "bob", "carol", "exchange", "risky"]: + try: + bal = get_balance(w) + print(f" {w}: {bal} BTC") + except Exception as e: + print(f" {w}: ERROR - {e}") diff --git a/backend/script/lib/bitcoin_rpc_testnet.py b/backend/script/lib/bitcoin_rpc_testnet.py new file mode 100644 index 0000000..3d95ffd --- /dev/null +++ b/backend/script/lib/bitcoin_rpc_testnet.py @@ -0,0 +1,127 @@ +""" +bitcoin_rpc.py — Thin wrapper around bitcoin-cli for Python tests. +Connection settings are read from config.ini in the same directory. +""" + +import json +import subprocess +import time +import os +import requests +from lib.clis import cli_testnet + + + +def cli(*args, wallet=None): + return cli_testnet(*args, wallet=wallet) + # return cli_regtest(*args, wallet=wallet) + +def mine_blocks(n=1): + """Mine n blocks on regtest using generatetoaddress.""" + miner_addr = cli("getnewaddress", "", "bech32", wallet="miner") + cli("generatetoaddress", n, miner_addr) + return int(cli("getblockcount")) + + +def fund_wallet(wallet_name, amount=1.0, from_wallet="miner"): + """Send `amount` BTC from `from_wallet` to a new address in `wallet_name`.""" + addr = cli("getnewaddress", "", "bech32", wallet=wallet_name) + txid = cli("sendtoaddress", addr, f"{amount:.8f}", wallet=from_wallet) + return txid, addr + + +def wait_for_mempool_empty(timeout=60): + """Wait until mempool is empty (all txs mined).""" + for _ in range(timeout * 2): + info = cli("getmempoolinfo") + if info["size"] == 0: + return True + time.sleep(0.5) + return False + + +def get_tx(txid): + """Get decoded transaction.""" + return cli("getrawtransaction", txid, "true") + + +def get_utxos(wallet_name, min_conf=0): + """List unspent outputs for a wallet.""" + return cli("listunspent", min_conf, wallet=wallet_name) + + +def get_balance(wallet_name): + """Get wallet balance.""" + return float(cli("getbalance", wallet=wallet_name)) + + +def send_raw(hex_tx): + """Broadcast a raw transaction.""" + return cli("sendrawtransaction", hex_tx) + + +def create_funded_psbt(wallet_name, inputs, outputs, options=None): + """Create a funded PSBT.""" + args = ["walletcreatefundedpsbt", json.dumps(inputs), json.dumps(outputs), 0] + if options: + args.append(json.dumps(options)) + return cli(*args, wallet=wallet_name) + + +def process_psbt(wallet_name, psbt): + """Sign a PSBT.""" + return cli("walletprocesspsbt", psbt, wallet=wallet_name) + + +def finalize_psbt(psbt): + """Finalize a PSBT.""" + return cli("finalizepsbt", psbt) + + +def decode_psbt(psbt): + """Decode a PSBT.""" + return cli("decodepsbt", psbt) + + +def create_raw_tx(inputs, outputs): + """Create a raw transaction.""" + return cli("createrawtransaction", json.dumps(inputs), json.dumps(outputs)) + + +def sign_raw_tx(wallet_name, hex_tx): + """Sign a raw transaction.""" + return cli("signrawtransactionwithwallet", hex_tx, wallet=wallet_name) + + +def decode_raw_tx(hex_tx): + """Decode a raw transaction.""" + return cli("decoderawtransaction", hex_tx) + + +def get_block_count(): + """Get current block height.""" + return int(cli("getblockcount")) + + +def get_new_address(wallet_name, addr_type="bech32"): + """Get a new address.""" + return cli("getnewaddress", "", addr_type, wallet=wallet_name) + + +def send_to_address(wallet_name, address, amount): + """Send BTC to an address.""" + return cli("sendtoaddress", address, f"{amount:.8f}", wallet=wallet_name) + + +if __name__ == "__main__": + print("Testing RPC connection...") + info = cli("getblockchaininfo") + print(f" Chain: {info['chain']}") + print(f" Blocks: {info['blocks']}") + + for w in ["miner", "alice", "bob", "carol", "exchange", "risky"]: + try: + bal = get_balance(w) + print(f" {w}: {bal} BTC") + except Exception as e: + print(f" {w}: ERROR - {e}") diff --git a/backend/script/lib/clis.py b/backend/script/lib/clis.py new file mode 100644 index 0000000..6fa1329 --- /dev/null +++ b/backend/script/lib/clis.py @@ -0,0 +1,167 @@ +""" +bitcoin_rpc.py — Thin wrapper around bitcoin-cli for Python tests. +Uses subprocess calls to bitcoin-cli -regtest. +""" + +import json +import subprocess +import time +import os + +CLI = "bitcoin-cli" +SIGNET_ARGS = [CLI] + +def cli_testnet(*args, wallet=None): + """Call bitcoin-cli -testnet4 [wallet] and return parsed JSON or string.""" + cmd = [CLI, "-testnet4"] + if wallet: + cmd.append(f"-rpcwallet={wallet}") + + normalized_args = [] + if len(args) == 1 and isinstance(args[0], str): + normalized_args.extend(args[0].split()) + else: + normalized_args.extend(str(a) for a in args) + cmd.extend(normalized_args) + + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + if result.returncode != 0: + raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + + output = result.stdout.strip() + if not output: + return None + try: + return json.loads(output) + except json.JSONDecodeError: + return output + +def cli_mainnet(*args, wallet=None): + """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" + cmd = list(SIGNET_ARGS) + if wallet: + cmd.append(f"-rpcwallet={wallet}") + cmd.extend(str(a) for a in args) + cmd = ' '.join(cmd) + ssh_pass = os.environ.get("SSHPASS") or os.environ.get("SSH_PASS") + if not ssh_pass: + raise RuntimeError("Environment variable SSHPASS or SSH_PASS must be set") + thecli = f"sshpass -p {ssh_pass} ssh root@95.111.247.57 '{cmd}'" + result = subprocess.run(thecli, shell=True, capture_output=True, text=True, timeout=60) + if result.returncode != 0: + raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + + output = result.stdout.strip() + if not output: + return None + try: + return json.loads(output) + except json.JSONDecodeError: + return output + +def cli_regtest(*args, wallet=None): + """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" + cmd = [CLI, "-regtest"] + if wallet: + cmd.append(f"-rpcwallet={wallet}") + for arg in args[0].split(' '): + cmd.append(str(arg)) + # cmd.extend(str(a) for a in args) + # print(f"Running command: {cmd}") + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + + # print(f"Command return code: {result.returncode}") + # print(f"result: {result}") + + if result.returncode != 0: + raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + + output = result.stdout.strip() + # print(f"Command output: {output}") + if not output: + # print("No output from command.") + return None + try: + js= json.loads(output) + # print(f"json from command: {js} (type {type(js)})") + return js + except json.JSONDecodeError: + # print("text from command.") + return output + + +# CLI = "bitcoin-cli" +# SIGNET_ARGS = [CLI] + +# def cli_testnet(*args, wallet=None): +# """Call Bitcoin RPC endpoint via HTTP and return parsed JSON or string.""" + +# url = "https://bitcoin-testnet-rpc.publicnode.com" +# args=args[0].split(' ') +# print(f"args: {args}") +# def parse_param(p): +# try: +# return json.loads(p) +# except (json.JSONDecodeError, TypeError): +# return p + +# payload = { +# "jsonrpc": "1.0", +# "id": "cli", +# "method": args[0] if args else "getblockcount", +# "params": [parse_param(p) for p in args[1:]] if len(args) > 1 else [] +# } +# print(f"RPC request: {payload}") + +# try: +# response = requests.post(url, json=payload, timeout=60) +# response.raise_for_status() +# result = response.json() + +# if "error" in result and result["error"]: +# raise RuntimeError(f"bitcoin-cli error: {result['error']}") + +# return result.get("result") +# except requests.RequestException as e: +# raise RuntimeError(f"RPC request failed: {str(e)}") + +# def cli_mainnet(*args, wallet=None): +# """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" +# cmd = list(SIGNET_ARGS) +# if wallet: +# cmd.append(f"-rpcwallet={wallet}") +# cmd.extend(str(a) for a in args) +# cmd = ' '.join(cmd) +# ssh_pass = os.environ.get("SSHPASS") or os.environ.get("SSH_PASS") +# if not ssh_pass: +# raise RuntimeError("Environment variable SSHPASS or SSH_PASS must be set") +# thecli = f"sshpass -p {ssh_pass} ssh root@95.111.247.57 '{cmd}'" +# result = subprocess.run(thecli, shell=True, capture_output=True, text=True, timeout=60) +# if result.returncode != 0: +# raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + +# output = result.stdout.strip() +# if not output: +# return None +# try: +# return json.loads(output) +# except json.JSONDecodeError: +# return output + +# def cli_regtest(*args, wallet=None): +# """Call bitcoin-cli -regtest [wallet] and return parsed JSON or string.""" +# cmd = [CLI, "-regtest"] +# if wallet: +# cmd.append(f"-rpcwallet={wallet}") +# cmd.extend(str(a) for a in args) +# result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) +# if result.returncode != 0: +# raise RuntimeError(f"bitcoin-cli error: {result.stderr.strip()}\n cmd: {' '.join(cmd)}") + +# output = result.stdout.strip() +# if not output: +# return None +# try: +# return json.loads(output) +# except json.JSONDecodeError: +# return output diff --git a/backend/script/load.py b/backend/script/load.py new file mode 100644 index 0000000..bf74815 --- /dev/null +++ b/backend/script/load.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python3 +import argparse +import getpass +import hashlib +from embit import bip32 +from embit.networks import NETWORKS +from embit.script import p2wpkh +from embit.ec import PrivateKey + +parser = argparse.ArgumentParser(description="Derive WIF and address from seed phrase") +parser.add_argument("seed", nargs="?", help="Seed phrase (if omitted, you'll be prompted)") +parser.add_argument("--network", choices=["test", "main"], default="test", help="Network") +args = parser.parse_args() + +seed_phrase = args.seed or getpass.getpass("Seed phrase: ") + +seed_bytes = hashlib.pbkdf2_hmac( + "sha512", + seed_phrase.encode("utf-8"), + b"electrum", + iterations=2048, +) + +root = bip32.HDKey.from_seed(seed_bytes, version=NETWORKS[args.network]["xprv"]) +key = root.derive("m/84h/1h/0h/0/0") + +privkey = PrivateKey(key.key.secret) +print("WIF:", privkey.wif(NETWORKS[args.network])) +print("Address:", p2wpkh(key.to_public()).address(NETWORKS[args.network])) \ No newline at end of file diff --git a/backend/script/reproduce.py b/backend/script/reproduce.py index eeb1449..a17b905 100644 --- a/backend/script/reproduce.py +++ b/backend/script/reproduce.py @@ -18,64 +18,139 @@ import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from bitcoin_rpc import ( - cli, mine_blocks, get_tx, get_utxos, get_balance, - get_new_address, send_to_address, create_raw_tx, sign_raw_tx, - send_raw, get_block_count, create_funded_psbt, - process_psbt, finalize_psbt, + cli, + mine_blocks, + get_tx, + get_utxos, + get_balance, + get_new_address, + send_to_address, + create_raw_tx, + sign_raw_tx, + send_raw, + get_block_count, + create_funded_psbt, + process_psbt, + finalize_psbt, + IS_REGTEST, ) + +# Original budget: bob 15, alice 5, exchange 5, carol 2, risky 2 BTC +# VALUE_FACTOR scales these down for testnet (÷200): +# bob ~0.075, alice ~0.025, exchange ~0.025, carol ~0.01, risky ~0.01 BTC +# Total testnet budget ≈ 0.145 BTC +VALUE_FACTOR = 0.005 + +# Fixed fee deduction for raw transactions (2000 sats — ~2-5 sat/vB) +FEE = 0.00002 + +# Use 0-conf on testnet (can't mine), 1-conf on regtest +MIN_CONF = 1 if IS_REGTEST else 0 + + # ═══════════════════════════════════════════════════════════════════════════════ # Formatting helpers # ═══════════════════════════════════════════════════════════════════════════════ -G = "\033[92m"; Y = "\033[93m"; C = "\033[96m"; B = "\033[1m"; R = "\033[0m" +G = "\033[92m" +Y = "\033[93m" +C = "\033[96m" +B = "\033[1m" +R = "\033[0m" + def header(num, title): print(f"\n{'═'*78}") print(f"{B}{C} REPRODUCE {num}: {title}{R}") print(f"{'═'*78}") + def ok(msg): print(f" {G}✓{R} {msg}") + def info(msg): print(f" {Y}ℹ{R} {msg}") + def ensure_funds(wallet, min_btc=0.5): bal = get_balance(wallet) if bal < min_btc: - addr = get_new_address(wallet, "bech32") - send_to_address("miner", addr, min_btc + 0.5) - mine_blocks(1) + if IS_REGTEST: + addr = get_new_address(wallet, "bech32") + send_to_address("miner", addr, round(min_btc * 2, 8)) + mine_and_confirm() + else: + info(f"{wallet} balance ({bal:.8f}) < needed ({min_btc:.8f}) — fund externally") + def mine_and_confirm(): - mine_blocks(1) - time.sleep(0.5) + if IS_REGTEST: + mine_blocks(1) + time.sleep(0.5) + else: + # On testnet we cannot mine — wait for broadcast propagation + time.sleep(5) + + +def send_transaction(tx): + txid = send_raw(tx) + mine_and_confirm() + return txid + + +WALLETS = ["alice", "bob", "carol", "exchange", "risky"] + + +def setup_wallets(): + """Ensure all required wallets are loaded (create if missing).""" + wallets = list(WALLETS) + if IS_REGTEST: + wallets.append("miner") + for w in wallets: + try: + cli("getwalletinfo", wallet=w) + except Exception: + try: + cli("loadwallet", w) + ok(f"Loaded wallet: {w}") + except Exception: + try: + cli("createwallet", w) + ok(f"Created wallet: {w}") + except Exception as e: + if "already" in str(e).lower(): + pass + else: + info(f"Could not setup wallet '{w}': {e}") + # ═══════════════════════════════════════════════════════════════════════════════ # 1. Address Reuse # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_01(): header(1, "Address Reuse") - ensure_funds("bob", 1.0) + ensure_funds("bob", 1.0 * VALUE_FACTOR) reused_addr = get_new_address("alice", "bech32") - txid1 = send_to_address("bob", reused_addr, 0.01) - txid2 = send_to_address("bob", reused_addr, 0.02) + txid1 = send_to_address("bob", reused_addr, 0.01 * VALUE_FACTOR) + txid2 = send_to_address("bob", reused_addr, 0.02 * VALUE_FACTOR) mine_and_confirm() ok(f"Sent to same address {reused_addr} twice: TX {txid1[:16]}… and {txid2[:16]}…") + # ═══════════════════════════════════════════════════════════════════════════════ # 2. Multi-input / CIOH # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_02(): header(2, "Multi-input / CIOH (Common Input Ownership Heuristic)") - ensure_funds("bob", 2.0) + ensure_funds("bob", 2.0 * VALUE_FACTOR) for _ in range(5): addr = get_new_address("alice", "bech32") - send_to_address("bob", addr, 0.005) + send_to_address("bob", addr, 0.005 * VALUE_FACTOR) mine_and_confirm() - utxos = get_utxos("alice", 1) - small = [u for u in utxos if 0.004 < u["amount"] < 0.006][:5] + utxos = get_utxos("alice", MIN_CONF) + small = [u for u in utxos if 0.004 * VALUE_FACTOR < u["amount"] < 0.006 * VALUE_FACTOR][:5] if len(small) < 2: info("Not enough small UTXOs, skipping consolidation step") return @@ -83,8 +158,10 @@ def reproduce_02(): dest = get_new_address("bob", "bech32") total = sum(u["amount"] for u in small) psbt_result = create_funded_psbt( - "alice", inputs, [{dest: round(total - 0.001, 8)}], - {"subtractFeeFromOutputs": [0], "add_inputs": False} + "alice", + inputs, + [{dest: round(total - 0.001 * VALUE_FACTOR, 12)}], + {"subtractFeeFromOutputs": [0], "add_inputs": False}, ) signed = process_psbt("alice", psbt_result["psbt"]) final = finalize_psbt(signed["psbt"]) @@ -92,271 +169,317 @@ def reproduce_02(): mine_and_confirm() ok(f"Consolidated {len(small)} inputs in TX {txid[:16]}… (CIOH trigger)") + # ═══════════════════════════════════════════════════════════════════════════════ # 3. Dust UTXO Detection # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_03(): header(3, "Dust UTXO Detection") - ensure_funds("bob", 1.0) + ensure_funds("bob", 1.0 * VALUE_FACTOR) dust1 = get_new_address("alice", "bech32") dust2 = get_new_address("alice", "bech32") - bob_utxos = get_utxos("bob", 1) + bob_utxos = get_utxos("bob", MIN_CONF) big = max(bob_utxos, key=lambda u: u["amount"]) change = get_new_address("bob", "bech32") - change_amt = round(big["amount"] - 0.00001000 - 0.00000546 - 0.0001, 8) + change_amt = round(big["amount"] - 0.00001000 - 0.00000546 - FEE, 8) raw = create_raw_tx( [{"txid": big["txid"], "vout": big["vout"]}], - [{dust1: 0.00001000}, {dust2: 0.00000546}, {change: change_amt}] + [{dust1: 0.00001000}, {dust2: 0.00000546}, {change: change_amt}], ) signed = sign_raw_tx("bob", raw) txid = send_raw(signed["hex"]) mine_and_confirm() ok(f"Created 1000-sat and 546-sat dust outputs to Alice in TX {txid[:16]}…") + # ═══════════════════════════════════════════════════════════════════════════════ # 4. Spending Dust with Normal Inputs # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_04(): header(4, "Spending Dust with Normal Inputs") - ensure_funds("alice", 0.5) - utxos = get_utxos("alice", 1) + ensure_funds("alice", 0.5 * VALUE_FACTOR) + utxos = get_utxos("alice", MIN_CONF) dust_utxos = [u for u in utxos if u["amount"] <= 0.00001] - normal_utxos = [u for u in utxos if u["amount"] > 0.001] + normal_utxos = [u for u in utxos if u["amount"] > 0.0001] if not dust_utxos: info("No dust UTXOs, creating one first…") - ensure_funds("bob", 1.0) + ensure_funds("bob", 1.0 * VALUE_FACTOR) a = get_new_address("alice", "bech32") - bu = get_utxos("bob", 1) + bu = get_utxos("bob", MIN_CONF) big = max(bu, key=lambda u: u["amount"]) ch = get_new_address("bob", "bech32") raw = create_raw_tx( [{"txid": big["txid"], "vout": big["vout"]}], - [{a: 0.00001000}, {ch: round(big["amount"] - 0.00001 - 0.0001, 8)}] + [{a: 0.00001000}, {ch: round(big["amount"] - 0.00001000 - FEE, 8)}], ) signed = sign_raw_tx("bob", raw) send_raw(signed["hex"]) mine_and_confirm() - utxos = get_utxos("alice", 1) + utxos = get_utxos("alice", MIN_CONF) dust_utxos = [u for u in utxos if u["amount"] <= 0.00001] - normal_utxos = [u for u in utxos if u["amount"] > 0.001] + normal_utxos = [u for u in utxos if u["amount"] > 0.0001] if not normal_utxos: - ensure_funds("alice", 0.5) + ensure_funds("alice", 0.5 * VALUE_FACTOR) mine_and_confirm() - utxos = get_utxos("alice", 1) - normal_utxos = [u for u in utxos if u["amount"] > 0.001] + utxos = get_utxos("alice", MIN_CONF) + normal_utxos = [u for u in utxos if u["amount"] > 0.0001] dust = dust_utxos[0] normal = normal_utxos[0] dest = get_new_address("bob", "bech32") total = dust["amount"] + normal["amount"] raw = create_raw_tx( - [{"txid": dust["txid"], "vout": dust["vout"]}, - {"txid": normal["txid"], "vout": normal["vout"]}], - [{dest: round(total - 0.0001, 8)}] + [ + {"txid": dust["txid"], "vout": dust["vout"]}, + {"txid": normal["txid"], "vout": normal["vout"]}, + ], + [{dest: round(total - FEE, 8)}], ) signed = sign_raw_tx("alice", raw) txid = send_raw(signed["hex"]) mine_and_confirm() - ok(f"Spent dust ({int(dust['amount']*1e8)} sats) + normal ({normal['amount']:.8f}) together in TX {txid[:16]}…") + ok( + f"Spent dust ({int(dust['amount']*1e8)} sats) + normal ({normal['amount']:.8f}) together in TX {txid[:16]}…" + ) + # ═══════════════════════════════════════════════════════════════════════════════ # 5. Change Detection # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_05(): header(5, "Change Detection — Round Payment") - ensure_funds("alice", 1.0) + ensure_funds("alice", 1.0 * VALUE_FACTOR) bob_addr = get_new_address("bob", "bech32") - txid = send_to_address("alice", bob_addr, 0.05) + txid = send_to_address("alice", bob_addr, 0.05 * VALUE_FACTOR) mine_and_confirm() - ok(f"Alice paid Bob 0.05 BTC (round amount) in TX {txid[:16]}… — change output is obvious") + ok( + f"Alice paid Bob {0.05 * VALUE_FACTOR:.8f} BTC (round amount) in TX {txid[:16]}… — change output is obvious" + ) + # ═══════════════════════════════════════════════════════════════════════════════ # 6. Consolidation Origin # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_06(): header(6, "Consolidation Origin") - ensure_funds("bob", 2.0) + ensure_funds("bob", 2.0 * VALUE_FACTOR) for _ in range(4): addr = get_new_address("alice", "bech32") - send_to_address("bob", addr, 0.003) + send_to_address("bob", addr, 0.003 * VALUE_FACTOR) mine_and_confirm() - utxos = get_utxos("alice", 1) - small = [u for u in utxos if 0.002 < u["amount"] < 0.004][:4] + utxos = get_utxos("alice", MIN_CONF) + small = [u for u in utxos if 0.002 * VALUE_FACTOR < u["amount"] < 0.004 * VALUE_FACTOR][:4] if len(small) < 3: info(f"Only {len(small)} small UTXOs, creating more…") for _ in range(4): addr = get_new_address("alice", "bech32") - send_to_address("bob", addr, 0.003) + send_to_address("bob", addr, 0.003 * VALUE_FACTOR) mine_and_confirm() - utxos = get_utxos("alice", 1) - small = [u for u in utxos if 0.002 < u["amount"] < 0.004][:4] + utxos = get_utxos("alice", MIN_CONF) + small = [u for u in utxos if 0.002 * VALUE_FACTOR < u["amount"] < 0.004 * VALUE_FACTOR][:4] inputs = [{"txid": u["txid"], "vout": u["vout"]} for u in small] consol_addr = get_new_address("alice", "bech32") total = sum(u["amount"] for u in small) - raw = create_raw_tx(inputs, [{consol_addr: round(total - 0.0001, 8)}]) + raw = create_raw_tx(inputs, [{consol_addr: round(total - FEE, 8)}]) signed = sign_raw_tx("alice", raw) consol_txid = send_raw(signed["hex"]) mine_and_confirm() ok(f"Consolidated {len(small)} UTXOs → 1 in TX {consol_txid[:16]}…") # Now spend the consolidated output - utxos = get_utxos("alice", 1) + utxos = get_utxos("alice", MIN_CONF) cu = [u for u in utxos if u["txid"] == consol_txid] if cu: dest = get_new_address("carol", "bech32") raw = create_raw_tx( [{"txid": cu[0]["txid"], "vout": cu[0]["vout"]}], - [{dest: round(cu[0]["amount"] - 0.0001, 8)}] + [{dest: round(cu[0]["amount"] - FEE, 8)}], ) signed = sign_raw_tx("alice", raw) txid2 = send_raw(signed["hex"]) mine_and_confirm() - ok(f"Spent consolidated UTXO in TX {txid2[:16]}… — carries full cluster history") + ok( + f"Spent consolidated UTXO in TX {txid2[:16]}… — carries full cluster history" + ) + # ═══════════════════════════════════════════════════════════════════════════════ # 7. Script Type Mixing # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_07(): header(7, "Script Type Mixing") - ensure_funds("bob", 2.0) + ensure_funds("bob", 2.0 * VALUE_FACTOR) wpkh = get_new_address("alice", "bech32") tr = get_new_address("alice", "bech32m") - send_to_address("bob", wpkh, 0.005) - send_to_address("bob", tr, 0.005) + send_to_address("bob", wpkh, 0.005 * VALUE_FACTOR) + send_to_address("bob", tr, 0.005 * VALUE_FACTOR) mine_and_confirm() - utxos = get_utxos("alice", 1) + utxos = get_utxos("alice", MIN_CONF) + def is_wpkh(addr): - return addr and not addr.startswith(("tb1p","bc1p","bcrt1p")) and addr.startswith(("tb1q","bc1q","bcrt1q")) + return ( + addr + and not addr.startswith(("tb1p", "bc1p", "bcrt1p")) + and addr.startswith(("tb1q", "bc1q", "bcrt1q")) + ) + def is_tr(addr): - return addr and addr.startswith(("tb1p","bc1p","bcrt1p")) - wu = next((u for u in utxos if is_wpkh(u.get("address","")) and u["amount"] >= 0.004), None) - tu = next((u for u in utxos if is_tr(u.get("address","")) and u["amount"] >= 0.004), None) + return addr and addr.startswith(("tb1p", "bc1p", "bcrt1p")) + + wu = next( + (u for u in utxos if is_wpkh(u.get("address", "")) and u["amount"] >= 0.004 * VALUE_FACTOR), + None, + ) + tu = next( + (u for u in utxos if is_tr(u.get("address", "")) and u["amount"] >= 0.004 * VALUE_FACTOR), None + ) if not wu or not tu: info("Could not find both UTXO types") return dest = get_new_address("bob", "bech32") total = wu["amount"] + tu["amount"] raw = create_raw_tx( - [{"txid": wu["txid"], "vout": wu["vout"]}, - {"txid": tu["txid"], "vout": tu["vout"]}], - [{dest: round(total - 0.0002, 8)}] + [ + {"txid": wu["txid"], "vout": wu["vout"]}, + {"txid": tu["txid"], "vout": tu["vout"]}, + ], + [{dest: round(total - FEE, 8)}], ) signed = sign_raw_tx("alice", raw) txid = send_raw(signed["hex"]) mine_and_confirm() ok(f"Mixed P2WPKH + P2TR inputs in TX {txid[:16]}… — script type fingerprint") + # ═══════════════════════════════════════════════════════════════════════════════ # 8. Cluster Merge # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_08(): header(8, "Cluster Merge") - ensure_funds("bob", 2.0) - ensure_funds("carol", 2.0) + ensure_funds("bob", 2.0 * VALUE_FACTOR) + ensure_funds("carol", 2.0 * VALUE_FACTOR) a_addr = get_new_address("alice", "bech32") b_addr = get_new_address("alice", "bech32") - txid_a = send_to_address("bob", a_addr, 0.004) - txid_b = send_to_address("carol", b_addr, 0.004) + txid_a = send_to_address("bob", a_addr, 0.004 * VALUE_FACTOR) + txid_b = send_to_address("carol", b_addr, 0.004 * VALUE_FACTOR) mine_and_confirm() - utxos = get_utxos("alice", 1) + utxos = get_utxos("alice", MIN_CONF) ua = next((u for u in utxos if u["txid"] == txid_a), None) ub = next((u for u in utxos if u["txid"] == txid_b), None) - if not ua: ua = next((u for u in utxos if u.get("address") == a_addr), None) - if not ub: ub = next((u for u in utxos if u.get("address") == b_addr), None) + if not ua: + ua = next((u for u in utxos if u.get("address") == a_addr), None) + if not ub: + ub = next((u for u in utxos if u.get("address") == b_addr), None) if not ua or not ub: info("Could not find both cluster UTXOs") return dest = get_new_address("bob", "bech32") total = ua["amount"] + ub["amount"] raw = create_raw_tx( - [{"txid": ua["txid"], "vout": ua["vout"]}, - {"txid": ub["txid"], "vout": ub["vout"]}], - [{dest: round(total - 0.0002, 8)}] + [ + {"txid": ua["txid"], "vout": ua["vout"]}, + {"txid": ub["txid"], "vout": ub["vout"]}, + ], + [{dest: round(total - FEE, 8)}], ) signed = sign_raw_tx("alice", raw) txid = send_raw(signed["hex"]) mine_and_confirm() ok(f"Merged Bob-cluster and Carol-cluster UTXOs in TX {txid[:16]}…") + # ═══════════════════════════════════════════════════════════════════════════════ # 9. Lookback Depth # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_09(): header(9, "Lookback Depth / UTXO Age") + funder = "miner" if IS_REGTEST else "bob" + ensure_funds(funder, 1.0 * VALUE_FACTOR) old_addr = get_new_address("alice", "bech32") - send_to_address("miner", old_addr, 0.01) - mine_blocks(20) + send_to_address(funder, old_addr, 0.01 * VALUE_FACTOR) + if IS_REGTEST: + mine_blocks(20) + else: + info("Testnet: waiting for first TX to confirm (age spread will be smaller)…") + mine_and_confirm() new_addr = get_new_address("alice", "bech32") - send_to_address("miner", new_addr, 0.01) + send_to_address(funder, new_addr, 0.01 * VALUE_FACTOR) mine_and_confirm() - ok(f"Created old UTXO (20+ blocks ago) and new UTXO (just now) for Alice") + ok(f"Created old UTXO and new UTXO for Alice") + # ═══════════════════════════════════════════════════════════════════════════════ # 10. Exchange Origin # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_10(): header(10, "Exchange Origin — Batch Withdrawal") - ensure_funds("exchange", 5.0) + ensure_funds("exchange", 5.0 * VALUE_FACTOR) batch = {} wallets = ["alice", "bob", "carol", "alice", "bob", "carol", "alice", "bob"] for i in range(8): addr = get_new_address(wallets[i], "bech32") - batch[addr] = round(0.01 + i * 0.001, 8) + batch[addr] = round((0.01 + i * 0.001) * VALUE_FACTOR, 8) txid = cli("sendmany", "", json.dumps(batch), wallet="exchange") mine_and_confirm() ok(f"Exchange batch withdrawal to 8 recipients in TX {txid[:16]}…") + # ═══════════════════════════════════════════════════════════════════════════════ # 11. Tainted UTXOs # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_11(): header(11, "Tainted UTXOs / Dirty Money") - ensure_funds("risky", 2.0) - ensure_funds("bob", 1.0) + ensure_funds("risky", 2.0 * VALUE_FACTOR) + ensure_funds("bob", 1.0 * VALUE_FACTOR) ta = get_new_address("alice", "bech32") - taint_txid = send_to_address("risky", ta, 0.01) + taint_txid = send_to_address("risky", ta, 0.01 * VALUE_FACTOR) ca = get_new_address("alice", "bech32") - clean_txid = send_to_address("bob", ca, 0.01) + clean_txid = send_to_address("bob", ca, 0.01 * VALUE_FACTOR) mine_and_confirm() - utxos = get_utxos("alice", 1) + utxos = get_utxos("alice", MIN_CONF) tu = next((u for u in utxos if u["txid"] == taint_txid), None) cu = next((u for u in utxos if u["txid"] == clean_txid), None) - if not tu: tu = next((u for u in utxos if u.get("address") == ta), None) - if not cu: cu = next((u for u in utxos if u.get("address") == ca), None) + if not tu: + tu = next((u for u in utxos if u.get("address") == ta), None) + if not cu: + cu = next((u for u in utxos if u.get("address") == ca), None) if not tu or not cu: info("Could not locate tainted + clean UTXOs") return dest = get_new_address("carol", "bech32") total = tu["amount"] + cu["amount"] raw = create_raw_tx( - [{"txid": tu["txid"], "vout": tu["vout"]}, - {"txid": cu["txid"], "vout": cu["vout"]}], - [{dest: round(total - 0.0002, 8)}] + [ + {"txid": tu["txid"], "vout": tu["vout"]}, + {"txid": cu["txid"], "vout": cu["vout"]}, + ], + [{dest: round(total - FEE, 8)}], ) signed = sign_raw_tx("alice", raw) txid = send_raw(signed["hex"]) mine_and_confirm() ok(f"Merged tainted + clean UTXOs in TX {txid[:16]}… — taint propagation") + # ═══════════════════════════════════════════════════════════════════════════════ # 12. Behavioral Fingerprinting # ═══════════════════════════════════════════════════════════════════════════════ def reproduce_12(): header(12, "Behavioral Fingerprinting") - ensure_funds("alice", 3.0) - ensure_funds("bob", 3.0) + ensure_funds("alice", 3.0 * VALUE_FACTOR) + ensure_funds("bob", 3.0 * VALUE_FACTOR) info("Alice's pattern: round amounts, always bech32…") for i in range(5): dest = get_new_address("carol", "bech32") - send_to_address("alice", dest, 0.01 * (i + 1)) + send_to_address("alice", dest, 0.01 * (i + 1) * VALUE_FACTOR) mine_and_confirm() @@ -364,7 +487,7 @@ def reproduce_12(): for i in range(5): atype = "bech32m" if i % 2 == 0 else "bech32" dest = get_new_address("carol", atype) - send_to_address("bob", dest, round(0.00723 * (i + 1) + 0.00011, 8)) + send_to_address("bob", dest, round((0.00723 * (i + 1) + 0.00011) * VALUE_FACTOR, 8)) mine_and_confirm() ok("Created distinguishable behavioral patterns for Alice and Bob") @@ -388,6 +511,7 @@ ALL = [ (12, "Behavioral Fingerprint", reproduce_12), ] + def main(): filt = None if "-k" in sys.argv: @@ -400,6 +524,8 @@ def main(): print(f"{B}{C} Custom Signet — {get_block_count()} blocks{R}") print(f"{B}{'═'*78}{R}") + setup_wallets() + for num, name, fn in ALL: if filt and str(num) != filt: continue @@ -407,12 +533,15 @@ def main(): fn() except Exception as e: print(f" \033[91m✗ ERROR in {name}: {e}\033[0m") - import traceback; traceback.print_exc() + import traceback + + traceback.print_exc() print(f"\n{B}{'═'*78}{R}") print(f" {G}Done. All vulnerability scenarios have been created on-chain.{R}") print(f" Now run: python3 detect.py ") print(f"{B}{'═'*78}{R}\n") + if __name__ == "__main__": main()