yttrx-welcomebot/app/suspicious_sweep.py

"""Suspicious-signup sweep.

Suspends (SUSPICIOUS_ACTION) accounts that were flagged at signup by either
IP-scrutiny (datacenter/hosting IP) or email-domain scrutiny (disposable/
high-risk domain) and show zero new posts AND zero new follows within
SUSPICIOUS_GRACE_HOURS of going live.

Only acts on accounts app.main.maybe_start_suspicious_watch already recorded
in the `suspicious_watch` table — this module doesn't touch signup
classification itself, just resolves the grace-period clock those signals
started.

Run on a schedule (cron on admin.yttrx.com, see CLAUDE.md):

    docker exec yttrx-welcomebot python -m app.suspicious_sweep

Requires ABUSE_BOT_TOKEN to carry the admin:read:accounts scope in addition
to its existing scopes (see CLAUDE.md's moderator-token-gotcha section) —
without it, fetching current account state 403s and that account is skipped
(logged as an error) until the token is re-minted.
"""

from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone

import httpx

from app.main import (
    ABUSE_ALLOWLIST,
    MASTODON_BASE_URL,
    SUSPICIOUS_ACTION,
    SUSPICIOUS_DRY_RUN,
    SUSPICIOUS_GRACE_HOURS,
    SUSPICIOUS_SWEEP_ENABLED,
    ABUSE_SKIP_PRIVILEGED,
    _admin_headers,
    _db,
    apply_action,
    dm_moderator,
)

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


def _pending_due() -> list[tuple]:
    cutoff = (datetime.now(timezone.utc) - timedelta(hours=SUSPICIOUS_GRACE_HOURS)).isoformat()
    with _db() as conn:
        return conn.execute(
            "SELECT account_id, acct, reasons, baseline_statuses_count, "
            "       baseline_following_count "
            "FROM suspicious_watch WHERE status = 'pending' AND started_at <= ?",
            (cutoff,),
        ).fetchall()


def _resolve(account_id: str, status: str) -> None:
    with _db() as conn:
        conn.execute(
            "UPDATE suspicious_watch SET status = ?, resolved_at = CURRENT_TIMESTAMP "
            "WHERE account_id = ?",
            (status, account_id),
        )


def _fetch_admin_account(account_id: str) -> dict | None:
    """GET the Admin::Account entity (role, suspended/silenced, nested public
    account with live statuses_count/following_count). Returns None if the
    account no longer exists (e.g. deleted since being flagged)."""
    resp = httpx.get(
        f"{MASTODON_BASE_URL}/api/v1/admin/accounts/{account_id}",
        headers=_admin_headers(), timeout=15.0,
    )
    if resp.status_code == 404:
        return None
    resp.raise_for_status()
    return resp.json()


def _is_privileged(admin_acct: dict) -> bool:
    role = admin_acct.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
    return role_id is not None and role_id > 0


def _is_allowlisted(acct: str) -> bool:
    return acct.split("@")[0].lower() in ABUSE_ALLOWLIST or acct.lower() in ABUSE_ALLOWLIST


def sweep() -> None:
    if not SUSPICIOUS_SWEEP_ENABLED:
        log.info("SUSPICIOUS_SWEEP_ENABLED=false, nothing to do")
        return

    due = _pending_due()
    log.info("%d flagged signup(s) past the %dh grace window", len(due), SUSPICIOUS_GRACE_HOURS)

    for account_id, acct, reasons, baseline_statuses, baseline_following in due:
        try:
            admin_acct = _fetch_admin_account(account_id)
        except httpx.HTTPError as exc:
            log.error("acct=%s: failed to fetch admin account, will retry next sweep: %s", acct, exc)
            continue

        if admin_acct is None:
            log.info("acct=%s no longer exists, clearing watch", acct)
            _resolve(account_id, "skipped-not-found")
            continue

        if admin_acct.get("suspended") or admin_acct.get("silenced"):
            log.info("acct=%s already limited by another mechanism, clearing watch", acct)
            _resolve(account_id, "skipped-already-limited")
            continue

        if (_is_privileged(admin_acct) and ABUSE_SKIP_PRIVILEGED) or _is_allowlisted(acct):
            log.info("acct=%s is staff/allowlisted, clearing watch", acct)
            _resolve(account_id, "skipped-privileged")
            continue

        public = admin_acct.get("account") or {}
        statuses_count = public.get("statuses_count", 0)
        following_count = public.get("following_count", 0)

        if statuses_count > baseline_statuses or following_count > baseline_following:
            log.info(
                "acct=%s showed activity since flagging (posts %d->%d, following %d->%d), clearing watch",
                acct, baseline_statuses, statuses_count, baseline_following, following_count,
            )
            _resolve(account_id, "cleared")
            continue

        note = (
            f"Auto-{SUSPICIOUS_ACTION} by suspicious-sweep: flagged at signup ({reasons}), "
            f"no post or follow in {SUSPICIOUS_GRACE_HOURS}h since going live."
        )

        if SUSPICIOUS_DRY_RUN:
            log.warning("[DRY-RUN] would %s acct=%s (reasons=%s, no activity in %dh)",
                        SUSPICIOUS_ACTION, acct, reasons, SUSPICIOUS_GRACE_HOURS)
            _resolve(account_id, f"dry-run:{SUSPICIOUS_ACTION}")
            dm_moderator(f"[DRY-RUN] would {SUSPICIOUS_ACTION} @{acct} — flagged signup "
                         f"({reasons}), no activity in {SUSPICIOUS_GRACE_HOURS}h.")
            continue

        try:
            apply_action(account_id, SUSPICIOUS_ACTION, note)
        except httpx.HTTPError as exc:
            log.error("acct=%s: failed to %s, will retry next sweep: %s", acct, SUSPICIOUS_ACTION, exc)
            continue

        _resolve(account_id, SUSPICIOUS_ACTION)
        log.warning("auto-%sd acct=%s — flagged signup (%s), no activity in %dh",
                    SUSPICIOUS_ACTION, acct, reasons, SUSPICIOUS_GRACE_HOURS)
        dm_moderator(f"🚨 Auto-{SUSPICIOUS_ACTION}d @{acct} — flagged signup ({reasons}), "
                     f"no post or follow in {SUSPICIOUS_GRACE_HOURS}h since going live.")


if __name__ == "__main__":
    sweep()