yttrx-welcomebot/app/main.py

"""yttrx welcome-bot + abuse-bot webhook server.

Receives Mastodon admin webhook deliveries, verifies the HMAC signature,
and dispatches by event:

* ``account.created`` / ``account.approved`` — direct-messages each new local
  account a welcome toot via the welcome bot's token (scope ``write:statuses``).
  Both events are handled so the welcome works whether or not registration
  approval is enabled; the dedup store welcomes each account exactly once.
  ``account.created`` also classifies the signup's IP (datacenter/vpn/proxy/
  tor/abuser via ipapi.is) and email domain (disposable/high-risk via
  check-mail.org); a flagged
  signup gets more scrutiny (held welcome, auto ip_block/email_domain_block,
  moderator DM) before falling through to a normal welcome.
* ``report.created`` — evaluates the reported account and, when enough
  *distinct* reporters have open reports against a young or dormant account,
  auto-**silences** it (reversible) and DMs a moderator for review. Uses a
  separate moderator bot token with admin scopes. Special case: a report
  against an account whose email domain is high-risk auto-**suspends** it
  immediately (no reporter-count threshold) and blocks the domain.

Mastodon signs every delivery with:

    X-Hub-Signature: sha256=<hex hmac-sha256 of the raw body, keyed by the
                            webhook's secret>

One Mastodon webhook subscribing to both events points at /webhook; both
share the single WEBHOOK_SECRET. See README.md for the deploy + Mastodon-side
configuration runbook.
"""

from __future__ import annotations

import hashlib
import hmac
import ipaddress
import logging
import os
import sqlite3
import threading
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone

import httpx
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
log = logging.getLogger("welcomebot")

# --- Configuration (from environment / .env) -------------------------------

WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "")
BOT_ACCESS_TOKEN = os.environ.get("BOT_ACCESS_TOKEN", "")
MASTODON_BASE_URL = os.environ.get("MASTODON_BASE_URL", "https://yttrx.com").rstrip("/")
# {acct} is substituted with the new account's handle (e.g. "alice").
WELCOME_MESSAGE = os.environ.get(
    "WELCOME_MESSAGE",
    "👋 Welcome to yttrx, @{acct}! Glad to have you here. "
    "If you have any questions, check out https://help.yttrx.com or just reply to this message.",
)
# Only welcome accounts that are local to this instance (domain is null/empty).
LOCAL_ONLY = os.environ.get("LOCAL_ONLY", "true").lower() in ("1", "true", "yes")
DB_PATH = os.environ.get("DB_PATH", "/data/welcomed.db")

# --- Abuse-bot configuration -----------------------------------------------
# Moderator bot token. The account holding it must have a role with the
# "Manage Users" + "Manage Reports" permissions, and the token these scopes:
#   admin:write:accounts  admin:read:reports  read:statuses  write:statuses
ABUSE_BOT_TOKEN = os.environ.get("ABUSE_BOT_TOKEN", "")
# Master switch; report handling is also disabled if ABUSE_BOT_TOKEN is empty.
ABUSE_ENABLED = os.environ.get("ABUSE_ENABLED", "true").lower() in ("1", "true", "yes")
# Don't actually silence — just log + DM what *would* happen. Safe for rollout.
ABUSE_DRY_RUN = os.environ.get("ABUSE_DRY_RUN", "false").lower() in ("1", "true", "yes")
# Moderation action to take. "silence" is reversible and the agreed default;
# "suspend" is also accepted.
ABUSE_ACTION = os.environ.get("ABUSE_ACTION", "silence").lower()
# Only auto-act on accounts local to this instance.
ABUSE_LOCAL_ONLY = os.environ.get("ABUSE_LOCAL_ONLY", "true").lower() in ("1", "true", "yes")
# An account is "young" if its OLDEST post is newer than this many days.
ABUSE_YOUNG_MAX_DAYS = int(os.environ.get("ABUSE_YOUNG_MAX_DAYS", "30"))
# An account is "dormant" if its NEWEST post is older than this many days.
ABUSE_DORMANT_MIN_DAYS = int(os.environ.get("ABUSE_DORMANT_MIN_DAYS", "30"))
# Distinct-reporter thresholds: young/dormant (or no-posts) vs normally-active.
ABUSE_SOURCES_NEWDORMANT = int(os.environ.get("ABUSE_SOURCES_NEWDORMANT", "2"))
ABUSE_SOURCES_ACTIVE = int(os.environ.get("ABUSE_SOURCES_ACTIVE", "3"))
# Safety cap on the backward status scan (40 statuses/page).
ABUSE_MAX_STATUS_PAGES = int(os.environ.get("ABUSE_MAX_STATUS_PAGES", "5"))
# Comma-separated handles (acct, no leading @) never to auto-act on.
ABUSE_ALLOWLIST = {
    h.strip().lstrip("@").lower()
    for h in os.environ.get("ABUSE_ALLOWLIST", "").split(",")
    if h.strip()
}
# Automatically never auto-act on accounts that hold a staff role
# (Admin/Owner/Moderator — any assigned, non-default UserRole). Read from the
# report payload's target_account.role, so new staff are protected with no
# allowlist upkeep. Belt-and-suspenders with ABUSE_ALLOWLIST.
ABUSE_SKIP_PRIVILEGED = os.environ.get(
    "ABUSE_SKIP_PRIVILEGED", "true"
).lower() in ("1", "true", "yes")
# Handle (acct, no @) to DM with a review summary. Empty disables the DM.
MOD_ALERT_ACCT = os.environ.get("MOD_ALERT_ACCT", "").strip().lstrip("@")
# Help/appeals page linked in the DM sent to a silenced user.
ABUSE_HELP_URL = os.environ.get(
    "ABUSE_HELP_URL", "https://welcome.yttrx.com/posts/account-limited/"
)
# DM sent to the silenced user. {acct} and {help_url} are substituted; the
# message must mention @{acct} for it to reach them. Empty disables the DM.
ABUSE_USER_DM = os.environ.get(
    "ABUSE_USER_DM",
    "Hi @{acct} — your yttrx account has been temporarily limited while we "
    "review some reports. This is reversible and a moderator will take a look. "
    "Here's what happened and how to appeal: {help_url}",
)

# --- IP-based signup scrutiny -----------------------------------------------
# Classifies each new signup's IP (already delivered free on the
# account.created payload as Admin::Account.ip) via ipapi.is (free, keyless,
# up to 1000 req/day), and flags it if ipapi.is reports it as datacenter,
# vpn, proxy, tor, or an independently-scored abuser.
IP_SCRUTINY_ENABLED = os.environ.get("IP_SCRUTINY_ENABLED", "true").lower() in ("1", "true", "yes")
# Optional ipapi.is API key for higher rate limits (anonymous is capped at
# 1000 req/day, shared across all callers of that IP from this host). Empty
# uses the anonymous tier.
IP_SCRUTINY_IPAPI_KEY = os.environ.get("IP_SCRUTINY_IPAPI_KEY", "")
# Log/DM only — no held welcome, no ip_blocks write. Rollout safety, same role
# as ABUSE_DRY_RUN.
IP_SCRUTINY_DRY_RUN = os.environ.get("IP_SCRUTINY_DRY_RUN", "true").lower() in ("1", "true", "yes")
# Hold the welcome DM for a flagged signup until account.approved fires
# (i.e. until a human clears the existing approval-required registration
# gate), instead of welcoming immediately on account.created.
IP_SCRUTINY_HOLD_WELCOME = os.environ.get("IP_SCRUTINY_HOLD_WELCOME", "true").lower() in ("1", "true", "yes")
# Distinct-reporter threshold used INSTEAD of the tier's usual threshold (via
# min()) when the reported account's signup IP was flagged.
IP_SCRUTINY_ABUSE_THRESHOLD = int(os.environ.get("IP_SCRUTINY_ABUSE_THRESHOLD", "1"))
# Auto-register flagged IPs into Mastodon's native Admin::IpBlock.
IP_SCRUTINY_AUTO_IPBLOCK = os.environ.get("IP_SCRUTINY_AUTO_IPBLOCK", "true").lower() in ("1", "true", "yes")
# severity: sign_up_requires_approval | sign_up_block | no_access
IP_SCRUTINY_IPBLOCK_SEVERITY = os.environ.get("IP_SCRUTINY_IPBLOCK_SEVERITY", "sign_up_requires_approval")

# --- Disposable/high-risk email signup scrutiny (check-mail.org) -----------
# Every account.created delivery already carries the signup email for free
# (Admin::Account.email). On each new local signup, the bot classifies the
# *domain only* (never the full address) via check-mail.org.
CHECK_MAIL_ENABLED = os.environ.get("CHECK_MAIL_ENABLED", "true").lower() in ("1", "true", "yes")
CHECK_MAIL_API_KEY = os.environ.get("CHECK_MAIL_API_KEY", "")
# Log/DM only — no held welcome, no email_domain_block write. Same rollout-
# safety role as IP_SCRUTINY_DRY_RUN.
CHECK_MAIL_DRY_RUN = os.environ.get("CHECK_MAIL_DRY_RUN", "true").lower() in ("1", "true", "yes")
# A domain is flagged if check-mail.org marks it is_disposable OR its risk
# score (0-100) is at or above this threshold.
CHECK_MAIL_RISK_THRESHOLD = int(os.environ.get("CHECK_MAIL_RISK_THRESHOLD", "80"))
# Hold the welcome DM for a flagged signup until account.approved fires, same
# semantics as IP_SCRUTINY_HOLD_WELCOME.
CHECK_MAIL_HOLD_WELCOME = os.environ.get("CHECK_MAIL_HOLD_WELCOME", "true").lower() in ("1", "true", "yes")
# Auto-register a flagged signup's email domain into Mastodon's native
# Admin::EmailDomainBlock (blocks future signups from that domain).
CHECK_MAIL_AUTO_DOMAIN_BLOCK = os.environ.get("CHECK_MAIL_AUTO_DOMAIN_BLOCK", "true").lower() in ("1", "true", "yes")

# --- Suspicious-signup sweep (flagged at signup, no activity in grace window) ---
# Watches every account flagged by EITHER IP-scrutiny or email-domain scrutiny
# at signup. Once the account goes live — account.created if signups are open
# (or an already-approved account), account.approved if this instance requires
# moderator approval, same dual event handling the welcome flow already uses —
# a baseline post/follow count is recorded (see maybe_start_suspicious_watch).
# A separate scheduled job (bin/welcomebot-suspicious sweep, see CLAUDE.md) run
# via `python -m app.suspicious_sweep` looks for accounts past
# SUSPICIOUS_GRACE_HOURS with zero new posts AND zero new follows since that
# baseline, and takes SUSPICIOUS_ACTION on them.
SUSPICIOUS_SWEEP_ENABLED = os.environ.get("SUSPICIOUS_SWEEP_ENABLED", "true").lower() in ("1", "true", "yes")
# Hours of grace given to a flagged signup to show organic activity (a new
# post or a new follow) before it's considered dormant/bot-like.
SUSPICIOUS_GRACE_HOURS = int(os.environ.get("SUSPICIOUS_GRACE_HOURS", "24"))
# "suspend" (agreed default — a flagged signup with zero organic activity in
# the grace window is a strong bulk/bot-signup signal, stronger than the
# report-driven abuse-bot's default "silence") or "silence".
SUSPICIOUS_ACTION = os.environ.get("SUSPICIOUS_ACTION", "suspend").lower()
# Rollout safety switch, same role as ABUSE_DRY_RUN — ships "false" (live)
# here per an explicit decision to skip the usual dry-run rollout step for
# this feature. Flip to "true" to pause without redeploying.
SUSPICIOUS_DRY_RUN = os.environ.get("SUSPICIOUS_DRY_RUN", "false").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:
    log.warning("BOT_ACCESS_TOKEN is empty — the bot cannot post welcome messages.")
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.")

app = FastAPI(title="yttrx welcome-bot + abuse-bot")

# --- Dedup / action store --------------------------------------------------
# Mastodon retries deliveries on any non-2xx response and may deliver more
# than once; persist what we've done so we never act twice.

_db_lock = threading.Lock()


def _init_db() -> None:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "CREATE TABLE IF NOT EXISTS welcomed ("
            "  account_id TEXT PRIMARY KEY,"
            "  acct       TEXT,"
            "  welcomed_at TEXT DEFAULT CURRENT_TIMESTAMP"
            ")"
        )
        conn.execute(
            "CREATE TABLE IF NOT EXISTS abuse_actions ("
            "  target_id  TEXT PRIMARY KEY,"
            "  acct       TEXT,"
            "  action     TEXT,"
            "  tier       TEXT,"
            "  sources    INTEGER,"
            "  report_id  TEXT,"
            "  acted_at   TEXT DEFAULT CURRENT_TIMESTAMP"
            ")"
        )
        conn.execute(
            "CREATE TABLE IF NOT EXISTS signup_ip ("
            "  account_id TEXT PRIMARY KEY,"
            "  acct       TEXT,"
            "  ip         TEXT,"
            "  classification TEXT,"
            "  org        TEXT,"
            "  flagged    INTEGER,"
            "  ipblock_registered INTEGER DEFAULT 0,"
            "  classified_at TEXT DEFAULT CURRENT_TIMESTAMP"
            ")"
        )
        conn.execute(
            "CREATE TABLE IF NOT EXISTS ipapi_cache ("
            "  ip            TEXT PRIMARY KEY,"
            "  is_datacenter INTEGER,"
            "  is_vpn        INTEGER,"
            "  is_proxy      INTEGER,"
            "  is_tor        INTEGER,"
            "  is_abuser     INTEGER,"
            "  org           TEXT,"
            "  cached_at     TEXT DEFAULT CURRENT_TIMESTAMP"
            ")"
        )
        conn.execute(
            "CREATE TABLE IF NOT EXISTS signup_email ("
            "  account_id TEXT PRIMARY KEY,"
            "  acct       TEXT,"
            "  domain     TEXT,"
            "  is_disposable INTEGER,"
            "  risk       INTEGER,"
            "  flagged    INTEGER,"
            "  domain_blocked INTEGER DEFAULT 0,"
            "  classified_at TEXT DEFAULT CURRENT_TIMESTAMP"
            ")"
        )
        conn.execute(
            "CREATE TABLE IF NOT EXISTS check_mail_cache ("
            "  domain     TEXT PRIMARY KEY,"
            "  is_disposable INTEGER,"
            "  risk       INTEGER,"
            "  cached_at  TEXT DEFAULT CURRENT_TIMESTAMP"
            ")"
        )
        conn.execute(
            "CREATE TABLE IF NOT EXISTS suspicious_watch ("
            "  account_id TEXT PRIMARY KEY,"
            "  acct       TEXT,"
            "  reasons    TEXT,"
            "  started_at TEXT DEFAULT CURRENT_TIMESTAMP,"
            "  baseline_statuses_count  INTEGER,"
            "  baseline_following_count INTEGER,"
            "  status     TEXT DEFAULT 'pending',"
            "  resolved_at TEXT"
            ")"
        )


@contextmanager
def _db():
    with _db_lock:
        conn = sqlite3.connect(DB_PATH)
        try:
            yield conn
            conn.commit()
        finally:
            conn.close()


def already_welcomed(account_id: str) -> bool:
    with _db() as conn:
        row = conn.execute(
            "SELECT 1 FROM welcomed WHERE account_id = ?", (account_id,)
        ).fetchone()
        return row is not None


def mark_welcomed(account_id: str, acct: str) -> None:
    with _db() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO welcomed (account_id, acct) VALUES (?, ?)",
            (account_id, acct),
        )


def already_actioned(target_id: str) -> bool:
    with _db() as conn:
        row = conn.execute(
            "SELECT 1 FROM abuse_actions WHERE target_id = ?", (target_id,)
        ).fetchone()
        return row is not None


def record_action(target_id: str, acct: str, action: str, tier: str,
                  sources: int, report_id: str) -> None:
    with _db() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO abuse_actions "
            "(target_id, acct, action, tier, sources, report_id) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (target_id, acct, action, tier, sources, report_id),
        )


def record_signup_ip(account_id: str, acct: str, ip: str, classification: str,
                     org: str, flagged: bool) -> None:
    with _db() as conn:
        conn.execute(
            "INSERT OR REPLACE INTO signup_ip "
            "(account_id, acct, ip, classification, org, flagged, "
            " ipblock_registered) "
            "VALUES (?, ?, ?, ?, ?, ?, "
            "  COALESCE((SELECT ipblock_registered FROM signup_ip WHERE account_id = ?), 0))",
            (account_id, acct, ip, classification, org, int(flagged), account_id),
        )


def get_signup_flag(account_id: str) -> bool:
    with _db() as conn:
        row = conn.execute(
            "SELECT flagged FROM signup_ip WHERE account_id = ?", (account_id,)
        ).fetchone()
        return bool(row and row[0])


def mark_ipblock_registered(account_id: str) -> None:
    with _db() as conn:
        conn.execute(
            "UPDATE signup_ip SET ipblock_registered = 1 WHERE account_id = ?",
            (account_id,),
        )


def cached_ip_intel(ip: str) -> dict | None:
    with _db() as conn:
        row = conn.execute(
            "SELECT is_datacenter, is_vpn, is_proxy, is_tor, is_abuser, org "
            "FROM ipapi_cache WHERE ip = ?", (ip,)
        ).fetchone()
        if row is None:
            return None
        return {
            "is_datacenter": bool(row[0]), "is_vpn": bool(row[1]),
            "is_proxy": bool(row[2]), "is_tor": bool(row[3]),
            "is_abuser": bool(row[4]), "org": row[5] or "",
        }


def cache_ip_intel(ip: str, intel: dict) -> None:
    with _db() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO ipapi_cache "
            "(ip, is_datacenter, is_vpn, is_proxy, is_tor, is_abuser, org) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            (ip, int(intel["is_datacenter"]), int(intel["is_vpn"]),
             int(intel["is_proxy"]), int(intel["is_tor"]),
             int(intel["is_abuser"]), intel["org"]),
        )


def record_signup_email(account_id: str, acct: str, domain: str, is_disposable: bool,
                        risk: int, flagged: bool) -> None:
    with _db() as conn:
        conn.execute(
            "INSERT OR REPLACE INTO signup_email "
            "(account_id, acct, domain, is_disposable, risk, flagged, domain_blocked) "
            "VALUES (?, ?, ?, ?, ?, ?, "
            "  COALESCE((SELECT domain_blocked FROM signup_email WHERE account_id = ?), 0))",
            (account_id, acct, domain, int(is_disposable), risk, int(flagged), account_id),
        )


def get_signup_email(account_id: str) -> tuple[str, bool]:
    """Return (domain, flagged) for a signup's classified email domain, or
    ("", False) if it was never classified (check-mail disabled/unavailable)."""
    with _db() as conn:
        row = conn.execute(
            "SELECT domain, flagged FROM signup_email WHERE account_id = ?", (account_id,)
        ).fetchone()
        return (row[0], bool(row[1])) if row else ("", False)


def mark_email_domain_blocked(account_id: str) -> None:
    with _db() as conn:
        conn.execute(
            "UPDATE signup_email SET domain_blocked = 1 WHERE account_id = ?",
            (account_id,),
        )


def cached_check_mail(domain: str) -> tuple[bool, int] | None:
    with _db() as conn:
        row = conn.execute(
            "SELECT is_disposable, risk FROM check_mail_cache WHERE domain = ?", (domain,)
        ).fetchone()
        return (bool(row[0]), row[1]) if row else None


def cache_check_mail(domain: str, is_disposable: bool, risk: int) -> None:
    with _db() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO check_mail_cache (domain, is_disposable, risk) VALUES (?, ?, ?)",
            (domain, int(is_disposable), risk),
        )


def already_watched_suspicious(account_id: str) -> bool:
    with _db() as conn:
        row = conn.execute(
            "SELECT 1 FROM suspicious_watch WHERE account_id = ?", (account_id,)
        ).fetchone()
        return row is not None


def start_suspicious_watch(account_id: str, acct: str, reasons: str,
                           statuses_count: int, following_count: int) -> None:
    with _db() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO suspicious_watch "
            "(account_id, acct, reasons, baseline_statuses_count, baseline_following_count) "
            "VALUES (?, ?, ?, ?, ?)",
            (account_id, acct, reasons, statuses_count, following_count),
        )


_init_db()

# --- Signature verification ------------------------------------------------


def verify_signature(raw_body: bytes, header_value: str | None) -> bool:
    """Constant-time check of the X-Hub-Signature header."""
    if not WEBHOOK_SECRET or not header_value:
        return False
    if not header_value.startswith("sha256="):
        return False
    sent = header_value[len("sha256=") :].strip()
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), raw_body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(sent, expected)


# --- Mastodon API (welcome) ------------------------------------------------


def send_welcome(account_id: str, acct: str) -> None:
    """Post a direct-visibility welcome status mentioning the new user."""
    if already_welcomed(account_id):
        log.info("already welcomed account_id=%s acct=%s, skipping", account_id, acct)
        return

    message = WELCOME_MESSAGE.format(acct=acct)
    try:
        resp = httpx.post(
            f"{MASTODON_BASE_URL}/api/v1/statuses",
            headers={"Authorization": f"Bearer {BOT_ACCESS_TOKEN}"},
            data={"status": message, "visibility": "direct"},
            timeout=15.0,
        )
        resp.raise_for_status()
    except httpx.HTTPError as exc:
        # Leave it un-marked so a Mastodon retry (or manual replay) can succeed.
        log.error("failed to send welcome to acct=%s: %s", acct, exc)
        return

    mark_welcomed(account_id, acct)
    log.info("welcomed acct=%s account_id=%s status_id=%s",
             acct, account_id, resp.json().get("id"))


def classify_email_domain(domain: str) -> tuple[bool, int]:
    """Classify a signup's email domain via check-mail.org.

    Domain-only query (never the full email) — see CHECK_MAIL_API_KEY comment.
    Cached indefinitely in sqlite (a domain's disposable/risk classification
    doesn't meaningfully change hour to hour, and the free tier is capped at
    1,000 req/month). API failure yields (False, 0) — never flag on our own
    lookup errors, same philosophy as classify_signup_ip's ipapi.is failure path.
    """
    cached = cached_check_mail(domain)
    if cached is not None:
        return cached

    try:
        resp = httpx.post(
            "https://api.check-mail.org/v2/",
            headers={"Authorization": f"Bearer {CHECK_MAIL_API_KEY}"},
            data={"domain": domain},
            timeout=10.0,
        )
        resp.raise_for_status()
        data = resp.json()
        is_disposable = bool(data.get("is_disposable") or data.get("block"))
        risk = int(data.get("risk") or 0)
    except (httpx.HTTPError, ValueError) as exc:
        log.warning("check-mail.org lookup failed for domain=%s: %s", domain, exc)
        return False, 0

    cache_check_mail(domain, is_disposable, risk)
    return is_disposable, risk


def register_email_domain_block(domain: str, acct: str) -> None:
    """Register a flagged signup's email domain in Mastodon's native
    Admin::EmailDomainBlock, blocking future signups from that domain.

    Requires the admin:write:email_domain_blocks OAuth scope on
    ABUSE_BOT_TOKEN — see CLAUDE.md's moderator-token-gotcha section.
    """
    try:
        resp = httpx.post(
            f"{MASTODON_BASE_URL}/api/v1/admin/email_domain_blocks",
            headers=_admin_headers(),
            data={"domain": domain},
            timeout=15.0,
        )
        if resp.status_code == 422:
            log.info("email_domain_block for %s already exists (or invalid), treating as handled", domain)
        else:
            resp.raise_for_status()
    except httpx.HTTPError as exc:
        log.error("failed to register email_domain_block for %s (acct=%s): %s", domain, acct, exc)


def process_signup(account_id: str, acct: str, ip: str, email: str = "") -> None:
    """Classify a new signup's IP and email domain, and act on account.created.

    Non-flagged (or both signals disabled/unavailable) signups are welcomed
    immediately, same as before these features existed. A signup flagged by
    either signal (datacenter IP and/or high-risk email domain) gets a
    moderator DM, the matching auto-block(s) (ip_block / email_domain_block),
    and — unless dry-run — has its welcome held until account.approved fires
    (see _handle_account_created).
    """
    reasons: list[str] = []
    hold = False

    if IP_SCRUTINY_ENABLED and ip:
        classification, org, ip_flagged = classify_signup_ip(ip)
        record_signup_ip(account_id, acct, ip, classification, org, ip_flagged)

        if ip_flagged:
            log.warning("flagged signup acct=%s ip=%s classification=%s org=%s",
                        acct, ip, classification, org)
            if IP_SCRUTINY_AUTO_IPBLOCK and not IP_SCRUTINY_DRY_RUN:
                register_ip_block(ip, acct, org)
                mark_ipblock_registered(account_id)
            prefix = "[DRY-RUN] " if IP_SCRUTINY_DRY_RUN else ""
            reasons.append(f"{prefix}IP {ip} ({org or 'unknown org'}, {classification})")
            hold = hold or (IP_SCRUTINY_HOLD_WELCOME and not IP_SCRUTINY_DRY_RUN)

    if CHECK_MAIL_ENABLED and CHECK_MAIL_API_KEY and email and "@" in email:
        domain = email.rsplit("@", 1)[-1].strip().lower()
        is_disposable, risk = classify_email_domain(domain)
        email_flagged = is_disposable or risk >= CHECK_MAIL_RISK_THRESHOLD
        record_signup_email(account_id, acct, domain, is_disposable, risk, email_flagged)

        if email_flagged:
            log.warning("flagged signup acct=%s email_domain=%s is_disposable=%s risk=%d",
                        acct, domain, is_disposable, risk)
            if CHECK_MAIL_AUTO_DOMAIN_BLOCK and not CHECK_MAIL_DRY_RUN:
                register_email_domain_block(domain, acct)
                mark_email_domain_blocked(account_id)
            prefix = "[DRY-RUN] " if CHECK_MAIL_DRY_RUN else ""
            reasons.append(f"{prefix}email domain {domain} (disposable={is_disposable}, risk={risk})")
            hold = hold or (CHECK_MAIL_HOLD_WELCOME and not CHECK_MAIL_DRY_RUN)

    if not reasons:
        send_welcome(account_id, acct)
        return

    dm_moderator(
        f"⚠️ Flagged signup @{acct}: {'; '.join(reasons)}."
        + (" Welcome DM held pending approval." if hold else "")
    )

    if not hold:
        send_welcome(account_id, acct)
        maybe_start_suspicious_watch(account_id, acct)
    # else: held — sent later if/when account.approved fires (human review),
    # which is also where maybe_start_suspicious_watch gets called.


def maybe_start_suspicious_watch(account_id: str, acct: str) -> None:
    """Start the suspicious-sweep grace-period clock for a flagged signup.

    Called at the moment the account actually goes live — immediately in
    process_signup() when the welcome isn't held, or from the
    account.approved handler once a human clears the approval queue for a
    held one. Either path is a no-op if the account was never flagged (by
    IP-scrutiny or email-domain scrutiny) or is already being watched, so
    it's safe to call from both places unconditionally.
    """
    if not SUSPICIOUS_SWEEP_ENABLED or already_watched_suspicious(account_id):
        return

    ip_flagged = get_signup_flag(account_id)
    _, email_flagged = get_signup_email(account_id)
    if not (ip_flagged or email_flagged):
        return

    reasons = ",".join(
        r for r, flagged in (("ip", ip_flagged), ("email", email_flagged)) if flagged
    )
    try:
        statuses_count, following_count = fetch_account_counts(account_id)
    except httpx.HTTPError as exc:
        log.error("failed to snapshot suspicious-watch baseline for acct=%s: %s", acct, exc)
        return

    start_suspicious_watch(account_id, acct, reasons, statuses_count, following_count)
    log.info("started suspicious watch acct=%s reasons=%s baseline=(statuses=%d, following=%d)",
             acct, reasons, statuses_count, following_count)


def fetch_account_counts(account_id: str) -> tuple[int, int]:
    """Return (statuses_count, following_count) for an account via the public
    account API. Split out from maybe_start_suspicious_watch as a single seam
    to stub in tests."""
    resp = httpx.get(f"{MASTODON_BASE_URL}/api/v1/accounts/{account_id}", timeout=15.0)
    resp.raise_for_status()
    data = resp.json()
    return data.get("statuses_count", 0), data.get("following_count", 0)


# --- Mastodon API (abuse) --------------------------------------------------


def _parse_ts(value: str | None) -> datetime | None:
    """Parse a Mastodon ISO-8601 timestamp into an aware UTC datetime."""
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        log.warning("could not parse timestamp %r", value)
        return None


def _admin_headers() -> dict:
    return {"Authorization": f"Bearer {ABUSE_BOT_TOKEN}"}


def classify_account(target_id: str) -> dict:
    """Walk the target's authored statuses (newest→oldest) and classify.

    Returns {has_posts, young, dormant, newest_at, oldest_seen, truncated}.

    * ``young``   — the oldest authored post is newer than ABUSE_YOUNG_MAX_DAYS.
                    Decided by paging backward and bailing out the moment we
                    see a post older than the window (so established accounts
                    cost one page).
    * ``dormant`` — the newest authored post is older than ABUSE_DORMANT_MIN_DAYS.
    * ``truncated`` — hit the page cap while everything was still inside the
                    young window, so ``young`` is treated as unknown→False.
    """
    now = datetime.now(timezone.utc)
    young_cutoff = now - timedelta(days=ABUSE_YOUNG_MAX_DAYS)
    dormant_cutoff = now - timedelta(days=ABUSE_DORMANT_MIN_DAYS)

    newest_at: datetime | None = None
    oldest_seen: datetime | None = None
    found_older_than_young = False
    truncated = False
    max_id: str | None = None

    for page_num in range(ABUSE_MAX_STATUS_PAGES):
        params = {"limit": 40, "exclude_reblogs": "true"}
        if max_id:
            params["max_id"] = max_id
        resp = httpx.get(
            f"{MASTODON_BASE_URL}/api/v1/accounts/{target_id}/statuses",
            headers=_admin_headers(), params=params, timeout=15.0,
        )
        resp.raise_for_status()
        page = resp.json()
        if not page:
            break

        for status in page:
            ts = _parse_ts(status.get("created_at"))
            if ts is None:
                continue
            if newest_at is None or ts > newest_at:
                newest_at = ts
            if oldest_seen is None or ts < oldest_seen:
                oldest_seen = ts
            if ts < young_cutoff:
                found_older_than_young = True

        max_id = str(page[-1].get("id"))
        if found_older_than_young or len(page) < 40:
            break
        if page_num == ABUSE_MAX_STATUS_PAGES - 1 and not found_older_than_young:
            # Still all-recent at the cap — can't prove the true oldest.
            truncated = True

    has_posts = newest_at is not None
    young = has_posts and not found_older_than_young and not truncated
    dormant = has_posts and newest_at < dormant_cutoff
    if truncated:
        log.info("status scan truncated at %d pages for target_id=%s; "
                 "treating young=unknown→False", ABUSE_MAX_STATUS_PAGES, target_id)
    return {
        "has_posts": has_posts, "young": young, "dormant": dormant,
        "newest_at": newest_at, "oldest_seen": oldest_seen, "truncated": truncated,
    }


def classify_signup_ip(ip: str) -> tuple[str, str, bool]:
    """Classify a signup IP via ipapi.is, cached indefinitely in sqlite (an
    IP's owning org/abuse posture doesn't change on the timescale that
    matters here).

    Returns (classification, org, flagged). classification is a "+"-joined
    list of every matched signal (datacenter/vpn/proxy/tor/abuser), or
    "clean" if none matched. flagged is True if any signal matched. API
    failure yields ("unknown", "", False) — scrutiny should never trigger on
    our own lookup errors.
    """
    intel = cached_ip_intel(ip)
    if intel is None:
        params = {"q": ip}
        if IP_SCRUTINY_IPAPI_KEY:
            params["key"] = IP_SCRUTINY_IPAPI_KEY
        try:
            resp = httpx.get("https://api.ipapi.is", params=params, timeout=10.0)
            resp.raise_for_status()
            data = resp.json()
        except (httpx.HTTPError, ValueError) as exc:
            log.warning("ipapi.is lookup failed for ip=%s: %s", ip, exc)
            return "unknown", "", False
        intel = {
            "is_datacenter": bool(data.get("is_datacenter")),
            "is_vpn": bool(data.get("is_vpn")),
            "is_proxy": bool(data.get("is_proxy")),
            "is_tor": bool(data.get("is_tor")),
            "is_abuser": bool(data.get("is_abuser")),
            "org": ((data.get("company") or {}).get("name")
                    or (data.get("asn") or {}).get("org") or ""),
        }
        cache_ip_intel(ip, intel)

    reasons = [name for name, key in (
        ("datacenter", "is_datacenter"),
        ("vpn", "is_vpn"),
        ("proxy", "is_proxy"),
        ("tor", "is_tor"),
        ("abuser", "is_abuser"),
    ) if intel[key]]
    classification = "+".join(reasons) if reasons else "clean"
    return classification, intel["org"], bool(reasons)


def register_ip_block(ip: str, acct: str, org: str) -> None:
    """Register a flagged signup IP in Mastodon's native Admin::IpBlock."""
    try:
        prefix_len = 32 if ipaddress.ip_address(ip).version == 4 else 128
    except ValueError:
        log.warning("skipping ip_block registration for unparseable ip=%r (acct=%s)", ip, acct)
        return
    cidr = f"{ip}/{prefix_len}"
    comment = f"welcomebot: flagged signup @{acct} ({org or 'unknown org'})"[:200]
    try:
        resp = httpx.post(
            f"{MASTODON_BASE_URL}/api/v1/admin/ip_blocks",
            headers=_admin_headers(),
            data={"ip": cidr, "severity": IP_SCRUTINY_IPBLOCK_SEVERITY, "comment": comment},
            timeout=15.0,
        )
        if resp.status_code == 422:
            log.info("ip_block for %s already exists, treating as registered", cidr)
        else:
            resp.raise_for_status()
    except httpx.HTTPError as exc:
        log.error("failed to register ip_block for %s (acct=%s): %s", cidr, acct, exc)
        return


def count_distinct_reporters(target_id: str) -> int:
    """Count distinct accounts with an OPEN report against the target.

    Self-reports (reporter == target) are excluded.
    """
    resp = httpx.get(
        f"{MASTODON_BASE_URL}/api/v1/admin/reports",
        headers=_admin_headers(),
        params={"target_account_id": target_id, "resolved": "false", "limit": 200},
        timeout=15.0,
    )
    resp.raise_for_status()
    reports = resp.json()
    if len(reports) >= 200:
        log.warning("hit 200-report cap counting reporters for target_id=%s; "
                    "count may be undercounted", target_id)
    reporters = set()
    for rep in reports:
        reporter = (rep.get("account") or {}).get("id")
        if reporter and str(reporter) != str(target_id):
            reporters.add(str(reporter))
    return len(reporters)


def apply_action(target_id: str, action: str, text: str) -> None:
    """POST the moderation action WITHOUT report_id, so the report stays open."""
    resp = httpx.post(
        f"{MASTODON_BASE_URL}/api/v1/admin/accounts/{target_id}/action",
        headers=_admin_headers(),
        data={"type": action, "text": text},
        timeout=15.0,
    )
    resp.raise_for_status()


def _post_direct(text: str, *, what: str) -> None:
    """Post a direct-visibility status from the moderator bot."""
    try:
        resp = httpx.post(
            f"{MASTODON_BASE_URL}/api/v1/statuses",
            headers=_admin_headers(),
            data={"status": text, "visibility": "direct"},
            timeout=15.0,
        )
        resp.raise_for_status()
    except httpx.HTTPError as exc:
        log.error("failed to send %s DM: %s", what, exc)


def dm_moderator(message: str) -> None:
    """DM the configured moderator account."""
    if not MOD_ALERT_ACCT:
        return
    _post_direct(f"@{MOD_ALERT_ACCT} {message}", what="moderator")


def dm_silenced_user(acct: str) -> None:
    """DM the silenced user a link to the appeals/help page."""
    if not ABUSE_USER_DM:
        return
    _post_direct(ABUSE_USER_DM.format(acct=acct, help_url=ABUSE_HELP_URL),
                 what="silenced-user")


def handle_report(report_id: str, target_id: str, acct: str, is_local: bool,
                  suspended: bool, silenced: bool, privileged: bool = False,
                  email: str = "") -> None:
    """Evaluate a reported account and silence it if the policy is met.

    Special case: if the reported account's email domain is classified
    high-risk by check-mail.org, it's auto-**suspended** (not merely silenced)
    on this first report alone — no distinct-reporter threshold — and the
    email domain is registered in Mastodon's native email_domain_blocks.
    Classified LIVE from this report delivery's target_account.email (a full
    Admin::Account, always present on the payload) rather than from any
    account.created-time record, so this also covers accounts that signed up
    before this feature existed or while CHECK_MAIL was disabled/down.

    Gated by ABUSE_DRY_RUN **or** CHECK_MAIL_DRY_RUN (either one holds it) —
    deliberately independent of ABUSE_DRY_RUN alone, so this new/unverified
    action path ships inert by CHECK_MAIL_DRY_RUN's own default even once the
    general abuse-bot is live, same rollout-safety pattern as every other
    signal here.
    """
    if not (ABUSE_ENABLED and ABUSE_BOT_TOKEN):
        return
    if ABUSE_LOCAL_ONLY and not is_local:
        log.info("report %s: target acct=%s is remote, skipping", report_id, acct)
        return
    if privileged and ABUSE_SKIP_PRIVILEGED:
        log.info("report %s: target acct=%s holds a staff role, skipping", report_id, acct)
        return
    if acct.split("@")[0].lower() in ABUSE_ALLOWLIST or acct.lower() in ABUSE_ALLOWLIST:
        log.info("report %s: target acct=%s is allowlisted, skipping", report_id, acct)
        return
    if suspended or silenced:
        log.info("report %s: target acct=%s already limited, skipping", report_id, acct)
        return
    if already_actioned(target_id):
        log.info("report %s: target acct=%s already auto-actioned, skipping",
                 report_id, acct)
        return

    report_url = f"{MASTODON_BASE_URL}/admin/reports/{report_id}"

    if CHECK_MAIL_ENABLED and CHECK_MAIL_API_KEY and email and "@" in email:
        domain = email.rsplit("@", 1)[-1].strip().lower()
        is_disposable, risk = classify_email_domain(domain)
        if is_disposable or risk >= CHECK_MAIL_RISK_THRESHOLD:
            note = (f"Auto-suspend by abuse-bot: reported account's email domain "
                    f"'{domain}' is flagged high-risk by check-mail.org "
                    f"(disposable={is_disposable}, risk={risk}). Pending human "
                    f"review (report left open).")

            if ABUSE_DRY_RUN or CHECK_MAIL_DRY_RUN:
                log.warning("[DRY-RUN] would suspend acct=%s (flagged email domain=%s, "
                            "risk=%d). %s", acct, domain, risk, report_url)
                record_action(target_id, acct, "dry-run:suspend", "email-risk", 1, report_id)
                dm_moderator(f"[DRY-RUN] would suspend @{acct} — email domain "
                             f"'{domain}' is flagged high-risk (risk={risk}). "
                             f"Review: {report_url}")
                return

            try:
                apply_action(target_id, "suspend", note)
            except httpx.HTTPError as exc:
                log.error("report %s: failed to suspend acct=%s (email-risk): %s",
                          report_id, acct, exc)
                return

            record_action(target_id, acct, "suspend", "email-risk", 1, report_id)
            log.warning("auto-suspended acct=%s (flagged email domain=%s, risk=%d); "
                        "report left open", acct, domain, risk)
            dm_moderator(f"🚨 Auto-suspended @{acct} — email domain '{domain}' flagged "
                         f"high-risk by check-mail.org (risk={risk}). Report left open "
                         f"for review: {report_url}")

            if CHECK_MAIL_AUTO_DOMAIN_BLOCK:
                record_signup_email(target_id, acct, domain, is_disposable, risk, True)
                register_email_domain_block(domain, acct)
                mark_email_domain_blocked(target_id)
            return

    try:
        profile = classify_account(target_id)
        sources = count_distinct_reporters(target_id)
    except httpx.HTTPError as exc:
        log.error("report %s: Mastodon API error evaluating acct=%s: %s",
                  report_id, acct, exc)
        return

    if not profile["has_posts"]:
        tier, required = "no-posts", ABUSE_SOURCES_NEWDORMANT
    elif profile["young"] or profile["dormant"]:
        tier = "young" if profile["young"] else "dormant"
        required = ABUSE_SOURCES_NEWDORMANT
    else:
        tier, required = "active", ABUSE_SOURCES_ACTIVE

    flagged_ip = get_signup_flag(target_id)
    if flagged_ip:
        required = min(required, IP_SCRUTINY_ABUSE_THRESHOLD)

    log.info("report %s: acct=%s tier=%s distinct_sources=%d required=%d%s",
             report_id, acct, tier, sources, required,
             " (lowered: flagged signup IP)" if flagged_ip else "")

    if sources < required:
        log.info("report %s: acct=%s below threshold (%d/%d), leaving for review",
                 report_id, acct, sources, required)
        return

    ip_note = " Signup IP was flagged (datacenter/hosting) — threshold lowered." if flagged_ip else ""
    note = (f"Auto-{ABUSE_ACTION} by abuse-bot: {sources} distinct reporters, "
            f"account tier={tier}. Pending human review (report left open).{ip_note}")

    if ABUSE_DRY_RUN:
        log.warning("[DRY-RUN] would %s acct=%s (tier=%s, sources=%d). %s",
                    ABUSE_ACTION, acct, tier, sources, report_url)
        record_action(target_id, acct, f"dry-run:{ABUSE_ACTION}", tier, sources, report_id)
        dm_moderator(f"[DRY-RUN] would {ABUSE_ACTION} @{acct} — {sources} distinct "
                     f"reporters, tier={tier}.{ip_note} Review: {report_url}")
        return

    try:
        apply_action(target_id, ABUSE_ACTION, note)
    except httpx.HTTPError as exc:
        log.error("report %s: failed to %s acct=%s: %s",
                  report_id, ABUSE_ACTION, acct, exc)
        return

    record_action(target_id, acct, ABUSE_ACTION, tier, sources, report_id)
    log.warning("auto-%sd acct=%s (tier=%s, %d distinct reporters); report left open",
                ABUSE_ACTION, acct, tier, sources)
    dm_moderator(f"🚨 Auto-{ABUSE_ACTION}d @{acct} — {sources} distinct reporters, "
                 f"account tier={tier}.{ip_note} Report left open for review: {report_url}")
    dm_silenced_user(acct)


# --- Routes ----------------------------------------------------------------


@app.get("/healthz")
def healthz():
    return {"ok": True}


@app.post("/webhook")
async def webhook(
    request: Request,
    background: BackgroundTasks,
    x_hub_signature: str | None = Header(default=None),
):
    raw = await request.body()

    if not verify_signature(raw, x_hub_signature):
        log.warning("rejected delivery: bad/missing signature")
        raise HTTPException(status_code=401, detail="invalid signature")

    try:
        payload = await request.json()
    except Exception:
        raise HTTPException(status_code=400, detail="invalid JSON")

    event = payload.get("event")
    obj = payload.get("object") or {}

    # Welcome on either event so it works whether registration approval is on
    # (account.created fires at signup, account.approved later) or off. The
    # dedup store ensures each account is welcomed exactly once.
    if event in ("account.created", "account.approved"):
        return _handle_account_created(obj, background, event)
    if event == "report.created":
        return _handle_report_created(obj, background)

    log.info("ignoring event=%s", event)
    return {"ok": True, "ignored": event}


def _handle_account_created(obj: dict, background: BackgroundTasks, event: str):
    # object is an Admin::Account; the public Account is nested under "account".
    nested = obj.get("account") or {}
    account_id = str(nested.get("id") or obj.get("id") or "")
    acct = nested.get("acct") or obj.get("username") or ""

    if not account_id or not acct:
        log.warning("could not extract account from payload: %s", obj)
        raise HTTPException(status_code=422, detail="no account in payload")

    if LOCAL_ONLY and ("@" in acct or obj.get("domain")):
        log.info("skipping non-local account acct=%s domain=%s", acct, obj.get("domain"))
        return {"ok": True, "skipped": "remote account"}

    # Respond 200 immediately; do the API call out-of-band so Mastodon's
    # delivery doesn't block on our outbound request.
    if event == "account.created":
        # First sighting of this account — classify its signup IP
        # (Admin::Account.ip) and email domain (Admin::Account.email) and
        # decide whether to welcome now or hold for account.approved.
        ip = obj.get("ip") or ""
        email = obj.get("email") or ""
        background.add_task(process_signup, account_id, acct, ip, email)
    else:
        # account.approved: a human has cleared the approval-required
        # registration gate. Always welcome (dedup makes this safe) — this is
        # what lifts a held welcome for a previously-flagged signup. Also the
        # moment a held, flagged signup's suspicious-watch clock should start;
        # a no-op if it was never flagged or already started (open-signup
        # case, where account.approved may fire right alongside
        # account.created).
        background.add_task(send_welcome, account_id, acct)
        background.add_task(maybe_start_suspicious_watch, account_id, acct)
    return {"ok": True, "queued": acct}


def _handle_report_created(obj: dict, background: BackgroundTasks):
    # object is an Admin::Report; target_account is the reported Admin::Account.
    report_id = str(obj.get("id") or "")
    target = obj.get("target_account") or {}
    nested = target.get("account") or {}
    target_id = str(target.get("id") or nested.get("id") or "")
    acct = nested.get("acct") or target.get("username") or ""
    is_local = not (target.get("domain") or "@" in acct)

    # Admin::Account.role: regular users get the default "everyone" role
    # (id -99); staff hold an assigned role with a positive id.
    role = target.get("role") or {}
    try:
        role_id = int(role.get("id")) if role.get("id") is not None else None
    except (TypeError, ValueError):
        role_id = None
    privileged = role_id is not None and role_id > 0

    if not report_id or not target_id or not acct:
        log.warning("could not extract report/target from payload: %s", obj)
        raise HTTPException(status_code=422, detail="no report/target in payload")

    # target_account is a full Admin::Account (per Mastodon's entity docs), so
    # its email is present here even if we never saw/recorded this account's
    # original signup (e.g. it predates this feature or CHECK_MAIL was down).
    email = target.get("email") or ""

    background.add_task(
        handle_report, report_id, target_id, acct, is_local,
        bool(target.get("suspended")), bool(target.get("silenced")), privileged,
        email,
    )
    return {"ok": True, "report": report_id, "target": acct}