Block the whole network for a flagged signup, not just its /32

register_ip_block always blocked the exact signup IP as a /32 (or /128
for v6), which is close to pointless for the datacenter/VPN/proxy space
these flags fire on: a repeat bad actor from the same provider almost
never reuses the exact same address, but very often reuses a different
one in the same block.

classify_signup_ip now returns a fourth value, block_cidr — ipapi.is's
asn.route CIDR when it's a valid network that actually contains the
signup IP (already being tracked in ipapi_range_cache purely for lookup
caching), falling back to the address's own /32 or /128 when no usable
route exists. register_ip_block blocks that instead of always deriving
a /32 itself.

Added range_cache_tests() covering the cache/fallback logic directly
(pure sqlite + ipaddress, no network), and updated the existing
ip_scrutiny_tests()/combined_signal_tests() mocks for the new 4-tuple
classify_signup_ip return and register_ip_block arity.

app/main.py

@@ -136,7 +136,13 @@ IP_SCRUTINY_HOLD_WELCOME = os.environ.get("IP_SCRUTINY_HOLD_WELCOME", "true").lo
 # Distinct-reporter threshold used INSTEAD of the tier's usual threshold (via
 # min()) when the reported account's signup IP was flagged.
 IP_SCRUTINY_ABUSE_THRESHOLD = int(os.environ.get("IP_SCRUTINY_ABUSE_THRESHOLD", "1"))
-# Auto-register flagged IPs into Mastodon's native Admin::IpBlock.
+# Auto-register a flagged signup's network into Mastodon's native
+# Admin::IpBlock. Blocks the ASN/route-level CIDR ipapi.is reports the IP
+# belonging to (see classify_signup_ip's block_cidr), not just the single
+# address — a flagged signup is almost always datacenter/VPN/proxy space, and
+# a repeat bad actor is far more likely to come back from a different address
+# in the same block than the exact same one. Falls back to a single-address
+# (/32 or /128) block if no usable route was returned/cached.
 IP_SCRUTINY_AUTO_IPBLOCK = os.environ.get("IP_SCRUTINY_AUTO_IPBLOCK", "true").lower() in ("1", "true", "yes")
 # severity: sign_up_requires_approval | sign_up_block | no_access
 IP_SCRUTINY_IPBLOCK_SEVERITY = os.environ.get("IP_SCRUTINY_IPBLOCK_SEVERITY", "sign_up_requires_approval")
@@ -403,7 +409,7 @@ def cached_ip_intel(ip: str) -> dict | None:
     family, packed, _, _ = bounds
     with _db() as conn:
         row = conn.execute(
-            "SELECT is_datacenter, is_vpn, is_proxy, is_tor, is_abuser, org "
+            "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),
@@ -414,6 +420,7 @@ def cached_ip_intel(ip: str) -> dict | None:
             "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],
         }
 
 
@@ -627,14 +634,14 @@ def process_signup(account_id: str, acct: str, ip: str, email: str = "",
     email_flagged = False
 
     if IP_SCRUTINY_ENABLED and ip:
-        classification, org, ip_flagged = classify_signup_ip(ip)
+        classification, org, ip_flagged, block_cidr = classify_signup_ip(ip)
         record_signup_ip(account_id, acct, ip, classification, org, ip_flagged)
 
         if ip_flagged:
             log.warning("flagged signup acct=%s ip=%s classification=%s org=%s",
                         acct, ip, classification, org)
             if IP_SCRUTINY_AUTO_IPBLOCK and not IP_SCRUTINY_DRY_RUN:
-                register_ip_block(ip, acct, org)
+                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})")
@@ -844,16 +851,21 @@ def classify_account(target_id: str) -> dict:
     }
 
 
-def classify_signup_ip(ip: str) -> tuple[str, str, bool]:
+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). classification is a "+"-joined
-    list of every matched signal (datacenter/vpn/proxy/tor/abuser), or
-    "clean" if none matched. flagged is True if any signal matched. API
-    failure yields ("unknown", "", False) — scrutiny should never trigger on
-    our own lookup errors.
+    Returns (classification, org, flagged, block_cidr). classification is a
+    "+"-joined list of every matched signal (datacenter/vpn/proxy/tor/abuser),
+    or "clean" if none matched. flagged is True if any signal matched.
+    block_cidr is the network register_ip_block should block when flagged —
+    ipapi.is's asn.route when it's a valid network that actually contains ip,
+    else ip's own /32 or /128 — since a flagged signup is almost always
+    datacenter/VPN/proxy space, where a repeat bad actor is far more likely to
+    come back from a different address in the same block than the exact same
+    one. API failure yields ("unknown", "", False, "<ip>/32|128") — scrutiny
+    should never trigger on our own lookup errors.
     """
     intel = cached_ip_intel(ip)
     if intel is None:
@@ -866,7 +878,8 @@ def classify_signup_ip(ip: str) -> tuple[str, str, bool]:
             data = resp.json()
         except (httpx.HTTPError, ValueError) as exc:
             log.warning("ipapi.is lookup failed for ip=%s: %s", ip, exc)
-            return "unknown", "", False
+            bounds = _ip_range_bounds(ip)
+            return "unknown", "", False, bounds[3] if bounds else f"{ip}/32"
         intel = {
             "is_datacenter": bool(data.get("is_datacenter")),
             "is_vpn": bool(data.get("is_vpn")),
@@ -877,6 +890,8 @@ def classify_signup_ip(ip: str) -> tuple[str, str, bool]:
                     or (data.get("asn") or {}).get("org") or ""),
         }
         route = (data.get("asn") or {}).get("route") or ""
+        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 (
@@ -887,17 +902,25 @@ def classify_signup_ip(ip: str) -> tuple[str, str, bool]:
         ("abuser", "is_abuser"),
     ) if intel[key]]
     classification = "+".join(reasons) if reasons else "clean"
-    return classification, intel["org"], bool(reasons)
+    return classification, intel["org"], bool(reasons), intel.get("cidr") or f"{ip}/32"
 
 
-def register_ip_block(ip: str, acct: str, org: str) -> None:
-    """Register a flagged signup IP in Mastodon's native Admin::IpBlock."""
-    try:
-        prefix_len = 32 if ipaddress.ip_address(ip).version == 4 else 128
-    except ValueError:
-        log.warning("skipping ip_block registration for unparseable ip=%r (acct=%s)", ip, acct)
-        return
-    cidr = f"{ip}/{prefix_len}"
+def register_ip_block(ip: str, acct: str, org: str, cidr: str = "") -> None:
+    """Register a flagged signup's network in Mastodon's native Admin::IpBlock.
+
+    cidr should be classify_signup_ip's block_cidr — the ASN/route-level
+    network the signup IP belongs to, so a repeat bad actor from the same
+    datacenter/VPN/proxy block is blocked too, not just this one address.
+    Falls back to a single-address block if no cidr is given or ip itself
+    doesn't parse (defensive only; classify_signup_ip always supplies one).
+    """
+    if not cidr:
+        try:
+            prefix_len = 32 if ipaddress.ip_address(ip).version == 4 else 128
+        except ValueError:
+            log.warning("skipping ip_block registration for unparseable ip=%r (acct=%s)", ip, acct)
+            return
+        cidr = f"{ip}/{prefix_len}"
     comment = f"welcomebot: flagged signup @{acct} ({org or 'unknown org'})"[:200]
     try:
         resp = httpx.post(