187 lines
7.1 KiB
Python
187 lines
7.1 KiB
Python
#!/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 -- <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),
|
|
}
|
|
|
|
|
|
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 ``<report_dir>/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
|