Cap how wide a single auto-registered IP block may be
docker-build-push / build-push (push) Successful in 5s

ipapi.is reports the flagged address's whole ASN route, and register_ip_block
blocked it whatever its width. That is the right unit for hosting space, where
a repeat bad actor comes back from a different address in the same VPS block.
It is badly wrong for an ISP aggregate, and the route cache holds routes as
wide as /8, /9 and /11: one `abuser` flag on a residential or mobile address
inside one of those would have walled off millions of consumer addresses.

That was not hypothetical -- 197.211.32.0/19 is registered against
GLOMOBILE-GPRS, a Benin mobile carrier, from a single `abuser` flag: 8,192
addresses of consumer mobile space.

Anything wider than /24 (v4) or /48 (v6) now falls back to blocking the single
address. This costs blocking strength against large hosting allocations -- an
OVH /16 blocks one address now, so a repeat offender from that block needs a
fresh block each time -- and in exchange makes the collateral-damage case
impossible. Both ceilings are env-tunable.

The cap applies to what gets BLOCKED, not to what gets CACHED: the wide route
is still the right cache key, since it genuinely belongs to one operator and
one lookup answers for the whole of it. Narrowing the cache key would multiply
provider lookups for no benefit.

Applied at registration time, so blocks already in force are untouched.
This commit is contained in:
pmb committed 2026-09-13 16:08:37 -07:00
1 parent 57234ca548
commit 147673fd49
3 files changed
+131 -4

No files matched your search

+62 -4
View File
@@ -173,6 +173,27 @@ IP_SCRUTINY_IPBLOCK_SEVERITY = os.environ.get("IP_SCRUTINY_IPBLOCK_SEVERITY", "s
# suspends immediately at signup, regardless of the email-domain signal,
# instead of just the held-welcome/auto-ipblock/lowered-threshold treatment
# every other flagged signup gets.
# Widest network a single auto-registered block may cover, as a prefix
# length. ipapi.is reports the flagged address's whole ASN route, which is the
# right unit for hosting space (a bad actor comes back from a different
# address in the same VPS block) but badly wrong for an ISP aggregate: routes
# as wide as /8, /9 and /11 show up in the cache, and one `abuser` flag on a
# residential or mobile address inside one of those would otherwise wall off
# millions of consumer addresses. A route wider than this is treated as "not a
# meaningful unit to block" and falls back to the single address.
#
# This deliberately costs blocking strength against large hosting allocations
# — anything wider than a /24 (which is most of the VPS ranges registered so
# far) now blocks one address rather than the range, so a repeat offender out
# of the same block needs a fresh block each time — in exchange for making the
# mobile/residential collateral-damage case impossible.
# Applied at registration time, not retroactively — blocks already in force
# keep whatever width they were registered with.
IP_SCRUTINY_MAX_BLOCK_PREFIX_V4 = int(os.environ.get("IP_SCRUTINY_MAX_BLOCK_PREFIX_V4", "24"))
# /48 is the v6 analogue: the standard end-site allocation, and the rough
# equivalent of "one customer's block" that a /24 represents in v4.
IP_SCRUTINY_MAX_BLOCK_PREFIX_V6 = int(os.environ.get("IP_SCRUTINY_MAX_BLOCK_PREFIX_V6", "48"))
IP_SCRUTINY_STRONG_SIGNALS = frozenset({"datacenter", "proxy", "tor", "abuser"})
IP_SCRUTINY_STRONG_SUSPEND_ENABLED = os.environ.get(
"IP_SCRUTINY_STRONG_SUSPEND_ENABLED", "true"
@@ -697,6 +718,41 @@ def start_suspicious_watch(account_id: str, acct: str, reasons: str,
)
def _single_address_cidr(ip: str) -> str:
"""The /32 or /128 for ip, or "" if it doesn't parse."""
try:
addr = ipaddress.ip_address(ip)
except ValueError:
return ""
return f"{addr}/{32 if addr.version == 4 else 128}"
def capped_block_cidr(ip: str, cidr: str) -> str:
"""The network to actually register a block for: cidr, unless it's wider
than IP_SCRUTINY_MAX_BLOCK_PREFIX_V4/V6, in which case the single address.
Applied when deciding what to BLOCK, not when caching the classification —
the wide ASN route is still the right cache key (the whole route really
does belong to one operator, so one lookup answers for all of it), it is
just the wrong thing to hand to Admin::IpBlock.
"""
fallback = _single_address_cidr(ip)
if not cidr:
return fallback
try:
net = ipaddress.ip_network(cidr, strict=False)
except ValueError:
return fallback
cap = (IP_SCRUTINY_MAX_BLOCK_PREFIX_V4 if net.version == 4
else IP_SCRUTINY_MAX_BLOCK_PREFIX_V6)
if net.prefixlen < cap:
log.info("block for ip=%s narrowed from route %s (%d addresses) to %s — "
"wider than the /%d ceiling", ip, cidr, net.num_addresses,
fallback or "the single address", cap)
return fallback or str(net)
return str(net)
def record_reputation_query(kind: str, value: str, classification: str, org: str,
risk: int, flagged: bool, target: str, blocked: bool,
source: str) -> None:
@@ -1294,7 +1350,9 @@ def classify_signup_ip(ip: str) -> tuple[str, str, bool, str]:
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
one. A route wider than the IP_SCRUTINY_MAX_BLOCK_PREFIX_V4/V6 ceiling is
narrowed back to the single address (see capped_block_cidr) — that width
means an ISP aggregate, not a hosting allocation. API failure yields ("unknown", "", False, "<ip>/32|128") — scrutiny
should never trigger on our own lookup errors.
"""
intel = cached_ip_intel(ip)
@@ -1308,8 +1366,7 @@ def classify_signup_ip(ip: str) -> tuple[str, str, bool, str]:
data = resp.json()
except (httpx.HTTPError, ValueError) as exc:
log.warning("ipapi.is lookup failed for ip=%s: %s", ip, exc)
bounds = _ip_range_bounds(ip)
return "unknown", "", False, bounds[3] if bounds else f"{ip}/32"
return "unknown", "", False, _single_address_cidr(ip) or f"{ip}/32"
# ipapi.is returns company/asn as flat strings, not nested dicts
company = data.get("company")
org_name = company if isinstance(company, str) else (company or {}).get("name", "")
@@ -1345,7 +1402,8 @@ def classify_signup_ip(ip: str) -> tuple[str, str, bool, str]:
("abuser", "is_abuser"),
) if intel[key]]
classification = "+".join(reasons) if reasons else "clean"
return classification, intel["org"], bool(reasons), intel.get("cidr") or f"{ip}/32"
block_cidr = capped_block_cidr(ip, intel.get("cidr") or "")
return classification, intel["org"], bool(reasons), block_cidr or f"{ip}/32"
def register_ip_block(ip: str, acct: str, org: str, cidr: str = "",