Add read-only blocklist API (off by default)
docker-build-push / build-push (push) Successful in 5s

Publishes the IP ranges and email domains the bot has auto-blocked as
JSON: GET /api/blocks, /api/blocks/ip-ranges, /api/blocks/email-domains,
each taking an optional ?limit=N.

The endpoints are unauthenticated, so BLOCKS_API_ENABLED gates them and
defaults to false — with it unset the paths 404 as if they didn't exist.
Publishing which ranges are already blocked also tells a spam signup
which ones to avoid, so opting in is a per-deployment call; yttrx.com
sets it true in the box's .env.

Responses assert only that a block exists: the blocked range/domain, a
signups count, and first/last seen. No account identifiers, and none of
the ipapi.is or check-mail.org classification that drove the decision —
republishing a paid reputation feed's verdicts on an open endpoint would
be redistributing their data rather than ours.

Supporting changes:

- signup_ip gains a block_cidr column (migrated in place on open) so the
  API reports the network actually blocked rather than the signup IP.
  Rows predating it fall back to that IP's /32 or /128, which is exactly
  what was registered before range-blocking started 2026-07-15.
- register_ip_block / register_email_domain_block now return whether the
  block landed, and the caller only marks it registered on success, so a
  failed POST is no longer published as a block that exists.
This commit is contained in:
pmb committed 2026-09-13 00:41:48 -07:00
1 parent 1a3e112946
commit eb6d7cad46
6 files changed
+422 -22

No files matched your search

+87 -6
View File
@@ -224,7 +224,7 @@ def ip_scrutiny_tests():
ipblocks = []
main.send_welcome = lambda account_id, acct: sent.append((account_id, acct))
main.dm_moderator = lambda message: dms.append(message)
main.register_ip_block = lambda ip, acct, org, cidr: ipblocks.append((ip, acct, org, cidr))
main.register_ip_block = lambda ip, acct, org, cidr: ipblocks.append((ip, acct, org, cidr)) or True
# A flagged-but-not-held signup starts the suspicious watch inline, which
# would otherwise hit the network for a baseline snapshot — stub it.
main.fetch_account_counts = lambda account_id: (0, 0)
@@ -315,7 +315,7 @@ def email_scrutiny_tests():
domain_blocks = []
main.send_welcome = lambda account_id, acct: sent.append((account_id, acct))
main.dm_moderator = lambda message: dms.append(message)
main.register_email_domain_block = lambda domain, acct: domain_blocks.append((domain, acct))
main.register_email_domain_block = lambda domain, acct: domain_blocks.append((domain, acct)) or True
# A flagged-but-not-held signup starts the suspicious watch inline, which
# would otherwise hit the network for a baseline snapshot — stub it.
main.fetch_account_counts = lambda account_id: (0, 0)
@@ -454,8 +454,8 @@ def combined_signal_tests():
main.send_welcome = lambda account_id, acct: sent.append((account_id, acct))
main.dm_moderator = lambda message: dms.append(message)
main.apply_action = lambda target_id, action, text: actions.append((target_id, action))
main.register_ip_block = lambda ip, acct, org, cidr: ipblocks.append((ip, acct, org, cidr))
main.register_email_domain_block = lambda domain, acct: domain_blocks.append((domain, acct))
main.register_ip_block = lambda ip, acct, org, cidr: ipblocks.append((ip, acct, org, cidr)) or True
main.register_email_domain_block = lambda domain, acct: domain_blocks.append((domain, acct)) or True
main.fetch_account_counts = lambda account_id: (0, 0)
main.IP_SCRUTINY_ENABLED = True
@@ -546,8 +546,8 @@ def strong_ip_suspend_tests():
main.send_welcome = lambda account_id, acct: sent.append((account_id, acct))
main.dm_moderator = lambda message: dms.append(message)
main.apply_action = lambda target_id, action, text: actions.append((target_id, action))
main.register_ip_block = lambda ip, acct, org, cidr: ipblocks.append((ip, acct, org, cidr))
main.register_email_domain_block = lambda domain, acct: None
main.register_ip_block = lambda ip, acct, org, cidr: ipblocks.append((ip, acct, org, cidr)) or True
main.register_email_domain_block = lambda domain, acct: True
main.fetch_account_counts = lambda account_id: (0, 0)
main.IP_SCRUTINY_ENABLED = True
@@ -780,6 +780,86 @@ def suspicious_sweep_tests():
assert any("[DRY-RUN]" in d and "dryrunbot" in d for d in dms), dms
def blocks_api_tests():
"""Drive the read-only blocklist API: the BLOCKS_API_ENABLED gate, the
response shape, the deliberate omission of provider classification data
and account identifiers, and the pre-block_cidr fallback."""
# Seed deterministic rows rather than relying on what earlier tests left
# in the shared db; assertions below check for presence, not exact lists.
main.record_signup_ip("900", "blockedone", "198.18.0.90", "datacenter",
"Cloud Inc", True, "198.18.0.0/24")
main.mark_ipblock_registered("900")
# A second signup out of the same range collapses into one entry.
main.record_signup_ip("901", "blockedtwo", "198.18.0.91", "datacenter",
"Cloud Inc", True, "198.18.0.0/24")
main.mark_ipblock_registered("901")
# Classified but never blocked -> must not appear at all.
main.record_signup_ip("902", "notblockedacct", "203.0.113.199", "clean", "", False)
# A row predating the block_cidr column (NULL) -> falls back to the /32
# that was in fact registered for it back then.
main.record_signup_ip("903", "legacyacct", "192.0.2.77", "datacenter", "Old Corp", True)
main.mark_ipblock_registered("903")
main.record_signup_email("910", "spamone", "blocked-example.test", True, 95, True)
main.mark_email_domain_blocked("910")
main.record_signup_email("911", "spamtwo", "blocked-example.test", True, 95, True)
main.mark_email_domain_blocked("911")
main.record_signup_email("912", "cleanacct", "notblocked-example.test", False, 0, False)
paths = ("/api/blocks", "/api/blocks/ip-ranges", "/api/blocks/email-domains")
# A. Disabled (the shipped default) -> 404, indistinguishable from a path
# that doesn't exist.
main.BLOCKS_API_ENABLED = False
for path in paths:
assert client.get(path).status_code == 404, path
# B. Enabled -> 200, with the ranges and domains actually blocked.
main.BLOCKS_API_ENABLED = True
r = client.get("/api/blocks")
assert r.status_code == 200, r.status_code
assert r.headers["cache-control"] == f"public, max-age={main.BLOCKS_API_CACHE_SECONDS}"
body = r.json()
ranges = {e["cidr"]: e for e in body["ip_ranges"]["items"]}
domains = {e["domain"]: e for e in body["email_domains"]["items"]}
assert body["ip_ranges"]["count"] == len(body["ip_ranges"]["items"])
assert body["email_domains"]["count"] == len(body["email_domains"]["items"])
assert "198.18.0.0/24" in ranges, ranges
assert ranges["198.18.0.0/24"]["signups"] == 2, "repeat signups collapse into one entry"
assert "192.0.2.77/32" in ranges, "a NULL block_cidr row falls back to its /32"
assert "blocked-example.test" in domains, domains
assert domains["blocked-example.test"]["signups"] == 2
assert "notblocked-example.test" not in domains, "an unblocked domain must not be listed"
# C. Nothing that identifies an account, and nothing sourced from the
# ipapi.is / check-mail.org feeds, may appear in the response.
raw = r.text
for leak in ("blockedone", "legacyacct", "notblockedacct", "spamone", "cleanacct",
"203.0.113.199", "datacenter", "Cloud Inc", "Old Corp"):
assert leak not in raw, f"{leak!r} must not be exposed by the blocklist API"
for entry in body["ip_ranges"]["items"]:
assert set(entry) == {"cidr", "signups", "first_blocked_at", "last_blocked_at"}, entry
for entry in body["email_domains"]["items"]:
assert set(entry) == {"domain", "signups", "first_blocked_at", "last_blocked_at"}, entry
# D. The per-list endpoints agree with the combined document.
ips_only = client.get("/api/blocks/ip-ranges").json()
domains_only = client.get("/api/blocks/email-domains").json()
assert ips_only["items"] == body["ip_ranges"]["items"]
assert domains_only["items"] == body["email_domains"]["items"]
# E. ?limit truncates; a nonsensical limit is treated as "no limit".
assert len(client.get("/api/blocks/ip-ranges?limit=1").json()["items"]) == 1
assert (client.get("/api/blocks/ip-ranges?limit=-5").json()["items"]
== ips_only["items"]), "a negative limit must not empty the list"
# F. BLOCKS_API_MAX_ITEMS caps the caller's limit, not the other way round.
main.BLOCKS_API_MAX_ITEMS = 1
assert len(client.get("/api/blocks/ip-ranges?limit=999").json()["items"]) == 1
main.BLOCKS_API_MAX_ITEMS = 5000
if __name__ == "__main__":
routing_tests()
policy_tests()
@@ -790,4 +870,5 @@ if __name__ == "__main__":
strong_ip_suspend_tests()
suspicious_watch_tests()
suspicious_sweep_tests()
blocks_api_tests()
print("ALL TESTS PASSED")