Add a reputation query API for pre-signup blocking

Everything the bot does today is reactive: Mastodon creates the account,
the account.created webhook arrives, we classify the signup IP and email
domain, and then we block the range and suspend what already exists.

POST /api/reputation is the pre-emptive half. A caller asks about an IP
and/or email domain; we classify it, record the answer, register the
matching Admin::IpBlock / EmailDomainBlock, and only then return. A
caller that runs its own block lookup immediately afterwards therefore
finds a block that did not exist when its request started. The intended
caller is Mastodon's signup path, which does exactly that.

It writes, so unlike /api/blocks it is authenticated and refuses to serve
without a key (REPUTATION_API_KEY): an unauthenticated endpoint that
installs IP blocks is a denial-of-service lever, so there is no open mode.

Notes on the pieces that aren't obvious:

- REPUTATION_API_MAX_LOOKUPS_PER_HOUR caps *fresh* provider lookups. The
  caller hits this on every view of the signup form, not just on
  submission, so without a ceiling a crawler walking /auth/sign_up from a
  new range each time would set our ipapi.is bill. Over budget we answer
  "unknown" and install nothing -- never "flagged". Cache hits are free
  and uncounted, and the webhook path never spends from this budget.

- Blocks installed here are recorded in their own table, keyed by the
  CIDR/domain rather than by account (there is no account yet), and are
  folded into /api/blocks and the reconcile pass so neither one
  under-reports as blocking shifts to the pre-signup path. The published
  entry shape is unchanged; a range blocked before anyone signed up from
  it simply reads "signups": 0.

- A repeat query for an already-installed block skips the Mastodon round
  trip. This runs on a live signup request, so that is saved latency.

.env.example

@@ -258,3 +258,59 @@ BLOCKS_API_PUBLIC_URL=https://hooks.yttrx.com
 # changes when a signup is flagged, so a few minutes of client/intermediate
 # caching costs nothing.
 BLOCKS_API_CACHE_SECONDS=300
+
+# ---------------------------------------------------------------------------
+# Reputation query API (pre-signup pre-blocking)
+# ---------------------------------------------------------------------------
+# POST /api/reputation turns the bot from reactive into pre-emptive. Mastodon
+# calls it during signup, immediately BEFORE running its own IpBlock /
+# EmailDomainBlock lookup:
+#
+#   POST /api/reputation            {"ip": "...", "email": "...", "source": "..."}
+#   X-Reputation-Key: <REPUTATION_API_KEY>
+#
+# The handler classifies the IP (ipapi.is) and/or the email's domain
+# (check-mail.org), records the answer, registers the matching Admin::IpBlock
+# / EmailDomainBlock, and only THEN returns — so the lookup Mastodon makes
+# microseconds later finds the block this call just installed and rejects the
+# signup outright, instead of the bot cleaning up an account that already
+# exists. Both request fields are optional (the IP choke point has no address
+# to offer); at least one is required.
+#
+# The caller is the `call-welcomebot-reputation-api` patch on yttrx's Mastodon
+# fork — see yttrx-documentation/mastodon-fork-patches.md. It fails open: if
+# this endpoint is slow, down or disabled, Mastodon proceeds with its own
+# lookup exactly as before.
+#
+# Unlike the blocklist API above this one WRITES (it installs blocks), so it
+# is authenticated and refuses to serve without a key.
+REPUTATION_API_ENABLED=false
+
+# Shared secret the caller presents as the X-Reputation-Key header; must match
+# WELCOMEBOT_REPUTATION_KEY on the Mastodon side. Generate with
+# `openssl rand -hex 32`. Empty = the route 404s even with ENABLED=true: an
+# unauthenticated endpoint that installs IP blocks is a denial-of-service
+# lever, so there is deliberately no "open" mode.
+REPUTATION_API_KEY=
+
+# Whether a flagged lookup actually registers the block, or only classifies,
+# records and answers. false is the dry-run equivalent for this path — the way
+# to watch what it *would* pre-block before letting it act. The per-signal
+# IP_SCRUTINY_DRY_RUN / CHECK_MAIL_DRY_RUN and IP_SCRUTINY_AUTO_IPBLOCK /
+# CHECK_MAIL_AUTO_DOMAIN_BLOCK switches still apply on top of this.
+REPUTATION_API_INSTALL_BLOCKS=true
+
+# Ceiling on *fresh* (cache-missing) provider lookups per rolling hour.
+# Mastodon calls this on every view of the signup form, not just on
+# submission, so without a ceiling a crawler walking /auth/sign_up from a new
+# range each time would set the ipapi.is / check-mail.org bill. Over budget an
+# uncached value answers "unknown" and installs nothing — never "flagged".
+# Cache hits are free and are never counted, and the webhook path never spends
+# from this budget (account.created is authoritative and must not be starved).
+# 0 = no ceiling.
+REPUTATION_API_MAX_LOOKUPS_PER_HOUR=500
+
+# DM MOD_ALERT_ACCT when this endpoint installs a block it has never installed
+# before. Bounded by how many genuinely new bad ranges/domains appear, not by
+# request volume: repeat hits on an already-installed block are silent.
+REPUTATION_API_DM_MODERATOR=true

app/main.py

@@ -34,6 +34,16 @@ Alongside the webhook, an optional read-only blocklist API publishes what the
 bot has auto-blocked — ``GET /api/blocks``, ``/api/blocks/ip-ranges``,
 ``/api/blocks/email-domains``. It is unauthenticated and therefore off unless
 ``BLOCKS_API_ENABLED`` is set; see that config entry.
+
+Everything above is *reactive*: it acts on an account Mastodon has already
+created. ``POST /api/reputation`` is the pre-emptive counterpart — Mastodon
+asks about an IP/email domain during signup, before creating anything, and
+this classifies it, records it, and registers the Admin::IpBlock /
+EmailDomainBlock *before returning*, so the lookup Mastodon runs immediately
+afterwards rejects the signup outright. It writes, so it is authenticated
+(``X-Reputation-Key``) and off unless ``REPUTATION_API_ENABLED`` and
+``REPUTATION_API_KEY`` are both set. The caller is the
+``call-welcomebot-reputation-api`` patch on yttrx's Mastodon fork.
 """
 
 from __future__ import annotations
@@ -45,12 +55,14 @@ import logging
 import os
 import sqlite3
 import threading
+import time
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 
 import httpx
 from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request
 from fastapi.responses import JSONResponse, PlainTextResponse
+from pydantic import BaseModel
 
 logging.basicConfig(
     level=logging.INFO,
@@ -248,6 +260,54 @@ BLOCKS_API_PUBLIC_URL = os.environ.get(
 # intermediate cache hold it for a few minutes costs nothing.
 BLOCKS_API_CACHE_SECONDS = int(os.environ.get("BLOCKS_API_CACHE_SECONDS", "300"))
 
+# --- Reputation query API (pre-signup pre-blocking) -------------------------
+# POST /api/reputation lets Mastodon ask "what do you know about this IP /
+# email domain?" *before* it runs its own IpBlock / EmailDomainBlock lookup at
+# signup time. The handler classifies, records the answer, registers the block
+# on the Mastodon side, and only then returns — so the lookup Mastodon makes
+# immediately afterwards finds the block this call just installed. That turns
+# the bot from purely reactive (suspend after the account exists) into
+# pre-emptive: a bad range is rejected on first contact rather than cleaned up
+# afterwards.
+#
+# Unlike the blocklist API this endpoint *writes* (it installs blocks), so it
+# is authenticated and refuses to serve at all without a key. See the
+# yttrx patch `call-welcomebot-reputation-api` on the Mastodon fork for the
+# caller.
+REPUTATION_API_ENABLED = os.environ.get(
+    "REPUTATION_API_ENABLED", "false").lower() in ("1", "true", "yes")
+# Shared secret the caller presents as `X-Reputation-Key`. No key, no service:
+# an unauthenticated endpoint that installs IP blocks is a denial-of-service
+# lever, so an empty key disables the route entirely (404) rather than
+# defaulting to open.
+REPUTATION_API_KEY = os.environ.get("REPUTATION_API_KEY", "")
+# Whether a flagged lookup actually registers the block on the Mastodon side.
+# Off = classify, record and answer only, installing nothing — the dry-run
+# equivalent for this path, and the way to watch what it *would* block before
+# letting it act. The per-signal IP_SCRUTINY_DRY_RUN / CHECK_MAIL_DRY_RUN and
+# IP_SCRUTINY_AUTO_IPBLOCK / CHECK_MAIL_AUTO_DOMAIN_BLOCK switches still apply
+# on top of this, so a signal that is dry-run for the webhook path is dry-run
+# here too.
+REPUTATION_API_INSTALL_BLOCKS = os.environ.get(
+    "REPUTATION_API_INSTALL_BLOCKS", "true").lower() in ("1", "true", "yes")
+# Ceiling on *fresh* (cache-missing) external provider lookups this endpoint
+# may make per rolling hour. Mastodon calls it on every view of the signup
+# form, not just on submission, so a crawl of /auth/sign_up from many distinct
+# ranges would otherwise burn paid ipapi.is / check-mail.org quota at whatever
+# rate the crawler chooses. Over budget, an uncached value answers "unknown"
+# (never "flagged") and nothing is installed. Cache hits are free and never
+# counted; 0 disables the ceiling.
+#
+# The webhook path deliberately does NOT consult this budget — account.created
+# is low-volume, authoritative, and must never be starved by crawler traffic.
+REPUTATION_API_MAX_LOOKUPS_PER_HOUR = int(
+    os.environ.get("REPUTATION_API_MAX_LOOKUPS_PER_HOUR", "500"))
+# DM MOD_ALERT_ACCT when this endpoint installs a block it has never installed
+# before. Bounded by how many genuinely new bad ranges/domains show up, not by
+# request volume — repeat hits on an already-installed block are silent.
+REPUTATION_API_DM_MODERATOR = os.environ.get(
+    "REPUTATION_API_DM_MODERATOR", "true").lower() in ("1", "true", "yes")
+
 if not WEBHOOK_SECRET:
     log.warning("WEBHOOK_SECRET is empty — signature verification will reject all requests.")
 if not BOT_ACCESS_TOKEN:
@@ -256,6 +316,8 @@ if ABUSE_ENABLED and not ABUSE_BOT_TOKEN:
     log.warning("ABUSE_BOT_TOKEN is empty — report.created handling is disabled.")
 if CHECK_MAIL_ENABLED and not CHECK_MAIL_API_KEY:
     log.warning("CHECK_MAIL_API_KEY is empty — disposable-email signup scrutiny is disabled.")
+if REPUTATION_API_ENABLED and not REPUTATION_API_KEY:
+    log.warning("REPUTATION_API_KEY is empty — the reputation query API stays disabled.")
 
 app = FastAPI(title="yttrx welcome-bot + abuse-bot")
 
@@ -368,6 +430,49 @@ def _init_db() -> None:
             "  cached_at  TEXT DEFAULT CURRENT_TIMESTAMP"
             ")"
         )
+        # Audit log of every /api/reputation answer: what was asked about,
+        # what the providers said, and whether it led to a block. The webhook
+        # path has signup_ip/signup_email for this; a pre-signup query has no
+        # account to hang a row off, so it gets its own table. Never stores a
+        # full email address — the domain only, same rule as signup_email.
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS reputation_query ("
+            "  id         INTEGER PRIMARY KEY AUTOINCREMENT,"
+            "  kind       TEXT NOT NULL,"   # 'ip' | 'email'
+            "  value      TEXT NOT NULL,"   # the IP, or the email domain
+            "  classification TEXT,"
+            "  org        TEXT,"
+            "  risk       INTEGER,"
+            "  flagged    INTEGER,"
+            "  target     TEXT,"            # cidr / domain the block covers
+            "  blocked    INTEGER DEFAULT 0,"
+            "  source     TEXT,"            # caller-supplied tag
+            "  queried_at TEXT DEFAULT CURRENT_TIMESTAMP"
+            ")"
+        )
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS reputation_query_kind_value "
+            "ON reputation_query (kind, value)"
+        )
+        # One row per block installed by the query API, keyed by what is
+        # actually blocked. Two jobs: skip re-POSTing a block we already
+        # installed (Mastodon answers 422, but the round trip is on the signup
+        # request path), and give /api/blocks + the reconcile pass the same
+        # view of these as of the signup-derived ones.
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS reputation_block ("
+            "  kind       TEXT NOT NULL,"   # 'ip' | 'email'
+            "  target     TEXT NOT NULL,"   # cidr, or domain
+            "  classification TEXT,"
+            "  org        TEXT,"
+            "  queries    INTEGER DEFAULT 1,"
+            "  first_blocked_at TEXT DEFAULT CURRENT_TIMESTAMP,"
+            "  last_seen_at     TEXT DEFAULT CURRENT_TIMESTAMP,"
+            "  block_active     INTEGER DEFAULT 1,"
+            "  block_checked_at TEXT,"
+            "  PRIMARY KEY (kind, target)"
+            ")"
+        )
         conn.execute(
             "CREATE TABLE IF NOT EXISTS suspicious_watch ("
             "  account_id TEXT PRIMARY KEY,"
@@ -592,6 +697,63 @@ def start_suspicious_watch(account_id: str, acct: str, reasons: str,
         )
 
 
+def record_reputation_query(kind: str, value: str, classification: str, org: str,
+                            risk: int, flagged: bool, target: str, blocked: bool,
+                            source: str) -> None:
+    """Append one /api/reputation answer to the audit log."""
+    with _db() as conn:
+        conn.execute(
+            "INSERT INTO reputation_query "
+            "(kind, value, classification, org, risk, flagged, target, blocked, source) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (kind, value, classification, org, int(risk), int(flagged),
+             target or None, int(blocked), source or None),
+        )
+
+
+def reputation_block_installed(kind: str, target: str) -> bool:
+    """True if the query API already installed this block and the reconcile
+    pass hasn't since found it gone. Lets a repeat query skip the Mastodon
+    round trip — this runs on the signup request path, so a saved POST is
+    saved latency."""
+    with _db() as conn:
+        row = conn.execute(
+            "SELECT 1 FROM reputation_block "
+            "WHERE kind = ? AND target = ? AND block_active = 1",
+            (kind, target),
+        ).fetchone()
+        return row is not None
+
+
+def record_reputation_block(kind: str, target: str, classification: str, org: str) -> None:
+    """Record that this block is installed, or that an installed one was hit
+    again. Reactivates a row the reconcile pass had marked gone, same as
+    mark_ipblock_registered does for the signup-derived rows."""
+    with _db() as conn:
+        conn.execute(
+            "INSERT INTO reputation_block "
+            "(kind, target, classification, org) VALUES (?, ?, ?, ?) "
+            "ON CONFLICT(kind, target) DO UPDATE SET "
+            "  queries = queries + 1,"
+            "  last_seen_at = CURRENT_TIMESTAMP,"
+            "  block_active = 1,"
+            "  classification = COALESCE(excluded.classification, classification),"
+            "  org = COALESCE(excluded.org, org)",
+            (kind, target, classification or None, org or None),
+        )
+
+
+def reputation_blocks(kind: str) -> list[tuple[str, str, str]]:
+    """(target, first_blocked_at, last_seen_at) for every still-active block
+    the query API installed, newest first."""
+    with _db() as conn:
+        return conn.execute(
+            "SELECT target, first_blocked_at, last_seen_at FROM reputation_block "
+            "WHERE kind = ? AND block_active = 1 ORDER BY last_seen_at DESC",
+            (kind,),
+        ).fetchall()
+
+
 def _fallback_cidr(ip: str) -> str:
     """The single-address CIDR for ip, used for signup_ip rows written before
     block_cidr existed — that's the block that was registered for them."""
@@ -644,7 +806,8 @@ def blocked_ip_ranges(limit: int = 0) -> list[dict]:
             entry["signups"] += 1
             entry["first_blocked_at"] = classified_at
 
-    return _capped(list(grouped.values()), limit)
+    _merge_reputation_blocks(grouped, "ip", "cidr")
+    return _capped(_newest_first(grouped), limit)
 
 
 def blocked_email_domains(limit: int = 0) -> list[dict]:
@@ -682,7 +845,44 @@ def blocked_email_domains(limit: int = 0) -> list[dict]:
             entry["signups"] += 1
             entry["first_blocked_at"] = classified_at
 
-    return _capped(list(grouped.values()), limit)
+    _merge_reputation_blocks(grouped, "email", "domain")
+    return _capped(_newest_first(grouped), limit)
+
+
+def _merge_reputation_blocks(grouped: dict[str, dict], kind: str, key: str) -> None:
+    """Fold the query API's blocks into a signup-derived blocklist, in place.
+
+    Both paths register the same kind of Admin::IpBlock / EmailDomainBlock, so
+    the published list would silently under-report as more blocking moves to
+    the pre-signup path. Entries keep the existing keys — no contract change
+    for anyone already consuming this — which means `signups` continues to
+    count *signups* specifically. A range blocked before anyone signed up from
+    it therefore reads `"signups": 0`, which is exactly right: blocked, and
+    nothing got through.
+    """
+    for target, first_at, last_at in reputation_blocks(kind):
+        if not target:
+            continue
+        entry = grouped.get(target)
+        if entry is None:
+            grouped[target] = {
+                key: target,
+                "signups": 0,
+                "first_blocked_at": first_at,
+                "last_blocked_at": last_at,
+            }
+        else:
+            entry["first_blocked_at"] = min(
+                filter(None, (entry["first_blocked_at"], first_at)), default=None)
+            entry["last_blocked_at"] = max(
+                filter(None, (entry["last_blocked_at"], last_at)), default=None)
+
+
+def _newest_first(grouped: dict[str, dict]) -> list[dict]:
+    """Most recently blocked first. The signup rows arrive in that order
+    already, but merged-in query-API entries have their own timestamps, so the
+    combined list has to be re-sorted to keep the documented ordering true."""
+    return sorted(grouped.values(), key=lambda e: e["last_blocked_at"] or "", reverse=True)
 
 
 def _capped(items: list[dict], limit: int) -> list[dict]:
@@ -1148,7 +1348,8 @@ def classify_signup_ip(ip: str) -> tuple[str, str, bool, str]:
     return classification, intel["org"], bool(reasons), intel.get("cidr") or f"{ip}/32"
 
 
-def register_ip_block(ip: str, acct: str, org: str, cidr: str = "") -> bool:
+def register_ip_block(ip: str, acct: str, org: str, cidr: str = "",
+                      comment: str = "") -> bool:
     """Register a flagged signup's network in Mastodon's native Admin::IpBlock.
 
     cidr should be classify_signup_ip's block_cidr — the ASN/route-level
@@ -1157,6 +1358,10 @@ def register_ip_block(ip: str, acct: str, org: str, cidr: str = "") -> bool:
     Falls back to a single-address block if no cidr is given or ip itself
     doesn't parse (defensive only; classify_signup_ip always supplies one).
 
+    comment overrides the moderator-facing note stored with the block; the
+    reputation query API passes its own, since a pre-signup block has no
+    account to name.
+
     Returns True if the range is blocked on the Mastodon side afterwards
     (including the 422 "already blocked" case), so the caller only records it
     as blocked when it really is — /api/blocks publishes those rows.
@@ -1168,7 +1373,8 @@ def register_ip_block(ip: str, acct: str, org: str, cidr: str = "") -> bool:
             log.warning("skipping ip_block registration for unparseable ip=%r (acct=%s)", ip, acct)
             return False
         cidr = f"{ip}/{prefix_len}"
-    comment = f"welcomebot: flagged signup @{acct} ({org or 'unknown org'})"[:200]
+    comment = (comment or
+               f"welcomebot: flagged signup @{acct} ({org or 'unknown org'})")[:200]
     try:
         resp = httpx.post(
             f"{MASTODON_BASE_URL}/api/v1/admin/ip_blocks",
@@ -1186,6 +1392,181 @@ def register_ip_block(ip: str, acct: str, org: str, cidr: str = "") -> bool:
     return True
 
 
+# --- Reputation query API (pre-signup pre-blocking) -------------------------
+# The pre-emptive half of the bot: Mastodon asks about an IP/email *before*
+# creating anything, and this installs the block in time for Mastodon's own
+# lookup to find it. Everything here runs on a live signup request, so it is
+# written to be fast on the common path (sqlite cache hit) and to fail open —
+# never to make a signup wait on, or fail because of, a provider outage.
+
+_lookup_budget_lock = threading.Lock()
+_lookup_budget: list[float] = []
+
+
+def _external_lookup_allowed() -> bool:
+    """Claim one slot from the rolling hourly budget for a fresh provider
+    lookup. Returns False when the budget is spent, in which case the caller
+    must answer "unknown" rather than making the call.
+
+    Only the query API spends budget — see REPUTATION_API_MAX_LOOKUPS_PER_HOUR.
+    Mastodon calls it on every *view* of the signup form, so without a ceiling
+    a crawler walking /auth/sign_up from a fresh range each time sets our paid
+    provider spend.
+    """
+    if REPUTATION_API_MAX_LOOKUPS_PER_HOUR <= 0:
+        return True
+    now = time.monotonic()
+    with _lookup_budget_lock:
+        cutoff = now - 3600
+        while _lookup_budget and _lookup_budget[0] < cutoff:
+            _lookup_budget.pop(0)
+        if len(_lookup_budget) >= REPUTATION_API_MAX_LOOKUPS_PER_HOUR:
+            return False
+        _lookup_budget.append(now)
+        return True
+
+
+def _reputation_ip(ip: str, source: str) -> dict | None:
+    """Classify one IP for the query API and install the block if it's flagged.
+
+    Returns None when the IP side is switched off or the address doesn't parse.
+    """
+    if not (IP_SCRUTINY_ENABLED and ip):
+        return None
+    try:
+        ipaddress.ip_address(ip)
+    except ValueError:
+        log.warning("reputation query: unparseable ip=%r", ip)
+        return None
+
+    cached = cached_ip_intel(ip) is not None
+    if not cached and not _external_lookup_allowed():
+        # Budget spent. Answer "unknown", never "flagged" — the whole point of
+        # the ceiling is to stop paying, and guessing in either direction here
+        # would either block innocents or waste the guard.
+        log.warning("reputation query: hourly lookup budget spent, ip=%s answered unknown", ip)
+        bounds = _ip_range_bounds(ip)
+        return {
+            "ip": ip, "classification": "unknown", "org": "", "flagged": False,
+            "cidr": bounds[3] if bounds else "", "blocked": False,
+            "cached": False, "deferred": True,
+        }
+
+    classification, org, flagged, cidr = classify_signup_ip(ip)
+    blocked = False
+
+    if flagged:
+        installable = (REPUTATION_API_INSTALL_BLOCKS and IP_SCRUTINY_AUTO_IPBLOCK
+                       and not IP_SCRUTINY_DRY_RUN)
+        if not installable:
+            log.info("reputation query: would block ip=%s cidr=%s (%s) — installing is off",
+                     ip, cidr, classification)
+        elif reputation_block_installed("ip", cidr):
+            # Already ours and still in force; Mastodon's own lookup will find
+            # it. Skip the round trip, just count the hit.
+            record_reputation_block("ip", cidr, classification, org)
+            blocked = True
+        elif register_ip_block(ip, "", org, cidr, comment=(
+                f"welcomebot: pre-signup block, {classification} "
+                f"({org or 'unknown org'})")):
+            record_reputation_block("ip", cidr, classification, org)
+            blocked = True
+            log.warning("reputation query: pre-blocked ip=%s cidr=%s classification=%s org=%s source=%s",
+                        ip, cidr, classification, org, source or "-")
+            if REPUTATION_API_DM_MODERATOR:
+                dm_moderator(
+                    f"🛡️ Pre-blocked {cidr} before signup — {classification} "
+                    f"({org or 'unknown org'})."
+                )
+
+    record_reputation_query("ip", ip, classification, org, 0, flagged, cidr, blocked, source)
+    return {
+        "ip": ip, "classification": classification, "org": org, "flagged": flagged,
+        "cidr": cidr, "blocked": blocked, "cached": cached, "deferred": False,
+    }
+
+
+def _reputation_email(email: str, source: str) -> dict | None:
+    """Classify one signup email's *domain* for the query API and install the
+    domain block if it's flagged. The full address is never stored or logged —
+    same rule as the webhook path (see classify_email_domain).
+
+    Returns None when the email side is switched off or there's no domain.
+    """
+    if not (CHECK_MAIL_ENABLED and CHECK_MAIL_API_KEY and email and "@" in email):
+        return None
+    domain = email.rsplit("@", 1)[-1].strip().lower()
+    if not domain:
+        return None
+
+    cached = cached_check_mail(domain) is not None
+    if not cached and not _external_lookup_allowed():
+        log.warning("reputation query: hourly lookup budget spent, domain=%s answered unknown",
+                    domain)
+        return {
+            "domain": domain, "disposable": False, "risk": 0, "flagged": False,
+            "blocked": False, "cached": False, "deferred": True,
+        }
+
+    is_disposable, risk = classify_email_domain(domain)
+    flagged = is_disposable or risk >= CHECK_MAIL_RISK_THRESHOLD
+    blocked = False
+
+    if flagged:
+        installable = (REPUTATION_API_INSTALL_BLOCKS and CHECK_MAIL_AUTO_DOMAIN_BLOCK
+                       and not CHECK_MAIL_DRY_RUN)
+        if not installable:
+            log.info("reputation query: would block domain=%s (disposable=%s risk=%d) "
+                     "— installing is off", domain, is_disposable, risk)
+        elif reputation_block_installed("email", domain):
+            record_reputation_block("email", domain, "", "")
+            blocked = True
+        elif register_email_domain_block(domain, "(pre-signup)"):
+            record_reputation_block("email", domain, "", "")
+            blocked = True
+            log.warning("reputation query: pre-blocked domain=%s disposable=%s risk=%d source=%s",
+                        domain, is_disposable, risk, source or "-")
+            if REPUTATION_API_DM_MODERATOR:
+                dm_moderator(
+                    f"🛡️ Pre-blocked email domain {domain} before signup "
+                    f"(disposable={is_disposable}, risk={risk})."
+                )
+
+    classification = "disposable" if is_disposable else ("high-risk" if flagged else "clean")
+    record_reputation_query("email", domain, classification, "", risk, flagged,
+                            domain, blocked, source)
+    return {
+        "domain": domain, "disposable": is_disposable, "risk": risk, "flagged": flagged,
+        "blocked": blocked, "cached": cached, "deferred": False,
+    }
+
+
+def reputation_check(ip: str = "", email: str = "", source: str = "") -> dict:
+    """Classify an IP and/or email domain, install any block they earn, and
+    report what happened. The write half of the bot's pre-emptive path.
+
+    Each side is independent: either may be absent from the request, switched
+    off by config, or fail its provider lookup, without affecting the other.
+    A provider failure classifies "unknown"/not-flagged (classify_signup_ip and
+    classify_email_domain both swallow their own errors) — we never block on
+    our own lookup errors.
+
+    `blocked` in the response means "something is blocked on the Mastodon side
+    as a result of this call", which is what the caller acts on: it is true
+    both for a block installed just now and for one we installed earlier that
+    is still in force.
+    """
+    result = {
+        "ip": _reputation_ip(ip.strip(), source),
+        "email": _reputation_email(email.strip(), source),
+        "checked_at": _now_iso(),
+    }
+    sides = [side for side in (result["ip"], result["email"]) if side]
+    result["flagged"] = any(side["flagged"] for side in sides)
+    result["blocked"] = any(side["blocked"] for side in sides)
+    return result
+
+
 def count_distinct_reporters(target_id: str) -> int:
     """Count distinct accounts with an OPEN report against the target.
 
@@ -1507,6 +1888,52 @@ def api_blocks_email_domains(limit: int = 0, format: str = "json"):
     })
 
 
+# --- Reputation query API ---
+# POST /api/reputation — authenticated, and it WRITES (installs blocks), which
+# is what separates it from the read-only /api/blocks above. Mastodon calls it
+# from the signup path immediately before its own IpBlock/EmailDomainBlock
+# lookup; see the `call-welcomebot-reputation-api` patch on the fork.
+
+
+class ReputationRequest(BaseModel):
+    """What Mastodon knows at the moment it asks. Both fields are optional —
+    the IP-block choke point has no email address to offer, the email
+    validator has both."""
+
+    ip: str = ""
+    email: str = ""
+    source: str = ""
+
+
+def _reputation_guard(key: str | None) -> None:
+    """404 unless the endpoint is switched on AND has a key (a route that
+    installs blocks must never be reachable unauthenticated); 401 on a bad
+    key. Disabled is indistinguishable from absent, same as /api/blocks."""
+    if not (REPUTATION_API_ENABLED and REPUTATION_API_KEY):
+        raise HTTPException(status_code=404, detail="Not Found")
+    if not key or not hmac.compare_digest(key, REPUTATION_API_KEY):
+        raise HTTPException(status_code=401, detail="invalid key")
+
+
+@app.post("/api/reputation")
+def api_reputation(
+    body: ReputationRequest,
+    x_reputation_key: str | None = Header(default=None),
+):
+    """Look up an IP and/or email domain, record the answer, install whatever
+    block it earns, and return — in that order, so that a caller running its
+    own block lookup straight afterwards sees the block from this call.
+
+    Declared as a plain `def` on purpose: FastAPI runs those in a threadpool,
+    so the blocking provider calls underneath never tie up the event loop and
+    stall webhook deliveries.
+    """
+    _reputation_guard(x_reputation_key)
+    if not body.ip and not body.email:
+        raise HTTPException(status_code=400, detail="ip or email is required")
+    return reputation_check(ip=body.ip, email=body.email, source=body.source)
+
+
 @app.post("/webhook")
 async def webhook(
     request: Request,

app/reconcile_blocks.py

@@ -12,6 +12,10 @@ This pass closes that gap: it fetches Mastodon's live `Admin::IpBlock` and
 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.
 
+It covers both sources of blocks: the signup-derived ones (`signup_ip` /
+`signup_email`, keyed by account) and the pre-signup ones the reputation query
+API installs (`reputation_block`, keyed by the blocked CIDR/domain itself).
+
 Run on a schedule (cron on admin.yttrx.com, see CLAUDE.md):
 
     docker exec yttrx-welcomebot python -m app.reconcile_blocks
@@ -150,6 +154,8 @@ def reconcile(dry_run: bool = False) -> int:
                     "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))
+        changed += _reconcile_reputation(
+            "ip", live_ips, now, dry_run, normalise=True)
 
     if live_domains is not None:
         with _db() as conn:
@@ -175,11 +181,50 @@ def reconcile(dry_run: bool = False) -> int:
                     "WHERE account_id = ?", updates)
         log.info("email_domain_blocks: %d live, %d bot rows checked",
                  len(live_domains), len(rows))
+        changed += _reconcile_reputation(
+            "email", live_domains, now, dry_run, normalise=False)
 
     log.info("%s%d row(s) changed state", "[DRY-RUN] " if dry_run else "", changed)
     return changed
 
 
+def _reconcile_reputation(kind: str, live: set[str], now: str, dry_run: bool,
+                          normalise: bool) -> int:
+    """Same reconcile, for the blocks the query API installed pre-signup.
+
+    Those live in `reputation_block` keyed by what is actually blocked (a CIDR
+    or a domain) rather than by account, since there is no account yet when
+    they're written — but they are the same Admin::IpBlock /
+    EmailDomainBlock records, so they go stale the same ways and /api/blocks
+    publishes them alongside the signup-derived ones.
+    """
+    with _db() as conn:
+        rows = conn.execute(
+            "SELECT target, block_active FROM reputation_block WHERE kind = ?",
+            (kind,),
+        ).fetchall()
+
+    changed = 0
+    updates = []
+    for target, active in rows:
+        key = _normalise(target) if normalise else (target or "").strip().lower()
+        present = bool(target) and key in live
+        if bool(active) != present:
+            changed += 1
+            log.info("reputation %s block %s: %s -> %s", kind, target,
+                     "active" if active else "inactive",
+                     "active" if present else "inactive")
+        updates.append((int(present), now, kind, target))
+
+    if not dry_run and updates:
+        with _db() as conn:
+            conn.executemany(
+                "UPDATE reputation_block SET block_active = ?, block_checked_at = ? "
+                "WHERE kind = ? AND target = ?", updates)
+    log.info("reputation %s blocks: %d rows checked", kind, len(rows))
+    return changed
+
+
 def _single_cidr(ip: str) -> str:
     try:
         addr = ipaddress.ip_address(ip)

nginx/hooks.yttrx.com.conf

@@ -43,6 +43,28 @@ server {
         proxy_set_header X-Forwarded-Proto $scheme;
     }
 
+    # Reputation query API (POST /api/reputation). Called by Mastodon on
+    # mammut during signup, so it has to be reachable from off-box — unlike
+    # /webhook this one is authenticated with a shared secret
+    # (X-Reputation-Key) rather than an HMAC over the body, and it is gated in
+    # the app by REPUTATION_API_ENABLED + REPUTATION_API_KEY: with either
+    # unset the app 404s this path.
+    #
+    # It sits on a live signup request, so the proxy timeouts are short: the
+    # caller gives up after ~2s anyway and proceeds without an answer, and a
+    # request left hanging here would just hold a Puma thread on mammut for no
+    # benefit.
+    location = /api/reputation {
+        proxy_pass http://127.0.0.1:8087;
+        proxy_set_header Host              $host;
+        proxy_set_header X-Real-IP         $remote_addr;
+        proxy_set_header X-Forwarded-For   $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+        proxy_connect_timeout 2s;
+        proxy_send_timeout    5s;
+        proxy_read_timeout    5s;
+    }
+
     location / {
         return 404;
     }

test_local.py

@@ -979,6 +979,209 @@ def reconcile_blocks_tests():
     assert "198.18.5.0/24" in active_ips(), "re-registering must reactivate the row"
 
 
+def reputation_api_tests():
+    """Drive POST /api/reputation — the pre-signup query path Mastodon calls
+    before its own IpBlock/EmailDomainBlock lookup. The load-bearing property
+    is ordering: the block must be installed by the time the call returns,
+    because the caller looks it up microseconds later."""
+    dms = []
+    ipblocks = []
+    domain_blocks = []
+    main.dm_moderator = lambda message: dms.append(message)
+    main.register_ip_block = (
+        lambda ip, acct, org, cidr="", comment="":
+        ipblocks.append((ip, cidr, comment)) or True)
+    main.register_email_domain_block = (
+        lambda domain, acct: domain_blocks.append((domain, acct)) or True)
+
+    classifications = {
+        "203.0.113.50": ("clean", "Example Residential ISP", False, "203.0.113.50/32"),
+        "198.51.100.50": ("datacenter", "Example Cloud Inc", True, "198.51.100.0/24"),
+        "198.51.100.51": ("datacenter", "Example Cloud Inc", True, "198.51.100.0/24"),
+        "198.51.100.60": ("proxy", "Example Proxy Co", True, "198.51.100.64/26"),
+    }
+    main.classify_signup_ip = lambda ip: classifications[ip]
+    main.classify_email_domain = lambda domain: {
+        "rep-clean.test": (False, 0),
+        "rep-spam.test": (True, 99),
+    }[domain]
+
+    main.IP_SCRUTINY_ENABLED = True
+    main.IP_SCRUTINY_DRY_RUN = False
+    main.IP_SCRUTINY_AUTO_IPBLOCK = True
+    main.CHECK_MAIL_ENABLED = True
+    main.CHECK_MAIL_API_KEY = "dummy"
+    main.CHECK_MAIL_DRY_RUN = False
+    main.CHECK_MAIL_AUTO_DOMAIN_BLOCK = True
+    main.CHECK_MAIL_RISK_THRESHOLD = 80
+    main.REPUTATION_API_INSTALL_BLOCKS = True
+    main.REPUTATION_API_DM_MODERATOR = True
+    main.REPUTATION_API_MAX_LOOKUPS_PER_HOUR = 0  # no ceiling except where tested
+
+    def ask(payload, key="rep-secret"):
+        headers = {"X-Reputation-Key": key} if key is not None else {}
+        return client.post("/api/reputation", json=payload, headers=headers)
+
+    # A. The gate. Disabled, or enabled with no key configured, is a 404 —
+    #    a route that installs blocks must never be reachable unauthenticated,
+    #    and "off" should look like "not a route".
+    main.REPUTATION_API_ENABLED = False
+    main.REPUTATION_API_KEY = "rep-secret"
+    assert ask({"ip": "203.0.113.50"}).status_code == 404
+
+    main.REPUTATION_API_ENABLED = True
+    main.REPUTATION_API_KEY = ""
+    assert ask({"ip": "203.0.113.50"}).status_code == 404, \
+        "enabled without a key must stay closed, not open"
+
+    main.REPUTATION_API_KEY = "rep-secret"
+    assert ask({"ip": "203.0.113.50"}, key="wrong").status_code == 401
+    assert ask({"ip": "203.0.113.50"}, key=None).status_code == 401
+
+    # B. Something to look up is required.
+    assert ask({}).status_code == 400
+    assert ask({"source": "registration"}).status_code == 400
+
+    # C. A clean IP: answered, recorded, nothing installed.
+    ipblocks.clear(); dms.clear()
+    r = ask({"ip": "203.0.113.50", "source": "registration"})
+    assert r.status_code == 200, r.status_code
+    body = r.json()
+    assert body["flagged"] is False and body["blocked"] is False, body
+    assert body["ip"]["classification"] == "clean"
+    assert body["email"] is None, "no email asked about, no email answer"
+    assert ipblocks == [], "a clean IP must not be blocked"
+    assert dms == [], "a clean IP must not wake a moderator"
+
+    # D. THE POINT OF THE FEATURE: a flagged IP is blocked *during* the call,
+    #    so the caller's own IpBlock lookup right afterwards finds it.
+    r = ask({"ip": "198.51.100.50", "source": "registration"})
+    body = r.json()
+    assert body["flagged"] is True and body["blocked"] is True, body
+    assert body["ip"]["cidr"] == "198.51.100.0/24"
+    assert len(ipblocks) == 1, ipblocks
+    assert ipblocks[0][1] == "198.51.100.0/24", "the range is blocked, not the address"
+    assert "pre-signup" in ipblocks[0][2], "the block says why it exists"
+    assert any("198.51.100.0/24" in d for d in dms), dms
+    assert main.reputation_block_installed("ip", "198.51.100.0/24")
+
+    # E. A second address in the same range: still blocked, but we don't
+    #    re-POST a block we already installed — this runs on the signup
+    #    request path, so the saved round trip is saved latency.
+    dms.clear()
+    r = ask({"ip": "198.51.100.51"})
+    assert r.json()["blocked"] is True, "an already-installed block still counts as blocked"
+    assert len(ipblocks) == 1, "an already-installed block must not be re-POSTed"
+    assert dms == [], "only a newly-installed block wakes a moderator"
+
+    # F. REPUTATION_API_INSTALL_BLOCKS=false classifies and records but
+    #    installs nothing — the dry-run equivalent for this path.
+    main.REPUTATION_API_INSTALL_BLOCKS = False
+    r = ask({"ip": "198.51.100.60"})
+    body = r.json()
+    assert body["flagged"] is True, body
+    assert body["blocked"] is False, "install-blocks off must not install"
+    assert len(ipblocks) == 1, ipblocks
+    main.REPUTATION_API_INSTALL_BLOCKS = True
+
+    # G. The email side, and the standing rule that a full address is never
+    #    stored, logged or echoed — only its domain.
+    domain_blocks.clear(); dms.clear()
+    r = ask({"ip": "203.0.113.50", "email": "[email protected]", "source": "sign_up"})
+    body = r.json()
+    assert body["email"]["domain"] == "rep-spam.test", "domain is normalised to lower case"
+    assert body["email"]["flagged"] is True and body["email"]["blocked"] is True
+    assert body["blocked"] is True, "either side flagging makes the whole answer blocked"
+    assert body["ip"]["flagged"] is False, "the two sides are independent"
+    assert domain_blocks and domain_blocks[0][0] == "rep-spam.test", domain_blocks
+    assert "SpamGuy" not in r.text and "spamguy" not in r.text.lower(), \
+        "the local part must never come back in the response"
+    with main._db() as conn:
+        stored = conn.execute(
+            "SELECT value FROM reputation_query WHERE kind = 'email'").fetchall()
+    assert all("@" not in v[0] for v in stored), \
+        "reputation_query must store domains, never addresses"
+
+    # H. A clean domain is answered but not blocked.
+    domain_blocks.clear()
+    r = ask({"email": "[email protected]"})
+    assert r.json()["email"]["flagged"] is False
+    assert r.json()["blocked"] is False
+    assert domain_blocks == [], domain_blocks
+
+    # I. The hourly ceiling on fresh provider lookups. Over budget, an
+    #    uncached value answers "unknown" and blocks nothing — the guard
+    #    exists to stop paying, so it must never guess "flagged" instead.
+    #    Addresses here must fall outside every range cached by earlier tests,
+    #    or they'd be cache hits and never reach the budget at all.
+    main._lookup_budget.clear()
+    main.REPUTATION_API_MAX_LOOKUPS_PER_HOUR = 1
+    classifications["198.51.102.10"] = ("datacenter", "Example Cloud Inc", True,
+                                        "198.51.102.0/25")
+    classifications["198.51.102.200"] = ("datacenter", "Other Cloud Inc", True,
+                                         "198.51.102.128/25")
+    before = len(ipblocks)
+    first = ask({"ip": "198.51.102.10"}).json()["ip"]
+    assert first["cached"] is False and first["deferred"] is False, first
+    assert len(ipblocks) == before + 1, "the in-budget lookup still works normally"
+
+    over = ask({"ip": "198.51.102.200"}).json()["ip"]
+    assert over["deferred"] is True, over
+    assert over["classification"] == "unknown" and over["flagged"] is False, over
+    assert over["blocked"] is False
+    assert len(ipblocks) == before + 1, "an over-budget lookup must install nothing"
+    assert not main.reputation_block_installed("ip", "198.51.102.128/25")
+
+    #    A cache hit costs nothing, so it is answered normally even with the
+    #    budget spent — the ceiling limits provider spend, not service.
+    cached_hit = ask({"ip": "198.51.100.51"}).json()["ip"]
+    assert cached_hit["cached"] is True and cached_hit["deferred"] is False, cached_hit
+    assert cached_hit["blocked"] is True, "a cached flagged range still answers blocked"
+    main.REPUTATION_API_MAX_LOOKUPS_PER_HOUR = 0
+    main._lookup_budget.clear()
+
+    # J. Blocks installed here show up in the public blocklist alongside the
+    #    signup-derived ones — otherwise the published list under-reports as
+    #    blocking shifts to the pre-signup path. `signups` still counts
+    #    signups, so a range blocked before anyone signed up reads 0.
+    main.BLOCKS_API_ENABLED = True
+    ranges = {e["cidr"]: e for e in client.get("/api/blocks/ip-ranges").json()["items"]}
+    domains = {e["domain"]: e for e in client.get("/api/blocks/email-domains").json()["items"]}
+    assert "198.51.100.0/24" in ranges, ranges
+    assert ranges["198.51.100.0/24"]["signups"] == 0, \
+        "nothing signed up from this range — it was blocked first"
+    assert "rep-spam.test" in domains, domains
+    assert set(ranges["198.51.100.0/24"]) == {
+        "cidr", "signups", "first_blocked_at", "last_blocked_at"}, \
+        "merging must not change the published entry shape"
+
+    # K. ...and the classification behind them stays out of the public list,
+    #    same rule as for the signup-derived entries.
+    raw = client.get("/api/blocks").text
+    for leak in ("Example Cloud Inc", "Example Proxy Co", "datacenter", "proxy"):
+        assert leak not in raw, f"{leak!r} must not reach the public blocklist"
+
+    # L. The reconcile pass covers these rows too: a block a moderator removes
+    #    stops being published, and reappears if it's registered again.
+    import app.reconcile_blocks as rec
+    rec.live_ip_cidrs = lambda: set()
+    rec.live_email_domains = lambda: set()
+    rec.reconcile()
+    active = {e["cidr"] for e in client.get("/api/blocks/ip-ranges").json()["items"]}
+    assert "198.51.100.0/24" not in active, \
+        "a query-API block removed on the Mastodon side must stop being published"
+    assert not main.reputation_block_installed("ip", "198.51.100.0/24")
+
+    ipblocks.clear()
+    r = ask({"ip": "198.51.100.50"})
+    assert r.json()["blocked"] is True
+    assert len(ipblocks) == 1, "a reconciled-away block is re-installed, not assumed present"
+    active = {e["cidr"] for e in client.get("/api/blocks/ip-ranges").json()["items"]}
+    assert "198.51.100.0/24" in active, "re-installing republishes it"
+
+    main.REPUTATION_API_ENABLED = False
+
+
 if __name__ == "__main__":
     routing_tests()
     policy_tests()
@@ -991,4 +1194,5 @@ if __name__ == "__main__":
     suspicious_sweep_tests()
     blocks_api_tests()
     reconcile_blocks_tests()
+    reputation_api_tests()
     print("ALL TESTS PASSED")