#!/usr/bin/env python3 """Git-backed cumulative blocklist history. The consolidated blocklist lives in git-tracked files (``reports/app/data/blocklist_ipv4.txt`` / ``blocklist_ipv6.txt``). To make the MikroTik blocklist grow-only, the "complete set of all blocked IPs" is reconstructed from **every git revision** of those files (the union of all lines ever present), plus the current working files and the latest scan targets. This keeps IPs blocked even after they leave the NPM ~30-day log window, and only ever removes router entries that were never blocked by the system. """ from __future__ import annotations import ipaddress import os import re import subprocess from datetime import datetime from typing import Any, Iterable from .files import write_json from .reporting import timestamp_name #: A ``+ 1.2.3.4/32`` added line in a git patch; ``+++ b/...`` is the file #: header and is excluded by the negative lookahead. _ADDED_LINE_RE = re.compile(r"^\+(?!\+)(.*)$") def _run_git(*args: str) -> subprocess.CompletedProcess[str] | None: """Run a git command, returning None on any failure (never raises).""" try: return subprocess.run( ["git", *args], capture_output=True, text=True, check=False) except (OSError, subprocess.SubprocessError): return None def _valid_line(line: str) -> str | None: """Normalize a blocklist line to a canonical CIDR string, or None.""" line = line.strip() if not line: return None try: return str(ipaddress.ip_network(line, strict=False)) except ValueError: return None def committed_lines(path: str) -> set[str]: """Return the union of all lines ever present in ``path`` across git. Uses a single ``git log --all -p -- `` and collects the ``+`` added lines (excluding the ``+++`` file header). A line present in any revision appears as an added line in some commit's diff, so the union equals "all lines ever present". Returns an empty set when git is unavailable, the path is untracked, or the repository has no history — callers then fall back to the working file. """ proc = _run_git("log", "--all", "-p", "--", path) if proc is None or proc.returncode != 0: return set() lines: set[str] = set() for raw in proc.stdout.splitlines(): m = _ADDED_LINE_RE.match(raw) if not m: continue clean = _valid_line(m.group(1)) if clean is not None: lines.add(clean) return lines def current_file_lines(path: str) -> set[str]: """Return the validated lines currently in ``path`` (missing file -> {}).""" try: with open(path, "r", encoding="utf-8") as f: return {clean for clean in (_valid_line(x) for x in f) if clean is not None} except OSError: return set() def cumulative_blocklist(v4_path: str, v6_path: str, desired_v4: Iterable[str], desired_v6: Iterable[str], ) -> tuple[set[str], set[str]]: """Complete set of all blocked targets (history | working file | desired). Returns ``(complete_v4, complete_v6)`` of canonical CIDR strings. """ complete_v4 = ( committed_lines(v4_path) | current_file_lines(v4_path) | {c for c in (_valid_line(x) for x in desired_v4) if c is not None}) complete_v6 = ( committed_lines(v6_path) | current_file_lines(v6_path) | {c for c in (_valid_line(x) for x in desired_v6) if c is not None}) return complete_v4, complete_v6 def compute_changes(current_v4: set[str], current_v6: set[str], complete_v4: set[str], complete_v6: set[str], whitelisted_v4: set[str], whitelisted_v6: set[str], ) -> dict[str, Any]: """Compute the add/remove diff for a cumulative (grow-only) blocklist. Entries are added when present in the complete set but not yet on the router and not whitelisted. Entries are removed only when they sit on the router but are absent from the complete set — i.e. they were never blocked by the system (manual/rogue entries). Callers may attach extra keys (e.g. ``dry_run``). """ return { "added_ipv4": sorted(complete_v4 - current_v4 - whitelisted_v4), "added_ipv6": sorted(complete_v6 - current_v6 - whitelisted_v6), "removed_ipv4": sorted(current_v4 - complete_v4), "removed_ipv6": sorted(current_v6 - complete_v6), } def blocklist_delta(history_v4: set[str], history_v6: set[str], prev_v4: set[str], prev_v6: set[str], complete_v4: set[str], complete_v6: set[str], ) -> dict[str, list[str]]: """Complete-set delta between the previous and current blocklist state. ``added`` are entries newly part of the blocklist (this run's targets not blocked before); ``removed`` are entries no longer part of it — normally empty in the cumulative model. """ before_v4 = history_v4 | prev_v4 before_v6 = history_v6 | prev_v6 return { "added_ipv4": sorted(complete_v4 - before_v4), "added_ipv6": sorted(complete_v6 - before_v6), "removed_ipv4": sorted(before_v4 - complete_v4), "removed_ipv6": sorted(before_v6 - complete_v6), } def write_blocklist_report(report_dir: str, *, history_v4: set[str], history_v6: set[str], prev_v4: set[str], prev_v6: set[str], complete_v4: set[str], complete_v6: set[str], firewall_changes: dict | None = None, dry_run: bool = False, errors: Iterable[str] = (), ) -> str: """Write the timestamped blocklist report to ``/app/``. Records the complete-set delta (added/removed) plus the full cumulative blocked list (git history | working files | scan targets), and — when supplied — the router sync changes. Returns the written file path. """ delta = blocklist_delta( history_v4, history_v6, prev_v4, prev_v6, complete_v4, complete_v6) payload = { "generated_at": datetime.now().astimezone().isoformat( timespec="seconds"), "report_dir": report_dir, "dry_run": dry_run, "summary": { "added": len(delta["added_ipv4"]) + len(delta["added_ipv6"]), "removed": len(delta["removed_ipv4"]) + len(delta["removed_ipv6"]), "total_blocked": len(complete_v4) + len(complete_v6), }, "added_ipv4": delta["added_ipv4"], "added_ipv6": delta["added_ipv6"], "removed_ipv4": delta["removed_ipv4"], "removed_ipv6": delta["removed_ipv6"], "blocked_ipv4": sorted(complete_v4), "blocked_ipv6": sorted(complete_v6), "firewall_changes": firewall_changes, "errors": list(errors), } app_dir = os.path.join(report_dir, "app") os.makedirs(app_dir, exist_ok=True) path = os.path.join(app_dir, f"{timestamp_name()}.json") write_json(path, payload) return path