Bitcoin Core watch-only descriptor watchdog -> Gotify

bitcoin_core_watchdog.py

#!/usr/bin/env python3
"""Bitcoin Core watch-only descriptor watchdog -> Gotify.

bitcoin.conf (minimal example):
  keypool=5000
  walletnotify=env GOTIFY_URL=https://gotify.example GOTIFY_TOKEN_FILE=/path/to/token /usr/local/bin/bitcoin_core_watchdog.py notify %w %s %b %h

Optional environment:
  WATCHDOG_EVENTS=both|mempool|confirmed   (default: both)
  MEMPOOL_URL=https://mempool.space        (default; only used to build links)

Examples:
  bitcoin_core_watchdog.py add cold-storage segwit 'zpub...' --timestamp 0
  bitcoin_core_watchdog.py add taproot-vault taproot 'xpub...' --timestamp now
  bitcoin_core_watchdog.py test-gotify

The script is stateless. Bitcoin Core owns and autoloads the descriptor wallets.
Amounts are wallet-local: collaborative transaction inputs/outputs from other parties are ignored.
Do not import seed phrases or private keys: use account-level xpub/zpubs only.
MEMPOOL_URL is never queried by this script; it is only placed in notifications.
"""

from __future__ import annotations

import argparse
import decimal
import hashlib
import json
import os
import re
import shlex
import subprocess
import sys
import time
import urllib.error
import urllib.request

BITCOIN_CLI = os.environ.get("BITCOIN_CLI", "bitcoin-cli")
BITCOIN_CLI_ARGS = shlex.split(os.environ.get("BITCOIN_CLI_ARGS", ""))
MEMPOOL_URL = os.environ.get("MEMPOOL_URL", "https://mempool.space").rstrip("/")
GOTIFY_PRIORITY = int(os.environ.get("GOTIFY_PRIORITY", "7"))
DEFAULT_LOOKAHEAD = 5000
# A walletnotify process can start just as another block arrives. Allow a very
# small tip lag while still suppressing historical rescan notifications.
MAX_LIVE_CONFIRMATION_LAG = 2

XPUB_VERSION = bytes.fromhex("0488b21e")
ZPUB_VERSION = bytes.fromhex("04b24746")
B58_ALPHABET = b"123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"
B58_MAP = {c: i for i, c in enumerate(B58_ALPHABET)}
WALLET_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
TXID_RE = re.compile(r"^[0-9a-fA-F]{64}$")
BLOCKHASH_RE = TXID_RE


def die(message: str, code: int = 1) -> None:
    print(f"error: {message}", file=sys.stderr)
    raise SystemExit(code)


def cli(*args: str, wallet: str | None = None) -> str:
    command = [BITCOIN_CLI, *BITCOIN_CLI_ARGS]
    if wallet:
        command.append(f"-rpcwallet={wallet}")
    command.extend(args)
    result = subprocess.run(
        command,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if result.returncode:
        detail = result.stderr.strip() or result.stdout.strip()
        rendered = " ".join(shlex.quote(x) for x in command)
        raise RuntimeError(f"{rendered} failed: {detail}")
    return result.stdout.strip()


def cli_json(*args: str, wallet: str | None = None):
    output = cli(*args, wallet=wallet)
    return json.loads(output) if output else None


def b58decode_check(value: str) -> bytes:
    number = 0
    try:
        for char in value.encode("ascii"):
            number = number * 58 + B58_MAP[char]
    except (KeyError, UnicodeEncodeError) as exc:
        raise ValueError("invalid Base58 extended key") from exc

    raw = number.to_bytes((number.bit_length() + 7) // 8, "big") if number else b""
    raw = b"\x00" * (len(value) - len(value.lstrip("1"))) + raw
    if len(raw) < 5:
        raise ValueError("invalid Base58Check extended key")

    payload, checksum = raw[:-4], raw[-4:]
    expected = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4]
    if checksum != expected:
        raise ValueError("bad extended-key checksum")
    return payload


def b58encode_check(payload: bytes) -> str:
    checksum = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4]
    raw = payload + checksum
    number = int.from_bytes(raw, "big")
    encoded = bytearray()
    while number:
        number, remainder = divmod(number, 58)
        encoded.append(B58_ALPHABET[remainder])
    leading_zeroes = len(raw) - len(raw.lstrip(b"\x00"))
    return (b"1" * leading_zeroes + bytes(reversed(encoded))).decode("ascii")


def normalize_key(key: str, kind: str) -> str:
    payload = b58decode_check(key)
    if len(payload) != 78:
        raise ValueError("expected a BIP32 extended public key")

    version = payload[:4]
    if version == XPUB_VERSION:
        return key
    if version == ZPUB_VERSION and kind == "segwit":
        return b58encode_check(XPUB_VERSION + payload[4:])
    if version == ZPUB_VERSION:
        raise ValueError("Taproot watches should use an account-level xpub, not zpub")
    raise ValueError("unsupported key type; use a mainnet xpub or, for SegWit, zpub")


def wallet_exists(name: str) -> bool:
    wallets = cli_json("listwalletdir").get("wallets", [])
    return any(item.get("name") == name for item in wallets)


def ensure_wallet(name: str) -> None:
    if not WALLET_RE.fullmatch(name):
        die("wallet name may contain only letters, numbers, _, . and -")

    if not wallet_exists(name):
        cli_json(
            "-named",
            "createwallet",
            f"wallet_name={name}",
            "disable_private_keys=true",
            "blank=true",
            "descriptors=true",
            "load_on_startup=true",
        )
    elif name not in cli_json("listwallets"):
        cli_json("loadwallet", name, "true")


def parse_timestamp(value: str):
    if value == "now":
        return "now"
    try:
        timestamp = int(value)
    except ValueError:
        die("--timestamp must be 'now' or a Unix timestamp; use 0 for the full chain")
    if timestamp < 0:
        die("--timestamp cannot be negative")
    return timestamp


def add_watch(args) -> None:
    ensure_wallet(args.name)
    try:
        xpub = normalize_key(args.key, args.kind)
    except ValueError as exc:
        die(str(exc))

    wrapper = "wpkh" if args.kind == "segwit" else "tr"
    raw_descriptor = f"{wrapper}({xpub}/<0;1>/*)"
    descriptor = cli_json("getdescriptorinfo", raw_descriptor)["descriptor"]

    request = [{
        "desc": descriptor,
        "active": True,
        "range": [0, args.lookahead - 1],
        "timestamp": parse_timestamp(args.timestamp),
    }]
    result = cli_json(
        "importdescriptors",
        json.dumps(request, separators=(",", ":")),
        wallet=args.name,
    )
    if not result or not result[0].get("success"):
        die(f"importdescriptors failed: {json.dumps(result, indent=2)}")

    print(f"Added watch-only wallet: {args.name}")
    print(f"Descriptor: {descriptor}")
    print(f"Initial lookahead: {args.lookahead}")
    print(f"Tip: set keypool={args.lookahead} in bitcoin.conf for a similar rolling lookahead.")


def gotify_config() -> tuple[str, str]:
    url = os.environ.get("GOTIFY_URL", "").strip().rstrip("/")
    if not url:
        raise RuntimeError("GOTIFY_URL is required")

    token = os.environ.get("GOTIFY_TOKEN", "").strip()
    token_file = os.environ.get("GOTIFY_TOKEN_FILE", "").strip()
    if not token and token_file:
        try:
            with open(token_file, "r", encoding="utf-8") as handle:
                token = handle.read().strip()
        except OSError as exc:
            raise RuntimeError(f"cannot read GOTIFY_TOKEN_FILE: {exc}") from exc
    if not token:
        raise RuntimeError("set GOTIFY_TOKEN or GOTIFY_TOKEN_FILE")
    return url, token


def event_mode() -> str:
    mode = os.environ.get("WATCHDOG_EVENTS", "both").strip().lower()
    if mode not in {"both", "mempool", "confirmed"}:
        raise RuntimeError("WATCHDOG_EVENTS must be one of: both, mempool, confirmed")
    return mode


def send_notification(title: str, message: str, click_url: str) -> None:
    """Send one notification. Replace this function to use another provider."""
    gotify_url, gotify_token = gotify_config()
    payload = json.dumps({
        "title": title,
        "message": message,
        "priority": GOTIFY_PRIORITY,
        "extras": {
            "client::display": {"contentType": "text/plain"},
            "client::notification": {"click": {"url": click_url}},
        },
    }).encode("utf-8")

    request = urllib.request.Request(
        f"{gotify_url}/message",
        data=payload,
        headers={"Content-Type": "application/json", "X-Gotify-Key": gotify_token},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            if not 200 <= response.status < 300:
                raise RuntimeError(f"Gotify returned HTTP {response.status}")
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", "replace")[:500]
        raise RuntimeError(f"Gotify HTTP {exc.code}: {detail}") from exc
    except urllib.error.URLError as exc:
        raise RuntimeError(f"Gotify request failed: {exc}") from exc


def btc(value) -> decimal.Decimal:
    return decimal.Decimal(str(value))


def format_btc(value: decimal.Decimal) -> str:
    text = format(value, "f").rstrip("0").rstrip(".")
    return text or "0"


def vout_address(vout: dict) -> str | None:
    script = vout.get("scriptPubKey", {})
    address = script.get("address")
    if isinstance(address, str):
        return address
    # Compatibility with older decoderawtransaction-style output.
    addresses = script.get("addresses")
    if isinstance(addresses, list) and len(addresses) == 1 and isinstance(addresses[0], str):
        return addresses[0]
    return None


def wallet_owns_address(wallet: str, address: str, cache: dict[str, bool]) -> bool:
    if address not in cache:
        try:
            info = cli_json("getaddressinfo", address, wallet=wallet)
            cache[address] = bool(info.get("ismine") or info.get("iswatchonly"))
        except RuntimeError:
            cache[address] = False
    return cache[address]


def wallet_flow(wallet: str, tx: dict) -> tuple[decimal.Decimal, decimal.Decimal]:
    """Return (our_inputs, our_outputs), ignoring other participants.

    Output ownership is checked with getaddressinfo. For each input we look up
    its parent transaction in this wallet and inspect only the referenced vout.
    This avoids treating every non-wallet output in a collaborative transaction
    as money sent by this wallet.
    """
    decoded = tx.get("decoded")
    if not isinstance(decoded, dict):
        raise RuntimeError("gettransaction verbose output did not contain decoded transaction")

    ownership_cache: dict[str, bool] = {}
    parent_cache: dict[str, dict | None] = {}
    our_inputs = decimal.Decimal("0")
    our_outputs = decimal.Decimal("0")

    for vout in decoded.get("vout", []):
        address = vout_address(vout)
        if address and wallet_owns_address(wallet, address, ownership_cache):
            our_outputs += btc(vout.get("value", 0))

    for vin in decoded.get("vin", []):
        parent_txid = vin.get("txid")
        parent_vout = vin.get("vout")
        if not isinstance(parent_txid, str) or not isinstance(parent_vout, int):
            continue  # coinbase or malformed/nonstandard input

        if parent_txid not in parent_cache:
            try:
                parent_cache[parent_txid] = cli_json(
                    "gettransaction", parent_txid, "true", "true", wallet=wallet
                )
            except RuntimeError:
                # An unrelated participant's parent transaction normally isn't
                # present in this wallet, which is exactly what we want here.
                parent_cache[parent_txid] = None

        parent = parent_cache[parent_txid]
        if not parent:
            continue
        parent_decoded = parent.get("decoded", {})
        prevout = next(
            (v for v in parent_decoded.get("vout", []) if v.get("n") == parent_vout),
            None,
        )
        if not prevout:
            continue
        address = vout_address(prevout)
        if address and wallet_owns_address(wallet, address, ownership_cache):
            our_inputs += btc(prevout.get("value", 0))

    return our_inputs, our_outputs


def tx_summary(wallet: str, tx: dict) -> str:
    our_inputs, our_outputs = wallet_flow(wallet, tx)
    net = our_outputs - our_inputs

    if our_inputs:
        parts = [
            f"our inputs {format_btc(our_inputs)} BTC",
            f"our outputs {format_btc(our_outputs)} BTC",
        ]
        if net > 0:
            parts.append(f"net +{format_btc(net)} BTC")
        elif net < 0:
            parts.append(f"net -{format_btc(-net)} BTC")
        else:
            parts.append("net 0 BTC")
    elif our_outputs:
        parts = [f"received {format_btc(our_outputs)} BTC"]
    else:
        parts = ["wallet transaction"]

    if tx.get("bip125-replaceable") == "yes":
        parts.append("RBF")
    return " · ".join(parts)


def in_local_mempool(txid: str) -> bool:
    # walletnotify and mempool RPC visibility can race very briefly.
    for delay in (0, 0.1, 0.25, 0.5):
        if delay:
            time.sleep(delay)
        try:
            cli_json("getmempoolentry", txid)
            return True
        except RuntimeError:
            pass
    return False


def get_wallet_tx(wallet: str, txid: str) -> dict:
    last_error = None
    for delay in (0, 0.1, 0.25, 0.5):
        if delay:
            time.sleep(delay)
        try:
            return cli_json("gettransaction", txid, "true", "true", wallet=wallet)
        except RuntimeError as exc:
            last_error = exc
    raise RuntimeError(f"gettransaction failed: {last_error}")


def is_live_confirmation(height: int) -> bool:
    info = cli_json("getblockchaininfo")
    if info.get("initialblockdownload", False):
        return False
    tip = int(info["blocks"])
    return 0 <= tip - height <= MAX_LIVE_CONFIRMATION_LAG


def notify(wallet: str, txid: str, block: str, height_text: str) -> None:
    if not WALLET_RE.fullmatch(wallet):
        die("unsafe wallet name received from walletnotify")
    if not TXID_RE.fullmatch(txid):
        die("invalid txid received from walletnotify")

    mode = event_mode()
    link = f"{MEMPOOL_URL}/tx/{txid}"

    if block == "unconfirmed":
        if mode == "confirmed":
            return
        if not in_local_mempool(txid):
            return
        tx = get_wallet_tx(wallet, txid)
        send_notification(
            f"Bitcoin mempool activity: {wallet}",
            f"{tx_summary(wallet, tx)}\nUnconfirmed / in local mempool\n{link}",
            link,
        )
        return

    if mode == "mempool":
        return
    if not BLOCKHASH_RE.fullmatch(block):
        die("invalid block hash received from walletnotify")
    try:
        height = int(height_text)
    except ValueError:
        die("invalid block height received from walletnotify")
    if height < 0:
        die("negative block height received for confirmed transaction")

    # Suppress historical imports/rescans. A transaction first delivered in a
    # freshly connected block (including private miner submission) is at/near tip.
    if not is_live_confirmation(height):
        return

    tx = get_wallet_tx(wallet, txid)
    send_notification(
        f"Bitcoin confirmed activity: {wallet}",
        f"{tx_summary(wallet, tx)}\nConfirmed at block {height}\n{link}",
        link,
    )


def test_gotify() -> None:
    send_notification("Bitcoin watchdog test", "Notifications are working.", MEMPOOL_URL)
    print("Gotify test notification sent")


def main() -> None:
    parser = argparse.ArgumentParser(description="Bitcoin Core xpub/zpub watchdog")
    commands = parser.add_subparsers(dest="command", required=True)

    add = commands.add_parser("add", help="create an autoloading watch-only descriptor wallet")
    add.add_argument("name", help="wallet / alert name")
    add.add_argument("kind", choices=("segwit", "taproot"))
    add.add_argument("key", help="account-level xpub, or zpub for SegWit")
    add.add_argument(
        "--timestamp",
        default="now",
        help="wallet birthday: 'now', Unix timestamp, or 0 for full chain",
    )
    add.add_argument(
        "--lookahead",
        type=int,
        default=DEFAULT_LOOKAHEAD,
        help=f"initial address range (default: {DEFAULT_LOOKAHEAD})",
    )

    notification = commands.add_parser("notify", help="walletnotify entry point")
    notification.add_argument("wallet", help="%%w from Bitcoin Core walletnotify")
    notification.add_argument("txid", help="%%s from Bitcoin Core walletnotify")
    notification.add_argument("block", help="%%b: block hash or 'unconfirmed'")
    notification.add_argument("height", help="%%h: block height or -1")

    commands.add_parser("test-gotify", help="send a test notification")

    args = parser.parse_args()
    if getattr(args, "lookahead", 1) < 1:
        die("--lookahead must be at least 1")

    try:
        if args.command == "add":
            add_watch(args)
        elif args.command == "notify":
            notify(args.wallet, args.txid, args.block, args.height)
        elif args.command == "test-gotify":
            test_gotify()
    except RuntimeError as exc:
        die(str(exc))


if __name__ == "__main__":
    main()