#!/usr/bin/env python3
"""
REST API bridge between Tauri frontend and wallet engine.
Exposes wallet operations over HTTP for the React frontend.
Security:
- API key authentication required via X-API-Key header
- Private keys stripped from all responses
- Wallet creation requires password (no plaintext seed)
- RPC credentials from environment variables only
"""
import hmac
import json
import logging
import os
import re
import sys
import time
from collections import defaultdict
from http.server import HTTPServer, BaseHTTPRequestHandler
# Add wallet engine to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from wallet import WalletEngine
logger = logging.getLogger("btc-wallet-api")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
stream=sys.stderr,
)
# ── Input validation ───────────────────────────────────────────────────
# Bitcoin address regex: bech32 (bc1...), bech32m (bc1p...), legacy (1...), P2SH (3...)
BTC_ADDRESS_RE = re.compile(r"^(bc1|[13])[a-zA-HJ-NP-Z0-9]{25,}$")
# BTC amount bounds (in BTC)
MAX_BTC = 21_000_000.0
# Fee rate bounds (sats/vB)
MIN_FEE_RATE = 1.0
MAX_FEE_RATE = 10_000.0
def validate_btc_address(address: str) -> bool:
"""Validate Bitcoin address format (bech32/bech32m/legacy/P2SH)."""
return bool(BTC_ADDRESS_RE.match(address))
def validate_amount(amount) -> tuple:
"""Validate BTC amount. Returns (is_valid, error_message)."""
try:
val = float(amount)
except (TypeError, ValueError):
return False, "Invalid amount: must be a number"
if val <= 0:
return False, f"Amount must be positive, got {amount}"
if val > MAX_BTC:
return False, f"Amount exceeds max ({MAX_BTC} BTC)"
return True, None
def validate_fee_rate(fee_rate) -> tuple:
"""Validate fee rate (sats/vB). Returns (is_valid, error_message)."""
try:
val = float(fee_rate)
except (TypeError, ValueError):
return False, "Invalid fee rate: must be a number"
if val < MIN_FEE_RATE:
return False, f"Fee rate too low (min {MIN_FEE_RATE} sats/vB)"
if val > MAX_FEE_RATE:
return False, f"Fee rate too high (max {MAX_FEE_RATE} sats/vB)"
return True, None
# Bitcoin Core RPC config — NO fallback defaults with real passwords
RPC_HOST = os.environ.get("BTC_RPC_HOST", "127.0.0.1")
RPC_PORT = int(os.environ.get("BTC_RPC_PORT", "8332"))
RPC_USER = os.environ.get("BTC_RPC_USER", "")
RPC_PASS = os.environ.get("BTC_RPC_PASS", "")
if not RPC_USER or not RPC_PASS:
raise RuntimeError(
"BTC_RPC_USER and BTC_RPC_PASS environment variables are required. "
"Do not store credentials in source code."
)
# API key authentication
API_KEY = os.environ.get("WALLET_API_KEY", "")
if not API_KEY:
raise RuntimeError(
"WALLET_API_KEY environment variable is required. "
"Generate one with: python3 -c 'import secrets; print(secrets.token_hex(32))'"
)
ENGINE = WalletEngine(
rpc_user=RPC_USER,
rpc_password=RPC_PASS,
rpc_host=RPC_HOST,
rpc_port=RPC_PORT,
)
# ── Rate limiting ──────────────────────────────────────────────────────
class RateLimiter:
"""Simple sliding-window rate limiter per client IP.
Default: 60 requests per 60 seconds per IP.
"""
def __init__(self, max_requests: int = 60, window_seconds: int = 60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self._requests: dict[str, list[float]] = defaultdict(list)
def is_allowed(self, client_ip: str) -> bool:
"""Check if a request from client_ip is allowed. Returns False if rate limited."""
now = time.time()
cutoff = now - self.window_seconds
# Prune old entries
self._requests[client_ip] = [
t for t in self._requests[client_ip] if t > cutoff
]
if len(self._requests[client_ip]) >= self.max_requests:
return False
self._requests[client_ip].append(now)
return True
RATE_LIMITER = RateLimiter(max_requests=60, window_seconds=60)
class WalletHandler(BaseHTTPRequestHandler):
def _check_api_key(self) -> bool:
"""Verify API key from X-API-Key header. Returns True if valid."""
api_key = self.headers.get("X-API-Key", "")
if not api_key or not hmac.compare_digest(api_key, API_KEY):
self._send_json({"error": "Unauthorized: invalid or missing API key"}, 401)
return False
return True
def _send_json(self, data: dict, status: int = 200):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type, X-API-Key")
self.end_headers()
self.wfile.write(json.dumps(data).encode())
def do_OPTIONS(self):
self.send_response(200)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type, X-API-Key")
self.end_headers()
def _rpc_unavailable_error(self) -> dict:
"""Return a user-friendly error for RPC unavailability."""
return {
"error": "Bitcoin Core RPC is currently unavailable. "
"Check that bitcoind is running and BTC_RPC_* env vars are correct.",
"code": "RPC_UNAVAILABLE",
}
def _is_rpc_error(self, exc: Exception) -> bool:
"""Check if an exception is due to RPC connection failure."""
error_str = str(exc).lower()
return any(kw in error_str for kw in [
"connection refused", "connection reset", "timeout",
"network is unreachable", "nodename nor servname",
"rpc_unavailable", "cannot connect",
])
def do_POST(self):
# Authenticate every request
if not self._check_api_key():
return
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length)
try:
data = json.loads(body) if body else {}
except json.JSONDecodeError:
self._send_json({"error": "Invalid JSON"}, 400)
return
if self.path == "/api/generate_seed":
strength = data.get("strength", 128)
try:
seed = ENGINE.generate_mnemonic(strength=strength)
self._send_json({"seed": seed})
except Exception as e:
logger.exception("generate_seed")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/health":
try:
block_count = ENGINE.get_block_count()
self._send_json({
"status": "connected",
"block_count": block_count,
})
except Exception as e:
if self._is_rpc_error(e):
self._send_json({
"status": "disconnected",
"error": "Bitcoin Core RPC unavailable",
}, 503)
else:
logger.exception("health check")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/create_wallet":
seed = data.get("seed")
password = data.get("password")
wallet_name = data.get("wallet_name", "Wallet 1")
passphrase = data.get("passphrase", "")
if not seed:
self._send_json({"error": "Seed phrase required"}, 400)
return
if not password:
self._send_json({"error": "Password is required — wallets cannot be stored unencrypted"}, 400)
return
try:
wallet = ENGINE.create_wallet(wallet_name, seed, password, passphrase)
# Derive initial addresses
addresses = ENGINE.get_addresses(wallet, count=5, derivation_type="bech32", password=password)
addresses = ENGINE.sanitize_addresses(addresses)
# Return sanitized wallet (no seed) + addresses
safe_wallet = ENGINE.sanitize_wallet(wallet)
self._send_json({"wallet": safe_wallet, "addresses": addresses})
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("create_wallet")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/derive_addresses":
wallet = data.get("wallet")
password = data.get("password", "")
count = data.get("count", data.get("num_addresses", 5))
derivation_type = data.get("derivation_type", "bech32")
if not wallet:
self._send_json({"error": "Wallet data required"}, 400)
return
try:
addresses = ENGINE.get_addresses(
wallet, count=count, derivation_type=derivation_type, password=password
)
# Already sanitized by get_addresses(), but double-check
addresses = ENGINE.sanitize_addresses(addresses)
self._send_json({"addresses": addresses})
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("derive_addresses")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_balance":
# Accept both singular "address" and plural "addresses" for compatibility
addrs = data.get("addresses") or [data.get("address")]
if not addrs or (isinstance(addrs, list) and len(addrs) == 0):
self._send_json({"error": "Address required"}, 400)
return
if isinstance(addrs, str):
addrs = [addrs]
for addr in addrs:
if not validate_btc_address(addr):
self._send_json({"error": f"Invalid Bitcoin address format: {addr}"}, 400)
return
try:
# Sum balances across all addresses
total_sats = 0
for addr in addrs:
b = ENGINE.get_balance(addr)
total_sats += b.get("balance_sats", 0)
self._send_json({
"balance_btc": total_sats / 100_000_000,
"balance_sats": total_sats,
})
except Exception as e:
if self._is_rpc_error(e):
logger.warning("get_balance: RPC unavailable")
self._send_json(self._rpc_unavailable_error(), 503)
else:
logger.exception("get_balance failed")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_block_count":
try:
count = ENGINE.get_block_count()
self._send_json({"block_count": count})
except Exception as e:
if self._is_rpc_error(e):
logger.warning("get_block_count: RPC unavailable")
self._send_json(self._rpc_unavailable_error(), 503)
else:
logger.exception("get_block_count")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_network_info":
try:
info = ENGINE.get_network_info()
self._send_json(info)
except Exception as e:
if self._is_rpc_error(e):
logger.warning("get_network_info: RPC unavailable")
self._send_json(self._rpc_unavailable_error(), 503)
else:
logger.exception("get_network_info")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/validate_seed":
seed = data.get("seed", "")
valid = ENGINE.validate_mnemonic(seed)
self._send_json({"valid": valid})
elif self.path == "/api/import_wallet":
xpub = data.get("xpub")
wallet_name = data.get("wallet_name", "watchonly")
if not xpub:
self._send_json({"error": "xpub required"}, 400)
return
try:
result = ENGINE.import_wallet_watchonly(xpub, wallet_name)
self._send_json(result)
except Exception as e:
logger.exception("import_wallet")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_utxos":
addresses = data.get("addresses", [])
if not addresses:
self._send_json({"error": "addresses required"}, 400)
return
try:
utxos = ENGINE.get_utxos(addresses)
self._send_json({"utxos": utxos})
except Exception as e:
logger.exception("get_utxos")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/create_psbt":
wallet = data.get("wallet")
outputs = data.get("outputs", [])
fee_rate = data.get("fee_rate", 10.0)
password = data.get("password")
addresses = data.get("addresses", None)
coin_selection = data.get("coin_selection", "largest")
selected_utxos = data.get("selected_utxos", None)
if not wallet or not outputs:
self._send_json({"error": "wallet and outputs required"}, 400)
return
if not password:
self._send_json({"error": "Password required for encrypted wallet"}, 400)
return
# Validate fee rate
valid, err = validate_fee_rate(fee_rate)
if not valid:
self._send_json({"error": err}, 400)
return
# Validate each output
for i, o in enumerate(outputs):
addr = o.get("address", "")
if not validate_btc_address(addr):
self._send_json({"error": f"Output {i}: invalid Bitcoin address format"}, 400)
return
ok, err = validate_amount(o.get("amount"))
if not ok:
self._send_json({"error": f"Output {i}: {err}"}, 400)
return
try:
psbt = ENGINE.create_psbt(
wallet, outputs, fee_rate, addresses, password,
coin_selection, selected_utxos
)
self._send_json(psbt)
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("create_psbt failed")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/sign_psbt":
wallet = data.get("wallet")
psbt_hex = data.get("psbt")
password = data.get("password")
if not wallet or not psbt_hex:
self._send_json({"error": "wallet and psbt required"}, 400)
return
if not password:
self._send_json({"error": "Password required for signing"}, 400)
return
try:
result = ENGINE.sign_psbt(wallet, psbt_hex, password)
self._send_json(result)
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("sign_psbt")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/finalize_psbt":
psbt_hex = data.get("psbt")
if not psbt_hex:
self._send_json({"error": "psbt required"}, 400)
return
try:
result = ENGINE.finalize_psbt(psbt_hex)
self._send_json(result)
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("finalize_psbt")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/sign_and_finalize_psbt":
wallet = data.get("wallet")
psbt_hex = data.get("psbt")
password = data.get("password")
if not wallet or not psbt_hex:
self._send_json({"error": "wallet and psbt required"}, 400)
return
if not password:
self._send_json({"error": "Password required for signing"}, 400)
return
try:
result = ENGINE.sign_and_finalize_psbt(wallet, psbt_hex, password)
self._send_json(result)
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("sign_and_finalize_psbt")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/broadcast":
raw_tx_hex = data.get("raw_tx_hex")
if not raw_tx_hex:
self._send_json({"error": "raw_tx_hex required"}, 400)
return
try:
result = ENGINE.broadcast_tx(raw_tx_hex)
self._send_json(result)
except Exception as e:
logger.exception("sign_and_finalize_psbt")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/decode_psbt":
psbt_hex = data.get("psbt")
if not psbt_hex:
self._send_json({"error": "psbt required"}, 400)
return
try:
result = ENGINE.decode_psbt(psbt_hex)
self._send_json(result)
except Exception as e:
logger.exception("broadcast")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_tx_history":
address = data.get("address")
count = data.get("count", 20)
if not address:
self._send_json({"error": "address required"}, 400)
return
if not validate_btc_address(address):
self._send_json({"error": "Invalid Bitcoin address format"}, 400)
return
try:
txs = ENGINE.get_tx_history(address, count=count)
self._send_json(txs)
except Exception as e:
logger.exception("get_tx_history failed")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/estimate_fee":
blocks = data.get("blocks", 6)
try:
result = ENGINE.estimate_fee(blocks)
self._send_json(result)
except Exception as e:
logger.exception("get_tx_history")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/save_wallet":
wallet = data.get("wallet")
wallet_dir = data.get("wallet_dir")
if not wallet:
self._send_json({"error": "wallet required"}, 400)
return
try:
fpath = ENGINE.save_wallet(wallet, wallet_dir)
self._send_json({"path": fpath})
except Exception as e:
logger.exception("estimate_fee")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/load_wallet":
fpath = data.get("fpath")
password = data.get("password", "")
if not fpath:
self._send_json({"error": "fpath required"}, 400)
return
try:
wallet = ENGINE.load_wallet(fpath, password)
safe = ENGINE.sanitize_wallet(wallet)
# Derive addresses if wallet has a seed
addresses = []
if safe.get("seed") or (wallet.get("seed") and not safe.get("seed")):
addresses = ENGINE.get_addresses(wallet, count=5, derivation_type="bech32", password=password)
addresses = ENGINE.sanitize_addresses(addresses)
self._send_json({"wallet": safe, "addresses": addresses})
except FileNotFoundError as e:
self._send_json({"error": str(e)}, 404)
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("load_wallet")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/list_wallets":
try:
wallets = ENGINE.list_wallets()
self._send_json({"wallets": wallets})
except Exception as e:
logger.exception("load_wallet")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/set_label":
wallet = data.get("wallet")
kind = data.get("kind")
identifier = data.get("identifier")
label = data.get("label", "")
if not wallet or not kind or not identifier:
self._send_json({"error": "wallet, kind, identifier required"}, 400)
return
try:
updated = ENGINE.set_label(wallet, kind, identifier, label)
self._send_json({"labels": updated.get("labels", {})})
except Exception as e:
logger.exception("list_wallets")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_utxo_tree":
utxos = data.get("utxos", [])
tx_history = data.get("tx_history", [])
try:
tree = ENGINE.get_utxo_tree(utxos, tx_history)
self._send_json({"tree": tree})
except Exception as e:
logger.exception("set_label")
self._send_json({"error": "Internal server error"}, 500)
# ── Multisig endpoints ─────────────────────────────
elif self.path == "/api/create_multisig_wallet":
m = data.get("m")
n = data.get("n")
participants = data.get("participants", [])
wallet_name = data.get("wallet_name", "Multisig")
password = data.get("password", "")
if not m or not n or not participants:
self._send_json({"error": "m, n, and participants required"}, 400)
return
try:
wallet = ENGINE.create_multisig_wallet(
m, n, participants, wallet_name, password
)
balance = ENGINE.get_multisig_balance(wallet)
# Sanitize wallet response
safe_wallet = ENGINE.sanitize_wallet(wallet)
self._send_json({"wallet": safe_wallet, "balance": balance})
except ValueError as e:
self._send_json({"error": str(e)}, 400)
except Exception as e:
logger.exception("create_multisig_wallet")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_multisig_balance":
wallet = data.get("wallet")
if not wallet:
self._send_json({"error": "wallet required"}, 400)
return
try:
balance = ENGINE.get_multisig_balance(wallet)
self._send_json(balance)
except Exception as e:
logger.exception("create_multisig_wallet")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_multisig_utxos":
wallet = data.get("wallet")
if not wallet:
self._send_json({"error": "wallet required"}, 400)
return
try:
utxos = ENGINE.get_multisig_utxos(wallet)
self._send_json({"utxos": utxos})
except Exception as e:
logger.exception("create_multisig_wallet")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/get_multisig_addresses":
wallet = data.get("wallet")
count = data.get("count", 5)
change = data.get("change", 0)
if not wallet:
self._send_json({"error": "wallet required"}, 400)
return
try:
addresses = ENGINE.get_multisig_addresses(wallet, count, change)
self._send_json({"addresses": addresses})
except Exception as e:
logger.exception("get_multisig_utxos")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/create_multisig_psbt":
wallet = data.get("wallet")
outputs = data.get("outputs", [])
fee_rate = data.get("fee_rate", 10.0)
utxos = data.get("utxos", None)
if not wallet or not outputs:
self._send_json({"error": "wallet and outputs required"}, 400)
return
try:
psbt = ENGINE.create_multisig_psbt(wallet, outputs, fee_rate, utxos)
self._send_json(psbt)
except Exception as e:
logger.exception("get_multisig_addresses")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/sign_multisig_psbt":
psbt_hex = data.get("psbt")
wallet = data.get("wallet")
password = data.get("password")
if not psbt_hex or not wallet:
self._send_json({"error": "psbt and wallet required"}, 400)
return
if not password:
self._send_json({"error": "Password required for signing"}, 400)
return
try:
result = ENGINE.sign_multisig_psbt(psbt_hex, wallet, password)
self._send_json(result)
except Exception as e:
logger.exception("create_multisig_psbt")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/combine_psbt":
psbt_hex_list = data.get("psbt_hex_list", [])
if len(psbt_hex_list) < 2:
self._send_json({"error": "Need at least 2 PSBTs to combine"}, 400)
return
try:
result = ENGINE.combine_psbt(psbt_hex_list)
self._send_json(result)
except Exception as e:
logger.exception("sign_multisig_psbt")
self._send_json({"error": "Internal server error"}, 500)
elif self.path == "/api/finalize_multisig_psbt":
psbt_hex = data.get("psbt")
if not psbt_hex:
self._send_json({"error": "psbt required"}, 400)
return
try:
result = ENGINE.finalize_multisig_psbt(psbt_hex)
self._send_json(result)
except Exception as e:
logger.exception("sign_multisig_psbt")
self._send_json({"error": "Internal server error"}, 500)
else:
self._send_json({"error": "Unknown endpoint"}, 404)
def log_message(self, format, *args):
pass # Suppress console spam
def main():
import ssl as ssl_module
port = int(os.environ.get("WALLET_API_PORT", "14195"))
cert_file = os.environ.get("WALLET_API_CERT")
key_file = os.environ.get("WALLET_API_KEY")
server = HTTPServer(("0.0.0.0", port), WalletHandler)
if cert_file and key_file:
# C9 fix — TLS via environment-configured certs
context = ssl_module.SSLContext(ssl_module.PROTOCOL_TLS_SERVER)
context.load_cert_chain(cert_file, key_file)
context.minimum_version = ssl_module.TLSVersion.TLSv1_2
server.socket = context.wrap_socket(server.socket, server_side=True)
print(f"Wallet API server running on https://127.0.0.1:{port} (TLS)", flush=True)
else:
print(f"Wallet API server running on http://127.0.0.1:{port}", flush=True)
print("WARNING: No TLS configured. Set WALLET_API_CERT and WALLET_API_KEY env vars.", flush=True)
print("Endpoints:")
print(" POST /api/generate_seed - Generate BIP39 mnemonic")
print(" POST /api/create_wallet - Create encrypted wallet")
print(" POST /api/derive_addresses - Derive addresses from wallet")
print(" POST /api/get_balance - Get balance for address")
print(" POST /api/get_block_count - Get current block height")
print(" POST /api/get_network_info - Get Bitcoin Core network info")
print(" POST /api/validate_seed - Validate mnemonic phrase", flush=True)
server.serve_forever()
if __name__ == "__main__":
main()