Add text format, CORS, and an importer so other instances can use the list
docker-build-push / build-push (push) Successful in 4s

Mastodon has no RBL/DNSBL support — nothing in it consults an outside
list when deciding whether to accept a signup, confirmed against v4.7.1
(the only outbound lookup in that path is EmailMxValidator's MX
resolution, which feeds the local email_domain_blocks table). So a
shared blocklist has to be imported rather than queried live, the same
way Garden Fence and other community lists propagate. These are the
pieces that make ours importable.

- ?format=txt on both per-list endpoints: one entry per line under a '#'
  comment header, the form blocklists are traded in and the form a cron
  job can consume without a JSON parser. The combined /api/blocks stays
  JSON-only — a text rendering would mix CIDRs with domains in one file,
  defeating the point. An unrecognised ?format is a 400 rather than a
  silent fallback to JSON, and the enabled-gate still runs first so a
  disabled deployment 404s either way.
- Access-Control-Allow-Origin: * on blocklist responses, so browser-side
  admin tooling can read them. The data is already public.
- bin/yttrx-blocklist-import: standard-library-only importer another
  admin can run against their own instance. Dry run unless --apply;
  422 from Mastodon counts as "already blocked", not an error.

The importer defaults IP entries to sign_up_requires_approval rather
than the sign_up_block yttrx uses itself. The entries are whole ASN
routes, which suits yttrx because a flagged signup here is nearly always
datacenter/VPN/proxy space — that calibration isn't transferable, so an
importer gets the reviewable severity by default and has to opt into the
hard one.
This commit is contained in:
pmb committed 2026-09-13 00:59:12 -07:00
1 parent af49aab79c
commit 8966de2a1d
5 files changed
+331 -9

No files matched your search

+67 -9
View File
@@ -50,7 +50,7 @@ from datetime import datetime, timedelta, timezone
import httpx
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.responses import JSONResponse, PlainTextResponse
logging.basicConfig(
level=logging.INFO,
@@ -238,6 +238,11 @@ BLOCKS_API_ENABLED = os.environ.get("BLOCKS_API_ENABLED", "false").lower() in ("
# asks for — this endpoint is unauthenticated, so the response size can't be
# left to the caller.
BLOCKS_API_MAX_ITEMS = int(os.environ.get("BLOCKS_API_MAX_ITEMS", "5000"))
# Public base URL this API is reached at, used only to print a "source:" line
# in the text format's header comments so a downloaded list says where it came
# from. Not used for routing.
BLOCKS_API_PUBLIC_URL = os.environ.get(
"BLOCKS_API_PUBLIC_URL", "https://hooks.yttrx.com").rstrip("/")
# Cache-Control max-age (seconds) on blocklist responses. The underlying data
# changes only when a signup is flagged, so letting clients and any
# intermediate cache hold it for a few minutes costs nothing.
@@ -1380,11 +1385,46 @@ def _blocks_guard(limit: int) -> int:
return max(limit, 0)
def _blocks_headers() -> dict:
return {
"Cache-Control": f"public, max-age={BLOCKS_API_CACHE_SECONDS}",
# The list is public anyway, and other instances' admin tooling may
# want to read it from a browser context.
"Access-Control-Allow-Origin": "*",
}
def _blocks_response(payload: dict) -> JSONResponse:
return JSONResponse(
payload,
headers={"Cache-Control": f"public, max-age={BLOCKS_API_CACHE_SECONDS}"},
)
return JSONResponse(payload, headers=_blocks_headers())
def _blocks_text(items: list[dict], key: str, what: str, path: str) -> PlainTextResponse:
"""Render one list as a plain-text blocklist: comment header, then one
entry per line. Blocklists are traded as text files, so this is the form
another admin's cron job can consume without a JSON parser — strip the
header with `grep -v '^#'`."""
lines = [
f"# yttrx.com — {what} automatically blocked at signup",
f"# source: {BLOCKS_API_PUBLIC_URL}{path}",
f"# generated: {_now_iso()}",
f"# entries: {len(items)}",
"#",
"# One entry per line; lines starting with '#' are comments. The JSON",
"# form of this URL adds first/last-seen timestamps and signup counts.",
]
lines.extend(str(entry[key]) for entry in items)
return PlainTextResponse("\n".join(lines) + "\n", headers=_blocks_headers())
def _blocks_format(value: str) -> str:
"""Normalise the ?format= query param, rejecting anything unrecognised
rather than silently serving JSON."""
fmt = (value or "json").strip().lower()
if fmt in ("txt", "text", "plain"):
return "txt"
if fmt == "json":
return "json"
raise HTTPException(status_code=400, detail="format must be 'json' or 'txt'")
def _now_iso() -> str:
@@ -1392,9 +1432,21 @@ def _now_iso() -> str:
@app.get("/api/blocks")
def api_blocks(limit: int = 0):
"""Both blocklists in one document."""
def api_blocks(limit: int = 0, format: str = "json"):
"""Both blocklists in one document.
JSON only: a text rendering would put CIDRs and domains in one file, and
the whole point of the text format is that it pipes straight into a tool
that expects one kind of entry. Callers wanting text use the per-list
endpoints.
"""
limit = _blocks_guard(limit)
if _blocks_format(format) == "txt":
raise HTTPException(
status_code=400,
detail="text format is per-list: use /api/blocks/ip-ranges?format=txt "
"or /api/blocks/email-domains?format=txt",
)
ip_ranges = blocked_ip_ranges(limit)
email_domains = blocked_email_domains(limit)
return _blocks_response({
@@ -1405,18 +1457,24 @@ def api_blocks(limit: int = 0):
@app.get("/api/blocks/ip-ranges")
def api_blocks_ip_ranges(limit: int = 0):
def api_blocks_ip_ranges(limit: int = 0, format: str = "json"):
limit = _blocks_guard(limit)
fmt = _blocks_format(format)
items = blocked_ip_ranges(limit)
if fmt == "txt":
return _blocks_text(items, "cidr", "IP ranges", "/api/blocks/ip-ranges")
return _blocks_response({
"generated_at": _now_iso(), "count": len(items), "items": items,
})
@app.get("/api/blocks/email-domains")
def api_blocks_email_domains(limit: int = 0):
def api_blocks_email_domains(limit: int = 0, format: str = "json"):
limit = _blocks_guard(limit)
fmt = _blocks_format(format)
items = blocked_email_domains(limit)
if fmt == "txt":
return _blocks_text(items, "domain", "email domains", "/api/blocks/email-domains")
return _blocks_response({
"generated_at": _now_iso(), "count": len(items), "items": items,
})