diff --git a/.gitignore b/.gitignore index 7b38716..6d2425b 100644 --- a/.gitignore +++ b/.gitignore @@ -13,4 +13,6 @@ data/scan_state.json # reports data reports/copyparty/ reports/mikrotik/ -reports/npm/ \ No newline at end of file +reports/npm/ +# timestamped app blocklist change reports (reports/app/data/ stays tracked) +reports/app/*.json \ No newline at end of file diff --git a/README.md b/README.md index c1e4637..4947330 100644 --- a/README.md +++ b/README.md @@ -204,6 +204,11 @@ time and is never trimmed automatically. (`/blocklist` → remove) or directly on the router; the automated sync will not re-add entries that are not part of a scan. +Each consolidated scan with `--update-firewall` also writes a timestamped +change report to `reports/app/YYYY-MM-DD-hh-mm-ss.json`: the number and list +of entries added / removed (complete-set delta), the full cumulative blocked +list, and the router sync changes. + ## Web UI `python app.py` starts a **localhost-only** Flask web UI (default diff --git a/app/blocklist_history.py b/app/blocklist_history.py index e3eb7a8..b325ad2 100644 --- a/app/blocklist_history.py +++ b/app/blocklist_history.py @@ -18,8 +18,12 @@ 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"^\+(?!\+)(.*)$") @@ -117,3 +121,66 @@ def compute_changes(current_v4: set[str], current_v6: set[str], "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 diff --git a/app/scan.py b/app/scan.py index d175001..2f6e344 100644 --- a/app/scan.py +++ b/app/scan.py @@ -8,8 +8,15 @@ and outputs a single consolidated MikroTik RouterOS blocklist. from __future__ import annotations import argparse +import os from dataclasses import dataclass, field +from .blocklist_history import ( + committed_lines, + cumulative_blocklist, + current_file_lines, + write_blocklist_report, +) from .constants import ( COPYPARTY_LOG_PATH, COPYPARTY_REPORT_DIR, @@ -70,6 +77,7 @@ class ScanResult: blocklist_v6_path: str | None = None firewall_connected: bool = False firewall_changes: dict | None = None + blocklist_report_path: str | None = None errors: list[str] = field(default_factory=list) @@ -173,12 +181,22 @@ def run_scan(params: ScanParams) -> ScanResult: print(f"Total target blocks/subnets: {len(final_blocked_targets)}") # --- Step 6: write consolidated blocklists to reports/app/data ---------- - blocklist_v4_path, blocklist_v6_path = write_blocklist_files( - blocked_ipv4, blocked_ipv6, params.report_dir) + # Capture the previous blocklist state BEFORE the files are overwritten so + # the change report can compute the complete-set delta. + app_data_dir = os.path.join(params.report_dir, "app", "data") + blocklist_v4_path = os.path.join(app_data_dir, "blocklist_ipv4.txt") + blocklist_v6_path = os.path.join(app_data_dir, "blocklist_ipv6.txt") + prev_v4 = current_file_lines(blocklist_v4_path) + prev_v6 = current_file_lines(blocklist_v6_path) + history_v4 = committed_lines(blocklist_v4_path) + history_v6 = committed_lines(blocklist_v6_path) + + write_blocklist_files(blocked_ipv4, blocked_ipv6, params.report_dir) print(f"blocklist files:\n {blocklist_v4_path}\n {blocklist_v6_path}") # --- Steps 5+7: compare current state with candidates and apply --------- firewall_changes: dict | None = None + blocklist_report_path: str | None = None if firewall is not None: try: firewall_changes = firewall.sync_blocklist( @@ -192,6 +210,20 @@ def run_scan(params: ScanParams) -> ScanResult: f"remove {len(firewall_changes['removed_ipv4'])} IPv4 and " f"{len(firewall_changes['removed_ipv6'])} IPv6 addresses.") + # Timestamped change report — only when the router was reachable. + complete_v4, complete_v6 = cumulative_blocklist( + blocklist_v4_path, blocklist_v6_path, + blocked_ipv4, blocked_ipv6) + blocklist_report_path = write_blocklist_report( + params.report_dir, + history_v4=history_v4, history_v6=history_v6, + prev_v4=prev_v4, prev_v6=prev_v6, + complete_v4=complete_v4, complete_v6=complete_v6, + firewall_changes=firewall_changes, + dry_run=params.dry_run, + errors=errors) + print(f"blocklist report: {blocklist_report_path}") + return ScanResult( ok=True, cp_exists=cp_exists, @@ -206,6 +238,7 @@ def run_scan(params: ScanParams) -> ScanResult: blocklist_v6_path=blocklist_v6_path, firewall_connected=firewall is not None, firewall_changes=firewall_changes, + blocklist_report_path=blocklist_report_path, errors=errors, )