Télécharger blocklist-report.py
#!/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())