yttrx-welcomebot/bin/yttrx-blocklist-import

#!/usr/bin/env python3
"""Import yttrx.com's published signup blocklist into another Mastodon instance.

Mastodon has no RBL/DNSBL support — nothing in it consults an outside list when
deciding whether to accept a signup (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, not queried live. This script
does that: it fetches the published list and POSTs each entry to your own
instance's admin API.

Standard library only — no pip install, no venv.

    # see what it would do (default)
    ./yttrx-blocklist-import --instance https://your.instance --token "$TOKEN"

    # actually write the blocks
    ./yttrx-blocklist-import --instance https://your.instance --token "$TOKEN" --apply

Your token needs `admin:write:ip_blocks` and `admin:write:email_domain_blocks`.
Mint one from the Rails console on your instance:

    bin/rails runner '
    acct = Account.find_local("youradmin")
    scopes = "admin:write:ip_blocks admin:write:email_domain_blocks"
    app = Doorkeeper::Application.create!(name: "blocklist import", scopes: scopes,
                                          redirect_uri: "urn:ietf:wg:oauth:2.0:oob")
    puts Doorkeeper::AccessToken.create!(application_id: app.id,
                                         resource_owner_id: acct.user.id,
                                         scopes: scopes).token'

A NOTE ON SEVERITY, worth reading before --apply:

The IP entries are whole networks (the ASN route a flagged signup's address
belongs to), because on yttrx a flagged signup is almost always datacenter,
VPN or proxy space where the next attempt comes from a neighbouring address.
That is calibrated to yttrx's traffic, not yours. This script therefore
defaults to `sign_up_requires_approval` — a flagged signup lands in your
moderation queue rather than being rejected outright — even though yttrx runs
these at `sign_up_block` itself. Pass --ip-severity sign_up_block if you want
the harder behaviour; think about whether you have users behind a VPN first.

Entries already present come back 422 from Mastodon and are counted as
"already blocked", not as errors.
"""

from __future__ import annotations

import argparse
import json
import sys
import urllib.error
import urllib.parse
import urllib.request

DEFAULT_SOURCE = "https://hooks.yttrx.com/api/blocks"
IP_SEVERITIES = ("sign_up_requires_approval", "sign_up_block", "no_access")


def fetch_source(url: str, timeout: float) -> dict:
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode("utf-8"))


def post(base: str, path: str, token: str, fields: dict, timeout: float) -> int:
    """POST form-encoded fields; returns the HTTP status (422 included, which
    Mastodon uses for 'this block already exists')."""
    req = urllib.request.Request(
        f"{base}{path}",
        data=urllib.parse.urlencode(fields).encode(),
        headers={"Authorization": f"Bearer {token}"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status
    except urllib.error.HTTPError as exc:
        return exc.code


def import_entries(entries, label, base, path, token, field_fn, timeout, apply_changes):
    created = existing = failed = 0
    for entry in entries:
        fields = field_fn(entry)
        shown = next(iter(fields.values()))
        if not apply_changes:
            print(f"  would block {shown}")
            created += 1
            continue
        status = post(base, path, token, fields, timeout)
        if status in (200, 201):
            created += 1
            print(f"  blocked {shown}")
        elif status == 422:
            existing += 1
        elif status in (401, 403):
            print(f"  ERROR {status} on {shown} — token is missing the required "
                  f"admin write scope; stopping.", file=sys.stderr)
            failed += 1
            break
        else:
            failed += 1
            print(f"  ERROR {status} on {shown}", file=sys.stderr)
    verb = "would block" if not apply_changes else "blocked"
    print(f"{label}: {created} {verb}, {existing} already blocked, {failed} failed")
    return failed


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Import yttrx.com's published signup blocklist into your Mastodon instance.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="Runs as a dry run unless --apply is given.",
    )
    ap.add_argument("--instance", required=True,
                    help="your instance's base URL, e.g. https://your.instance")
    ap.add_argument("--token", required=True,
                    help="admin token with admin:write:ip_blocks + admin:write:email_domain_blocks")
    ap.add_argument("--source", default=DEFAULT_SOURCE,
                    help=f"blocklist JSON to import (default: {DEFAULT_SOURCE})")
    ap.add_argument("--ip-severity", default="sign_up_requires_approval", choices=IP_SEVERITIES,
                    help="severity for imported IP ranges (default: %(default)s — see the note "
                         "at the top of this script before choosing sign_up_block)")
    ap.add_argument("--ips-only", action="store_true", help="import only the IP ranges")
    ap.add_argument("--domains-only", action="store_true", help="import only the email domains")
    ap.add_argument("--min-signups", type=int, default=1,
                    help="skip entries seen fewer than this many times (default: %(default)s)")
    ap.add_argument("--limit", type=int, default=0, help="import at most N of each kind")
    ap.add_argument("--timeout", type=float, default=30.0, help="HTTP timeout in seconds")
    ap.add_argument("--apply", action="store_true",
                    help="actually write the blocks (without this, nothing is changed)")
    args = ap.parse_args()

    if args.ips_only and args.domains_only:
        ap.error("--ips-only and --domains-only are mutually exclusive")

    base = args.instance.rstrip("/")
    try:
        data = fetch_source(args.source, args.timeout)
    except (urllib.error.URLError, ValueError) as exc:
        print(f"could not fetch {args.source}: {exc}", file=sys.stderr)
        return 1

    def pick(section):
        items = [e for e in data.get(section, {}).get("items", [])
                 if e.get("signups", 1) >= args.min_signups]
        return items[:args.limit] if args.limit > 0 else items

    ip_ranges = [] if args.domains_only else pick("ip_ranges")
    domains = [] if args.ips_only else pick("email_domains")

    print(f"source    : {args.source}")
    print(f"generated : {data.get('generated_at', 'unknown')}")
    print(f"target    : {base}")
    print(f"mode      : {'APPLY — writing blocks' if args.apply else 'dry run (pass --apply to write)'}")
    print(f"ip ranges : {len(ip_ranges)} (severity {args.ip_severity})")
    print(f"domains   : {len(domains)}")
    print()

    failed = 0
    if ip_ranges:
        failed += import_entries(
            ip_ranges, "IP ranges", base, "/api/v1/admin/ip_blocks", args.token,
            lambda e: {"ip": e["cidr"], "severity": args.ip_severity,
                       "comment": f"imported from {args.source}"},
            args.timeout, args.apply)
    if domains:
        failed += import_entries(
            domains, "email domains", base, "/api/v1/admin/email_domain_blocks", args.token,
            lambda e: {"domain": e["domain"]},
            args.timeout, args.apply)

    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())