pmb/yttrx-welcomebot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add disposable/high-risk email signup scrutiny via check-mail.org

Browse Source 
Classifies each signup's email domain (domain-only, GDPR-friendly) alongside
the existing IP scrutiny signal, with matching held-welcome and auto
email_domain_block behavior. A report against an account with a flagged
domain suspends immediately (no reporter-count threshold) and blocks the
domain, classified live from the report payload rather than any signup-time
record so it also covers pre-existing accounts. Ships CHECK_MAIL_DRY_RUN=true
by default, independent of ABUSE_DRY_RUN, so the new report-triggered suspend
path stays inert until watched.
This commit is contained in:
waffle2k
2026-07-03 09:08:30 -07:00
parent 6ad9a93f38
commit 656eec09c0
5 changed files with 452 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/main.py
+262 -34
View File
@@ -7,10 +7,16 @@ and dispatches by event:
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/hosting via
RDAP) and email domain (disposable/high-risk via check-mail.org); a flagged
signup gets more scrutiny (held welcome, auto ip_block/email_domain_block,
moderator DM) before falling through to a normal welcome.
* ``report.created`` — evaluates the reported account and, when enough
*distinct* reporters have open reports against a young or dormant account,
auto-**silences** it (reversible) and DMs a moderator for review. Uses a
separate moderator bot token with admin scopes.
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:
@@ -145,12 +151,33 @@ IP_SCRUTINY_MOBILE_RE = re.compile(os.environ.get(
r"vodafone|telstra|sprint|three\b|orange mobile|deutsche telekom|o2\b",
), re.I)
# --- 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")
if not WEBHOOK_SECRET:
log.warning("WEBHOOK_SECRET is empty — signature verification will reject all requests.")
if not BOT_ACCESS_TOKEN:
log.warning("BOT_ACCESS_TOKEN is empty — the bot cannot post welcome messages.")
if ABUSE_ENABLED and not ABUSE_BOT_TOKEN:
log.warning("ABUSE_BOT_TOKEN is empty — report.created handling is disabled.")
if CHECK_MAIL_ENABLED and not CHECK_MAIL_API_KEY:
log.warning("CHECK_MAIL_API_KEY is empty — disposable-email signup scrutiny is disabled.")
app = FastAPI(title="yttrx welcome-bot + abuse-bot")
@@ -201,6 +228,26 @@ def _init_db() -> None:
" cached_at TEXT DEFAULT CURRENT_TIMESTAMP"
")"
)
conn.execute(
"CREATE TABLE IF NOT EXISTS signup_email ("
" account_id TEXT PRIMARY KEY,"
" acct TEXT,"
" domain TEXT,"
" is_disposable INTEGER,"
" risk INTEGER,"
" flagged INTEGER,"
" domain_blocked INTEGER DEFAULT 0,"
" classified_at TEXT DEFAULT CURRENT_TIMESTAMP"
")"
)
conn.execute(
"CREATE TABLE IF NOT EXISTS check_mail_cache ("
" domain TEXT PRIMARY KEY,"
" is_disposable INTEGER,"
" risk INTEGER,"
" cached_at TEXT DEFAULT CURRENT_TIMESTAMP"
")"
)
@contextmanager
@@ -294,6 +341,52 @@ def cache_whois_org(ip: str, org: str) -> None:
)
def record_signup_email(account_id: str, acct: str, domain: str, is_disposable: bool,
risk: int, flagged: bool) -> None:
with _db() as conn:
conn.execute(
"INSERT OR REPLACE INTO signup_email "
"(account_id, acct, domain, is_disposable, risk, flagged, domain_blocked) "
"VALUES (?, ?, ?, ?, ?, ?, "
" COALESCE((SELECT domain_blocked FROM signup_email WHERE account_id = ?), 0))",
(account_id, acct, domain, int(is_disposable), risk, int(flagged), account_id),
)
def get_signup_email(account_id: str) -> tuple[str, bool]:
"""Return (domain, flagged) for a signup's classified email domain, or
("", False) if it was never classified (check-mail disabled/unavailable)."""
with _db() as conn:
row = conn.execute(
"SELECT domain, flagged FROM signup_email WHERE account_id = ?", (account_id,)
).fetchone()
return (row[0], bool(row[1])) if row else ("", False)
def mark_email_domain_blocked(account_id: str) -> None:
with _db() as conn:
conn.execute(
"UPDATE signup_email SET domain_blocked = 1 WHERE account_id = ?",
(account_id,),
)
def cached_check_mail(domain: str) -> tuple[bool, int] | None:
with _db() as conn:
row = conn.execute(
"SELECT is_disposable, risk FROM check_mail_cache WHERE domain = ?", (domain,)
).fetchone()
return (bool(row[0]), row[1]) if row else None
def cache_check_mail(domain: str, is_disposable: bool, risk: int) -> None:
with _db() as conn:
conn.execute(
"INSERT OR IGNORE INTO check_mail_cache (domain, is_disposable, risk) VALUES (?, ?, ?)",
(domain, int(is_disposable), risk),
)
_init_db()
# --- Signature verification ------------------------------------------------
@@ -340,39 +433,110 @@ def send_welcome(account_id: str, acct: str) -> None:
acct, account_id, resp.json().get("id"))
def process_signup(account_id: str, acct: str, ip: str) -> None:
"""Classify a new signup's IP and act on account.created.
def classify_email_domain(domain: str) -> tuple[bool, int]:
"""Classify a signup's email domain via check-mail.org.
Non-flagged (or IP-scrutiny disabled / no ip in payload) signups are
welcomed immediately, same as before this feature existed. Flagged
(datacenter) signups get a moderator DM, an optional ip_blocks
registration, and — unless dry-run — have their welcome held until
account.approved fires (see _handle_account_created).
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 RDAP failure path.
"""
if not (IP_SCRUTINY_ENABLED and ip):
cached = cached_check_mail(domain)
if cached is not None:
return cached
try:
resp = httpx.post(
"https://api.check-mail.org/v2/",
headers={"Authorization": f"Bearer {CHECK_MAIL_API_KEY}"},
data={"domain": domain},
timeout=10.0,
)
resp.raise_for_status()
data = resp.json()
is_disposable = bool(data.get("is_disposable") or data.get("block"))
risk = int(data.get("risk") or 0)
except (httpx.HTTPError, ValueError) as exc:
log.warning("check-mail.org lookup failed for domain=%s: %s", domain, exc)
return False, 0
cache_check_mail(domain, is_disposable, risk)
return is_disposable, risk
def register_email_domain_block(domain: str, acct: str) -> None:
"""Register a flagged signup's email domain in Mastodon's native
Admin::EmailDomainBlock, blocking future signups from that domain.
Requires the admin:write:email_domain_blocks OAuth scope on
ABUSE_BOT_TOKEN — see CLAUDE.md's moderator-token-gotcha section.
"""
try:
resp = httpx.post(
f"{MASTODON_BASE_URL}/api/v1/admin/email_domain_blocks",
headers=_admin_headers(),
data={"domain": domain},
timeout=15.0,
)
if resp.status_code == 422:
log.info("email_domain_block for %s already exists (or invalid), treating as handled", domain)
else:
resp.raise_for_status()
except httpx.HTTPError as exc:
log.error("failed to register email_domain_block for %s (acct=%s): %s", domain, acct, exc)
def process_signup(account_id: str, acct: str, ip: str, email: str = "") -> None:
"""Classify a new signup's IP and email domain, and act on account.created.
Non-flagged (or both signals disabled/unavailable) signups are welcomed
immediately, same as before these features existed. A signup flagged by
either signal (datacenter IP and/or high-risk email domain) gets a
moderator DM, the matching auto-block(s) (ip_block / email_domain_block),
and — unless dry-run — has its welcome held until account.approved fires
(see _handle_account_created).
"""
reasons: list[str] = []
hold = False
if IP_SCRUTINY_ENABLED and ip:
classification, org = classify_signup_ip(ip)
ip_flagged = classification == "datacenter"
record_signup_ip(account_id, acct, ip, classification, org, ip_flagged)
if ip_flagged:
log.warning("flagged signup acct=%s ip=%s classification=%s org=%s",
acct, ip, classification, org)
if IP_SCRUTINY_AUTO_IPBLOCK and not IP_SCRUTINY_DRY_RUN:
register_ip_block(ip, acct, org)
mark_ipblock_registered(account_id)
prefix = "[DRY-RUN] " if IP_SCRUTINY_DRY_RUN else ""
reasons.append(f"{prefix}IP {ip} ({org or 'unknown org'}, datacenter/hosting)")
hold = hold or (IP_SCRUTINY_HOLD_WELCOME and not IP_SCRUTINY_DRY_RUN)
if CHECK_MAIL_ENABLED and CHECK_MAIL_API_KEY and email and "@" in email:
domain = email.rsplit("@", 1)[-1].strip().lower()
is_disposable, risk = classify_email_domain(domain)
email_flagged = is_disposable or risk >= CHECK_MAIL_RISK_THRESHOLD
record_signup_email(account_id, acct, domain, is_disposable, risk, email_flagged)
if email_flagged:
log.warning("flagged signup acct=%s email_domain=%s is_disposable=%s risk=%d",
acct, domain, is_disposable, risk)
if CHECK_MAIL_AUTO_DOMAIN_BLOCK and not CHECK_MAIL_DRY_RUN:
register_email_domain_block(domain, acct)
mark_email_domain_blocked(account_id)
prefix = "[DRY-RUN] " if CHECK_MAIL_DRY_RUN else ""
reasons.append(f"{prefix}email domain {domain} (disposable={is_disposable}, risk={risk})")
hold = hold or (CHECK_MAIL_HOLD_WELCOME and not CHECK_MAIL_DRY_RUN)
if not reasons:
send_welcome(account_id, acct)
return
classification, org = classify_signup_ip(ip)
flagged = classification == "datacenter"
record_signup_ip(account_id, acct, ip, classification, org, flagged)
if not flagged:
send_welcome(account_id, acct)
return
log.warning("flagged signup acct=%s ip=%s classification=%s org=%s",
acct, ip, classification, org)
if IP_SCRUTINY_AUTO_IPBLOCK and not IP_SCRUTINY_DRY_RUN:
register_ip_block(ip, acct, org)
mark_ipblock_registered(account_id)
hold = IP_SCRUTINY_HOLD_WELCOME and not IP_SCRUTINY_DRY_RUN
prefix = "[DRY-RUN] " if IP_SCRUTINY_DRY_RUN else ""
dm_moderator(
f"{prefix}⚠️ Flagged signup @{acct} from {ip} ({org or 'unknown org'}, "
f"datacenter/hosting)."
f"⚠️ Flagged signup @{acct}: {'; '.join(reasons)}."
+ (" Welcome DM held pending approval." if hold else "")
)
@@ -581,8 +745,25 @@ def dm_silenced_user(acct: str) -> None:
def handle_report(report_id: str, target_id: str, acct: str, is_local: bool,
suspended: bool, silenced: bool, privileged: bool = False) -> None:
"""Evaluate a reported account and silence it if the policy is met."""
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:
@@ -602,6 +783,46 @@ def handle_report(report_id: str, target_id: str, acct: str, is_local: bool,
report_id, acct)
return
report_url = f"{MASTODON_BASE_URL}/admin/reports/{report_id}"
if CHECK_MAIL_ENABLED and CHECK_MAIL_API_KEY and email and "@" in email:
domain = email.rsplit("@", 1)[-1].strip().lower()
is_disposable, risk = classify_email_domain(domain)
if is_disposable or risk >= CHECK_MAIL_RISK_THRESHOLD:
note = (f"Auto-suspend by abuse-bot: reported account's email domain "
f"'{domain}' is flagged high-risk by check-mail.org "
f"(disposable={is_disposable}, risk={risk}). Pending human "
f"review (report left open).")
if ABUSE_DRY_RUN or CHECK_MAIL_DRY_RUN:
log.warning("[DRY-RUN] would suspend acct=%s (flagged email domain=%s, "
"risk=%d). %s", acct, domain, risk, report_url)
record_action(target_id, acct, "dry-run:suspend", "email-risk", 1, report_id)
dm_moderator(f"[DRY-RUN] would suspend @{acct} — email domain "
f"'{domain}' is flagged high-risk (risk={risk}). "
f"Review: {report_url}")
return
try:
apply_action(target_id, "suspend", note)
except httpx.HTTPError as exc:
log.error("report %s: failed to suspend acct=%s (email-risk): %s",
report_id, acct, exc)
return
record_action(target_id, acct, "suspend", "email-risk", 1, report_id)
log.warning("auto-suspended acct=%s (flagged email domain=%s, risk=%d); "
"report left open", acct, domain, risk)
dm_moderator(f"🚨 Auto-suspended @{acct} — email domain '{domain}' flagged "
f"high-risk by check-mail.org (risk={risk}). Report left open "
f"for review: {report_url}")
if CHECK_MAIL_AUTO_DOMAIN_BLOCK:
record_signup_email(target_id, acct, domain, is_disposable, risk, True)
register_email_domain_block(domain, acct)
mark_email_domain_blocked(target_id)
return
try:
profile = classify_account(target_id)
sources = count_distinct_reporters(target_id)
@@ -631,7 +852,6 @@ def handle_report(report_id: str, target_id: str, acct: str, is_local: bool,
report_id, acct, sources, required)
return
report_url = f"{MASTODON_BASE_URL}/admin/reports/{report_id}"
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}")
@@ -716,10 +936,12 @@ def _handle_account_created(obj: dict, background: BackgroundTasks, event: str):
# 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 decide whether to welcome now or hold for account.approved.
# 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 ""
background.add_task(process_signup, account_id, acct, ip)
email = obj.get("email") or ""
background.add_task(process_signup, account_id, acct, ip, email)
else:
# account.approved: a human has cleared the approval-required
# registration gate. Always welcome (dedup makes this safe) — this is
@@ -750,8 +972,14 @@ def _handle_report_created(obj: dict, background: BackgroundTasks):
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}