#!/usr/bin/env python3
"""
Wallet Engine — Core Bitcoin wallet operations.
BIP39 mnemonic → BIP32 HD wallet → BIP44/BIP84/BIP86 addresses.
Bitcoin Core RPC for balance/tx checks.
Argon2 + AES encryption for key persistence.
Stack: mnemonic (seed gen) + bip-utils (HD derivation) + python-bitcoinlib (RPC)
"""
import hashlib
import json
import os
import secrets
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import argon2
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from bip_utils import (
Bip44,
Bip44Changes,
Bip44Coins,
Bip84,
Bip84Coins,
Bip86,
Bip86Coins,
)
from mnemonic import Mnemonic
class WalletEngine:
"""HD wallet engine with Bitcoin Core RPC integration."""
def __init__(
self,
network: str = "mainnet",
rpc_user: str = "",
rpc_password: str = "",
rpc_host: str = "127.0.0.1",
rpc_port: int = 8332,
):
if not rpc_user or not rpc_password:
raise ValueError(
"RPC credentials required. Set BTC_RPC_USER and BTC_RPC_PASS environment variables."
)
self.network = network
self.rpc_user = rpc_user
self.rpc_password = rpc_password
self.rpc_host = rpc_host
self.rpc_port = rpc_port
self.mn = Mnemonic("english")
# ------------------------------------------------------------------
# Mnemonic operations
# ------------------------------------------------------------------
def generate_mnemonic(self, strength: int = 128) -> str:
"""Generate a new BIP39 mnemonic phrase.
Args:
strength: Entropy bits — 128 (12 words), 256 (24 words).
Returns:
Space-separated mnemonic words.
"""
return self.mn.generate(strength)
def validate_mnemonic(self, mnemonic: str) -> bool:
"""Validate a BIP39 mnemonic phrase."""
return self.mn.check(mnemonic)
def mnemonic_to_seed(self, mnemonic: str, passphrase: str = "") -> bytes:
"""Convert mnemonic to seed using PBKDF2-HMAC-SHA512."""
return self.mn.to_seed(mnemonic, passphrase)
# ------------------------------------------------------------------
# HD derivation — BIP84 (Native SegWit)
# ------------------------------------------------------------------
def derive_bip84(
self,
mnemonic: str,
passphrase: str = "",
account: int = 0,
) -> dict:
"""Derive BIP84 (Native SegWit) HD wallet structure.
Returns account-level keys + derived addresses.
"""
seed = self.mnemonic_to_seed(mnemonic, passphrase)
bip84 = Bip84.FromSeed(seed, Bip84Coins.BITCOIN)
account_node = bip84.Purpose().Coin().Account(account)
account_obj = account_node.Bip32Object()
xpriv = account_obj.PrivateKey().ToExtended()
xpub = account_obj.PublicKey().ToExtended()
fingerprint = account_node.Bip32Object().FingerPrint().ToHex()
return {
"derivation_path": f"m/84'/0'/{account}'",
"fingerprint": fingerprint,
"xpriv": xpriv,
"xpub": xpub,
"coin": "BTC",
"standard": "BIP84",
}
def derive_bip84_address(
self,
mnemonic: str,
passphrase: str = "",
account: int = 0,
address_index: int = 0,
internal: bool = False,
) -> dict:
"""Derive a single BIP84 address.
Args:
internal: False = receiving chain (0), True = change chain (1).
"""
seed = self.mnemonic_to_seed(mnemonic, passphrase)
bip84 = Bip84.FromSeed(seed, Bip84Coins.BITCOIN)
change = Bip44Changes.CHAIN_INT if internal else Bip44Changes.CHAIN_EXT
addr_node = bip84.Purpose().Coin().Account(account).Change(change).AddressIndex(address_index)
addr_obj = addr_node.Bip32Object()
pubkey = bytes(addr_obj.PublicKey().RawCompressed())
privkey = bytes(addr_obj.PrivateKey().Raw())
path = f"m/84'/0'/{account}'/{1 if internal else 0}/{address_index}"
return {
"address": addr_node.PublicKey().ToAddress(),
"path": path,
"pubkey": pubkey.hex(),
"privkey_wif": self._private_key_to_wif(privkey),
"index": address_index,
"internal": internal,
}
# ------------------------------------------------------------------
# HD derivation — BIP44 (Legacy P2PKH)
# ------------------------------------------------------------------
def derive_bip44(
self,
mnemonic: str,
passphrase: str = "",
account: int = 0,
) -> dict:
"""Derive BIP44 (Legacy P2PKH) HD wallet structure."""
seed = self.mnemonic_to_seed(mnemonic, passphrase)
bip44 = Bip44.FromSeed(seed, Bip44Coins.BITCOIN)
account_node = bip44.Purpose().Coin().Account(account)
account_obj = account_node.Bip32Object()
return {
"derivation_path": f"m/44'/0'/{account}'",
"fingerprint": account_obj.FingerPrint().ToHex(),
"xpriv": account_obj.PrivateKey().ToExtended(),
"xpub": account_obj.PublicKey().ToExtended(),
"coin": "BTC",
"standard": "BIP44",
}
def derive_bip44_address(
self,
mnemonic: str,
passphrase: str = "",
account: int = 0,
address_index: int = 0,
internal: bool = False,
) -> dict:
"""Derive a single BIP44 (Legacy) address."""
seed = self.mnemonic_to_seed(mnemonic, passphrase)
bip44 = Bip44.FromSeed(seed, Bip44Coins.BITCOIN)
change = Bip44Changes.CHAIN_INT if internal else Bip44Changes.CHAIN_EXT
addr_node = bip44.Purpose().Coin().Account(account).Change(change).AddressIndex(address_index)
addr_obj = addr_node.Bip32Object()
pubkey = bytes(addr_obj.PublicKey().RawCompressed())
privkey = bytes(addr_obj.PrivateKey().Raw())
path = f"m/44'/0'/{account}'/{1 if internal else 0}/{address_index}"
return {
"address": addr_node.PublicKey().ToAddress(),
"path": path,
"pubkey": pubkey.hex(),
"privkey_wif": self._private_key_to_wif(privkey),
"index": address_index,
"internal": internal,
}
# ------------------------------------------------------------------
# HD derivation — BIP86 (Taproot)
# ------------------------------------------------------------------
def derive_bip86(
self,
mnemonic: str,
passphrase: str = "",
account: int = 0,
) -> dict:
"""Derive BIP86 (Taproot/SegWit v1) HD wallet structure."""
seed = self.mnemonic_to_seed(mnemonic, passphrase)
bip86 = Bip86.FromSeed(seed, Bip86Coins.BITCOIN)
account_node = bip86.Purpose().Coin().Account(account)
account_obj = account_node.Bip32Object()
return {
"derivation_path": f"m/86'/0'/{account}'",
"fingerprint": account_obj.FingerPrint().ToHex(),
"xpriv": account_obj.PrivateKey().ToExtended(),
"xpub": account_obj.PublicKey().ToExtended(),
"coin": "BTC",
"standard": "BIP86",
}
def derive_bip86_address(
self,
mnemonic: str,
passphrase: str = "",
account: int = 0,
address_index: int = 0,
internal: bool = False,
) -> dict:
"""Derive a single BIP86 (Taproot) address."""
seed = self.mnemonic_to_seed(mnemonic, passphrase)
bip86 = Bip86.FromSeed(seed, Bip86Coins.BITCOIN)
change = Bip44Changes.CHAIN_INT if internal else Bip44Changes.CHAIN_EXT
addr_node = bip86.Purpose().Coin().Account(account).Change(change).AddressIndex(address_index)
addr_obj = addr_node.Bip32Object()
pubkey = bytes(addr_obj.PublicKey().RawCompressed())
privkey = bytes(addr_obj.PrivateKey().Raw())
path = f"m/86'/0'/{account}'/{1 if internal else 0}/{address_index}"
return {
"address": addr_node.PublicKey().ToAddress(),
"path": path,
"pubkey": pubkey.hex(),
"privkey_wif": self._private_key_to_wif(privkey),
"index": address_index,
"internal": internal,
}
# ------------------------------------------------------------------
# Key utilities
# ------------------------------------------------------------------
@staticmethod
def _private_key_to_wif(privkey: bytes) -> str:
"""Convert raw private key bytes to WIF (Wallet Import Format)."""
# Prefix with 0x80 for mainnet
header = b"\x80"
checksum = hashlib.sha256(hashlib.sha256(header + privkey).digest()).digest()[:4]
encoded = header + privkey + checksum
return _base58_encode(encoded)
# ------------------------------------------------------------------
# Encryption
# ------------------------------------------------------------------
def encrypt_key(self, plaintext: str, password: str = "") -> dict:
"""Encrypt sensitive data with Argon2 + AES-256-GCM.
Args:
plaintext: Data to encrypt (mnemonic phrase).
password: Password for key derivation. If empty, uses random key
(stored in response — for testing only).
"""
salt = secrets.token_bytes(16)
iv = secrets.token_bytes(12)
if password:
# Derive key from password using Argon2id
import argon2.low_level as argon2_ll
key = argon2_ll.hash_secret_raw(
secret=password.encode(),
salt=salt,
time_cost=5,
memory_cost=65536,
parallelism=4,
hash_len=32,
type=argon2_ll.Type.ID,
)
else:
raise ValueError(
"Password is required — unencrypted wallets are not permitted"
)
# Encrypt
cipher = AES.new(key, AES.MODE_GCM, nonce=iv)
ciphertext, tag = cipher.encrypt_and_digest(
pad(plaintext.encode("utf-8"), AES.block_size)
)
result = {
"ciphertext": ciphertext.hex(),
"iv": iv.hex(),
"tag": tag.hex(),
"salt": salt.hex(),
}
if not password:
result["key"] = key.hex() # Only for testing
return result
def decrypt_key(self, encrypted: dict, password_or_key: str = "") -> str:
"""Decrypt data that was encrypted with encrypt_key.
Args:
encrypted: Dict from encrypt_key.
password_or_key: Password for derivation, or raw hex key if wallet is unencrypted.
"""
iv = bytes.fromhex(encrypted["iv"])
tag = bytes.fromhex(encrypted["tag"])
ciphertext = bytes.fromhex(encrypted["ciphertext"])
if "key" in encrypted:
raise ValueError(
"Cannot decrypt unencrypted wallet — legacy format no longer supported"
)
# Derive key from password
salt = bytes.fromhex(encrypted["salt"])
import argon2.low_level as argon2_ll
key = argon2_ll.hash_secret_raw(
secret=password_or_key.encode(),
salt=salt,
time_cost=5,
memory_cost=65536,
parallelism=4,
hash_len=32,
type=argon2_ll.Type.ID,
)
cipher = AES.new(key, AES.MODE_GCM, nonce=iv)
plaintext = unpad(
cipher.decrypt_and_verify(ciphertext, tag),
AES.block_size,
)
return plaintext.decode("utf-8")
# ------------------------------------------------------------------
# Wallet management
# ------------------------------------------------------------------
SENSITIVE_KEYS = {"xpriv", "privkey_wif", "privkey", "seed", "key", "encrypted_passphrase"}
@staticmethod
def sanitize_wallet(wallet: dict) -> dict:
"""Strip sensitive keys from a wallet dict before sending over API."""
if not isinstance(wallet, dict):
return wallet
return {
k: v
for k, v in wallet.items()
if k not in WalletEngine.SENSITIVE_KEYS
}
@staticmethod
def sanitize_addresses(addresses: list) -> list:
"""Strip private keys from address derivation results."""
cleaned = []
for addr in addresses:
if isinstance(addr, dict):
cleaned.append({
k: v
for k, v in addr.items()
if k not in WalletEngine.SENSITIVE_KEYS
})
else:
cleaned.append(addr)
return cleaned
def create_wallet(
self,
name: str,
mnemonic: str,
password: str,
passphrase: str = "",
) -> dict:
"""Create a wallet from a mnemonic phrase.
Password is mandatory — plaintext seed storage is no longer permitted.
Args:
name: Wallet name.
mnemonic: BIP39 mnemonic phrase.
password: Encryption password for the wallet.
passphrase: Optional BIP39 passphrase (different from encryption password).
This is the "25th word" that modifies seed derivation.
Returns wallet metadata with encrypted seed.
"""
if not self.validate_mnemonic(mnemonic):
raise ValueError("Invalid BIP39 mnemonic phrase")
if not password:
raise ValueError("Password is required — wallets cannot be stored unencrypted")
now = datetime.now(timezone.utc).isoformat()
# Derive fingerprint from BIP84 account 0 using passphrase
bip84_wallet = self.derive_bip84(mnemonic, passphrase=passphrase)
wallet = {
"name": name,
"fingerprint": bip84_wallet["fingerprint"],
"encrypted": True,
"created": now,
"xpub": bip84_wallet["xpub"],
"derivation_path": bip84_wallet["derivation_path"],
"encrypted_seed": self.encrypt_key(mnemonic, password),
"has_passphrase": bool(passphrase),
}
# If there's a BIP39 passphrase, encrypt and store it separately
if passphrase:
wallet["encrypted_passphrase"] = self.encrypt_key(passphrase, password)
return wallet
def get_addresses(
self,
wallet: dict,
count: int = 5,
derivation_type: str = "bech32",
password: str = "",
) -> list:
"""Derive multiple addresses from a wallet.
Args:
wallet: Wallet dict with encrypted_seed and metadata.
count: Number of addresses to derive.
derivation_type: "bech32" (BIP84), "legacy" (BIP44), "taproot" (BIP86).
password: Password to decrypt seed if wallet is encrypted.
"""
# Recover mnemonic from encrypted seed
encrypted_seed = wallet.get("encrypted_seed")
if encrypted_seed:
if not password:
raise ValueError("Password required for encrypted wallet")
mnemonic = self.decrypt_key(encrypted_seed, password)
else:
raise ValueError("No encrypted seed found in wallet data")
if not self.validate_mnemonic(mnemonic):
raise ValueError("Invalid mnemonic after decryption")
addresses = []
for i in range(count):
if derivation_type == "bech32":
addr = self.derive_bip84_address(mnemonic, address_index=i)
addr["type"] = "BIP84"
elif derivation_type == "legacy":
addr = self.derive_bip44_address(mnemonic, address_index=i)
addr["type"] = "BIP44"
elif derivation_type == "taproot":
addr = self.derive_bip86_address(mnemonic, address_index=i)
addr["type"] = "BIP86"
else:
raise ValueError(f"Unknown derivation type: {derivation_type}")
addresses.append(addr)
# Strip private keys before returning
return self.sanitize_addresses(addresses)
@staticmethod
def from_dict(data: dict) -> "WalletEngine":
"""Reconstruct a WalletEngine from a dict (no-op — engine is stateless)."""
# Engine is stateless — just return a fresh instance.
# All state lives in wallet dicts.
return WalletEngine()
# ------------------------------------------------------------------
# RPC operations
# ------------------------------------------------------------------
def get_block_count(self) -> int:
"""Get current block count from Bitcoin Core."""
result = self._rpc_call("getblockcount")
return result
def get_network_info(self) -> dict:
"""Get Bitcoin Core network information."""
result = self._rpc_call("getnetworkinfo")
return {
"version": result.get("version"),
"subversion": result.get("subversion"),
"protocol_version": result.get("protocolversion"),
"connections": result.get("connections"),
"network": result.get("network"),
"relay_fee": result.get("relayfee"),
}
def get_balance(self, address: str, wallet_name: str = "watchonly") -> dict:
"""Get address balance via watch-only wallet.
Uses listunspent on the imported watch-only wallet for instant results.
"""
try:
utxos = self._rpc_call("listunspent", [0, 999999, [address]], wallet_name=wallet_name)
total_sats = int(sum(u["amount"] for u in utxos) * 1e8)
return {
"address": address,
"balance_sats": total_sats,
"utxo_count": len(utxos),
"source": "wallet",
}
except (RuntimeError, ConnectionError):
return {
"address": address,
"balance_sats": 0,
"utxo_count": 0,
"source": "unavailable",
"note": "No watch-only wallet imported for this address",
}
def get_address_info(self, address: str) -> dict:
"""Get detailed address information.
Requires addrindex=1. Returns minimal info otherwise.
"""
try:
return self._rpc_call("getaddressinfo", [address])
except (RuntimeError, ConnectionError):
return {
"address": address,
"note": "Enable addrindex=1 in bitcoin.conf for full address info",
}
def list_transactions(self, address: str, count: int = 10, wallet_name: str = "watchonly") -> list:
"""List recent transactions for an address via watch-only wallet."""
try:
result = self._rpc_call("listtransactions", ["*", 50, 0, True], wallet_name=wallet_name)
filtered = [tx for tx in result if tx.get("address") == address]
return filtered[:count]
except (RuntimeError, ConnectionError):
try:
result = self._rpc_call("listreceivedbyaddress", [0, False], wallet_name=wallet_name)
filtered = [tx for tx in result if tx["address"] == address]
return filtered[:count]
except (RuntimeError, ConnectionError):
return []
# ------------------------------------------------------------------
# Watch-only wallet management
# ------------------------------------------------------------------
def import_wallet_watchonly(self, xpub: str, wallet_name: str = "watchonly") -> dict:
"""Import wallet as watch-only via importdescriptors.
Imports BIP84 receiving and change chains. Only public keys — node
never sees private keys.
Args:
xpub: Extended public key from BIP84 derivation.
wallet_name: Name for the watch-only wallet.
Returns:
Import result from Bitcoin Core.
"""
# Create watch-only wallet if it doesn't exist
try:
self._rpc_call("createwallet", [wallet_name, False, False, "", False, True, True, 0, True])
except RuntimeError:
# Wallet already exists — fine
pass
# Get descriptor info for validation
desc_receiving = f"wpkh({xpub}/0/*)"
desc_change = f"wpkh({xpub}/1/*)"
requests = [
{
"desc": desc_receiving,
"timestamp": "now",
"active": True,
"range": [0, 1000],
"next_index": 0,
},
{
"desc": desc_change,
"timestamp": "now",
"active": True,
"range": [0, 1000],
"next_index": 0,
},
]
result = self._rpc_call("importdescriptors", [requests], wallet_name=wallet_name)
return {
"wallet_name": wallet_name,
"imported": result,
"xpub": xpub,
}
def estimate_fee(self, blocks: int = 6) -> dict:
"""Estimate transaction fee rate for target confirmation in blocks."""
try:
result = self._rpc_call("estimatesmartfee", [blocks])
feerate = result.get("feerate")
if feerate is None:
return {
"blocks": blocks,
"feerate_btc": None,
"note": "Not enough data for fee estimation",
}
return {
"blocks": blocks,
"feerate_btc": feerate,
"feerate_sats_vb": feerate * 1e8,
}
except (RuntimeError, ConnectionError):
return {
"blocks": blocks,
"feerate_btc": None,
"note": "Fee estimation unavailable",
}
def get_tx_history(self, address: str, count: int = 20) -> list:
"""Get transaction history for an address."""
return self.list_transactions(address, count=count)
def get_utxos(self, addresses: list) -> list:
"""Get all UTXOs for a list of addresses from the watch-only wallet.
Args:
addresses: List of addresses to query.
Returns:
List of UTXO dicts with txid, vout, amount, address.
"""
try:
utxos = self._rpc_call("listunspent", [0, 999999, addresses])
return utxos
except (RuntimeError, ConnectionError):
return []
# ------------------------------------------------------------------
# Coin selection
# ------------------------------------------------------------------
def coin_select(self, utxos: list, target_btc: float, fee_rate: float, strategy: str = "largest") -> list:
"""Select UTXOs to cover target amount + fees.
Strategies:
largest: Greedy largest-first (fast, usually fine)
bnb: Branch-and-bound knapsack (optimal but slower for large sets)
Args:
utxos: List of UTXO dicts from listunspent (have 'amount' key in BTC).
target_btc: Target output amount in BTC (excludes fees).
fee_rate: Fee rate in sats/vB.
strategy: 'largest' or 'bnb'.
Returns:
List of selected UTXOs.
Raises:
ValueError: If no combination covers the target.
"""
total_needed = target_btc
# Pre-compute with estimated fees for selected UTXOs
# Each input costs ~140 vB for P2WPKH witness sig, 0.00000014 * fee_rate BTC
input_fee_btc = 140 * fee_rate / 1e8
if strategy == "bnb" and len(utxos) <= 25:
return _bnb_coin_select(utxos, target_btc, input_fee_btc)
# Default: largest-first (works for any set size)
sorted_utxos = sorted(utxos, key=lambda u: u["amount"], reverse=True)
selected = []
running_sum = 0.0
for u in sorted_utxos:
selected.append(u)
running_sum += u["amount"] - input_fee_btc
if running_sum >= target_btc:
break
# Final check: does selection actually cover target + fees?
total_input = sum(u["amount"] for u in selected)
total_fees = len(selected) * input_fee_btc
if total_input - total_fees < target_btc:
raise ValueError(
f"Insufficient balance after coin selection. "
f"Have {total_input:.8f} BTC, need {target_btc + total_fees:.8f} BTC"
)
return selected
# Bitcoin address prefix validation (C12 fix)
_ADDRESS_PREFIXES = ("1", "3", "bc1")
@staticmethod
def validate_bitcoin_address(address: str) -> bool:
"""Validate Bitcoin address format.
Accepts P2PKH (1...), P2SH (3...), and Bech32 (bc1...) addresses.
"""
if not isinstance(address, str) or len(address) < 25:
return False
return address.startswith(WalletEngine._ADDRESS_PREFIXES)
def create_psbt(
self,
wallet: dict,
outputs: list,
fee_rate: float = 10.0,
addresses: list = None,
password: str = "",
coin_selection: str = "largest",
selected_utxos: list = None,
) -> dict:
"""Create a PSBT (Partially Signed Bitcoin Transaction).
Uses createpsbt RPC to build an unsigned transaction.
Args:
wallet: Wallet dict with seed/encrypted_seed.
outputs: List of {"address": str, "amount": float (BTC)}.
fee_rate: Satoshis per vbyte.
addresses: List of wallet addresses to scan for UTXOs.
password: Password if wallet is encrypted.
coin_selection: 'largest', 'bnb', or 'manual'.
selected_utxos: Pre-selected UTXOs (for manual mode).
Returns:
Dict with PSBT string and transaction details.
"""
# C11: Validate amounts and fee rate
if not isinstance(outputs, list) or len(outputs) == 0:
raise ValueError("outputs must be a non-empty list")
for i, out in enumerate(outputs):
if "address" not in out or "amount" not in out:
raise ValueError(
f"Output {i} missing required 'address' or 'amount' field"
)
amount = out["amount"]
if not isinstance(amount, (int, float)):
raise ValueError(f"Output {i}: amount must be a number")
if amount <= 0:
raise ValueError(
f"Output {i}: amount must be positive, got {amount}"
)
max_sats = 21_000_000 * 1e8 # 21 million BTC
if amount * 1e8 > max_sats:
raise ValueError(
f"Output {i}: amount exceeds Bitcoin maximum (21M BTC)"
)
# C12: Validate address formats
for i, out in enumerate(outputs):
if not self.validate_bitcoin_address(out["address"]):
raise ValueError(
f"Output {i}: invalid Bitcoin address format: {out['address'][:20]}..."
)
# Validate fee rate bounds (M10 fix)
if not isinstance(fee_rate, (int, float)):
raise ValueError("fee_rate must be a number")
if fee_rate <= 0:
raise ValueError("fee_rate must be positive")
if fee_rate > 10_000: # 10,000 sats/vB sanity cap
raise ValueError(
f"fee_rate suspiciously high: {fee_rate} sats/vB. "
"This is likely a unit error (sats vs BTC)."
)
if addresses is None:
addrs = self.get_addresses(wallet, count=25, password=password)
addresses = [a["address"] for a in addrs]
all_utxos = self.get_utxos(addresses)
if not all_utxos:
raise ValueError("No UTXOs found for wallet addresses")
total_output = sum(o["amount"] for o in outputs)
# Coin selection
if selected_utxos is not None:
# Manual selection — user picked specific UTXOs
utxos = selected_utxos
elif coin_selection == "manual":
raise ValueError("Manual coin selection requires selected_utxos")
else:
utxos = self.coin_select(all_utxos, total_output, fee_rate, strategy=coin_selection)
total_input = sum(u["amount"] for u in utxos)
# Build initial PSBT without change to get accurate vsize from decodepsbt
inputs_for_psbt = [{"txid": u["txid"], "vout": u["vout"]} for u in utxos]
outputs_no_change = {}
for o in outputs:
outputs_no_change[o["address"]] = o["amount"]
# Use decodepsbt to get accurate vsize instead of guessing
psbt_hex = self._rpc_call(
"createpsbt",
[inputs_for_psbt, outputs_no_change, 0, True], # locktime=0, replaceable=True (RBF)
)
try:
decoded = self._rpc_call("decodepsbt", [psbt_hex])
# decodepsbt returns size/vsize in the psbt section
tx_vsize = decoded.get("psbt", {}).get("vsize", None)
if tx_vsize is None:
# Fallback: use global vsize from the extracted transaction
tx_vsize = decoded.get("psbt", {}).get("global", {}).get("vsize", None)
except Exception:
tx_vsize = None
if tx_vsize is None:
# Final fallback: estimate formula
num_outputs = len(outputs) + 1 # +1 for potential change
tx_vsize = 250 + len(utxos) * 100 + num_outputs * 43
fee_btc = (tx_vsize * fee_rate) / 1e8
if total_input < (total_output + fee_btc):
raise ValueError(
f"Insufficient balance. Have {total_input:.8f} BTC, "
f"need {total_output + fee_btc:.8f} BTC (output + fee)"
)
inputs = [{"txid": u["txid"], "vout": u["vout"]} for u in utxos]
outputs_dict = {}
for o in outputs:
outputs_dict[o["address"]] = o["amount"]
# Derive change address from internal chain (m/84'/0'/0'/1/*)
# instead of reusing a receiving address
encrypted_seed = wallet.get("encrypted_seed")
mnemonic_for_change = self.decrypt_key(encrypted_seed, password)
change_addr_info = self.derive_bip84_address(
mnemonic_for_change, address_index=0, internal=True
)
change_addr = change_addr_info["address"]
change_amount = total_input - total_output - fee_btc
if change_amount > 0.00000546:
outputs_dict[change_addr] = round(change_amount, 8)
psbt_hex = self._rpc_call(
"createpsbt",
[inputs, outputs_dict, 0, True], # locktime=0, replaceable=True (RBF)
)
return {
"psbt": psbt_hex,
"inputs": len(inputs),
"outputs": len(outputs_dict),
"total_input": round(total_input, 8),
"total_output": round(total_output, 8),
"fee_btc": round(fee_btc, 8),
"change_address": change_addr if change_amount > 0.00000546 else None,
"change_amount": round(change_amount, 8) if change_amount > 0.00000546 else 0,
"selected_utxos": utxos,
"coin_selection": coin_selection if selected_utxos is None else "manual",
}
def sign_psbt(
self,
wallet: dict,
psbt_hex: str,
password: str,
) -> dict:
"""Sign a PSBT using the wallet's private keys.
Uses a temporary Bitcoin Core descriptor wallet for signing —
keys are derived lazily by Core, never bulk-precomputed in memory.
The passphrase (BIP39 "25th word") is respected from the wallet record.
Args:
wallet: Wallet dict with encrypted_seed and optional encrypted_passphrase.
psbt_hex: PSBT hex string (BIP174).
password: Password to decrypt wallet (required).
Returns:
Dict with signed PSBT hex and completion status.
"""
encrypted_seed = wallet.get("encrypted_seed")
if not encrypted_seed:
raise ValueError("No encrypted seed found in wallet data")
if not password:
raise ValueError("Password required for signing")
# Decrypt mnemonic
mnemonic = self.decrypt_key(encrypted_seed, password)
# Decrypt BIP39 passphrase if present (C6 fix — no longer silently dropped)
passphrase = ""
enc_passphrase = wallet.get("encrypted_passphrase")
if enc_passphrase:
passphrase = self.decrypt_key(enc_passphrase, password)
# Derive seed with passphrase and xpriv for descriptor wallet (C7 fix)
seed = self.mnemonic_to_seed(mnemonic, passphrase)
bip84 = Bip84.FromSeed(seed, Bip84Coins.BITCOIN)
account_node = bip84.Purpose().Coin().Account(0)
account_obj = account_node.Bip32Object()
xpriv = account_obj.PrivateKey().ToExtended()
# Use a temporary Bitcoin Core descriptor wallet for signing
# This avoids pre-deriving 200 keys in memory (C7 fix)
btc_wallet_name = "temp_sign_" + wallet.get("fingerprint", "wallet")[:8]
self._ensure_bip84_signing_wallet(btc_wallet_name, xpriv, password)
sign_result = self._rpc_call(
"signrawtransactionwithwallet",
[psbt_hex],
wallet_name=btc_wallet_name,
)
signed_psbt = sign_result.get("psbt", psbt_hex)
complete = sign_result.get("complete", False)
return {
"psbt": signed_psbt,
"complete": complete,
"all_signed": complete,
"finalized": False,
"raw_tx_hex": None,
}
def finalize_psbt(self, psbt_hex: str) -> dict:
"""Finalize a fully-signed PSBT and extract the raw transaction.
Args:
psbt_hex: Signed PSBT hex string from sign_psbt().
Returns:
Dict with raw_tx_hex and completion status.
"""
result = self._rpc_call("finalizepsbt", [psbt_hex, True])
return {
"raw_tx_hex": result.get("hex"),
"complete": result.get("complete", False),
"psbt": result.get("psbt"),
}
def sign_and_finalize_psbt(
self,
wallet: dict,
psbt_hex: str,
password: str,
) -> dict:
"""Sign and finalize a PSBT in one call.
Convenience method that chains sign_psbt → finalize_psbt.
Args:
wallet: Wallet dict with seed/encrypted_seed.
psbt_hex: PSBT hex string (BIP174).
password: Password for encrypted wallet — required, never empty.
Returns:
Dict with raw_tx_hex ready for broadcast.
"""
if not password:
raise ValueError("Password required for signing")
sign_result = self.sign_psbt(wallet, psbt_hex, password)
if not sign_result.get("complete"):
return {
"raw_tx_hex": None,
"complete": False,
"error": "PSBT signing incomplete — not all inputs signed",
}
final_result = self.finalize_psbt(sign_result["psbt"])
return final_result
def broadcast_tx(self, raw_tx_hex: str) -> dict:
"""Broadcast a raw transaction to the Bitcoin network.
Args:
raw_tx_hex: Raw transaction hex string.
Returns:
Dict with txid, broadcast status, and verification result.
"""
txid = self._rpc_call("sendrawtransaction", [raw_tx_hex, 1])
# Verify tx made it into mempool (13a/13b)
verified = False
mempool_position = None
try:
# getmempoolentry confirms tx is in local mempool
mempool = self._rpc_call("getmempoolentry", [txid])
verified = True
mempool_position = {
"fee_rate_sats_vb": mempool.get("fees", {}).get("base", 0) * 1e8,
"vsize": mempool.get("vsize"),
"time": mempool.get("time"),
"height": mempool.get("height"),
}
except Exception:
# Tx might still be propagating — not a hard failure
pass
return {
"txid": txid,
"broadcast": True,
"verified": verified,
"mempool": mempool_position,
}
def decode_psbt(self, psbt_hex: str) -> dict:
"""Decode a PSBT and return human-readable info.
Uses decoderawtransaction RPC to get transaction details.
Args:
psbt_hex: PSBT hex string.
Returns:
Dict with decoded transaction details.
"""
try:
result = self._rpc_call("decoderawtransaction", [psbt_hex])
return result
except RuntimeError:
# If it's a PSBT hex, try to finalize and decode
try:
final = self._rpc_call("finalizepsbt", [psbt_hex])
hex_str = final.get("hex", "")
if hex_str:
result = self._rpc_call("decoderawtransaction", [hex_str])
return result
except RuntimeError:
pass
return {
"type": "PSBT",
"psbt": psbt_hex,
"note": "Raw PSBT — finalize first to decode",
}
# RPC methods that require a wallet context (/wallet/{name})
_WALLET_RPC_METHODS = frozenset({
"importdescriptors", "listunspent", "listtransactions",
"listreceivedbyaddress", "getbalance", "sendfrom",
"settxfee", "getreceivedbyaddress", "abandontransaction",
})
# ------------------------------------------------------------------
# Wallet persistence
# ------------------------------------------------------------------
def save_wallet(self, wallet: dict, wallet_dir: str = None) -> str:
"""Save wallet to disk as encrypted JSON.
Args:
wallet: Wallet dict from create_wallet().
wallet_dir: Directory to store wallet files. Defaults to ~/.btc-wallet/.
Returns:
Path to saved wallet file.
"""
import pathlib
if wallet_dir is None:
wallet_dir = pathlib.Path.home() / ".btc-wallet"
wallet_path = pathlib.Path(wallet_dir)
wallet_path.mkdir(parents=True, exist_ok=True)
# Restrict wallet directory to owner-only (700)
os.chmod(str(wallet_path), 0o700)
name = wallet.get("name", "wallet").replace(" ", "_")
fp = wallet.get("fingerprint", "unknown")
fname = f"{name}_{fp}.json"
fpath = wallet_path / fname
now = datetime.now(timezone.utc).isoformat()
data = {
"version": 2,
"wallet": wallet,
"labels": wallet.get("labels", {"addresses": {}, "transactions": {}}),
"created": wallet.get("created", now),
"modified": now,
}
fpath.write_text(json.dumps(data, indent=2))
# Restrict wallet file to owner-only (600)
os.chmod(str(fpath), 0o600)
return str(fpath)
def load_wallet(self, fpath: str, password: str = "") -> dict:
"""Load wallet from disk.
Args:
fpath: Path to wallet JSON file.
password: Password if wallet was created with encryption.
Returns:
Wallet dict with labels merged in.
Raises:
FileNotFoundError: If path doesn't exist.
ValueError: If password is wrong or wallet is corrupted.
"""
import pathlib
p = pathlib.Path(fpath)
if not p.exists():
raise FileNotFoundError(f"Wallet file not found: {fpath}")
data = json.loads(p.read_text())
if data.get("version") != 2:
raise ValueError(f"Unsupported wallet version: {data.get('version')}")
wallet = data["wallet"]
# If wallet was encrypted with a password and no seed is present,
# we need the password to decrypt
if wallet.get("encrypted") and not wallet.get("seed"):
if not password:
raise ValueError("Password required for encrypted wallet")
# Decrypt the encrypted_seed
enc = wallet.get("encrypted_seed")
if enc:
mnemonic = self.decrypt_key(enc, password)
wallet["seed"] = mnemonic
wallet["encrypted"] = False
# Merge labels into wallet
wallet["labels"] = data.get("labels", {"addresses": {}, "transactions": {}})
wallet["file_path"] = str(p)
wallet["modified"] = data.get("modified", "")
return wallet
def list_wallets(self, wallet_dir: str = None) -> list:
"""List all saved wallets on disk.
Args:
wallet_dir: Directory to scan. Defaults to ~/.btc-wallet/.
Returns:
List of wallet metadata dicts with path, name, fingerprint, modified time.
"""
import pathlib
if wallet_dir is None:
wallet_dir = pathlib.Path.home() / ".btc-wallet"
d = pathlib.Path(wallet_dir)
if not d.exists():
return []
wallets = []
for f in sorted(d.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True):
try:
data = json.loads(f.read_text())
w = data.get("wallet", {})
wallets.append({
"path": str(f),
"name": w.get("name", "unknown"),
"type": w.get("type", "single-sig"),
"fingerprint": w.get("fingerprint", "unknown"),
"encrypted": w.get("encrypted", False),
"modified": data.get("modified", ""),
"created": data.get("created", ""),
})
except (json.JSONDecodeError, KeyError):
continue
return wallets
# ------------------------------------------------------------------
# Labels
# ------------------------------------------------------------------
def set_label(self, wallet: dict, kind: str, identifier: str, label: str) -> dict:
"""Set a label on an address or transaction.
Args:
wallet: Wallet dict with "labels" dict.
kind: "address" or "transaction".
identifier: Address string or txid.
label: Label text (empty string to remove).
Returns:
Updated wallet dict.
"""
labels = wallet.get("labels", {"addresses": {}, "transactions": {}})
if kind == "address":
if label:
labels["addresses"][identifier] = label
elif identifier in labels["addresses"]:
del labels["addresses"][identifier]
elif kind == "transaction":
if label:
labels["transactions"][identifier] = label
elif identifier in labels["transactions"]:
del labels["transactions"][identifier]
wallet["labels"] = labels
return wallet
def get_label(self, wallet: dict, kind: str, identifier: str) -> str:
"""Get a label for an address or transaction.
Returns:
Label string or empty string if not set.
"""
labels = wallet.get("labels", {})
if kind == "address":
return labels.get("addresses", {}).get(identifier, "")
elif kind == "transaction":
return labels.get("transactions", {}).get(identifier, "")
return ""
# ------------------------------------------------------------------
# UTXO tree
# ------------------------------------------------------------------
def get_utxo_tree(self, utxos: list, tx_history: list) -> list:
"""Build a UTXO tree showing spending relationships.
Args:
utxos: List of UTXO dicts from get_utxos().
tx_history: List of tx dicts from get_tx_history().
Returns:
List of tree nodes with spending info.
"""
# Build txid -> tx mapping
tx_map = {}
for tx in tx_history:
txid = tx.get("txid") or tx.get("txid")
if txid:
tx_map[txid] = tx
# Map spent UTXOs
spent_txids = set()
spent_map = {} # txid:vout -> spending_txid
for tx in tx_history:
for inp in tx.get("vin", []):
prev_txid = inp.get("txid", "")
prev_vout = inp.get("vout")
if prev_txid and prev_vout is not None:
key = f"{prev_txid}:{prev_vout}"
spent_txids.add(key)
spent_map[key] = tx.get("txid", "")
# Build tree nodes
tree = []
for u in utxos:
key = f"{u['txid']}:{u['vout']}"
node = {
"txid": u["txid"],
"vout": u["vout"],
"amount": u["amount"],
"confirmations": u.get("confirmations", 0),
"spent": key in spent_txids,
"spent_by": spent_map.get(key),
"label": "",
}
tree.append(node)
return tree
# ------------------------------------------------------------------
# Multisig (m-of-n) — P2WSH with sortedmulti
# ------------------------------------------------------------------
def create_multisig_wallet(
self,
m: int,
n: int,
participants: list,
wallet_name: str,
password: str = "",
) -> dict:
"""Create a multisig wallet descriptor and P2WSH address.
Uses Bitcoin Core's createmultisig + wsh(sortedmulti(...)) descriptor
to produce a native SegWit multisig address (bc1p...).
Args:
m: Number of required signatures.
n: Total number of participants.
participants: List of dicts with keys:
- name (str)
- xpub (str) — BIP84 xpub (m/84'/0'/0')
- local (bool) — True if this keypair is stored locally
- mnemonic (str, optional) — mnemonic if local
wallet_name: Display name for the wallet.
password: Optional password to encrypt local mnemonics.
Returns:
Multisig wallet dict ready for persistence.
"""
import pathlib
if m > n or m < 1 or n < 1:
raise ValueError(f"Invalid m-of-n: m={m}, n={n}")
if len(participants) != n:
raise ValueError(
f"Expected {n} participants, got {len(participants)}"
)
# Derive xpubs for P2WSH multisig path m/48'/0'/0'/2' (BIP48)
# For participants with local keys, derive from mnemonic.
# For remote participants, use provided xpub.
xpubs = []
local_indices: list[int] = []
local_mnemonics: list[dict] = []
for i, p in enumerate(participants):
xpub = p.get("xpub", "").strip()
is_local = p.get("local", False)
mnemonic = p.get("mnemonic", "").strip()
if is_local and mnemonic:
# Password required for local keys — no plaintext storage
if not password:
raise ValueError(
"Password is required for wallets with local keys "
"(mnemonics cannot be stored unencrypted)"
)
# Derive BIP48 multisig path xpub from mnemonic
from bip_utils import Bip32Slip10Secp256k1, Bip32PathParser
seed = self.mnemonic_to_seed(mnemonic)
# BIP48: m/48'/0'/0'/2' (native SegWit multisig)
bip48_key = Bip32Slip10Secp256k1.FromSeedAndPath(
seed, Bip32PathParser.Parse("m/48'/0'/0'/2'")
)
xpub = bip48_key.PublicKey().ToExtended()
enc_mnemonic = self.encrypt_key(mnemonic, password)
local_mnemonics.append({
"participant_index": i,
"name": p["name"],
"encrypted_mnemonic": enc_mnemonic,
})
local_indices.append(i)
elif not xpub:
raise ValueError(
f"Participant {i} ({p['name']}): provide xpub or (local=True + mnemonic)"
)
xpubs.append(xpub)
# Build descriptor with derivation paths for the receive chain (index 0/*)
# Each xpub gets /0/* appended for the external (receive) chain
xpub_paths = [f"{xpub}/0/*" for xpub in xpubs]
xpubs_joined = ", ".join(xpub_paths)
descriptor = f"wsh(sortedmulti({m}, {xpubs_joined}))"
# getdescriptorinfo adds the checksum and returns witnessscript
# getdescriptorinfo adds the checksum and returns witnessscript
desc_info = self._rpc_call("getdescriptorinfo", [descriptor])
full_descriptor = desc_info.get("descriptor", descriptor)
# Derive witness script by getting raw pubkeys at m/0/0 from each xpub
# and calling createmultisig RPC which returns the redeemScript/witnessScript.
from bip_utils import Bip32Slip10Secp256k1
raw_pubkeys = []
for xpub in xpubs:
node = Bip32Slip10Secp256k1.FromExtendedKey(xpub)
child = node.DerivePath("0/0")
pk = bytes(child.PublicKey().RawCompressed())
raw_pubkeys.append(pk.hex())
witness_script = ""
try:
# Sort pubkeys to match sortedmulti in descriptor
sorted_pubkeys = sorted(raw_pubkeys)
cms_result = self._rpc_call("createmultisig", [m, sorted_pubkeys, "bech32"])
# For bech32 wtype, redeemScript is the witnessScript
witness_script = cms_result.get("redeemScript", "")
except (RuntimeError, ConnectionError):
pass # witness_script will be empty; PSBT can still work with descriptor
# Derive address at index 0 (first receive address)
addr_result = self._rpc_call(
"deriveaddresses",
[full_descriptor, 0],
)
address = addr_result[0] if addr_result else ""
now = datetime.now(timezone.utc).isoformat()
wallet = {
"type": "multisig",
"name": wallet_name,
"m": m,
"n": n,
"threshold": m,
"participants": [
{
"index": i,
"name": p["name"],
"xpub": xpubs[i],
"local": p.get("local", False),
}
for i, p in enumerate(participants)
],
"local_indices": local_indices,
"local_mnemonics": local_mnemonics if password else [],
"encrypted": bool(password),
"witness_script": witness_script,
"descriptor": full_descriptor,
"descriptor_hash": desc_info.get("hash160", ""),
"address": address,
"created": now,
}
# Save to disk
wallet_dir = pathlib.Path.home() / ".btc-wallet"
wallet_dir.mkdir(parents=True, exist_ok=True)
# Restrict wallet directory to owner-only (700)
os.chmod(str(wallet_dir), 0o700)
fname = f"multisig_{wallet_name.replace(' ', '_')}_{address[:10]}.json"
fpath = wallet_dir / fname
data = {
"version": 2,
"wallet": wallet,
"labels": {"addresses": {}, "transactions": {}},
"created": now,
"modified": now,
}
fpath.write_text(json.dumps(data, indent=2))
# Restrict wallet file to owner-only (600)
os.chmod(str(fpath), 0o600)
wallet["file_path"] = str(fpath)
return wallet
def get_multisig_balance(self, wallet: dict) -> dict:
"""Get balance of a multisig wallet address.
Args:
wallet: Multisig wallet dict from create_multisig_wallet().
Returns:
Dict with balance info in BTC and sats.
"""
address = wallet.get("address", "")
if not address:
return {"error": "No address in wallet"}
try:
# Get received amount
received = self._rpc_call(
"getreceivedbyaddress",
[address],
)
# Get unconfirmed
unconfirmed = self._rpc_call(
"getreceivedbyaddress",
[address, 0],
)
confirmed = received # getreceivedbyaddress defaults to 6 confs
return {
"address": address,
"balance_btc": float(confirmed),
"balance_sats": round(confirmed * 1e8),
"unconfirmed_btc": float(unconfirmed - confirmed) if unconfirmed > confirmed else 0,
}
except (RuntimeError, ConnectionError):
return {"address": address, "balance_btc": 0, "balance_sats": 0}
def get_multisig_utxos(self, wallet: dict) -> list:
"""Get UTXOs for a multisig wallet address.
Args:
wallet: Multisig wallet dict.
Returns:
List of UTXO dicts.
"""
address = wallet.get("address", "")
if not address:
return []
try:
utxos = self._rpc_call(
"listunspent",
[0, 9999999, [address]],
)
result = []
for u in utxos:
result.append({
"txid": u.get("txid"),
"vout": u.get("vout"),
"address": u.get("address"),
"amount": float(u.get("amount", 0)),
"confirmations": u.get("confirmations", 0),
"label": wallet.get("name", ""),
})
return result
except (RuntimeError, ConnectionError):
return []
def get_multisig_addresses(
self,
wallet: dict,
count: int = 5,
change: int = 0,
) -> list:
"""Derive additional addresses for a multisig wallet from its descriptor.
Uses Bitcoin Core's deriveaddresses RPC to generate addresses
at sequential indices on either the receive (change=0) or
internal (change=1) chain.
Args:
wallet: Multisig wallet dict with "descriptor" field.
count: Number of addresses to derive.
change: 0 for receive chain, 1 for change chain.
Returns:
List of {index, address, chain} dicts.
"""
descriptor = wallet.get("descriptor", "")
if not descriptor:
return []
try:
# The stored descriptor has /0/* (external chain) with a checksum.
# For change addresses, swap to /1/* and recompute checksum.
if change == 1:
# Strip checksum, replace chain, let Bitcoin Core add new checksum
desc_no_cksum = descriptor.rsplit("#", 1)[0]
desc_no_cksum = desc_no_cksum.replace("/0/*", "/1/*")
desc_info = self._rpc_call("getdescriptorinfo", [desc_no_cksum])
desc = desc_info.get("descriptor", desc_no_cksum)
else:
desc = descriptor
# deriveaddresses takes descriptor + range end (inclusive).
# To get count addresses starting from index 0, range = count - 1.
addresses = self._rpc_call("deriveaddresses", [desc, count - 1])
return [
{
"index": i,
"address": addresses[i] if isinstance(addresses, list) and i < len(addresses) else "",
"chain": "receive" if change == 0 else "change",
}
for i in range(count)
]
except (RuntimeError, ConnectionError):
return []
def create_multisig_psbt(
self,
wallet: dict,
outputs: list,
fee_rate: float,
utxos: list = None,
) -> dict:
"""Create a PSBT for a multisig transaction.
Uses Bitcoin Core's wallet to create a PSBT via bumptransaction or
sendtoaddress with PSBT output, then strips local signatures for
multi-signer workflow.
Args:
wallet: Multisig wallet dict.
outputs: List of {address, amount} dicts (amount in BTC).
fee_rate: Fee rate in sats/vB.
utxos: Optional list of UTXOs to spend (coin control).
Returns:
Dict with psbt hex string.
"""
address = wallet.get("address", "")
descriptor = wallet.get("descriptor", "")
witness_script = wallet.get("witness_script", "")
if not utxos:
utxos = self.get_multisig_utxos(wallet)
if not utxos:
return {"error": "No UTXOs available for spending"}
# Calculate total input
total_in = sum(u["amount"] for u in utxos)
total_out = sum(o["amount"] for o in outputs)
# Estimate fee (multisig inputs are larger)
# P2WSH input ~ 180 vBytes per input (witness stack)
input_vbytes = len(utxos) * 200 # conservative estimate
output_vbytes = len(outputs) * 43 + 10 # base tx
estimated_fee_btc = (input_vbytes + output_vbytes) * fee_rate / 1e8
change = total_in - total_out - estimated_fee_btc
if change < 0.00000546:
# Dust threshold for P2WSH ~ 0.00000546 BTC
change = 0
# Build outputs for Bitcoin Core
tx_outputs = {}
for o in outputs:
tx_outputs[o["address"]] = o["amount"]
if change > 0:
tx_outputs[address] = round(change, 8)
# Build inputs list
inputs = []
for u in utxos:
inputs.append({
"txid": u["txid"],
"vout": u["vout"],
"scriptPubKey": witness_script,
})
# Use Bitcoin Core to create PSBT via RPC
# We'll use createpsbt if available, otherwise fall back to raw tx construction
try:
psbt = self._rpc_call(
"createpsbt",
[
inputs,
tx_outputs,
0, # locktime
True, # replaceable (RBF)
],
)
return {
"psbt": psbt,
"inputs": len(inputs),
"outputs": len(tx_outputs),
"total_input_btc": round(total_in, 8),
"total_output_btc": round(sum(tx_outputs.values()), 8),
"fee_estimate_btc": round(estimated_fee_btc, 8),
}
except RuntimeError as e:
return {"error": f"Failed to create PSBT: {str(e)}"}
def sign_multisig_psbt(
self,
psbt_hex: str,
wallet: dict,
password: str = "",
) -> dict:
"""Sign a multisig PSBT with local keys.
Imports the multisig descriptor into a Bitcoin Core wallet with
private keys for local participants, then uses signrawtransactionwithwallet.
Before signing, validates:
- PSBT inputs belong to this multisig wallet
- Total output amount does not exceed total input amount
Args:
psbt_hex: PSBT hex string.
wallet: Multisig wallet dict with local mnemonics.
password: Password to decrypt local mnemonics.
Returns:
Dict with signed PSBT hex and signature count.
"""
from bitcoin.core.script import CScript
from coincurve import PrivateKey, PublicKey
from bip_utils import Bip32Slip10Secp256k1, Bip32PathParser
local_mnemonics = wallet.get("local_mnemonics", [])
m = wallet.get("m", 2)
n = wallet.get("n", 3)
participants = wallet.get("participants", [])
multisig_address = wallet.get("address", "")
if not local_mnemonics:
return {
"psbt": psbt_hex,
"local_signatures_added": 0,
"message": "No local keys configured for this wallet",
}
# ── Validate PSBT before signing ──────────────────────────────
validation = self._validate_psbt(psbt_hex, multisig_address)
if validation:
return validation
# Build a Bitcoin Core wallet name for this multisig
btc_wallet_name = "msig_" + wallet.get("name", "wallet").replace(" ", "_").lower()
# Ensure the descriptor wallet exists in Bitcoin Core with private keys
self._ensure_multisig_signing_wallet(btc_wallet_name, wallet, password)
# Sign with the Bitcoin Core wallet
try:
sign_result = self._rpc_call(
"signrawtransactionwithwallet",
[psbt_hex],
wallet_name=btc_wallet_name,
)
signed_psbt = sign_result.get("psbt", psbt_hex)
complete = sign_result.get("complete", False)
# Count signatures by comparing PSBTs
sigs_added = self._count_psbt_signatures(signed_psbt)
return {
"psbt": signed_psbt,
"local_signatures_added": sigs_added,
"complete": complete,
"required_sigs": m,
}
except RuntimeError as e:
return {
"error": f"Signing failed: {str(e)}",
"psbt": psbt_hex,
"local_signatures_added": 0,
}
def _ensure_bip84_signing_wallet(
self,
btc_wallet_name: str,
xpriv: str,
password: str = "",
) -> None:
"""Import BIP84 descriptor into Bitcoin Core for signing.
Creates a descriptor wallet with the xpriv for lazy key derivation
instead of pre-computing 200 keys in memory.
Args:
btc_wallet_name: Name for the temporary Bitcoin Core wallet.
xpriv: BIP84 account xpriv (m/84'/0'/0').
password: Optional password to encrypt the Core wallet.
"""
# Check if wallet already exists
try:
existing = self._rpc_call("listwallets")
if btc_wallet_name in existing:
return # Already imported
except RuntimeError:
pass
# Build taproot descriptor with xpriv and key origin
# [fingerprint/derivation]xpriv/0/* + [fingerprint/derivation]xpriv/1/*
# Bitcoin Core will lazily derive keys as needed
fingerprint = ""
try:
from bip_utils import Bip32Slip10Secp256k1
node = Bip32Slip10Secp256k1.FromExtendedKey(xpriv)
fingerprint = node.FingerPrint().ToHex()
except Exception:
pass
# Descriptor for BIP84 native SegWit with both chains
# Using two descriptors joined with | for receive and change chains
if fingerprint:
descriptor = f"wpkh([{fingerprint}/84'/0'/0']{xpriv}/0/*)|wpkh([{fingerprint}/84'/0'/0']{xpriv}/1/*)"
else:
descriptor = f"wpkh({xpriv}/0/*)|wpkh({xpriv}/1/*)"
try:
self._rpc_call("createwallet", [btc_wallet_name, False, False, password or None, False, True, 1])
except RuntimeError:
# Wallet might already exist
pass
# Load the descriptor into the wallet
try:
self._rpc_call("importdescriptor", [
btc_wallet_name,
[
{
"desc": descriptor,
"timestamp": "now",
"range": [0, 1000],
"active": True,
}
]
], wallet_name=btc_wallet_name,
)
except RuntimeError:
# May fail if already imported — ignore
pass
def _ensure_multisig_signing_wallet(
self,
btc_wallet_name: str,
wallet: dict,
password: str = "",
) -> None:
"""Import multisig descriptor into Bitcoin Core for signing.
Creates a descriptor wallet with private keys for local participants
and master key fingerprints for remote ones.
"""
# Check if wallet already exists
try:
existing = self._rpc_call("listwallets")
if btc_wallet_name in existing:
return # Already imported
except RuntimeError:
pass
# Create the wallet (blank, no descriptor yet)
try:
self._rpc_call("createwallet", [btc_wallet_name, False, False, password or None, False, True, 1])
except RuntimeError:
# Wallet might already exist
pass
# Decrypt local mnemonics and derive private keys
local_mnemonics = wallet.get("local_mnemonics", [])
participants = wallet.get("participants", [])
descriptor = wallet.get("descriptor", "")
# Build the signing descriptor with private key origins
# We need to reconstruct the descriptor with xprivs for local keys
# and xpubs with key origin for remote keys
# Parse the original descriptor to get xpub paths
# Format: wsh(sortedmulti(m, XPUB1/0/*, XPUB2/0/*, ...))#checksum
# Strip checksum for parsing
desc_no_checksum = descriptor.split("#")[0] if "#" in descriptor else descriptor
# Extract m and xpubs from the descriptor
inner = desc_no_checksum.replace("wsh(sortedmulti(", "").rstrip(")")
parts = inner.split(",")
m = int(parts[0].strip())
xpub_paths = [p.strip() for p in parts[1:]]
# Build new descriptor components with key origins
new_xpub_parts = []
for i, xp_part in enumerate(xpub_paths):
# xp_part = "xpub..." or "xpub.../0/*"
xpub = xp_part.split("/")[0]
participant = participants[i] if i < len(participants) else {}
is_local = participant.get("local", False)
if is_local:
# Find the matching local mnemonic
lm = None
for loc in local_mnemonics:
if loc.get("participant_index") == i:
lm = loc
break
if lm:
# Decrypt mnemonic
enc = lm.get("encrypted_mnemonic", {})
if isinstance(enc, dict) and enc.get("encrypted"):
mnemonic_str = self.decrypt_key(enc, password)
elif isinstance(enc, dict):
mnemonic_str = enc.get("key", "")
else:
mnemonic_str = str(enc)
# Derive BIP48 private key
seed = self.mnemonic_to_seed(mnemonic_str)
bip48_node = Bip32Slip10Secp256k1.FromSeedAndPath(
seed, Bip32PathParser.Parse("m/48'/0'/0'/2'")
)
xpriv = bip48_node.PrivateKey().ToExtended()
fingerprint = bip48_node.Bip32Object().FingerPrint().ToHex()
# Use key origin: [fingerprint/48h/0h/0h/2h]xpriv/0/*
# Keep the /0/* path suffix
path_suffix = xp_part[len(xpub):] # "/0/*"
new_xpub_parts.append(
f"[{fingerprint}/48'/0'/0'/2']{xpriv}{path_suffix}"
)
else:
new_xpub_parts.append(xp_part)
else:
# Remote participant — use xpub with origin if we have it
new_xpub_parts.append(xp_part)
# Build descriptor with private keys
signing_descriptor = f"wsh(sortedmulti({m},{','.join(new_xpub_parts)}))"
# Import the descriptor
try:
self._rpc_call(
"importdescriptors",
[
[
{
"desc": signing_descriptor,
"active": True,
"timestamp": 0,
"range": [0, 99],
"label": wallet.get("name", "multisig"),
}
]
],
wallet_name=btc_wallet_name,
)
except RuntimeError as e:
# If import fails (e.g., duplicate), try to continue
# The wallet might already have the keys
pass
def _validate_psbt(self, psbt_hex: str, multisig_address: str) -> Optional[dict]:
"""Validate a PSBT before signing.
Checks:
- All inputs belong to the multisig wallet address
- Total output amount does not exceed total input amount
Args:
psbt_hex: PSBT hex string to validate.
multisig_address: Expected multisig address.
Returns:
None if validation passes, or a dict with "error" key if it fails.
"""
try:
decoded = self._rpc_call("decodepsbt", [psbt_hex])
psbt_info = decoded.get("psbt", {})
tx = psbt_info.get("tx", {})
inputs = tx.get("vin", [])
outputs_dict = tx.get("vout", [])
except RuntimeError:
return {"error": "Failed to decode PSBT — invalid or unsupported format"}
# Check inputs belong to the multisig address
for i, inp in enumerate(inputs):
prev_txid = inp.get("txid", "")
prev_vout = inp.get("vout", -1)
# Check if the PSBT has previous txout info
input_info = decoded.get("inputs", [])
if i < len(input_info):
prev_txout = input_info[i].get("witness_utxo") or input_info[i].get("non_witness_utxo")
if prev_txout:
# Check the scriptPubKey matches our multisig address
# We need to verify the output pays to our multisig address
try:
addr_info = self._rpc_call("getaddressinfo", [multisig_address])
expected_script = addr_info.get("scriptPubKey", "")
actual_script = prev_txout.get("scriptPubKey", "")
if actual_script != expected_script:
return {
"error": f"PSBT input {i} does not belong to this multisig wallet. "
f"Expected script matching {multisig_address}"
}
except RuntimeError:
# If we can't verify the script, log but don't block
pass
# Check output amounts don't exceed input amounts
total_input_sats = 0
for i, inp in enumerate(inputs):
input_info_list = decoded.get("inputs", [])
if i < len(input_info_list):
prev_txout = input_info_list[i].get("witness_utxo") or input_info_list[i].get("non_witness_utxo")
if prev_txout and "value" in prev_txout:
total_input_sats += int(prev_txout["value"] * 100_000_000)
total_output_sats = 0
for output in outputs_dict:
total_output_sats += int(output.get("value", 0) * 100_000_000)
if total_output_sats > total_input_sats and total_input_sats > 0:
return {
"error": f"Total output ({total_output_sats} sats) exceeds total input ({total_input_sats} sats). "
f"Invalid transaction."
}
return None
def _count_psbt_signatures(self, psbt_hex: str) -> int:
"""Count total partial signatures in a PSBT using Bitcoin Core decodepsbt."""
try:
decoded = self._rpc_call("decodepsbt", [psbt_hex])
total_sigs = 0
for inp in decoded.get("inputs", []):
partial_sig = inp.get("partial_signatures", {})
total_sigs += len(partial_sig)
return total_sigs
except (RuntimeError, ConnectionError):
return 0
def combine_psbt(self, psbt_hex_list: list) -> dict:
"""Combine multiple partially-signed PSBTs into one.
Args:
psbt_hex_list: List of PSBT hex strings from different signers.
Returns:
Dict with combined PSBT hex.
"""
if len(psbt_hex_list) < 2:
return {
"error": "Need at least 2 PSBTs to combine",
"psbt": psbt_hex_list[0] if psbt_hex_list else "",
}
try:
combined = self._rpc_call(
"combinepsbt",
psbt_hex_list,
)
return {
"psbt": combined,
"num_inputs": len(psbt_hex_list),
}
except RuntimeError as e:
return {"error": f"Failed to combine PSBTs: {str(e)}"}
def finalize_multisig_psbt(self, psbt_hex: str) -> dict:
"""Finalize a multisig PSBT with enough signatures.
Args:
psbt_hex: PSBT hex with sufficient signatures.
Returns:
Dict with raw_tx_hex and completion status.
"""
return self.finalize_psbt(psbt_hex)
def _rpc_call(
self,
method: str,
params: list = None,
wallet_name: str = None,
timeout: int = 30,
) -> dict:
"""Make JSON-RPC call to Bitcoin Core.
Args:
method: RPC method name.
params: Method parameters.
wallet_name: Wallet name for /wallet/{name} scoped calls.
If None and method is in _WALLET_RPC_METHODS,
defaults to "watchonly".
timeout: Request timeout in seconds.
"""
import json as rpc_json
import urllib.request
import urllib.error as urllib_err
# Determine URL path
if wallet_name:
url = f"http://{self.rpc_host}:{self.rpc_port}/wallet/{wallet_name}"
elif method in self._WALLET_RPC_METHODS:
url = f"http://{self.rpc_host}:{self.rpc_port}/wallet/watchonly"
else:
url = f"http://{self.rpc_host}:{self.rpc_port}"
payload = {
"jsonrpc": "1.0",
"id": "wallet-engine",
"method": method,
"params": params or [],
}
auth = f"{self.rpc_user}:{self.rpc_password}".encode()
auth_header = __import__("base64").b64encode(auth).decode()
req = urllib.request.Request(
url,
data=rpc_json.dumps(payload).encode(),
headers={
"Content-Type": "application/json",
"Authorization": f"Basic {auth_header}",
},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
result = rpc_json.loads(resp.read())
if "error" in result and result["error"] is not None:
raise RuntimeError(result["error"]["message"])
return result["result"]
except urllib_err.HTTPError as e:
# Bitcoin Core may return HTTP errors with JSON bodies
body = e.read()
try:
result = rpc_json.loads(body)
if "error" in result and result["error"] is not None:
raise RuntimeError(result["error"]["message"])
if "result" in result:
return result["result"]
except (rpc_json.JSONDecodeError, KeyError):
pass
raise ConnectionError(f"RPC connection failed: HTTP {e.code} for {method}") from e
except urllib_err.URLError as e:
raise ConnectionError(f"RPC connection failed: {e}") from e
# ------------------------------------------------------------------
# Coin selection helpers
# ------------------------------------------------------------------
import itertools
def _bnb_coin_select(utxos: list, target_btc: float, input_fee_btc: float) -> list:
"""Branch-and-bound knapsack coin selection.
Finds the smallest subset of UTXOs that covers target + fees.
Falls back to largest-first if no solution found within time limit.
Args:
utxos: List of UTXO dicts with 'amount' key (BTC).
target_btc: Target output amount (excludes fees).
input_fee_btc: Fee cost per input in BTC.
Returns:
List of selected UTXOs.
"""
import random
import time
timeout = 0.5 # seconds max
# Shuffle to avoid worst-case ordering
shuffled = list(utxos)
random.shuffle(shuffled)
n = len(shuffled)
best = None
best_cost = float('inf')
start = time.monotonic()
max_nodes = 50_000 # prevent exponential blowup on pathological inputs
nodes_visited = 0
def _bnb(index: int, current: list, remaining: float, available: list):
nonlocal best, best_cost, nodes_visited
# Node count limit — prevents exponential blowup
nodes_visited += 1
if nodes_visited > max_nodes:
return
# Timeout check
if time.monotonic() - start > timeout:
return
# Pruning: if remaining + available sum is insufficient, prune
if remaining > sum(a['amount'] - input_fee_btc for a in available) + 1e-10:
return
# Bound: if current cost >= best cost, prune
cost = len(current) # minimize input count
if cost >= best_cost:
return
# Base case: no more items
if index >= len(available):
# Check if we've met target
total = sum(u['amount'] for u in current) - len(current) * input_fee_btc
if total >= target_btc:
best = list(current)
best_cost = cost
return
u = available[index]
net = u['amount'] - input_fee_btc
# Include this UTXO
current.append(u)
_bnb(index + 1, current, remaining - net, available[index + 1:])
current.pop()
# Exclude this UTXO — only if remaining can still be met
_bnb(index + 1, current, remaining, available[index + 1:])
_bnb(0, [], target_btc, shuffled)
if best is not None:
return best
# Fallback: largest-first
sorted_utxos = sorted(utxos, key=lambda u: u['amount'], reverse=True)
selected = []
running = 0.0
for u in sorted_utxos:
selected.append(u)
running += u['amount'] - input_fee_btc
if running >= target_btc:
break
return selected
# ------------------------------------------------------------------
# Wallet file format version
# ------------------------------------------------------------------
WALLET_FILE_VERSION = 2
# Migration registry: maps source version → migration function
MIGRATION_REGISTRY: dict[int, "Callable[[dict], dict]"] = {}
def register_migration(version: int):
"""Decorator to register a wallet migration function.
Migrates wallet data from `version` to `version + 1`.
"""
def decorator(func):
MIGRATION_REGISTRY[version] = func
return func
return decorator
def migrate_wallet(data: dict) -> dict:
"""Migrate wallet data to the current schema version.
Applies migrations sequentially from current version to WALLET_FILE_VERSION.
"""
current_version = data.get("version", 1)
if current_version > WALLET_FILE_VERSION:
raise ValueError(
f"Wallet file version {current_version} is newer than supported "
f"(max {WALLET_FILE_VERSION}). Update btc-wallet."
)
if current_version == WALLET_FILE_VERSION:
return data
while current_version < WALLET_FILE_VERSION:
migration_fn = MIGRATION_REGISTRY.get(current_version)
if migration_fn is None:
raise ValueError(
f"No migration from version {current_version} to {current_version + 1}. "
"Wallet data may be incompatible."
)
data = migration_fn(data)
data["version"] = current_version + 1
current_version += 1
return data
# Example: Migration from v1 → v2
# (v2 added encrypted_seed instead of plaintext seed, labels dict, etc.)
@register_migration(1)
def migrate_v1_to_v2(data: dict) -> dict:
"""Migrate from v1 (plaintext seed) to v2 (encrypted seed).
Note: Requires the old password to re-encrypt. This migration path
is informational — v1 wallets cannot be automatically migrated because
the migration function cannot re-encrypt without the user's password.
Users are prompted to create a new wallet and transfer funds.
"""
wallet = data.get("wallet", {})
if wallet.get("seed") and not wallet.get("encrypted_seed"):
wallet.setdefault("migrated_from", "v1")
wallet.setdefault("migration_note", "v1 wallets require manual re-creation for security. See docs/migration.md")
return data
# ------------------------------------------------------------------
# Encoding helpers
# ------------------------------------------------------------------
def _base58_encode(data: bytes) -> str:
"""Base58 encoding for WIF addresses."""
alphabet = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
# Convert data to big integer
num = int.from_bytes(data, "big")
encoded = ""
while num > 0:
num, remainder = divmod(num, 58)
encoded = alphabet[remainder] + encoded
# Add leading '1's for each leading zero byte
for byte in data:
if byte == 0:
encoded = "1" + encoded
else:
break
return encoded
if __name__ == "__main__":
eng = WalletEngine()
# Test: generate mnemonic
print("=" * 60)
print("Wallet Engine Test")
print("=" * 60)
mnemonic = eng.generate_mnemonic()
print(f"\nGenerated mnemonic:\n {mnemonic}")
# Derive BIP84 wallet
wallet = eng.derive_bip84(mnemonic)
print(f"\nBIP84 Wallet:")
print(f" Path: {wallet['derivation_path']}")
print(f" Fingerprint: {wallet['fingerprint']}")
print(f" XPUB: {wallet['xpub']}")
# Derive first address
addr = eng.derive_bip84_address(mnemonic, address_index=0)
print(f"\nFirst BIP84 Address:")
print(f" Path: {addr['path']}")
print(f" Address: {addr['address']}")
# Test encryption (unencrypted mode)
encrypted = eng.encrypt_key(mnemonic)
decrypted = eng.decrypt_key(encrypted, encrypted["key"])
print(f"\nEncryption test (no password): {'PASS' if decrypted == mnemonic else 'FAIL'}")
# Test encryption (password mode)
enc_pass = eng.encrypt_key(mnemonic, password="test123")
dec_pass = eng.decrypt_key(enc_pass, "test123")
print(f"Encryption test (password): {'PASS' if dec_pass == mnemonic else 'FAIL'}")
# Test wallet creation
wallet = eng.create_wallet("Test Wallet", mnemonic)
print(f"\nWallet created: {wallet['name']}")
print(f" Fingerprint: {wallet['fingerprint']}")
print(f" Encrypted: {wallet['encrypted']}")
# Test address derivation
addresses = eng.get_addresses(wallet, count=3, derivation_type="bech32")
print(f"\nAddresses derived: {len(addresses)}")
for addr in addresses:
print(f" [{addr['index']}] {addr['address']} ({addr['type']})")
# Test RPC
try:
height = eng.get_block_count()
print(f"\nRPC test: block height = {height}")
except Exception as e:
print(f"\nRPC test: SKIP ({e})")