pmb/yttrx-welcomebot
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
233458d6b8a1c5228193ff04d759528d426ea27a
yttrx-welcomebot/app/main.py
T
pmb 233458d6b8
docker-build-push / build-push (push) Successful in 24s
Details
Never block narrower than a /24 
A /32 block is close to useless against what IP scrutiny fires on: a
datacenter/VPN/proxy customer has the rest of the range sitting right
there, so blocking the single address we saw stops nothing the same
actor does a minute later from the address next door. The width ceiling
added 2026-09-13 made this the common case rather than the exception:
its fallback for an over-wide ASN route was the bare address, so a
flagged signup out of a /18 hosting allocation got a /32.

Add IP_SCRUTINY_MIN_BLOCK_PREFIX_V4/_V6 (24/48) as a floor under that
ceiling. Anything narrower than the floor -- a missing route, the /32
recorded when a route didn't contain the IP, a small hosting allocation,
or the fallback from a route wider than the ceiling -- is widened to the
network of that width containing the address. At the defaults the floor
equals the ceiling, so every v4 block is now exactly the /24 containing
the flagged address.

State the collateral cost rather than waving at it: at the live settings
(sign_up_block severity, ~30d expiry) one flagged signup now hard-refuses
registration from 256 addresses for a month. Moving the severity to
sign_up_requires_approval or shortening the expiry buys that back, and
setting the floor to 32/128 restores the old behaviour entirely.
2026-09-21 22:33:00 -07:00

2241 lines
102 KiB
Python
Raw Blame History

"""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=<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.
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 the
# IP_SCRUTINY_MIN_BLOCK_PREFIX network containing the address (a /24 by
# default) 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 floor width below — the
# /24 containing the address, not the address alone.
#
# 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 a /24 rather than the whole range, so a repeat offender out
# of another part of the same allocation needs a fresh block — 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"))
# Narrowest network a single auto-registered block may cover, as a prefix
# length — the floor under the ceiling above. Blocking one address is close to
# useless against what this fires on: a datacenter/VPN/proxy customer has the
# rest of the range sitting right there, so a /32 blocks the address we saw
# and nothing the same actor moves to a minute later. Anything narrower than
# this — a missing route, the /32 recorded when a route didn't contain the ip,
# a small hosting allocation, or the fallback from a route wider than the
# ceiling — is widened to the network of this width containing the address.
#
# The collateral cost is real and worth stating plainly rather than waving
# at: how much it hurts an innocent neighbour is set by the two policies
# around this one. At what yttrx actually runs
# (IP_SCRUTINY_IPBLOCK_SEVERITY=sign_up_block,
# IP_SCRUTINY_IPBLOCK_EXPIRES_IN=2629800, ~30 days) one flagged signup
# hard-refuses registration from 256 addresses for a month. Moving the
# severity to sign_up_requires_approval (neighbour is queued for a human
# rather than refused) or shortening the expiry buys that back; widening the
# block without touching either is a deliberate choice to spend it.
#
# At the default this equals the ceiling, so every v4 block is exactly the /24
# containing the flagged address. Set it to 32 (128 for v6) to restore the old
# behaviour of blocking only what the route says, down to the single address.
IP_SCRUTINY_MIN_BLOCK_PREFIX_V4 = int(os.environ.get("IP_SCRUTINY_MIN_BLOCK_PREFIX_V4", "24"))
# /48 is the v6 analogue, matching the v6 ceiling for the same reason.
IP_SCRUTINY_MIN_BLOCK_PREFIX_V6 = int(os.environ.get("IP_SCRUTINY_MIN_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 _block_width_bounds(version: int) -> tuple[int, int]:
"""(ceiling, floor) prefix lengths allowed for a block on this IP version.
Both are prefix lengths, so the ceiling — the widest network allowed — is
the SMALLER number. A floor configured wider than the ceiling is nonsense,
and the ceiling wins: it is the one that exists to bound collateral damage.
"""
host_bits = 32 if version == 4 else 128
cap = (IP_SCRUTINY_MAX_BLOCK_PREFIX_V4 if version == 4
else IP_SCRUTINY_MAX_BLOCK_PREFIX_V6)
floor = (IP_SCRUTINY_MIN_BLOCK_PREFIX_V4 if version == 4
else IP_SCRUTINY_MIN_BLOCK_PREFIX_V6)
cap = max(0, min(cap, host_bits))
floor = max(cap, min(floor, host_bits))
return cap, floor
def _block_floor_cidr(ip: str) -> str:
"""The narrowest network the bot will block for ip: the
IP_SCRUTINY_MIN_BLOCK_PREFIX_V4/V6 network containing it (a /24 or /48 by
default), or "" if ip doesn't parse.
This is what a missing, unusable or too-wide route degrades to. It used to
be the single /32 or /128, which is the wrong unit for the space these
flags fire on — the next signup comes from the address next door.
"""
try:
addr = ipaddress.ip_address(ip)
except ValueError:
return ""
floor = _block_width_bounds(addr.version)[1]
return str(ipaddress.ip_network(f"{addr}/{floor}", strict=False))
def capped_block_cidr(ip: str, cidr: str) -> str:
"""The network to actually register a block for: cidr, clamped to the
width window between the IP_SCRUTINY_MAX_BLOCK_PREFIX_V4/V6 ceiling
(anything wider is not a meaningful unit to block, so it degrades to the
floor network containing ip) and the IP_SCRUTINY_MIN_BLOCK_PREFIX_V4/V6
floor (anything narrower, a single address included, is widened to it).
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.
"""
floor_net = _block_floor_cidr(ip)
if not cidr:
return floor_net
try:
net = ipaddress.ip_network(cidr, strict=False)
except ValueError:
return floor_net
cap, floor = _block_width_bounds(net.version)
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,
floor_net or "the route itself", cap)
return floor_net or str(net)
if net.prefixlen > floor and floor_net:
log.info("block for ip=%s widened from %s to %s — narrower than the "
"/%d floor", ip, cidr, floor_net, floor)
return floor_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
and falls inside the allowed width window, else the
IP_SCRUTINY_MIN_BLOCK_PREFIX_V4/V6 network containing ip (a /24 or /48) —
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 to that same floor
network rather than blocked as-is (see capped_block_cidr) — that width
means an ISP aggregate, not a hosting allocation. API failure yields
("unknown", "", False, "<ip>/24|48") — 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, _block_floor_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 the floor-width network containing ip (a /24 or /48, see
_block_floor_cidr) if no cidr is given; refuses only if ip itself doesn't
parse (defensive only; classify_signup_ip always supplies a cidr).
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:
cidr = _block_floor_cidr(ip)
if not cidr:
log.warning("skipping ip_block registration for unparseable ip=%r (acct=%s)", ip, acct)
return False
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, notify) -> 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:
notify(
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, notify) -> 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:
notify(
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 = "",
notify=None) -> 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.
`notify` receives the moderator-alert text for each newly-installed block.
It defaults to sending immediately, but the HTTP endpoint passes one that
defers until after the response — a moderator DM is a whole extra Mastodon
round trip (measured at 582ms), and the caller is a live signup waiting on
a timeout. Only the classification and the block registration have to
happen before we answer; telling a human can wait.
"""
if notify is None:
notify = dm_moderator
result = {
"ip": _reputation_ip(ip.strip(), source, notify),
"email": _reputation_email(email.strip(), source, notify),
"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,
background: BackgroundTasks,
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.
Moderator DMs are deferred to after the response. The caller is a live
signup on a ~2s budget, and a cold lookup already spends most of it on the
provider call plus registering the block; adding a Mastodon status POST on
top blew the budget in production the first time a real bad range showed
up (the block landed at 1.35s, the DM pushed the answer past 2.0s and the
caller timed out). Nothing the caller acts on depends on the DM.
"""
_reputation_guard(x_reputation_key)
if not body.ip and not body.email:
raise HTTPException(status_code=400, detail="ip or email is required")
alerts: list[str] = []
result = reputation_check(ip=body.ip, email=body.email, source=body.source,
notify=alerts.append)
for message in alerts:
background.add_task(dm_moderator, message)
return result
@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}
Reference in new issue View Git Blame Copy Permalink