Mirror/stealth
1
0
Fork 0
mirror of https://github.com/LORDBABUINO/stealth.git synced 2026-06-12 07:13:31 -07:00
Code Issues Packages Projects Releases Wiki Activity

move fast break things

Browse Source
This commit is contained in:
Renato Britto
2026-02-27 13:22:19 -03:00
parent b8e4f03695
commit 6188665f46
10 changed files with 1228 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/script/.env.template
+1
View File
@@ -0,0 +1 @@
SSHPASS=
backend/script/DONTRUN_mytest.py
+96
View File
@@ -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] <args> 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] <args> 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"))
backend/script/README.md
+285 -1
View File
@@ -26,4 +26,288 @@ kick
jaguar
paddle
void
dinner
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 -rpcwallet=alice importdescriptors \
'[{"desc":"wpkh(02439f42b07b13682a319c06cc209119d10f96f72db9ba5a0f1b843ef89ce7807c)#j50ja6ax","timestamp":"now","range":[0,100]}]' -->
<!-- bitcoin-cli -testnet4 -rpcwallet=alice importdescriptors \
'[{"desc":"wpkh(02439f42b07b13682a319c06cc209119d10f96f72db9ba5a0f1b843ef89ce7807c)#j50ja6ax","timestamp":"now"}]' -->
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
```
backend/script/bcli.py
+25
View File
@@ -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"))
backend/script/bitcoin_rpc.py
+27 -46
View File
@@ -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] <args> 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']}")
backend/script/lib/bitcoin_rpc_regtest.py
+248
View File
@@ -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] <args> 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] <args> 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] <args> 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}")
backend/script/lib/bitcoin_rpc_testnet.py
+127
View File
@@ -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}")
backend/script/lib/clis.py
+167
View File
@@ -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] <args> 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] <args> 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] <args> 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] <args> 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] <args> 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
backend/script/load.py
+29
View File
@@ -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]))
backend/script/reproduce.py
+223 -94
View File
@@ -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 <descriptor>")
print(f"{B}{'═'*78}{R}\n")
if __name__ == "__main__":
main()