feat: make MikroTik blocklist cumulative via git history

app/blocklist_history.py

@@ -0,0 +1,119 @@
+#!/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 typing import Any, Iterable
+
+#: 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 -- <path>`` 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),
+    }