#!/usr/bin/env python3 import argparse import concurrent.futures import json import os import re import shlex import subprocess import sys import tempfile import time import urllib.request from collections import defaultdict from pathlib import Path try: import dns.resolver except ImportError: dns = None VERSION = "1.0.0" SOURCES = [ "https://raw.githubusercontent.com/mehrtat/disposable-email-domain/main/domains.txt", "https://raw.githubusercontent.com/FFraud-com/disposable-email-domains/main/disposable-email-domains.txt", "https://disposable.github.io/disposable-email-domains/domains.txt", ] DOMAIN_RE = re.compile( r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+" r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$" ) def log(message=""): print(message, flush=True) def normalize_domain(value): if not value: return None value = value.strip().lower().rstrip(".") if not value or value.startswith("#"): return None if value.startswith("@"): value = value[1:] if value.startswith("*."): value = value[2:] try: value = value.encode("idna").decode("ascii") except Exception: return None return value if DOMAIN_RE.match(value) else None def read_domain_file(filename): path = Path(filename) if not path.exists(): return set() result = set() for line in path.read_text(encoding="utf-8", errors="ignore").splitlines(): domain = normalize_domain(line) if domain: result.add(domain) return result def download_source(url, timeout=120): request = urllib.request.Request( url, headers={"User-Agent": f"mastodon-email-blocker/{VERSION}"}, ) with urllib.request.urlopen(request, timeout=timeout) as response: content = response.read().decode("utf-8", errors="ignore") result = set() for line in content.splitlines(): domain = normalize_domain(line) if domain: result.add(domain) return result def chunks(values, size): values = list(values) for i in range(0, len(values), size): yield values[i:i + size] def save_json_atomic(filename, value): filename = Path(filename) filename.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=filename.name + ".", dir=str(filename.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(value, handle, indent=2, sort_keys=True) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, filename) finally: if os.path.exists(temporary): os.unlink(temporary) class MastodonBackend: def run_rails(self, code): raise NotImplementedError def run_tootctl_add(self, domains): raise NotImplementedError class NativeBackend(MastodonBackend): def __init__(self, mastodon_dir, ruby_bin=None): self.mastodon_dir = Path(mastodon_dir) self.ruby_bin = ruby_bin if not (self.mastodon_dir / "bin/rails").exists(): raise RuntimeError(f"bin/rails not found in {self.mastodon_dir}") if not (self.mastodon_dir / "bin/tootctl").exists(): raise RuntimeError(f"bin/tootctl not found in {self.mastodon_dir}") def _env(self): env = os.environ.copy() env["RAILS_ENV"] = "production" if self.ruby_bin: env["PATH"] = f"{self.ruby_bin}:{env.get('PATH', '/usr/local/bin:/usr/bin:/bin')}" return env def _run(self, command): process = subprocess.run( command, cwd=str(self.mastodon_dir), env=self._env(), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if process.returncode != 0: if process.stdout: sys.stderr.write(process.stdout) if process.stderr: sys.stderr.write(process.stderr) raise RuntimeError("Command failed: " + " ".join(command)) return process.stdout def run_rails(self, code): return self._run(["bin/rails", "runner", code]) def run_tootctl_add(self, domains): return self._run(["bin/tootctl", "email-domain-blocks", "add", *domains]) class DockerBackend(MastodonBackend): def __init__(self, compose_dir, compose_command="docker compose", service="web"): self.compose_dir = Path(compose_dir) self.compose_command = shlex.split(compose_command) self.service = service if not self.compose_dir.exists(): raise RuntimeError(f"Compose directory not found: {self.compose_dir}") def _run(self, command): full_command = [ *self.compose_command, "exec", "-T", "-e", "RAILS_ENV=production", self.service, *command, ] process = subprocess.run( full_command, cwd=str(self.compose_dir), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if process.returncode != 0: if process.stdout: sys.stderr.write(process.stdout) if process.stderr: sys.stderr.write(process.stderr) raise RuntimeError("Command failed: " + " ".join(full_command)) return process.stdout def run_rails(self, code): return self._run(["bin/rails", "runner", code]) def run_tootctl_add(self, domains): return self._run(["bin/tootctl", "email-domain-blocks", "add", *domains]) def get_blocks(backend): code = r''' rows = EmailDomainBlock.order(:id).pluck(:id, :domain, :allow_with_approval, :parent_id) STDOUT.write("__JSON__" + JSON.generate(rows)) ''' output = backend.run_rails(code) marker = "__JSON__" pos = output.rfind(marker) if pos < 0: raise RuntimeError("Could not read EmailDomainBlock from Rails") return json.loads(output[pos + len(marker):]) def force_deny(backend): code = r''' count = EmailDomainBlock.where(allow_with_approval: true).update_all( allow_with_approval: false, updated_at: Time.current ) STDOUT.write("__COUNT__#{count}") ''' output = backend.run_rails(code) marker = "__COUNT__" pos = output.rfind(marker) if pos < 0: raise RuntimeError("Could not convert approval blocks to deny blocks") return int(output[pos + len(marker):].strip()) def resolve_mx(domain, timeout): resolver = dns.resolver.Resolver(configure=True) resolver.timeout = timeout resolver.lifetime = timeout try: answers = resolver.resolve(domain, "MX") except Exception: return domain, [] mx_records = set() for answer in answers: mx = normalize_domain(str(answer.exchange)) if mx: mx_records.add(mx) return domain, sorted(mx_records) def build_backend(args): if args.backend == "native": if not args.mastodon_dir: raise RuntimeError("--mastodon-dir is required for the native backend") return NativeBackend(args.mastodon_dir, args.ruby_bin) if not args.compose_dir: raise RuntimeError("--compose-dir is required for the docker backend") return DockerBackend(args.compose_dir, args.compose_command, args.compose_service) def main(): parser = argparse.ArgumentParser(description="Maintain Mastodon disposable email domain blocks") parser.add_argument("--backend", choices=["native", "docker"], required=True) parser.add_argument("--mastodon-dir") parser.add_argument("--ruby-bin", help="Directory containing ruby, only needed when Ruby is not already in PATH") parser.add_argument("--compose-dir") parser.add_argument("--compose-command", default="docker compose") parser.add_argument("--compose-service", default="web") parser.add_argument("--allowlist", default="/etc/mastodon-email-blocker/allowlist.txt") parser.add_argument("--report", default="/var/lib/mastodon-email-blocker/last-report.json") parser.add_argument("--state", default="/var/lib/mastodon-email-blocker/state.json") parser.add_argument("--batch-size", type=int, default=250) parser.add_argument("--mx-mode", choices=["off", "report"], default="off") parser.add_argument("--mx-workers", type=int, default=40) parser.add_argument("--mx-timeout", type=float, default=4.0) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--no-force-deny", action="store_true") args = parser.parse_args() if args.mx_mode == "report" and dns is None: raise RuntimeError("MX reporting requires dnspython: pip install dnspython") backend = build_backend(args) allowlist = read_domain_file(args.allowlist) log(f"mastodon-email-blocker {VERSION}") log(f"backend: {args.backend}") log(f"mx mode: {args.mx_mode}") log() disposable = set() source_stats = {} successful_sources = 0 for url in SOURCES: log(f"[source] {url}") try: domains = download_source(url) source_stats[url] = len(domains) successful_sources += 1 disposable |= domains log(f" {len(domains):,} domains") except Exception as exc: source_stats[url] = f"ERROR: {exc}" log(f" ERROR: {exc}") if successful_sources == 0: raise RuntimeError("All blocklist sources failed; refusing to change Mastodon") disposable -= allowlist log(f"[source] union: {len(disposable):,} disposable domains") log("[mastodon] reading existing blocks") rows = get_blocks(backend) approvals = [row for row in rows if row[2] is True] converted = 0 if approvals and not args.no_force_deny: if args.dry_run: log(f"[dry-run] would convert {len(approvals):,} approval blocks to deny") else: converted = force_deny(backend) log(f"[mastodon] converted {converted:,} approval blocks to deny") rows = get_blocks(backend) existing = {normalize_domain(row[1]) for row in rows} existing.discard(None) missing = sorted(disposable - existing) log(f"[domains] existing blocks: {len(existing):,}") log(f"[domains] disposable domains to add: {len(missing):,}") domain_batches = list(chunks(missing, args.batch_size)) for number, batch in enumerate(domain_batches, 1): log(f"[domains] batch {number}/{len(domain_batches)} ({len(batch)})") if args.dry_run: continue backend.run_tootctl_add(batch) mx_to_domains = defaultdict(set) domains_with_mx = 0 domains_without_mx = 0 if args.mx_mode == "report": all_domains = sorted(disposable) log(f"[mx] resolving {len(all_domains):,} domains; MX records are report-only") with concurrent.futures.ThreadPoolExecutor(max_workers=args.mx_workers) as executor: futures = { executor.submit(resolve_mx, domain, args.mx_timeout): domain for domain in all_domains } completed = 0 for future in concurrent.futures.as_completed(futures): try: domain, records = future.result() except Exception: domain, records = futures[future], [] if records: domains_with_mx += 1 else: domains_without_mx += 1 for mx in records: mx_to_domains[mx].add(domain) completed += 1 if completed % 5000 == 0 or completed == len(all_domains): log(f"[mx] {completed:,}/{len(all_domains):,}") top_mx = sorted( ((mx, len(domains)) for mx, domains in mx_to_domains.items()), key=lambda item: (-item[1], item[0]), )[:100] final_rows = rows if args.dry_run else get_blocks(backend) final_deny = sum(1 for row in final_rows if row[2] is False) final_approval = sum(1 for row in final_rows if row[2] is True) report = { "version": VERSION, "timestamp": int(time.time()), "backend": args.backend, "dry_run": args.dry_run, "sources": source_stats, "successful_sources": successful_sources, "disposable_domains": len(disposable), "allowlist_entries": len(allowlist), "existing_before": len(rows), "approval_before": len(approvals), "approval_converted": converted, "missing_domains": len(missing), "domains_added_or_planned": len(missing), "mx_mode": args.mx_mode, "mx_autoblock": False, "domains_with_mx": domains_with_mx, "domains_without_mx": domains_without_mx, "mx_unique": len(mx_to_domains), "final_deny": final_deny, "final_approval": final_approval, "top_mx": [ {"mx": mx, "disposable_domains": count, "action": "REPORT_ONLY"} for mx, count in top_mx ], } save_json_atomic(args.report, report) save_json_atomic(args.state, { "version": VERSION, "last_run": int(time.time()), "backend": args.backend, "disposable_domains": len(disposable), "mx_unique": len(mx_to_domains), "mx_autoblock": False, "last_report": args.report, }) log() log("Result") log(f" disposable domains : {len(disposable):,}") log(f" new domains : {len(missing):,}") log(f" MX observed : {len(mx_to_domains):,}") log(" MX blocked : 0") log(f" deny blocks : {final_deny:,}") log(f" approval blocks : {final_approval:,}") log(f" report : {args.report}") if __name__ == "__main__": try: main() except KeyboardInterrupt: log("\nInterrupted") sys.exit(130) except Exception as exc: log(f"\nERROR: {exc}") sys.exit(1)