From 28c77691bb02494fea14472333a841aad6b87301 Mon Sep 17 00:00:00 2001 From: zwitschi Date: Wed, 5 Aug 2026 19:44:09 +0200 Subject: [PATCH] feat: write consolidated blocklists and mirror git-diffable router state - Add `write_blocklist_files()` to emit final IPv4/IPv6 blocklists to `reports/app/data/` - Add `APP_REPORT_DIR` / `APP_REPORT_DATA_DIR` constants for consolidated outputs - Write firewall state snapshots to `data/mikrotik/` for git diff auditing - Capture router state before scanning; fall back to scan-only if router is unreachable - Use `split_cidr_versions` for CIDR-aware subnet splitting - Update README with output layout and git-diffable state notes --- README.md | 11 +++++++++- app/constants.py | 4 ++++ app/firewall.py | 55 ++++++++++++++++++++++++++++++++++++++---------- app/scan.py | 47 +++++++++++++++++++++++++++++------------ 4 files changed, 91 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index 91780a9..e5c2c8a 100644 --- a/README.md +++ b/README.md @@ -173,7 +173,7 @@ python app.py --quiet --update-firewall --dry-run ## Output Each script writes to its own report directory. `app.py` also writes -firewall audit snapshots: +consolidated blocklists and firewall audit snapshots: ```text reports/ @@ -206,6 +206,10 @@ reports/ │ │ ├── host_summary.csv virtual host breakdown │ │ └── requests.csv │ └── 2026-08-02-12-00-00.txt +├── app/ +│ └── data/ +│ ├── blocklist_ipv4.txt final consolidated IPv4 blocklist (CIDRs) +│ └── blocklist_ipv6.txt final consolidated IPv6 blocklist (CIDRs) └── mikrotik/ ├── data/ │ ├── address_list_ipv4.json @@ -215,6 +219,11 @@ reports/ └── 2026-08-04-18-30-28.json firewall audit snapshot ``` +The current router state (address lists + firewall rules) is also mirrored to +`data/mikrotik/` on every `--update-firewall` run. Those files are overwritten +each run and are **git-visible**, so `git diff` shows how the firewall rules +and address lists change over time. + ## Detection ### Three detection layers diff --git a/app/constants.py b/app/constants.py index 739e033..cf43374 100644 --- a/app/constants.py +++ b/app/constants.py @@ -39,6 +39,10 @@ REPORT_DIR = os.path.join(PROJECT_ROOT, "reports") COPYPARTY_REPORT_DIR = os.path.join(REPORT_DIR, "copyparty") NPM_REPORT_DIR = os.path.join(REPORT_DIR, "npm") +# Consolidated application outputs (blocklists, shared data) +APP_REPORT_DIR = os.path.join(REPORT_DIR, "app") +APP_REPORT_DATA_DIR = os.path.join(APP_REPORT_DIR, "data") + # MikroTik reports + structured snapshots from each run MIKROTIK_REPORT_DIR = os.path.join(REPORT_DIR, "mikrotik") MIKROTIK_REPORT_DATA_DIR = os.path.join(MIKROTIK_REPORT_DIR, "data") diff --git a/app/firewall.py b/app/firewall.py index d2c0624..4dfc906 100644 --- a/app/firewall.py +++ b/app/firewall.py @@ -15,6 +15,7 @@ import shlex from typing import Iterable from .constants import ( BLOCKLIST_NAME, + MIKROTIK_DATA_DIR, REPORT_DIR, ROUTER_IP, ROUTER_PASSWORD, @@ -23,6 +24,7 @@ from .constants import ( SSH_TIMEOUT, WHITELIST_NAMES, ) +from .files import write_lines from .ip_utils import is_whitelisted @@ -130,9 +132,20 @@ class MikrotikFirewall: f"{path} firewall filter print terse") return self._parse_terse_output(output) + @staticmethod + def _write_json(path: str, obj) -> None: + with open(path, "w", encoding="utf-8") as file: + json.dump(obj, file, indent=2, sort_keys=True) + def save_state_report(self, report_dir: str = REPORT_DIR, changes: dict | None = None) -> str: - """Save current address lists and firewall rules for auditing.""" + """Save current address lists and firewall rules for auditing. + + Writes structured snapshots to ``/mikrotik/data/`` and to + the common ``data/mikrotik/`` directory (overwritten every run so git + can diff changes to the firewall state over time), plus a timestamped + JSON report under ``/mikrotik/``. + """ state = { "router": self.router_ip, "collected_at": datetime.now().astimezone().isoformat(), @@ -147,19 +160,22 @@ class MikrotikFirewall: } if changes is not None: state["changes"] = changes - data_dir = os.path.join(report_dir, "mikrotik", "data") - os.makedirs(data_dir, exist_ok=True) + + report_data_dir = os.path.join(report_dir, "mikrotik", "data") + os.makedirs(report_data_dir, exist_ok=True) + os.makedirs(MIKROTIK_DATA_DIR, exist_ok=True) for version in ("ipv4", "ipv6"): - with open(os.path.join(data_dir, f"address_list_{version}.json"), "w", encoding="utf-8") as file: - json.dump(state["address_lists"][version], - file, indent=2, sort_keys=True) - with open(os.path.join(data_dir, f"firewall_{version}.json"), "w", encoding="utf-8") as file: - json.dump(state["firewall_rules"][version], - file, indent=2, sort_keys=True) + for data_dir in (report_data_dir, MIKROTIK_DATA_DIR): + self._write_json( + os.path.join(data_dir, f"address_list_{version}.json"), + state["address_lists"][version]) + self._write_json( + os.path.join(data_dir, f"firewall_{version}.json"), + state["firewall_rules"][version]) + timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S") report_path = os.path.join(report_dir, "mikrotik", f"{timestamp}.json") - with open(report_path, "w", encoding="utf-8") as file: - json.dump(state, file, indent=2, sort_keys=True) + self._write_json(report_path, state) return report_path def sync_blocklist(self, blocked_ipv4: Iterable[str], @@ -213,6 +229,23 @@ class MikrotikFirewall: f"/ipv6 firewall address-list add list={name} address={ip}") +def write_blocklist_files(blocked_ipv4: Iterable[str], + blocked_ipv6: Iterable[str], + report_dir: str = REPORT_DIR) -> tuple[str, str]: + """Write the final consolidated blocklist to ``/app/data/``. + + Returns the paths of the IPv4 and IPv6 blocklist files + (``blocklist_ipv4.txt`` / ``blocklist_ipv6.txt``). + """ + data_dir = os.path.join(report_dir, "app", "data") + os.makedirs(data_dir, exist_ok=True) + v4_path = os.path.join(data_dir, "blocklist_ipv4.txt") + v6_path = os.path.join(data_dir, "blocklist_ipv6.txt") + write_lines(v4_path, sorted(blocked_ipv4)) + write_lines(v6_path, sorted(blocked_ipv6)) + return v4_path, v6_path + + def get_scan_report(report_dir: str = REPORT_DIR) -> dict[str, list[str]]: """Read the latest scan report `mikrotik-block.rsc` from the report directory and return the list of blocked IPs.""" report_path = os.path.join(report_dir, "mikrotik-block.rsc") diff --git a/app/scan.py b/app/scan.py index 395741c..0f9e36e 100644 --- a/app/scan.py +++ b/app/scan.py @@ -21,10 +21,11 @@ from .constants import ( REPORT_DIR, SUBNET_THRESHOLD, ) -from .firewall import MikrotikFirewall +from .firewall import MikrotikFirewall, write_blocklist_files from .copyparty import CopypartyAnalyzer from .npm import NpmAnalyzer -from .ip_utils import is_whitelisted, split_versions +from .ip_utils import is_whitelisted +from .mikrotik import split_cidr_versions from .whois import consolidate_by_subnet @@ -65,10 +66,24 @@ def main() -> None: args = parser.parse_args() - # 1. Initialize Analyzers - # We turn off inner write_rsc of individual analyzers to avoid duplicate/redundant rsc writes, - # or let them write their individual ones if they want to. Writing individual scripts is harmless. - # We'll match the individual tools' default write behavior, but generate the consolidated one. + # --- Step 1-3: connect to MikroTik RouterOS and capture current state --- + # If the router is unreachable we fall back to scan-only mode: still + # generate reports and blocklists, but skip the firewall sync. + firewall: MikrotikFirewall | None = None + if args.update_firewall: + firewall = MikrotikFirewall() + try: + firewall.connect() + # Read current firewall rules + address lists and snapshot them to + # reports/mikrotik/data/ and the git-diffable data/mikrotik/. + firewall.save_state_report(args.report_dir) + except Exception as e: + print(f"\nWARNING: could not read MikroTik router " + f"({firewall.router_ip}): {e}") + firewall.disconnect() + firewall = None + + # --- Step 4: run copyparty + NPM analyzers -> blocklist candidates ------ cp_analyzer = CopypartyAnalyzer( log_path=args.copyparty_log_path, report_dir=args.copyparty_report_dir, @@ -86,7 +101,6 @@ def main() -> None: error_ratio=args.error_ratio, ) - # 2. Run copyparty analysis print("=== Running copyparty scan ===") cp_exists = True try: @@ -95,7 +109,6 @@ def main() -> None: print(f"Error running copyparty scan: {e}") cp_exists = False - # 3. Run NPM analysis print("\n=== Running NPM scan ===") npm_exists = True try: @@ -106,9 +119,11 @@ def main() -> None: if not cp_exists and not npm_exists: print("\nBoth scans failed. Unable to generate consolidated blocklist.") + if firewall is not None: + firewall.disconnect() return - # 4. Consolidate block candidates + # Consolidate block candidates consolidated_candidates: set[str] = set() if cp_exists: consolidated_candidates.update(cp_analyzer.block_candidates()) @@ -119,11 +134,11 @@ def main() -> None: final_blocked_ips = { ip for ip in consolidated_candidates if not is_whitelisted(ip)} - # Group into subnets via WHOIS if multiple malicious IPs in same subnet (threshold = SUBNET_THRESHOLD) + # Group into subnets via WHOIS if multiple malicious IPs in same subnet print("\nGrouping blocked IPs into subnets using WHOIS...") final_blocked_targets = consolidate_by_subnet( final_blocked_ips, threshold=SUBNET_THRESHOLD) - blocked_ipv4, blocked_ipv6 = split_versions(final_blocked_targets) + blocked_ipv4, blocked_ipv6 = split_cidr_versions(final_blocked_targets) print("\n=== Consolidated Summary ===") print( @@ -133,9 +148,13 @@ def main() -> None: print(f"Total unified blocked IPs: {len(final_blocked_ips)}") print(f"Total target blocks/subnets: {len(final_blocked_targets)}") - if args.update_firewall: - firewall = MikrotikFirewall() - firewall.connect() + # --- Step 6: write consolidated blocklists to reports/app/data ---------- + blocklist_v4_path, blocklist_v6_path = write_blocklist_files( + blocked_ipv4, blocked_ipv6, args.report_dir) + print(f"blocklist files:\n {blocklist_v4_path}\n {blocklist_v6_path}") + + # --- Steps 5+7: compare current state with candidates and apply --------- + if firewall is not None: try: changes = firewall.sync_blocklist( blocked_ipv4, blocked_ipv6, dry_run=args.dry_run,