#!/usr/bin/env python3 """ coldcard_sweep_watch.py — watch YOUR OWN bitcoin addresses for a live sweep. WHY THIS EXISTS During the Coldcard entropy incident the thefts arrive as programmatic sweeps. Galaxy's Alex Thorn (@intangiblecoins) noted that similar transactions sit in the mempool unconfirmed for a period before they are mined. If your coins are in one of those unconfirmed transactions, you may have a short window to fee-bump a competing transaction and move them somewhere safe first. This script does one thing: it tells you, fast and loudly, that a spend from one of your addresses is sitting unconfirmed in the mempool. WHAT IT DELIBERATELY DOES NOT DO - It never asks for, reads, derives, or transmits a seed phrase, private key, or xpub. It takes ADDRESSES, which are already public. - It cannot move your coins. It has no signing code. Recovery is yours to do, in your own wallet, deliberately. - It does not enumerate, guess, or weaken any key. It only reads the public mempool for addresses YOU supply, all of which you already own. If some tool asks for your seed words to "check if you are affected", that is the theft. There is no legitimate check that needs them. USAGE python3 coldcard_sweep_watch.py addresses.txt # one address per line python3 coldcard_sweep_watch.py bc1q... bc1q... 3ABC... python3 coldcard_sweep_watch.py --once addresses.txt # single pass, no loop python3 coldcard_sweep_watch.py --selftest # offline, no network Stdlib only. Python 3.7+. No pip install. Reads public APIs, sends nothing else. DATA SOURCE mempool.space public REST API (override with --api for your own instance; running your own node is strictly better if you have one). """ import sys import json import time import argparse import urllib.request import urllib.error DEFAULT_API = "https://mempool.space/api" POLL_SECONDS = 20 UA = "coldcard-sweep-watch/1.0 (defensive; reads public mempool only)" # ---------------------------------------------------------------- http def _get(url, timeout=20): req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) # ---------------------------------------------------------------- core logic # Pure functions below, so --selftest can exercise them with no network. def outgoing_from(tx, address): """ Total sats this tx spends FROM `address` (i.e. address appears as an input). Returns 0 if the address is not an input. This is the theft direction — an address merely RECEIVING coins is not being swept. """ total = 0 for vin in tx.get("vin", []): prev = vin.get("prevout") or {} if prev.get("scriptpubkey_address") == address: total += int(prev.get("value") or 0) return total def incoming_to(tx, address): total = 0 for vout in tx.get("vout", []): if vout.get("scriptpubkey_address") == address: total += int(vout.get("value") or 0) return total def is_rbf_signalled(tx): """ BIP-125 opt-in: any input with nSequence < 0xfffffffe signals replaceability. Note this describes THEIR transaction, not your ability to respond — see describe_window(). Absent the field we return None rather than guessing. """ vins = tx.get("vin", []) if not vins: return None seqs = [v.get("sequence") for v in vins] if any(s is None for s in seqs): return None return any(s < 0xFFFFFFFE for s in seqs) def describe_window(tx): """ Plain-English read on what, if anything, can still be done. Deliberately conservative: this is not advice that your coins are recoverable. """ if tx.get("status", {}).get("confirmed"): return ("CONFIRMED", "This spend is already mined. It cannot be replaced. " "Nothing in this script can undo it.") rbf = is_rbf_signalled(tx) if rbf: return ("UNCONFIRMED_RBF", "Unconfirmed and the spender signalled RBF. A higher-fee " "conflicting transaction from you MAY replace it. Act now, in " "your own wallet.") return ("UNCONFIRMED", "Unconfirmed but no opt-in RBF signal. Replacement is not guaranteed " "and depends on miner policy; a higher-fee conflicting spend is still " "your only lever. Act now, in your own wallet.") def summarize(address, txs): """Build alert records for any tx that spends FROM address. Pure; no I/O.""" out = [] for tx in txs: sent = outgoing_from(tx, address) if sent <= 0: continue state, msg = describe_window(tx) out.append({ "address": address, "txid": tx.get("txid"), "sats_out": sent, "sats_returned": incoming_to(tx, address), "state": state, "guidance": msg, }) return out # ---------------------------------------------------------------- network passes def mempool_txs(api, address): return _get("%s/address/%s/txs/mempool" % (api, address)) def recent_txs(api, address): return _get("%s/address/%s/txs" % (api, address)) def alert(rec): bar = "!" * 68 print("\n" + bar) print(" SPEND DETECTED FROM YOUR ADDRESS") print(" address : %s" % rec["address"]) print(" txid : %s" % rec["txid"]) print(" leaving : %d sats (%.8f BTC)" % (rec["sats_out"], rec["sats_out"] / 1e8)) if rec["sats_returned"]: print(" change back to this address: %d sats" % rec["sats_returned"]) print(" state : %s" % rec["state"]) print(" %s" % rec["guidance"]) print(" inspect : https://mempool.space/tx/%s" % rec["txid"]) print(bar, flush=True) def watch(addresses, api, once): seen = set() print("watching %d address(es) via %s" % (len(addresses), api)) print("polling every %ds. Ctrl-C to stop." % POLL_SECONDS) print("NOTE: an alert here means coins are ALREADY LEAVING. This is a smoke") print(" alarm, not a lock. Move funds off an at-risk seed proactively.\n") first = True while True: for addr in addresses: try: txs = mempool_txs(api, addr) if first: try: txs = list(txs) + list(recent_txs(api, addr))[:10] except Exception: pass for rec in summarize(addr, txs): if rec["txid"] in seen: continue seen.add(rec["txid"]) alert(rec) except urllib.error.HTTPError as e: print(" [%s] HTTP %s — %s" % (addr[:14], e.code, "rate limited, backing off" if e.code == 429 else "check the address"), file=sys.stderr) if e.code == 429: time.sleep(30) except Exception as e: print(" [%s] %s: %s" % (addr[:14], type(e).__name__, e), file=sys.stderr) time.sleep(1.0) # be polite to a free public API first = False if once: print("\nsingle pass complete. No further polling (--once).") return time.sleep(POLL_SECONDS) # ---------------------------------------------------------------- self-test def selftest(): A = "bc1qexampleaddressaaaaaaaaaaaaaaaaaaaaaaa" B = "bc1qsomeoneelsebbbbbbbbbbbbbbbbbbbbbbbbbb" ok = True def chk(name, got, want): nonlocal ok good = got == want ok = ok and good print(" %-46s %s" % (name, "PASS" if good else "FAIL got=%r want=%r" % (got, want))) sweep = { "txid": "aa" * 32, "vin": [{"prevout": {"scriptpubkey_address": A, "value": 500000}, "sequence": 0xFFFFFFFD}], "vout": [{"scriptpubkey_address": B, "value": 499000}], "status": {"confirmed": False}, } deposit = { "txid": "bb" * 32, "vin": [{"prevout": {"scriptpubkey_address": B, "value": 900000}, "sequence": 0xFFFFFFFF}], "vout": [{"scriptpubkey_address": A, "value": 899000}], "status": {"confirmed": False}, } mined = dict(sweep, txid="cc" * 32, status={"confirmed": True}) final = { "txid": "dd" * 32, "vin": [{"prevout": {"scriptpubkey_address": A, "value": 100}, "sequence": 0xFFFFFFFF}], "vout": [], "status": {"confirmed": False}, } print("known-answer tests (no network):") chk("outgoing: sweep spends from A", outgoing_from(sweep, A), 500000) chk("outgoing: deposit does NOT spend from A", outgoing_from(deposit, A), 0) chk("incoming: deposit credits A", incoming_to(deposit, A), 899000) chk("rbf: sequence fffffffd signals", is_rbf_signalled(sweep), True) chk("rbf: sequence ffffffff does not", is_rbf_signalled(final), False) chk("rbf: no inputs -> unknown", is_rbf_signalled({"vin": []}), None) chk("state: unconfirmed+rbf", describe_window(sweep)[0], "UNCONFIRMED_RBF") chk("state: unconfirmed no-rbf", describe_window(final)[0], "UNCONFIRMED") chk("state: mined", describe_window(mined)[0], "CONFIRMED") chk("summarize: only outgoing alerts", len(summarize(A, [sweep, deposit])), 1) chk("summarize: deposit alone is silent", len(summarize(A, [deposit])), 0) chk("summarize: reports sats", summarize(A, [sweep])[0]["sats_out"], 500000) print("\n%s" % ("ALL PASS — logic verified offline." if ok else "FAILURES ABOVE.")) return 0 if ok else 1 # ---------------------------------------------------------------- cli def load_addresses(items): addrs = [] for it in items: try: with open(it, "r") as fh: for line in fh: line = line.split("#")[0].strip() if line: addrs.append(line) continue except (IOError, OSError): pass addrs.append(it.strip()) seen, uniq = set(), [] for a in addrs: if a and a not in seen: seen.add(a) uniq.append(a) return uniq def main(): p = argparse.ArgumentParser( description="Watch your own bitcoin addresses for an in-progress sweep.") p.add_argument("targets", nargs="*", help="addresses, or a file with one per line") p.add_argument("--api", default=DEFAULT_API, help="mempool REST base (default: mempool.space)") p.add_argument("--once", action="store_true", help="single pass, then exit") p.add_argument("--selftest", action="store_true", help="run offline known-answer tests") a = p.parse_args() if a.selftest: return selftest() if not a.targets: p.print_help() print("\nNo addresses given. Nothing was sent anywhere.") return 2 addrs = load_addresses(a.targets) if not addrs: print("No usable addresses found.", file=sys.stderr) return 2 if any(w in " ".join(addrs).lower() for w in ("abandon ", "zoo zoo", " mnemonic")): print("\nThat input looks like it may contain seed words. Stopping.\n" "This tool takes ADDRESSES only and never needs a seed phrase.\n" "Nothing was sent anywhere.", file=sys.stderr) return 2 try: watch(addrs, a.api.rstrip("/"), a.once) except KeyboardInterrupt: print("\nstopped.") return 0 if __name__ == "__main__": sys.exit(main())