git / moonbite-network / explorer/webhooks.py · raw
"""Webhook subscriptions + delivery for the MoonBite explorer.A single background poller watches the chain tip and delivers signed HTTPcallbacks to subscribers: * event="new_block" -> fired once per new block with a block summary. * event="address" -> fired when a new block contains a tx that pays to (or spends from) the subscribed address.Design notes------------* Storage is stdlib ``sqlite3`` (no extra dependency). The DB path is ephemeral on Railway by default, which is fine for v1.* Delivery is signed: each POST carries ``X-MoonBite-Signature: sha256=<hmac-sha256(secret, body)>`` so the receiver can verify authenticity. Failures back off and the hook auto-disables after ``WEBHOOK_MAX_FAILURES`` consecutive strikes.* SSRF guard: callback URLs must be https and must not resolve to a private / loopback / link-local address unless ``WEBHOOK_ALLOW_PRIVATE`` is set.* Single poller: gunicorn runs multiple workers, so the poller claims a lock row in SQLite (with a heartbeat) and only the holder polls. If the holder dies, another worker takes over once the lock goes stale.This module never touches keys or signs coin transactions; it only reads chainstate over RPC and POSTs notifications."""from __future__ import annotationsimport hashlibimport hmacimport ipaddressimport jsonimport secretsimport socketimport sqlite3import threadingimport timeimport urllib.errorimport urllib.requestimport uuidfrom urllib.parse import urlparseimport configfrom rpc import RpcClient, RPCConnectionError, RPCError_VALID_EVENTS = ("new_block", "address")_LOCK_STALE_SECONDS = 45 # a heartbeat older than this means the holder died.# ------------------------------------------------------------------------- ## Storage# ------------------------------------------------------------------------- #def _connect() -> sqlite3.Connection: conn = sqlite3.connect(config.WEBHOOK_DB_PATH, timeout=10) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") return conndef init_db() -> None: conn = _connect() try: conn.execute( """ CREATE TABLE IF NOT EXISTS webhooks ( id TEXT PRIMARY KEY, url TEXT NOT NULL, event TEXT NOT NULL, address TEXT, secret TEXT NOT NULL, active INTEGER NOT NULL DEFAULT 1, failures INTEGER NOT NULL DEFAULT 0, created INTEGER NOT NULL, last_delivery INTEGER ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS state ( key TEXT PRIMARY KEY, value TEXT ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS poller_lock ( id INTEGER PRIMARY KEY CHECK (id = 1), owner TEXT, heartbeat INTEGER ) """ ) conn.execute( "INSERT OR IGNORE INTO poller_lock (id, owner, heartbeat) VALUES (1, NULL, 0)" ) conn.commit() finally: conn.close()def _get_state(conn, key): row = conn.execute("SELECT value FROM state WHERE key = ?", (key,)).fetchone() return row["value"] if row else Nonedef _set_state(conn, key, value): conn.execute( "INSERT INTO state (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (key, str(value)), )# ------------------------------------------------------------------------- ## Registration / management (called from the API layer)# ------------------------------------------------------------------------- #class WebhookError(Exception): """Registration/validation failure; carries an HTTP status.""" def __init__(self, message, status=400): super().__init__(message) self.message = message self.status = statusdef _validate_callback_url(url: str) -> None: parsed = urlparse(url) if parsed.scheme not in ("http", "https"): raise WebhookError("callback url must be http or https") if not config.WEBHOOK_ALLOW_PRIVATE and parsed.scheme != "https": raise WebhookError("callback url must use https") host = parsed.hostname if not host: raise WebhookError("callback url has no host") if config.WEBHOOK_ALLOW_PRIVATE: return # Resolve every address the host maps to and reject anything non-public. try: infos = socket.getaddrinfo(host, parsed.port or None, proto=socket.IPPROTO_TCP) except socket.gaierror: raise WebhookError("callback host does not resolve") for info in infos: ip = ipaddress.ip_address(info[4][0]) if (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified): raise WebhookError("callback url must be a public address")def register(url: str, event: str, address: str | None) -> dict: """Create a subscription. Returns {id, secret, ...}. Raises WebhookError.""" url = (url or "").strip() event = (event or "").strip() address = (address or "").strip() or None if event not in _VALID_EVENTS: raise WebhookError(f"event must be one of {list(_VALID_EVENTS)}") if event == "address" and not address: raise WebhookError("event 'address' requires an 'address'") if not url: raise WebhookError("missing 'url'") _validate_callback_url(url) # Confirm the address is well-formed on the node (skipped in demo mode). if event == "address": client = RpcClient() if not client.is_demo(): try: info = client.validateaddress(address) if not info.get("isvalid"): raise WebhookError("invalid MoonBite address") except RPCConnectionError as exc: raise WebhookError(f"node unreachable: {exc}", 503) except RPCError as exc: raise WebhookError(str(exc), 502) conn = _connect() try: total = conn.execute("SELECT COUNT(*) AS c FROM webhooks").fetchone()["c"] if total >= config.WEBHOOK_MAX: raise WebhookError("webhook limit reached", 429) hook_id = uuid.uuid4().hex secret = secrets.token_hex(24) conn.execute( "INSERT INTO webhooks (id, url, event, address, secret, created) " "VALUES (?, ?, ?, ?, ?, ?)", (hook_id, url, event, address, secret, int(time.time())), ) conn.commit() finally: conn.close() return {"id": hook_id, "secret": secret, "url": url, "event": event, "address": address}def get(hook_id: str) -> dict | None: conn = _connect() try: row = conn.execute( "SELECT id, url, event, address, active, failures, created, last_delivery " "FROM webhooks WHERE id = ?", (hook_id,), ).fetchone() finally: conn.close() if not row: return None d = dict(row) d["active"] = bool(d["active"]) return ddef delete(hook_id: str, secret: str) -> bool: """Delete a subscription. Requires the secret issued at creation.""" conn = _connect() try: row = conn.execute( "SELECT secret FROM webhooks WHERE id = ?", (hook_id,) ).fetchone() if not row: return False if not hmac.compare_digest(row["secret"], secret or ""): raise WebhookError("invalid secret", 403) conn.execute("DELETE FROM webhooks WHERE id = ?", (hook_id,)) conn.commit() finally: conn.close() return True# ------------------------------------------------------------------------- ## Delivery# ------------------------------------------------------------------------- #class _NoRedirect(urllib.request.HTTPRedirectHandler): """Refuse to follow redirects during delivery. The registered URL is validated as public, but a 3xx Location (or an http:// downgrade) can point at an internal target — the SSRF bypass that made delivery-time validation necessary. Returning None here lets the 3xx surface as an HTTPError, which _deliver counts as a failed (non-2xx) delivery.""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None_no_redirect_opener = urllib.request.build_opener(_NoRedirect)def _deliver(hook: sqlite3.Row, payload: dict) -> bool: # Re-validate at delivery time: the callback host's DNS could have been # repointed to a private/loopback address since registration (rebinding), # and only the registration-time check would have caught it. Skipped when # private targets are explicitly allowed (self-hosted/testing). if not config.WEBHOOK_ALLOW_PRIVATE: try: _validate_callback_url(hook["url"]) except WebhookError: return False body = json.dumps(payload, separators=(",", ":")).encode("utf-8") sig = hmac.new(hook["secret"].encode(), body, hashlib.sha256).hexdigest() req = urllib.request.Request(hook["url"], data=body, method="POST") req.add_header("Content-Type", "application/json") req.add_header("User-Agent", "MoonBite-Webhooks/1.0") req.add_header("X-MoonBite-Event", payload.get("event", "")) req.add_header("X-MoonBite-Delivery", uuid.uuid4().hex) req.add_header("X-MoonBite-Signature", f"sha256={sig}") try: # No-redirect opener: a 3xx to an internal host is not followed and is # treated as a failed delivery rather than an SSRF into the host. with _no_redirect_opener.open(req, timeout=config.WEBHOOK_TIMEOUT) as resp: return 200 <= resp.status < 300 except urllib.error.HTTPError as exc: return 200 <= exc.code < 300 except (urllib.error.URLError, TimeoutError, ConnectionError, OSError): return Falsedef _record_result(conn, hook_id: str, ok: bool) -> None: if ok: conn.execute( "UPDATE webhooks SET failures = 0, last_delivery = ? WHERE id = ?", (int(time.time()), hook_id), ) else: conn.execute( "UPDATE webhooks SET failures = failures + 1, " "active = CASE WHEN failures + 1 >= ? THEN 0 ELSE active END " "WHERE id = ?", (config.WEBHOOK_MAX_FAILURES, hook_id), ) conn.commit()def _fanout(conn, event: str, address: str | None, payload: dict) -> None: if address is None: rows = conn.execute( "SELECT * FROM webhooks WHERE active = 1 AND event = ?", (event,) ).fetchall() else: rows = conn.execute( "SELECT * FROM webhooks WHERE active = 1 AND event = 'address' " "AND address = ?", (address,), ).fetchall() for hook in rows: ok = _deliver(hook, payload) _record_result(conn, hook["id"], ok)# ------------------------------------------------------------------------- ## Chain scanning# ------------------------------------------------------------------------- #def _block_summary(block: dict) -> dict: txids = block.get("tx") or [] tx_count = block.get("nTx") if tx_count is None: tx_count = len(txids) return { "height": block.get("height"), "hash": block.get("hash"), "time": block.get("time"), "tx_count": tx_count, "size": block.get("size"), "previousblockhash": block.get("previousblockhash"), }def _vout_addresses(vout: dict): spk = vout.get("scriptPubKey", {}) or {} if spk.get("address"): return [spk["address"]] return list(spk.get("addresses") or [])def _vin_addresses(vin: dict): prevout = vin.get("prevout") or {} spk = prevout.get("scriptPubKey", {}) or {} if spk.get("address"): return [spk["address"]] return list(spk.get("addresses") or [])def _process_block(conn, client: RpcClient, height: int) -> None: block_hash = client.getblockhash(height) # verbosity 2 embeds full tx data so we can scan without extra RPC calls. block = client.getblock(block_hash, 2) summary = _block_summary(block) _fanout(conn, "new_block", None, { "event": "new_block", "block": summary, "time": int(time.time()), }) # Which addresses does *any* active subscription care about? watched = { row["address"] for row in conn.execute( "SELECT DISTINCT address FROM webhooks " "WHERE active = 1 AND event = 'address' AND address IS NOT NULL" ).fetchall() } if not watched: return txs = block.get("tx") or [] for tx in txs: if not isinstance(tx, dict): continue # verbosity<2 fell back to txids; skip address scan. received = {} for vout in tx.get("vout", []): val = float(vout.get("value", 0) or 0) for addr in _vout_addresses(vout): if addr in watched: received[addr] = received.get(addr, 0.0) + val spent = set() for vin in tx.get("vin", []): for addr in _vin_addresses(vin): if addr in watched: spent.add(addr) for addr in watched: got = received.get(addr) gave = addr in spent if got is None and not gave: continue direction = "both" if (got is not None and gave) else ( "receive" if got is not None else "send") _fanout(conn, "address", addr, { "event": "address", "address": addr, "txid": tx.get("txid"), "direction": direction, "value": got, "block": {"height": summary["height"], "hash": summary["hash"], "time": summary["time"]}, })# ------------------------------------------------------------------------- ## Poller (single instance across workers via a SQLite lock)# ------------------------------------------------------------------------- #def _acquire_lock(conn, owner: str) -> bool: now = int(time.time()) row = conn.execute( "SELECT owner, heartbeat FROM poller_lock WHERE id = 1" ).fetchone() holder, beat = row["owner"], row["heartbeat"] or 0 if holder in (None, "", owner) or (now - beat) > _LOCK_STALE_SECONDS: # Claim only if the row still looks the way we just read it (optimistic). cur = conn.execute( "UPDATE poller_lock SET owner = ?, heartbeat = ? " "WHERE id = 1 AND (owner IS ? OR owner = ? OR heartbeat = ?)", (owner, now, holder, holder if holder else "", beat), ) conn.commit() return cur.rowcount == 1 return Falsedef _heartbeat(conn, owner: str) -> bool: now = int(time.time()) cur = conn.execute( "UPDATE poller_lock SET heartbeat = ? WHERE id = 1 AND owner = ?", (now, owner), ) conn.commit() return cur.rowcount == 1def _poll_once(conn, client: RpcClient, owner: str) -> None: if not _heartbeat(conn, owner): # Lost the lock (another worker took over); try to reclaim next tick. if not _acquire_lock(conn, owner): return tip = client.getblockcount() last = _get_state(conn, "last_height") if last is None: # First run: start from the tip so we don't replay all of history. _set_state(conn, "last_height", tip) conn.commit() return last = int(last) if tip <= last: return start = last + 1 if tip - last > config.WEBHOOK_MAX_CATCHUP: # Bound catch-up after a long outage; skip the older gap. start = tip - config.WEBHOOK_MAX_CATCHUP + 1 for height in range(start, tip + 1): _process_block(conn, client, height) _set_state(conn, "last_height", height) conn.commit()def _poller_loop() -> None: owner = uuid.uuid4().hex interval = max(2.0, config.WEBHOOK_POLL_INTERVAL) conn = _connect() try: _acquire_lock(conn, owner) while True: try: client = RpcClient() if client.is_demo(): # No live chain to watch; idle but keep the lock warm. _heartbeat(conn, owner) else: _poll_once(conn, client, owner) except (RPCError, RPCConnectionError): pass except Exception: # Never let the poller thread die on an unexpected error. pass time.sleep(interval) finally: conn.close()_started = False_start_lock = threading.Lock()def start_poller() -> None: """Start the background poller once per process.""" global _started if not config.WEBHOOKS_ENABLED: return with _start_lock: if _started: return init_db() thread = threading.Thread(target=_poller_loop, name="webhook-poller", daemon=True) thread.start() _started = True