coldcard-stolen-to-btclock.py

gistfile1.txt

#!/usr/bin/env python3
"""Show the Coldcard entropy-hack stolen BTC total on a BTClock.

Two public trackers publish a total; they measure different things:

  watch    coldcardwatch.com -- what was *stolen*. Sum of
           min(live_balance, attributed) over the 7 collector addresses,
           plus a snapshot of the wave-3 vaults. The attribution cap means
           the figure is a verified minimum and can only fall.

  tracker  coldcard-hack-tracker.vercel.app -- what its *headline* shows:
           the sum of a per-wave `stolenBtc` figure. These are editorial
           attributions, not chain reads, and they include a wave the site
           itself marks "pattern-match only (no victim report yet)".

  held     the same site's snapshot.json -- what the thief still holds.
           Sum of the live balances of its curated address list. This is
           what `tracker` used to mean, before the site was rebuilt around
           per-wave attribution.

The three answer different questions and no longer converge: as of Aug 5
they read 1366 / 1924 / 1397 BTC. `tracker` is the largest because it
counts coins that left the tracked addresses, and because 443 BTC of it is
an unconfirmed Wave 4.

`tracker` and `held` track the site on their own; `watch` needs COLLECTORS
below refreshed from its inline WALLETS array when its attribution changes.

Stdlib only. Python 3.9+.

    ./coldcard-stolen-to-btclock.py --host btclock-9d5530
    ./coldcard-stolen-to-btclock.py --host btclock-9d5530 --usd --loop 300
"""

import argparse
import base64
import json
import math
import re
import sys
import time
import urllib.error
import urllib.request

UA = "coldcard-stolen-to-btclock/1.0"

# --- coldcard-watch's model, transcribed from its inline script -------------
# `attributed` caps each address at the sats forensically traced to the theft,
# so an exchange or third-party wallet cannot inflate the total with coins that
# were never stolen.
COLLECTORS = [
    ("bc1qq85v2c926eg6pgxhwp6q7lf6cnsz80qs3fcu9r", 56_202_932_050),
    ("bc1qx76cae2706qd5q576feh7xq8rfcsjpf2htfhe3", 39_847_590_375),
    ("bc1q8jy96fe5lf8vfugydnte3cguk92gpev7kwtp3q", 8_962_327_890),
    ("bc1qnk4zh9qcnap2mycp56qjrgza3cc8ylrh8fecp0", 3_245_056_320),
    ("bc1qtfrwa4j6rmj9rsgspv6a0yjumkg39js2numu75", 4_590_251_994),
    ("bc1qhh4jkkj07vxpdt0zlvxctjlfhqmurhxa24x3h2", 19_153_809),
    ("bc1qmd5m5ktv7m5ffujxv4248fxv36myvdx79n8jp6", 3_018_476_329),
]

# follow() tuning, same constants the site uses
FOLLOW_FLOOR = 2_000_000     # never chase an output below 0.02 BTC
FOLLOW_SHARE = 0.02          # nor below 2% of what the spending address held
FOLLOW_PER_TX = 5            # only the largest few outputs per transaction
MAX_TRACKED = 40             # backstop against a peel chain exploding
# coldcard-watch.vercel.app 308-redirects here. urllib only follows 308 from
# Python 3.11, so point at the canonical host rather than rely on that.
WAVE3_URL = "https://coldcardwatch.com/wave3.js"
TRACKER_URL = "https://coldcard-hack-tracker.vercel.app"
SNAPSHOT_URL = TRACKER_URL + "/snapshot.json"
# Every backend below speaks the Esplora API. mempool.space regularly stalls
# for the full timeout under a burst, so the list is a fallthrough chain, not a
# preference. Promote one with --explorer.
EXPLORERS = [
    "https://mempool.space/api",
    "https://blockstream.info/api",
    "https://mempool.dbtc.link/api",
]
HTTP_TIMEOUT = 6  # short, because falling through beats waiting out a stall


def get_json(url, timeout=HTTP_TIMEOUT):
    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())


def esplora(path):
    last = None
    for base in list(EXPLORERS):
        try:
            d = get_json(f"{base}{path}")
        except (urllib.error.URLError, OSError, ValueError) as e:
            last = e
            continue
        # Stick to whatever answered. A stalling explorer otherwise costs the
        # full timeout on every single address instead of just the first.
        if EXPLORERS[0] != base:
            EXPLORERS[:] = [base] + [e for e in EXPLORERS if e != base]
        return d
    raise last


def address_balance(addr, cache):
    """Confirmed + unconfirmed balance, memoised so a 40-address trail costs
    one request per address rather than one per pass.

    Only the txo *sums* are read. An Electrum-backed instance reports the txo
    counts as zero, and the sums still net out to the right balance, so this
    works across every backend in EXPLORERS.
    """
    if addr not in cache:
        d = esplora(f"/address/{addr}")
        c, m = d.get("chain_stats", {}), d.get("mempool_stats", {})
        cache[addr] = ((c.get("funded_txo_sum", 0) - c.get("spent_txo_sum", 0))
                       + (m.get("funded_txo_sum", 0) - m.get("spent_txo_sum", 0)))
    return cache[addr]


def follow_chain(wallets, cache):
    """Chase spends onward the way coldcard-watch's follow() does, adding each
    destination with attributed = the output value.

    Note this over-counts: a spend that splits into several outputs adds every
    branch, and a hop into a shared address contributes that whole output again
    while the earlier hops are already at zero. It is reproduced here only so
    the number can be made to match what the site renders.
    """
    known = {w["addr"] for w in wallets}
    queue = list(wallets)
    while queue and len(wallets) < MAX_TRACKED:
        w = queue.pop(0)
        # An address still holding everything traced to it has not moved theft
        # proceeds. This is the site's own cold-load test, and unlike a spent-
        # output count it reads the same on every backend.
        if address_balance(w["addr"], cache) >= w["attributed"]:
            continue
        floor = max(FOLLOW_FLOOR, w["attributed"] * FOLLOW_SHARE)
        for tx in esplora(f"/address/{w['addr']}/txs"):
            if not any(v.get("prevout", {}).get("scriptpubkey_address") == w["addr"]
                       for v in tx.get("vin", [])):
                continue
            outs = [o for o in tx.get("vout", [])
                    if o.get("scriptpubkey_address")
                    and o["scriptpubkey_address"] not in known
                    and o["value"] >= floor]
            outs.sort(key=lambda o: -o["value"])
            for o in outs[:FOLLOW_PER_TX]:
                if len(wallets) >= MAX_TRACKED:
                    break
                known.add(o["scriptpubkey_address"])
                hop = {"addr": o["scriptpubkey_address"], "attributed": o["value"], "traced": True}
                wallets.append(hop)
                queue.append(hop)
    return wallets


def total_watch(follow=False):
    """Replicate coldcard-watch: capped collector balances + wave-3 snapshot."""
    cache = {}
    wallets = [{"addr": a, "attributed": att} for a, att in COLLECTORS]
    if follow:
        wallets = follow_chain(wallets, cache)

    held = 0
    for w in wallets:
        try:
            bal = address_balance(w["addr"], cache)
        except (urllib.error.URLError, OSError, ValueError) as e:
            print(f"  ! {w['addr'][:12]}... unreachable ({e}), using attributed", file=sys.stderr)
            bal = w["attributed"]
        held += max(0, min(bal, w["attributed"]))

    # The wave-3 vaults are one-per-victim (293 and climbing) and are not
    # live-pollable; the site ships them as a snapshot in window.WAVE3.
    req = urllib.request.Request(WAVE3_URL, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=15) as r:
        body = r.read().decode()
    payload = json.loads(re.sub(r"^\s*window\.WAVE3\s*=\s*", "", body).strip().rstrip(";"))
    held += payload.get("held", 0)
    return held, None


def total_held():
    """Sum of the live balances in coldcard-hack-tracker's snapshot.json."""
    snap = get_json(SNAPSHOT_URL)
    held = sum(a["balanceSats"] for a in snap["addresses"])
    return held, snap.get("usdPrice")


def tracker_waves():
    """Pull the per-wave `stolenBtc` table out of the site's JS bundle.

    The headline is `waves.reduce((a, w) => a + w.stolenBtc, 0)` and the table
    is compiled into the bundle -- there is no JSON feed for it, unlike
    snapshot.json. So this reads the bundle: resolve the hashed asset name
    from index.html, then pull each `stolenBtc:` value. Most are literals; one
    is a sum of `<obj>.<key>Btc` members, which are resolved by looking the
    keys up in the same bundle.

    Minified identifiers change every build, so this deliberately keys off the
    unminified property names the site's own data model uses. If a rebuild
    changes those, the sanity check below fails loudly rather than quietly
    reporting a wrong total.
    """
    req = urllib.request.Request(TRACKER_URL, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as r:
        index_html = r.read().decode()
    m = re.search(r'src="(/assets/index-[^"]+\.js)"', index_html)
    if not m:
        raise RuntimeError("tracker: no /assets/index-*.js in index.html")

    req = urllib.request.Request(TRACKER_URL.rstrip("/") + m.group(1),
                                 headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as r:
        bundle = r.read().decode()

    waves = []
    for label, expr in re.findall(r"label:`([^`]*)`,stolenBtc:([^,]+),date:", bundle):
        try:
            waves.append((label, float(expr)))
            continue
        except ValueError:
            pass
        # Composite like `x.smallVaultBtc+x.hopVaultBtc+x.laterVaultBtc`. Keys
        # are not unique across the bundle -- `hopVaultBtc` also appears on an
        # unrelated movement object -- so resolve them from the one object
        # literal that carries all of them, not from the first match anywhere.
        keys = [t.strip().split(".")[-1] for t in expr.split("+")]
        holder = next((m.group(0) for m in re.finditer(r"\{[^{}]*\}", bundle)
                       if all(re.search(rf"\b{re.escape(k)}:", m.group(0)) for k in keys)), None)
        if holder is None:
            raise RuntimeError(f"tracker: no object in the bundle carries all of {keys}")
        waves.append((label, sum(
            float(re.search(rf"\b{re.escape(k)}:([0-9]*\.?[0-9]+)", holder).group(1))
            for k in keys)))

    if len(waves) < 4:
        raise RuntimeError(f"tracker: only {len(waves)} waves parsed; the bundle "
                           "layout changed and the total would be wrong")
    return waves


def total_tracker():
    """The site's headline: every wave's attributed stolenBtc, summed."""
    waves = tracker_waves()
    return round(sum(btc for _, btc in waves) * 1e8), None


def btc_price_usd():
    return float(esplora("/v1/prices")["USD"])


# --- rendering --------------------------------------------------------------

def fit(candidates, width):
    """Pick the most precise rendering that fits, right-aligned in `width`."""
    for s in candidates:
        if len(s) <= width:
            return s.rjust(width)
    return candidates[-1][-width:]


def num(v, decimals, rounding):
    """Round to nearest, or truncate when the figure must never overstate."""
    if rounding == "down":
        scale = 10 ** decimals
        v = math.floor(v * scale) / scale
    return f"{v:.{decimals}f}"


# /api/show/custom sizes a one-glyph cell at 120px by default. The wrapper
# form {"cells":[...],"digitPx":N} raises that ceiling; the renderer still
# fits to panel width, so the maximum just means "as big as the panel allows"
# rather than "clipped". The firmware accepts 20..220 and silently ignores
# anything outside that band, falling back to the 120px default.
DIGIT_PX_MAX = 220


def build_cells(sats, price, num_screens, head, tail, usd, rounding="nearest"):
    """Panels 0 and N-1 are two-line labels; the panels between carry one
    character of the value each."""
    width = num_screens - 2
    btc = sats / 1e8
    if usd:
        m = btc * price / 1e6
        # The tail label already says USD, so dropping the $ to gain a decimal
        # is a fair trade when the panels run out.
        value = fit([f"${num(m, 2, rounding)}M", f"${num(m, 1, rounding)}M",
                     f"{num(m, 1, rounding)}M", f"${num(m, 0, rounding)}M",
                     f"{num(m, 0, rounding)}M"], width)
    else:
        value = fit([num(btc, 2, rounding), num(btc, 1, rounding),
                     num(btc, 0, rounding)], width)
    return [head] + list(value) + [tail], value.strip()


# --- device -----------------------------------------------------------------

def post(host, path, payload, user, password, timeout=15):
    data = json.dumps(payload).encode() if payload is not None else b""
    req = urllib.request.Request(
        f"http://{host}{path}", data=data, method="POST",
        headers={"User-Agent": UA, "Content-Type": "application/json"},
    )
    if user:
        token = base64.b64encode(f"{user}:{password}".encode()).decode()
        req.add_header("Authorization", f"Basic {token}")
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.status


def device_screens(host, user, password):
    req = urllib.request.Request(f"http://{host}/api/status", headers={"User-Agent": UA})
    if user:
        token = base64.b64encode(f"{user}:{password}".encode()).decode()
        req.add_header("Authorization", f"Basic {token}")
    with urllib.request.urlopen(req, timeout=15) as r:
        return json.loads(r.read().decode()).get("numScreens", 7)


def main():
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--host", default="btclock-9d5530", help="BTClock hostname or IP")
    p.add_argument("--source", choices=["watch", "tracker", "held"], default="watch",
                   help="watch = coldcardwatch.com attributed total (default), "
                        "tracker = hack-tracker headline (per-wave stolenBtc), "
                        "held = hack-tracker snapshot.json live balances")
    p.add_argument("--usd", action="store_true", help="show USD in millions instead of BTC")
    p.add_argument("--follow", action="store_true",
                   help="chase spends onward like the site's live page does; matches its "
                        "drifting number but double-counts across hops")
    p.add_argument("--round", dest="rounding", choices=["nearest", "down"], default="nearest",
                   help="nearest (default) or down, so a verified minimum never overstates")
    p.add_argument("--explorer", metavar="BASE_URL",
                   help="Esplora base URL to try first, e.g. https://mempool.dbtc.link/api")
    p.add_argument("--digit-px", type=int, default=DIGIT_PX_MAX, metavar="PX",
                   help=f"digit height ceiling, 20-{DIGIT_PX_MAX} (default {DIGIT_PX_MAX}, "
                        "the largest the panel allows); 0 keeps the firmware's 120px auto-size")
    p.add_argument("--label", default="COLDCARD/FUCKUP",
                   help="first panel label, '/' splits the two lines")
    p.add_argument("--tail-label", help="last panel label (default BTC/STOLEN, USD/STOLEN with --usd)")
    p.add_argument("--loop", type=int, metavar="SECONDS", help="refresh forever every N seconds")
    p.add_argument("--pause", action="store_true",
                   help="stop the screen rotation so the value stays up")
    p.add_argument("--user", help="basic auth username")
    p.add_argument("--password", default="", help="basic auth password")
    p.add_argument("--dry-run", action="store_true", help="print the payload, do not POST")
    args = p.parse_args()

    # Out-of-band values are dropped by the firmware without complaint, which
    # would look like the flag did nothing. Fail loudly instead.
    if args.digit_px and not 20 <= args.digit_px <= DIGIT_PX_MAX:
        sys.exit(f"--digit-px must be 0 or 20..{DIGIT_PX_MAX}; the firmware silently "
                 f"ignores anything else and falls back to its 120px default")

    if args.explorer:
        base = args.explorer.rstrip("/")
        EXPLORERS[:] = [base] + [e for e in EXPLORERS if e != base]

    head = args.label
    tail = args.tail_label or ("USD/DRAINED" if args.usd else "BTC/DRAINED")
    if args.source == "watch":
        def fetch():
            return total_watch(follow=args.follow)
    elif args.source == "tracker":
        fetch = total_tracker
    else:
        fetch = total_held

    if args.dry_run and args.source == "tracker":
        for label, btc in tracker_waves():
            print(f"    {label:<32} {btc:10.4f}")

    num_screens = 7
    if not args.dry_run:
        try:
            num_screens = device_screens(args.host, args.user, args.password)
        except (urllib.error.URLError, OSError) as e:
            sys.exit(f"cannot reach {args.host}: {e}")
        if args.pause:
            post(args.host, "/api/action/pause", None, args.user, args.password)

    while True:
        sats, price = fetch()
        if args.usd and price is None:
            price = btc_price_usd()

        cells, shown = build_cells(sats, price, num_screens, head, tail, args.usd, args.rounding)
        print(f"{time.strftime('%H:%M:%S')}  {sats / 1e8:.4f} BTC ({args.source})"
              f"  ->  {head.replace('/', ' ')} | {shown} | {tail.replace('/', ' ')}")

        # Bare array is the legacy wire format; the wrapper object is what
        # carries digitPx. Only send the wrapper when there is something to say.
        payload = cells
        if args.digit_px:
            payload = {"cells": cells, "digitPx": args.digit_px}

        if args.dry_run:
            print(json.dumps(payload))
        else:
            post(args.host, "/api/show/custom", payload, args.user, args.password)

        if not args.loop:
            break
        time.sleep(args.loop)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass