"""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. A signup flagged with a *strong* IP signal (datacenter/proxy/tor/abuser, not vpn alone) is suspended immediately regardless of the email signal. * ``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= 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. 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 import hashlib import hmac import ipaddress 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, 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 a flagged signup's network into Mastodon's native # Admin::IpBlock. Blocks the ASN/route-level CIDR ipapi.is reports the IP # belonging to (see classify_signup_ip's block_cidr), not just the single # address β€” a flagged signup is almost always datacenter/VPN/proxy space, and # a repeat bad actor is far more likely to come back from a different address # in the same block than the exact same one. Falls back to a single-address # (/32 or /128) block if no usable route was returned/cached. 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") # Signals stronger than vpn alone β€” vpn also flags plenty of privacy-conscious # real users, but datacenter/proxy/tor/independently-scored-abuser is a much # cleaner bulk/bot-signup indicator. A signup flagged with any of these # suspends immediately at signup, regardless of the email-domain signal, # instead of just the held-welcome/auto-ipblock/lowered-threshold treatment # every other flagged signup gets. # Widest network a single auto-registered block may cover, as a prefix # length. ipapi.is reports the flagged address's whole ASN route, which is the # right unit for hosting space (a bad actor comes back from a different # address in the same VPS block) but badly wrong for an ISP aggregate: routes # as wide as /8, /9 and /11 show up in the cache, and one `abuser` flag on a # residential or mobile address inside one of those would otherwise wall off # millions of consumer addresses. A route wider than this is treated as "not a # meaningful unit to block" and falls back to the single address. # # This deliberately costs blocking strength against large hosting allocations # β€” anything wider than a /24 (which is most of the VPS ranges registered so # far) now blocks one address rather than the range, so a repeat offender out # of the same block needs a fresh block each time β€” in exchange for making the # mobile/residential collateral-damage case impossible. # Applied at registration time, not retroactively β€” blocks already in force # keep whatever width they were registered with. IP_SCRUTINY_MAX_BLOCK_PREFIX_V4 = int(os.environ.get("IP_SCRUTINY_MAX_BLOCK_PREFIX_V4", "24")) # /48 is the v6 analogue: the standard end-site allocation, and the rough # equivalent of "one customer's block" that a /24 represents in v4. IP_SCRUTINY_MAX_BLOCK_PREFIX_V6 = int(os.environ.get("IP_SCRUTINY_MAX_BLOCK_PREFIX_V6", "48")) # How long a registered Admin::IpBlock lasts, in seconds. 0 = never expires. # # An IP block is a statement about who is *currently* behind an address, and # that goes stale: VPS addresses are recycled between customers constantly, # and with the /24 ceiling above most blocks now cover a single address, so a # permanent block increasingly punishes whoever gets that address next rather # than the actor we saw. 24h is long enough to break a signup run (which is # what the abuse actually looks like β€” bursts, not a steady trickle) and short # enough that the blocklist stays a live picture instead of accumulating # forever. # # Mastodon expires these itself (Scheduler::IpCleanupScheduler), and # app/reconcile_blocks.py already treats an elapsed expires_at as "gone", so # an expired block drops out of /api/blocks on its own. IP_SCRUTINY_IPBLOCK_EXPIRES_IN = int(os.environ.get("IP_SCRUTINY_IPBLOCK_EXPIRES_IN", "86400")) IP_SCRUTINY_STRONG_SIGNALS = frozenset({"datacenter", "proxy", "tor", "abuser"}) IP_SCRUTINY_STRONG_SUSPEND_ENABLED = os.environ.get( "IP_SCRUTINY_STRONG_SUSPEND_ENABLED", "true" ).lower() in ("1", "true", "yes") IP_SCRUTINY_STRONG_SUSPEND_ACTION = os.environ.get("IP_SCRUTINY_STRONG_SUSPEND_ACTION", "suspend").lower() # Rollout safety switch β€” ships dry-run-first, same reasoning as # SUSPICIOUS_COMBINED_DRY_RUN, since this is a brand-new action path. IP_SCRUTINY_STRONG_SUSPEND_DRY_RUN = os.environ.get( "IP_SCRUTINY_STRONG_SUSPEND_DRY_RUN", "true" ).lower() in ("1", "true", "yes") # --- 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") # --- Combined-signal immediate suspend (both IP AND email flagged) --- # The sweep above acts on EITHER signal after SUSPICIOUS_GRACE_HOURS of no # activity. A signup flagged by BOTH signals at once is a stronger indicator # than either alone, so it's acted on immediately in process_signup() instead # of waiting for the hourly sweep β€” this is additive, it doesn't change the # OR-based sweep's behavior for single-signal flags. SUSPICIOUS_COMBINED_ENABLED = os.environ.get("SUSPICIOUS_COMBINED_ENABLED", "true").lower() in ("1", "true", "yes") SUSPICIOUS_COMBINED_ACTION = os.environ.get("SUSPICIOUS_COMBINED_ACTION", "suspend").lower() # Rollout safety switch β€” ships dry-run-first (unlike SUSPICIOUS_DRY_RUN's # by-design "false" default above) since this is a brand-new action path. SUSPICIOUS_COMBINED_DRY_RUN = os.environ.get("SUSPICIOUS_COMBINED_DRY_RUN", "true").lower() in ("1", "true", "yes") # --- Public blocklist API --------------------------------------------------- # Read-only JSON of the blocks this bot has registered (GET /api/blocks and # friends). Served WITHOUT authentication when enabled, so it defaults to # "off" and must be opted into per-deployment: publishing which ranges and # domains are already blocked also tells a would-be spam signup which ones to # avoid, and that's a per-instance call, not a safe default. yttrx.com sets # BLOCKS_API_ENABLED=true in its .env; every other deployment 404s. BLOCKS_API_ENABLED = os.environ.get("BLOCKS_API_ENABLED", "false").lower() in ("1", "true", "yes") # Hard ceiling on how many entries one response may return, whatever ?limit # asks for β€” this endpoint is unauthenticated, so the response size can't be # left to the caller. BLOCKS_API_MAX_ITEMS = int(os.environ.get("BLOCKS_API_MAX_ITEMS", "5000")) # Public base URL this API is reached at, used only to print a "source:" line # in the text format's header comments so a downloaded list says where it came # from. Not used for routing. BLOCKS_API_PUBLIC_URL = os.environ.get( "BLOCKS_API_PUBLIC_URL", "https://hooks.yttrx.com").rstrip("/") # Cache-Control max-age (seconds) on blocklist responses. The underlying data # changes only when a signup is flagged, so letting clients and any # 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: 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.") 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") # --- 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 _column_exists(conn: sqlite3.Connection, table: str, column: str) -> bool: return any(row[1] == column for row in conn.execute(f"PRAGMA table_info({table})")) 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," " block_cidr TEXT," " block_active INTEGER DEFAULT 1," " block_checked_at TEXT," " classified_at TEXT DEFAULT CURRENT_TIMESTAMP" ")" ) # block_cidr was added after the table shipped (it records the network # register_ip_block actually blocks, which is what /api/blocks # publishes). Rows written before this column existed keep it NULL and # fall back to the single-address /32 or /128 of their signup IP β€” # which is exactly what was registered back then, since blocking the # ASN route CIDR only started 2026-07-15. if not _column_exists(conn, "signup_ip", "block_cidr"): conn.execute("ALTER TABLE signup_ip ADD COLUMN block_cidr TEXT") # block_active tracks whether the block we registered is STILL present # on the Mastodon side. ipblock_registered only ever meant "we posted # it once"; a moderator clearing blocks by hand (as happened to 24 # rows on 2026-07-21) or an expires_at elapsing would leave us # publishing entries that no longer block anything. Maintained by # app/reconcile_blocks.py; rows default to active until it says # otherwise, so a reconcile that never runs degrades to the old # behaviour rather than emptying the list. if not _column_exists(conn, "signup_ip", "block_active"): conn.execute("ALTER TABLE signup_ip ADD COLUMN block_active INTEGER DEFAULT 1") if not _column_exists(conn, "signup_ip", "block_checked_at"): conn.execute("ALTER TABLE signup_ip ADD COLUMN block_checked_at TEXT") conn.execute( "CREATE TABLE IF NOT EXISTS ipapi_range_cache (" " family INTEGER NOT NULL," # 4 or 6 " range_start BLOB NOT NULL," # packed address bytes, fixed width per family " range_end BLOB NOT NULL," " cidr TEXT," # human-readable, for inspection only " is_datacenter INTEGER," " is_vpn INTEGER," " is_proxy INTEGER," " is_tor INTEGER," " is_abuser INTEGER," " org TEXT," " cached_at TEXT DEFAULT CURRENT_TIMESTAMP," " PRIMARY KEY (family, range_start, range_end)" ")" ) 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," " block_active INTEGER DEFAULT 1," " block_checked_at TEXT," " classified_at TEXT DEFAULT CURRENT_TIMESTAMP" ")" ) if not _column_exists(conn, "signup_email", "block_active"): conn.execute("ALTER TABLE signup_email ADD COLUMN block_active INTEGER DEFAULT 1") if not _column_exists(conn, "signup_email", "block_checked_at"): conn.execute("ALTER TABLE signup_email ADD COLUMN block_checked_at TEXT") 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" ")" ) # 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," " expires_at TEXT," " PRIMARY KEY (kind, target)" ")" ) # expires_at arrived with IP_SCRUTINY_IPBLOCK_EXPIRES_IN. Rows written # before it keep NULL, which reads as "no expiry" β€” correct, since # they were registered without one. if not _column_exists(conn, "reputation_block", "expires_at"): conn.execute("ALTER TABLE reputation_block ADD COLUMN expires_at TEXT") 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, block_cidr: str = "") -> None: with _db() as conn: conn.execute( "INSERT OR REPLACE INTO signup_ip " "(account_id, acct, ip, classification, org, flagged, " " ipblock_registered, block_cidr) " "VALUES (?, ?, ?, ?, ?, ?, " " COALESCE((SELECT ipblock_registered FROM signup_ip WHERE account_id = ?), 0), ?)", (account_id, acct, ip, classification, org, int(flagged), account_id, block_cidr or None), ) 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: # block_active is reset too: if this range was reconciled away earlier and # we've just registered it again, it is live again. with _db() as conn: conn.execute( "UPDATE signup_ip SET ipblock_registered = 1, block_active = 1 " "WHERE account_id = ?", (account_id,), ) def _ip_range_bounds(ip: str, route: str = "") -> tuple[int, bytes, bytes, str] | None: """Compute the (family, range_start, range_end, cidr) to cache ip's classification under: route (ipapi.is's asn.route CIDR) when it's a valid network that actually contains ip, else a single-address range. bytes are fixed-width packed addresses (4 bytes for v4, 16 for v6), so lexicographic BLOB comparison in sqlite matches numeric address order. Returns None if ip itself doesn't parse. """ try: addr = ipaddress.ip_address(ip) except ValueError: log.warning("skipping ipapi range cache lookup for unparseable ip=%r", ip) return None if route: try: net = ipaddress.ip_network(route, strict=False) except ValueError: net = None if net is not None and net.version == addr.version and addr in net: return (net.version, net.network_address.packed, net.broadcast_address.packed, f"{net.network_address}/{net.prefixlen}") prefix_len = 32 if addr.version == 4 else 128 return addr.version, addr.packed, addr.packed, f"{addr}/{prefix_len}" def cached_ip_intel(ip: str) -> dict | None: bounds = _ip_range_bounds(ip) if bounds is None: return None family, packed, _, _ = bounds with _db() as conn: row = conn.execute( "SELECT is_datacenter, is_vpn, is_proxy, is_tor, is_abuser, org, cidr " "FROM ipapi_range_cache " "WHERE family = ? AND range_start <= ? AND range_end >= ? LIMIT 1", (family, packed, packed), ).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 "", "cidr": row[6] or bounds[3], } def cache_ip_intel(ip: str, intel: dict, route: str = "") -> None: bounds = _ip_range_bounds(ip, route) if bounds is None: return family, range_start, range_end, cidr = bounds with _db() as conn: conn.execute( "INSERT OR IGNORE INTO ipapi_range_cache " "(family, range_start, range_end, cidr, is_datacenter, is_vpn, " " is_proxy, is_tor, is_abuser, org) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (family, range_start, range_end, cidr, 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, block_active = 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), ) def _single_address_cidr(ip: str) -> str: """The /32 or /128 for ip, or "" if it doesn't parse.""" try: addr = ipaddress.ip_address(ip) except ValueError: return "" return f"{addr}/{32 if addr.version == 4 else 128}" def capped_block_cidr(ip: str, cidr: str) -> str: """The network to actually register a block for: cidr, unless it's wider than IP_SCRUTINY_MAX_BLOCK_PREFIX_V4/V6, in which case the single address. Applied when deciding what to BLOCK, not when caching the classification β€” the wide ASN route is still the right cache key (the whole route really does belong to one operator, so one lookup answers for all of it), it is just the wrong thing to hand to Admin::IpBlock. """ fallback = _single_address_cidr(ip) if not cidr: return fallback try: net = ipaddress.ip_network(cidr, strict=False) except ValueError: return fallback cap = (IP_SCRUTINY_MAX_BLOCK_PREFIX_V4 if net.version == 4 else IP_SCRUTINY_MAX_BLOCK_PREFIX_V6) if net.prefixlen < cap: log.info("block for ip=%s narrowed from route %s (%d addresses) to %s β€” " "wider than the /%d ceiling", ip, cidr, net.num_addresses, fallback or "the single address", cap) return fallback or str(net) return str(net) 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, the reconcile pass hasn't since found it gone, and it hasn't expired. Lets a repeat query skip the Mastodon round trip β€” this runs on the signup request path, so a saved POST is saved latency. The expiry check matters independently of the reconcile pass: IP blocks now last IP_SCRUTINY_IPBLOCK_EXPIRES_IN, and reconcile only runs hourly, so between a block lapsing and the next reconcile this would otherwise claim a block that Mastodon has already dropped β€” and skip re-installing it. A NULL expires_at means no expiry (email domain blocks never expire, and neither did IP blocks before this was introduced). """ with _db() as conn: row = conn.execute( "SELECT 1 FROM reputation_block " "WHERE kind = ? AND target = ? AND block_active = 1 " " AND (expires_at IS NULL OR expires_at > ?)", (kind, target, _now_iso()), ).fetchone() return row is not None def record_reputation_block(kind: str, target: str, classification: str, org: str, expires_in: int = 0) -> 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. expires_in mirrors what was sent to Mastodon, so our dedup record lapses at the same moment the block itself does; 0 means no expiry. """ expires_at = None if expires_in > 0: expires_at = (datetime.now(timezone.utc).replace(microsecond=0) + timedelta(seconds=expires_in)).isoformat() with _db() as conn: conn.execute( "INSERT INTO reputation_block " "(kind, target, classification, org, expires_at) VALUES (?, ?, ?, ?, ?) " "ON CONFLICT(kind, target) DO UPDATE SET " " queries = queries + 1," " last_seen_at = CURRENT_TIMESTAMP," " block_active = 1," " expires_at = excluded.expires_at," " classification = COALESCE(excluded.classification, classification)," " org = COALESCE(excluded.org, org)", (kind, target, classification or None, org or None, expires_at), ) 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 " " AND (expires_at IS NULL OR expires_at > ?) " "ORDER BY last_seen_at DESC", (kind, _now_iso()), ).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.""" try: addr = ipaddress.ip_address(ip) except ValueError: return "" return f"{addr}/{32 if addr.version == 4 else 128}" def blocked_ip_ranges(limit: int = 0) -> list[dict]: """Every network this bot has registered an Admin::IpBlock for that is still present on the Mastodon side, newest first, collapsed to one entry per distinct CIDR. Rows the reconcile pass found gone or expired are excluded β€” see app/reconcile_blocks.py. Repeat signups out of the same range are common (that's the point of blocking the range rather than the address), so entries carry a `signups` count and the first/last time the range was hit rather than appearing once per account. Entries state only *that* a range is blocked. The ipapi.is classification and org that drove the decision stay out of the response on purpose: this is an unauthenticated endpoint, and republishing a paid IP-reputation feed's verdicts would be redistributing their data, not ours. The same goes for identifying anything about the accounts involved. """ with _db() as conn: rows = conn.execute( "SELECT block_cidr, ip, classified_at " "FROM signup_ip WHERE ipblock_registered = 1 AND block_active = 1 " "ORDER BY classified_at DESC" ).fetchall() grouped: dict[str, dict] = {} for block_cidr, ip, classified_at in rows: cidr = block_cidr or _fallback_cidr(ip or "") if not cidr: continue entry = grouped.get(cidr) if entry is None: grouped[cidr] = { "cidr": cidr, "signups": 1, "first_blocked_at": classified_at, "last_blocked_at": classified_at, } else: # rows are newest-first, so every later row is an older sighting entry["signups"] += 1 entry["first_blocked_at"] = classified_at _merge_reputation_blocks(grouped, "ip", "cidr") return _capped(_newest_first(grouped), limit) def blocked_email_domains(limit: int = 0) -> list[dict]: """Every domain this bot has registered an Admin::EmailDomainBlock for that is still present on the Mastodon side, newest first, one entry per distinct domain. See app/reconcile_blocks.py. Domains only β€” the bot never stores or forwards a full signup address (see classify_email_domain), so there is none to expose here. As with blocked_ip_ranges, entries state only *that* a domain is blocked: check-mail.org's disposable flag and risk score stay out of the response so this doesn't republish a third-party reputation feed's verdicts. """ with _db() as conn: rows = conn.execute( "SELECT domain, classified_at " "FROM signup_email WHERE domain_blocked = 1 AND block_active = 1 " "ORDER BY classified_at DESC" ).fetchall() grouped: dict[str, dict] = {} for domain, classified_at in rows: if not domain: continue entry = grouped.get(domain) if entry is None: grouped[domain] = { "domain": domain, "signups": 1, "first_blocked_at": classified_at, "last_blocked_at": classified_at, } else: entry["signups"] += 1 entry["first_blocked_at"] = classified_at _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]: """Apply a caller's ?limit on top of BLOCKS_API_MAX_ITEMS. limit <= 0 means "as many as the ceiling allows".""" ceiling = max(BLOCKS_API_MAX_ITEMS, 0) n = ceiling if limit <= 0 else min(limit, ceiling) return items[:n] _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) -> bool: """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. Returns True if the domain 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. """ 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) return False return True def process_signup(account_id: str, acct: str, ip: str, email: str = "", approved: bool = False) -> 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). `approved` is the account's *current* Admin::Account.approved state at signup time: if it's already true (open registration, never queued), account.approved will never be delivered for it, so the hold is lifted immediately instead of waiting forever. A signup whose IP is flagged with a *strong* signal β€” datacenter, proxy, tor, or abuser, but NOT vpn alone (see IP_SCRUTINY_STRONG_SIGNALS) β€” is auto-actioned immediately here regardless of the email-domain signal, same early-return shape as the combined-signal path below. A vpn-only flag falls through to the normal hold/welcome path unless also combined with a flagged email domain. A signup flagged by BOTH signals at once (see SUSPICIOUS_COMBINED_ENABLED) is auto-actioned immediately here rather than falling through to the hold/welcome path β€” no welcome, no suspicious-watch entry, since the account is already suspended. This is additive to the existing hourly sweep (app/suspicious_sweep.py), which still separately handles single-signal flags after SUSPICIOUS_GRACE_HOURS of inactivity. """ reasons: list[str] = [] hold = False ip_flagged = False ip_strong_flagged = False email_flagged = False if IP_SCRUTINY_ENABLED and ip: classification, org, ip_flagged, block_cidr = classify_signup_ip(ip) record_signup_ip(account_id, acct, ip, classification, org, ip_flagged, block_cidr) if ip_flagged: ip_strong_flagged = bool(set(classification.split("+")) & IP_SCRUTINY_STRONG_SIGNALS) 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: if register_ip_block(ip, acct, org, block_cidr): 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: if 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) allowlisted = (acct.split("@")[0].lower() in ABUSE_ALLOWLIST or acct.lower() in ABUSE_ALLOWLIST) if ip_strong_flagged and IP_SCRUTINY_STRONG_SUSPEND_ENABLED and not allowlisted: action = IP_SCRUTINY_STRONG_SUSPEND_ACTION note = ( f"Auto-{action} at signup: signup IP flagged as {classification} " f"({'; '.join(reasons)})." ) if IP_SCRUTINY_STRONG_SUSPEND_DRY_RUN: log.warning("[DRY-RUN] would %s acct=%s immediately (strong IP signal: %s)", action, acct, classification) dm_moderator( f"[DRY-RUN] would {action} @{acct} immediately β€” strong IP signal " f"({'; '.join(reasons)})." ) else: try: apply_action(account_id, action, note) except httpx.HTTPError as exc: log.error("acct=%s: failed to immediately %s on strong IP signal, " "falling back to the normal flagged-signup path: %s", acct, action, exc) else: log.warning("auto-%sd acct=%s immediately β€” strong IP signup signal (%s)", action, acct, classification) dm_moderator( f"🚨 Auto-{action}d @{acct} immediately β€” strong IP signal " f"({'; '.join(reasons)})." ) return if ip_flagged and email_flagged and SUSPICIOUS_COMBINED_ENABLED and not allowlisted: note = ( f"Auto-{SUSPICIOUS_COMBINED_ACTION} at signup: flagged by BOTH " f"IP-scrutiny and email-domain scrutiny ({'; '.join(reasons)})." ) if SUSPICIOUS_COMBINED_DRY_RUN: log.warning("[DRY-RUN] would %s acct=%s immediately (IP+email combined signal)", SUSPICIOUS_COMBINED_ACTION, acct) dm_moderator( f"[DRY-RUN] would {SUSPICIOUS_COMBINED_ACTION} @{acct} immediately β€” " f"flagged by BOTH IP and email signals ({'; '.join(reasons)})." ) else: try: apply_action(account_id, SUSPICIOUS_COMBINED_ACTION, note) except httpx.HTTPError as exc: log.error("acct=%s: failed to immediately %s on combined signal, " "falling back to the normal flagged-signup path: %s", acct, SUSPICIOUS_COMBINED_ACTION, exc) else: log.warning("auto-%sd acct=%s immediately β€” combined IP+email signup signal", SUSPICIOUS_COMBINED_ACTION, acct) dm_moderator( f"🚨 Auto-{SUSPICIOUS_COMBINED_ACTION}d @{acct} immediately β€” " f"flagged by BOTH IP and email signals ({'; '.join(reasons)})." ) return 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 or approved: # Either never held, or held but Mastodon already auto-approved this # signup (open registration, no moderator queue) β€” account.approved # is not coming, so act now instead of waiting for it. send_welcome(account_id, acct) maybe_start_suspicious_watch(account_id, acct) # else: genuinely queued for a moderator's decision β€” 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 admin account API. Split out from maybe_start_suspicious_watch as a single seam to stub in tests. Uses the *admin* endpoint (like suspicious_sweep._fetch_admin_account), not the public one β€” the public accounts API 404s for any account that hasn't confirmed its email yet, which most flagged signups haven't at the moment account.created/approved fires, silently dropping them from the watch list before the grace-period clock ever starts. """ resp = httpx.get( f"{MASTODON_BASE_URL}/api/v1/admin/accounts/{account_id}", headers=_admin_headers(), timeout=15.0, ) resp.raise_for_status() data = resp.json() account = data.get("account") or {} return account.get("statuses_count", 0), account.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, str]: """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, block_cidr). 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. block_cidr is the network register_ip_block should block when flagged β€” ipapi.is's asn.route when it's a valid network that actually contains ip, else ip's own /32 or /128 β€” since a flagged signup is almost always datacenter/VPN/proxy space, where a repeat bad actor is far more likely to come back from a different address in the same block than the exact same one. A route wider than the IP_SCRUTINY_MAX_BLOCK_PREFIX_V4/V6 ceiling is narrowed back to the single address (see capped_block_cidr) β€” that width means an ISP aggregate, not a hosting allocation. API failure yields ("unknown", "", False, "/32|128") β€” 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, _single_address_cidr(ip) or f"{ip}/32" # ipapi.is returns company/asn as flat strings, not nested dicts company = data.get("company") org_name = company if isinstance(company, str) else (company or {}).get("name", "") if not org_name: asn_str = data.get("asn", "") if isinstance(asn_str, str) and " " in asn_str: org_name = asn_str.split(" ", 1)[1] # "AS15169 Google LLC" β†’ "Google LLC" asn_obj = data.get("asn") if isinstance(asn_obj, dict): route = asn_obj.get("route", "") else: route = "" # ipapi.is free tier returns asn as string, not nested object 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": org_name, } bounds = _ip_range_bounds(ip, route) intel["cidr"] = bounds[3] if bounds else f"{ip}/32" cache_ip_intel(ip, intel, route) 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" block_cidr = capped_block_cidr(ip, intel.get("cidr") or "") return classification, intel["org"], bool(reasons), block_cidr or f"{ip}/32" def _ip_block_params(cidr: str, comment: str) -> dict: """POST body for Admin::IpBlock. expires_in is omitted entirely when the expiry is switched off, rather than sent as 0 β€” Mastodon validates it against a fixed list of durations and would reject the 0.""" params = {"ip": cidr, "severity": IP_SCRUTINY_IPBLOCK_SEVERITY, "comment": comment} if IP_SCRUTINY_IPBLOCK_EXPIRES_IN > 0: params["expires_in"] = IP_SCRUTINY_IPBLOCK_EXPIRES_IN return params 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 network the signup IP belongs to, so a repeat bad actor from the same datacenter/VPN/proxy block is blocked too, not just this one address. 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. """ if not cidr: 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 False cidr = f"{ip}/{prefix_len}" 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", headers=_admin_headers(), data=_ip_block_params(cidr, 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 False 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, IP_SCRUTINY_IPBLOCK_EXPIRES_IN) 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, IP_SCRUTINY_IPBLOCK_EXPIRES_IN) 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. 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-suspend @{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) if 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} # --- Blocklist API --- # Read-only JSON view of what this bot has auto-blocked, served unauthenticated # when BLOCKS_API_ENABLED is on (off by default β€” see the config block). It # reports only the bot's own record: blocks a human added by hand in Mastodon's # admin UI are not in here. Responses assert only that a range or domain is # blocked β€” no account handles, account ids or email addresses, and none of # the ipapi.is / check-mail.org classification data behind the decision (see # blocked_ip_ranges). def _blocks_guard(limit: int) -> int: """404 unless the API is switched on for this deployment; a disabled endpoint should be indistinguishable from one that doesn't exist. Returns the sanitised limit.""" if not BLOCKS_API_ENABLED: raise HTTPException(status_code=404, detail="Not Found") return max(limit, 0) def _blocks_headers() -> dict: return { "Cache-Control": f"public, max-age={BLOCKS_API_CACHE_SECONDS}", # The list is public anyway, and other instances' admin tooling may # want to read it from a browser context. "Access-Control-Allow-Origin": "*", } def _blocks_response(payload: dict) -> JSONResponse: return JSONResponse(payload, headers=_blocks_headers()) def _blocks_text(items: list[dict], key: str, what: str, path: str) -> PlainTextResponse: """Render one list as a plain-text blocklist: comment header, then one entry per line. Blocklists are traded as text files, so this is the form another admin's cron job can consume without a JSON parser β€” strip the header with `grep -v '^#'`.""" lines = [ f"# yttrx.com β€” {what} automatically blocked at signup", f"# source: {BLOCKS_API_PUBLIC_URL}{path}", f"# generated: {_now_iso()}", f"# entries: {len(items)}", "#", "# One entry per line; lines starting with '#' are comments. The JSON", "# form of this URL adds first/last-seen timestamps and signup counts.", ] lines.extend(str(entry[key]) for entry in items) return PlainTextResponse("\n".join(lines) + "\n", headers=_blocks_headers()) def _blocks_format(value: str) -> str: """Normalise the ?format= query param, rejecting anything unrecognised rather than silently serving JSON.""" fmt = (value or "json").strip().lower() if fmt in ("txt", "text", "plain"): return "txt" if fmt == "json": return "json" raise HTTPException(status_code=400, detail="format must be 'json' or 'txt'") def _now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() @app.get("/api/blocks") def api_blocks(limit: int = 0, format: str = "json"): """Both blocklists in one document. JSON only: a text rendering would put CIDRs and domains in one file, and the whole point of the text format is that it pipes straight into a tool that expects one kind of entry. Callers wanting text use the per-list endpoints. """ limit = _blocks_guard(limit) if _blocks_format(format) == "txt": raise HTTPException( status_code=400, detail="text format is per-list: use /api/blocks/ip-ranges?format=txt " "or /api/blocks/email-domains?format=txt", ) ip_ranges = blocked_ip_ranges(limit) email_domains = blocked_email_domains(limit) return _blocks_response({ "generated_at": _now_iso(), "ip_ranges": {"count": len(ip_ranges), "items": ip_ranges}, "email_domains": {"count": len(email_domains), "items": email_domains}, }) @app.get("/api/blocks/ip-ranges") def api_blocks_ip_ranges(limit: int = 0, format: str = "json"): limit = _blocks_guard(limit) fmt = _blocks_format(format) items = blocked_ip_ranges(limit) if fmt == "txt": return _blocks_text(items, "cidr", "IP ranges", "/api/blocks/ip-ranges") return _blocks_response({ "generated_at": _now_iso(), "count": len(items), "items": items, }) @app.get("/api/blocks/email-domains") def api_blocks_email_domains(limit: int = 0, format: str = "json"): limit = _blocks_guard(limit) fmt = _blocks_format(format) items = blocked_email_domains(limit) if fmt == "txt": return _blocks_text(items, "domain", "email domains", "/api/blocks/email-domains") return _blocks_response({ "generated_at": _now_iso(), "count": len(items), "items": items, }) # --- 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, 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 "" # Admin::Account.approved is already true here for any signup that # Mastodon auto-approved as part of open registration (no moderator # queue involved) β€” account.approved never fires for those, so # process_signup must act now instead of waiting for it. approved = bool(obj.get("approved")) background.add_task(process_signup, account_id, acct, ip, email, approved) 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}