"""Blocklist reconcile pass. The bot's `signup_ip.ipblock_registered` / `signup_email.domain_blocked` flags only ever meant "we successfully posted this block once". They say nothing about whether the block is still there. A moderator clearing entries by hand (24 bot-registered `/32` rows were removed on 2026-07-21), or an `expires_at` elapsing, leaves the bot's record claiming blocks that no longer block anything — and the public /api/blocks endpoint publishing them. This pass closes that gap: it fetches Mastodon's live `Admin::IpBlock` and `Admin::EmailDomainBlock` lists, and flips `block_active` off for any of our rows whose block is gone (or has expired), back on for any that reappeared. The API then serves only entries that are actually in force. Run on a schedule (cron on admin.yttrx.com, see CLAUDE.md): docker exec yttrx-welcomebot python -m app.reconcile_blocks Requires ABUSE_BOT_TOKEN to carry `admin:read:ip_blocks` and `admin:read:email_domain_blocks` in addition to its existing scopes. Without them the fetch 403s and the pass exits without changing anything — rows keep whatever state they had, so a missing scope degrades to the old (over- reporting) behaviour rather than silently emptying the published list. `--dry-run` reports what would change without writing. """ from __future__ import annotations import argparse import ipaddress import logging import sys from datetime import datetime, timezone import httpx from app.main import ( MASTODON_BASE_URL, _admin_headers, _db, ) log = logging.getLogger("welcomebot.reconcile_blocks") def _fetch_all(path: str) -> list[dict] | None: """Page through an admin list endpoint. Returns None on failure, which the caller treats as "don't touch anything" — an outage or a missing scope must never be read as "every block was removed".""" results: list[dict] = [] url = f"{MASTODON_BASE_URL}{path}?limit=200" try: while url: resp = httpx.get(url, headers=_admin_headers(), timeout=30.0) if resp.status_code in (401, 403): log.error("%s returned %d — ABUSE_BOT_TOKEN is missing the matching " "admin read scope; leaving all rows untouched", path, resp.status_code) return None resp.raise_for_status() page = resp.json() if not page: break results.extend(page) url = resp.links.get("next", {}).get("url") except (httpx.HTTPError, ValueError) as exc: log.error("failed to fetch %s: %s — leaving all rows untouched", path, exc) return None return results def _expired(entry: dict) -> bool: raw = entry.get("expires_at") if not raw: return False try: when = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError: return False if when.tzinfo is None: when = when.replace(tzinfo=timezone.utc) return when <= datetime.now(timezone.utc) def live_ip_cidrs() -> set[str] | None: """Normalised CIDR strings of every non-expired Admin::IpBlock.""" entries = _fetch_all("/api/v1/admin/ip_blocks") if entries is None: return None cidrs = set() for entry in entries: if _expired(entry): continue raw = entry.get("ip") or "" try: cidrs.add(str(ipaddress.ip_network(raw, strict=False))) except ValueError: log.warning("skipping unparseable live ip_block %r", raw) return cidrs def live_email_domains() -> set[str] | None: entries = _fetch_all("/api/v1/admin/email_domain_blocks") if entries is None: return None return {(e.get("domain") or "").strip().lower() for e in entries if e.get("domain")} def _normalise(cidr: str) -> str: try: return str(ipaddress.ip_network(cidr, strict=False)) except ValueError: return cidr def reconcile(dry_run: bool = False) -> int: """Sync block_active against Mastodon. Returns the number of rows whose state changed (or would change, under dry_run).""" live_ips = live_ip_cidrs() live_domains = live_email_domains() if live_ips is None and live_domains is None: log.error("both fetches failed — nothing reconciled") return 0 changed = 0 now = datetime.now(timezone.utc).replace(microsecond=0).isoformat() if live_ips is not None: with _db() as conn: rows = conn.execute( "SELECT account_id, ip, block_cidr, block_active " "FROM signup_ip WHERE ipblock_registered = 1" ).fetchall() updates = [] for account_id, ip, block_cidr, active in rows: cidr = block_cidr or _single_cidr(ip or "") present = bool(cidr) and _normalise(cidr) in live_ips if bool(active) != present: changed += 1 log.info("ip_block %s: %s -> %s", cidr, "active" if active else "inactive", "active" if present else "inactive") updates.append((int(present), now, account_id)) if not dry_run and updates: with _db() as conn: conn.executemany( "UPDATE signup_ip SET block_active = ?, block_checked_at = ? " "WHERE account_id = ?", updates) log.info("ip_blocks: %d live, %d bot rows checked", len(live_ips), len(rows)) if live_domains is not None: with _db() as conn: rows = conn.execute( "SELECT account_id, domain, block_active " "FROM signup_email WHERE domain_blocked = 1" ).fetchall() updates = [] for account_id, domain, active in rows: present = bool(domain) and domain.strip().lower() in live_domains if bool(active) != present: changed += 1 log.info("email_domain_block %s: %s -> %s", domain, "active" if active else "inactive", "active" if present else "inactive") updates.append((int(present), now, account_id)) if not dry_run and updates: with _db() as conn: conn.executemany( "UPDATE signup_email SET block_active = ?, block_checked_at = ? " "WHERE account_id = ?", updates) log.info("email_domain_blocks: %d live, %d bot rows checked", len(live_domains), len(rows)) log.info("%s%d row(s) changed state", "[DRY-RUN] " if dry_run else "", changed) return changed def _single_cidr(ip: str) -> str: try: addr = ipaddress.ip_address(ip) except ValueError: return "" return f"{addr}/{32 if addr.version == 4 else 128}" def main() -> int: logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s", ) ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) ap.add_argument("--dry-run", action="store_true", help="report what would change without writing") args = ap.parse_args() reconcile(dry_run=args.dry_run) return 0 if __name__ == "__main__": sys.exit(main())