#!/usr/bin/env python3 """Best-effort Fail2ban -> IP Blocklist Manager reporter. This helper submits exactly one report per invocation. It intentionally performs no automatic retry after timeouts or HTTP 5xx because the server API has no idempotency key and a retry could create a duplicate historical report. """ from __future__ import annotations import argparse import ipaddress import json import logging import os import sys import time import urllib.error import urllib.request from pathlib import Path LOG = logging.getLogger("blocklist-report") def load_config(path: Path) -> dict: with path.open("r", encoding="utf-8") as fh: cfg = json.load(fh) if not isinstance(cfg, dict): raise ValueError("configuration root must be a JSON object") base = str(cfg.get("api_base", "")).rstrip("/") key_file = str(cfg.get("api_key_file", "")) timeout = int(cfg.get("timeout_seconds", 20)) if not base.startswith("https://") and not cfg.get("allow_http", False): raise ValueError("api_base must use HTTPS (or set allow_http=true for a trusted lab)") if not key_file: raise ValueError("api_key_file is required") if timeout < 1 or timeout > 120: raise ValueError("timeout_seconds must be between 1 and 120") return {"api_base": base, "api_key_file": key_file, "timeout_seconds": timeout} def parse_services(raw: str) -> list[dict]: if not raw: return [] result: list[dict] = [] for item in raw.split(","): item = item.strip() if not item: continue try: proto, port_text = item.split(":", 1) port = int(port_text) except Exception as exc: raise ValueError(f"invalid service {item!r}; expected tcp:22 or udp:5060") from exc proto = proto.lower() if proto not in {"tcp", "udp"}: raise ValueError(f"invalid protocol in {item!r}") if not 1 <= port <= 65535: raise ValueError(f"invalid port in {item!r}") result.append({"protocol": proto, "port": port}) # Stable de-duplication. seen = set() unique = [] for service in result: key = (service["protocol"], service["port"]) if key not in seen: unique.append(service) seen.add(key) if len(unique) > 100: raise ValueError("at most 100 services are allowed by the API") return unique def read_key(path: str) -> str: key = Path(path).read_text(encoding="utf-8").strip() if not key: raise ValueError("API key file is empty") return key def build_payload(args: argparse.Namespace) -> dict: # ip_network validates IPv4, IPv6 and CIDR. strict=False is intentional: # the API itself normalizes host bits for CIDR input. ipaddress.ip_network(args.network, strict=False) services = parse_services(args.services) metadata: dict[str, object] = { "producer": "fail2ban", "jail": args.jail, "event_epoch": int(time.time()), } if args.failures is not None: metadata["attempts"] = args.failures if args.bantime is not None: metadata["bantime_seconds"] = args.bantime if args.hostname: metadata["host"] = args.hostname return { "network": args.network, "category": args.category, "reason": f"Fail2ban : bannissement dans la jail {args.jail}", "services": services, "metadata": metadata, } def submit(cfg: dict, api_key: str, payload: dict) -> int: body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") if len(body) > 65536: raise ValueError("generated report exceeds the API 64 KiB limit") request = urllib.request.Request( cfg["api_base"] + "/api/v1/reports", data=body, method="POST", headers={ "Authorization": "Bearer " + api_key, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "blocklist-fail2ban-reporter/1.0", }, ) try: with urllib.request.urlopen(request, timeout=cfg["timeout_seconds"]) as response: status = response.getcode() content_type = response.headers.get("Content-Type", "") raw = response.read(65537) if status != 201: LOG.error("unexpected HTTP status %s", status) return 0 if "application/json" not in content_type.lower(): LOG.warning("report accepted with unexpected Content-Type %r", content_type) return 0 try: decoded = json.loads(raw.decode("utf-8")) report_id = decoded.get("data", {}).get("id") except Exception: report_id = None LOG.info("report accepted%s", f" id={report_id}" if report_id else "") return 0 except urllib.error.HTTPError as exc: # HTTP 429 is definitive: the request was rate-limited and no report was # accepted. We still avoid retrying here; a separate operator-controlled # retry policy can respect Retry-After without creating retry storms. retry_after = exc.headers.get("Retry-After") if exc.headers else None if exc.code == 429: LOG.warning("report rate-limited%s", f" Retry-After={retry_after}" if retry_after else "") else: LOG.error("report rejected HTTP %s", exc.code) return 0 except (urllib.error.URLError, TimeoutError, OSError) as exc: # Ambiguous outcome: the API documentation explicitly warns that a timeout # may have committed the report already. Never auto-retry here. LOG.error("report delivery outcome is uncertain: %s", exc.__class__.__name__) return 0 def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser() p.add_argument("--config", default="/etc/blocklist-reporter/config.json") p.add_argument("--network", required=True) p.add_argument("--category", required=True) p.add_argument("--jail", required=True) p.add_argument("--services", default="") p.add_argument("--failures", type=int) p.add_argument("--bantime", type=int) p.add_argument("--hostname", default=os.uname().nodename if hasattr(os, "uname") else "") return p.parse_args() def main() -> int: logging.basicConfig(level=logging.INFO, format="%(name)s: %(levelname)s: %(message)s") args = parse_args() try: cfg = load_config(Path(args.config)) api_key = read_key(cfg["api_key_file"]) payload = build_payload(args) return submit(cfg, api_key, payload) except Exception as exc: # Never print the API key or full request body. LOG.error("local reporter configuration/input error: %s", exc) return 0 # Reporting is auxiliary and must not break the primary Fail2ban ban action. if __name__ == "__main__": sys.exit(main())