#!/usr/bin/env python3 """Consolidated intrusion detection log scanner. Runs both copyparty and Nginx Proxy Manager (NPM) log analyzers, and outputs a single consolidated MikroTik RouterOS blocklist. """ from __future__ import annotations import argparse import os from dataclasses import dataclass, field from ..utils.blocklist_history import ( committed_lines, cumulative_blocklist, current_file_lines, write_blocklist_report, ) from ..utils.constants import ( COPYPARTY_LOG_PATH, COPYPARTY_REPORT_DIR, DEFAULT_ERROR_RATIO, DEFAULT_MIN_DENIED, DEFAULT_MIN_ERRORS, DEFAULT_TOP, NPM_LOG_PATH, NPM_REPORT_DIR, REPORT_DIR, REPORT_RETENTION_DAYS, REPORT_RETENTION_MAX_FILES, SUBNET_THRESHOLD, ) from ..firewall import MikrotikFirewall, write_blocklist_files from ..utils.retention import prune_report_dirs from ..copyparty import CopypartyAnalyzer from ..npm import NpmAnalyzer from ..utils.ip_utils import is_whitelisted from ..mikrotik import split_cidr_versions from ..utils.whois import consolidate_by_subnet def _sync_blocklist_to_db(connected: bool, desired_v4: list[str], desired_v6: list[str], added_by: str | None = None) -> None: """Best-effort grow-only sync of block targets into the SQLite ledger. Uses ``record_blocklist`` (add-only) so the ledger is the authoritative cumulative record: once an IP is ever blocked it stays active even after it leaves the scan window, matching the router's cumulative model. Returns the added delta or None if the DB is unavailable. """ try: from ..db import get_connection from ..db import repositories as repo conn = get_connection() return repo.record_blocklist( conn, list(desired_v4) + list(desired_v6), added_by=added_by) except Exception as exc: # never break the scan for a DB issue print(f"WARN: could not sync blocklist to db ({exc})") return None def _db_complete_blocklist(desired_v4: list[str], desired_v6: list[str]): """Complete cumulative blocklist from the DB ledger (B001). Returns ``(complete_v4, complete_v6)`` derived from the active blocklist ledger plus this scan's newly desired targets (in case the DB write for this scan has not run yet). Falls back to the git-history reconstruction when the DB is unavailable, so nothing regresses. """ try: from ..db import get_connection from ..db import repositories as repo conn = get_connection() active = [e["cidr"] for e in repo.active_blocklist(conn)] except Exception: active = [] union = set(active) | set(desired_v4) | set(desired_v6) v4, v6 = split_cidr_versions(union) return v4, v6 @dataclass class ScanParams: """Configuration for a consolidated scan. Shared by the CLI entry point and the web interface so both run the exact same pipeline. """ copyparty_log_path: str = COPYPARTY_LOG_PATH copyparty_report_dir: str = COPYPARTY_REPORT_DIR min_denied: int = DEFAULT_MIN_DENIED npm_log_path: str = NPM_LOG_PATH npm_report_dir: str = NPM_REPORT_DIR min_errors: int = DEFAULT_MIN_ERRORS error_ratio: float = DEFAULT_ERROR_RATIO report_dir: str = REPORT_DIR top: int = DEFAULT_TOP quiet: bool = False full: bool = False update_firewall: bool = False dry_run: bool = False @dataclass class ScanResult: """Structured outcome of a consolidated scan.""" ok: bool cp_exists: bool = False npm_exists: bool = False cp_candidates: int = 0 npm_candidates: int = 0 blocked_ips: int = 0 blocked_targets: int = 0 blocked_ipv4: list[str] = field(default_factory=list) blocked_ipv6: list[str] = field(default_factory=list) blocklist_v4_path: str | None = None 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) def run_scan(params: ScanParams) -> ScanResult: """Run the full consolidated scan pipeline and return structured results. Steps: connect to MikroTik (optional) -> snapshot state -> run copyparty + NPM analyzers -> consolidate candidates -> write blocklists -> compare and apply firewall changes (optional). """ errors: list[str] = [] # --- Step 0: open the SQLite store for this consolidated run ------------- db_scan_id: int | None = None try: from ..db import get_connection from ..db import repositories as repo from ..utils.whitelists import sync_whitelists_to_db conn = get_connection() sync_whitelists_to_db() params_map = { "full": params.full, "dry_run": params.dry_run, "update_firewall": params.update_firewall, "quiet": params.quiet, "top": params.top, } db_scan_id = repo.start_scan(conn, "ids", params=params_map, full=params.full, dry_run=params.dry_run) except Exception as exc: print(f"WARN: sqlite store unavailable ({exc})") # --- 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 params.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(params.report_dir) except Exception as e: message = f"could not read MikroTik router ({firewall.router_ip}): {e}" errors.append(message) print(f"\nWARNING: {message}") firewall.disconnect() firewall = None # --- Step 4: run copyparty + NPM analyzers -> blocklist candidates ------ cp_analyzer = CopypartyAnalyzer( log_path=params.copyparty_log_path, report_dir=params.copyparty_report_dir, top=params.top, quiet=params.quiet, full=params.full, min_denied=params.min_denied, ) npm_analyzer = NpmAnalyzer( log_path=params.npm_log_path, report_dir=params.npm_report_dir, top=params.top, quiet=params.quiet, full=params.full, min_errors=params.min_errors, error_ratio=params.error_ratio, ) print("=== Running copyparty scan ===") cp_exists = True try: cp_analyzer.run(db_scan_id=db_scan_id) except Exception as e: errors.append(f"copyparty scan failed: {e}") print(f"Error running copyparty scan: {e}") cp_exists = False print("\n=== Running NPM scan ===") npm_exists = True try: npm_analyzer.run(db_scan_id=db_scan_id) except Exception as e: errors.append(f"NPM scan failed: {e}") print(f"Error running NPM scan: {e}") npm_exists = False if not cp_exists and not npm_exists: message = "Both scans failed. Unable to generate consolidated blocklist." errors.append(message) print(f"\n{message}") if firewall is not None: firewall.disconnect() if db_scan_id is not None: try: from ..db import get_connection from ..db import repositories as repo repo.finish_scan(get_connection(), db_scan_id, status="error", error="; ".join(errors)) except Exception: pass return ScanResult(ok=False, errors=errors) # Consolidate block candidates consolidated_candidates: set[str] = set() if cp_exists: consolidated_candidates.update(cp_analyzer.block_candidates()) if npm_exists: consolidated_candidates.update(npm_analyzer.block_candidates()) # Exclude whitelisted IPs 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 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_cidr_versions(final_blocked_targets) cp_candidates = len(cp_analyzer.block_candidates()) if cp_exists else 0 npm_candidates = len(npm_analyzer.block_candidates()) if npm_exists else 0 print("\n=== Consolidated Summary ===") print(f"copyparty block candidates: {cp_candidates}") print(f"NPM block candidates: {npm_candidates}") print(f"Total unified blocked IPs: {len(final_blocked_ips)}") print(f"Total target blocks/subnets: {len(final_blocked_targets)}") # --- Step 6: write consolidated blocklists to reports/app/data ---------- # 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( blocked_ipv4, blocked_ipv6, dry_run=params.dry_run, report_dir=params.report_dir) finally: firewall.disconnect() action = "Would" if params.dry_run else "" print(f"{action} add {len(firewall_changes['added_ipv4'])} IPv4 and " f"{len(firewall_changes['added_ipv6'])} IPv6 addresses; " 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. # Source of truth for the cumulative set: the SQLite |ledger| first # (B001), falling back to the git-history reconstruction only if the # DB is unavailable. complete_v4, complete_v6 = _db_complete_blocklist( blocked_ipv4, blocked_ipv6) if not complete_v4 and not complete_v6: 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}") # --- Step 8: prune old reports (retention) ------------------------------ # Each report subdirectory keeps at most REPORT_RETENTION_MAX_FILES files, # all newer than REPORT_RETENTION_DAYS days. The blocklist files in # ``app/data`` are always protected. ``--dry-run`` only previews. pruned = prune_report_dirs( params.report_dir, params.copyparty_report_dir, params.npm_report_dir, dry_run=params.dry_run) total_pruned = sum(len(paths) for paths in pruned.values()) action = "Would prune" if params.dry_run else "Pruned" print(f"{action} {total_pruned} report file(s) " f"(keep {REPORT_RETENTION_DAYS}d / " f"max {REPORT_RETENTION_MAX_FILES} per dir)") # --- Step 9: persist consolidated results to SQLite --------------------- # Sync the blocklist ledger and close out the scan record regardless of # whether the router sync was attempted. _sync_blocklist_to_db( firewall is not None, blocked_ipv4, blocked_ipv6, added_by=("scan" if db_scan_id is None else f"scan:{db_scan_id}")) if db_scan_id is not None: try: from ..db import get_connection from ..db import repositories as repo conn = get_connection() result_payload = { "ok": True, "cp_candidates": cp_candidates, "npm_candidates": npm_candidates, "blocked_ips": len(final_blocked_ips), "blocked_targets": len(final_blocked_targets), "firewall_connected": firewall is not None, } repo.finish_scan( conn, db_scan_id, status="done", result=result_payload, log_files=0, request_count=(cp_analyzer and len(cp_analyzer.requests) or 0) + (npm_analyzer and len(npm_analyzer.requests) or 0), error="; ".join(errors) or None, ) except Exception as exc: print(f"WARN: could not finalize scan in db ({exc})") return ScanResult( ok=True, cp_exists=cp_exists, npm_exists=npm_exists, cp_candidates=cp_candidates, npm_candidates=npm_candidates, blocked_ips=len(final_blocked_ips), blocked_targets=len(final_blocked_targets), blocked_ipv4=blocked_ipv4, blocked_ipv6=blocked_ipv6, blocklist_v4_path=blocklist_v4_path, blocklist_v6_path=blocklist_v6_path, firewall_connected=firewall is not None, firewall_changes=firewall_changes, blocklist_report_path=blocklist_report_path, errors=errors, ) def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Consolidated IDS Scanner: run copyparty + NPM analyzers and generate unified blocklist." ) # copyparty arguments parser.add_argument("--copyparty-log-path", default=COPYPARTY_LOG_PATH, help=f"copyparty log directory (default: {COPYPARTY_LOG_PATH})") parser.add_argument("--copyparty-report-dir", default=COPYPARTY_REPORT_DIR, help=f"copyparty report directory (default: {COPYPARTY_REPORT_DIR})") parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED, help=f"minimum copyparty denied hits before flagging (default: {DEFAULT_MIN_DENIED})") # NPM arguments parser.add_argument("--npm-log-path", default=NPM_LOG_PATH, help=f"NPM log directory (default: {NPM_LOG_PATH})") parser.add_argument("--npm-report-dir", default=NPM_REPORT_DIR, help=f"NPM report directory (default: {NPM_REPORT_DIR})") parser.add_argument("--min-errors", type=int, default=DEFAULT_MIN_ERRORS, help=f"minimum NPM error responses to flag (default: {DEFAULT_MIN_ERRORS})") parser.add_argument("--error-ratio", type=float, default=DEFAULT_ERROR_RATIO, help=f"minimum NPM error ratio to flag (default: {DEFAULT_ERROR_RATIO})") # Shared arguments parser.add_argument("--report-dir", default=REPORT_DIR, help=f"consolidated reports directory (default: {REPORT_DIR})") parser.add_argument("--top", type=int, default=DEFAULT_TOP, help=f"rows in the top-IP tables (default: {DEFAULT_TOP})") parser.add_argument("--quiet", action="store_true", help="write files, suppress console reports") parser.add_argument("--full", action="store_true", help="rescan all log files from the start (ignore incremental state)") parser.add_argument("--update-firewall", action="store_true", help="synchronize the MikroTik Blocked list after scanning") parser.add_argument("--dry-run", action="store_true", help="preview firewall changes without applying them") return parser def main(argv: list[str] | None = None) -> None: args = _build_parser().parse_args(argv) params = ScanParams( copyparty_log_path=args.copyparty_log_path, copyparty_report_dir=args.copyparty_report_dir, min_denied=args.min_denied, npm_log_path=args.npm_log_path, npm_report_dir=args.npm_report_dir, min_errors=args.min_errors, error_ratio=args.error_ratio, report_dir=args.report_dir, top=args.top, quiet=args.quiet, full=args.full, update_firewall=args.update_firewall, dry_run=args.dry_run, ) run_scan(params) if __name__ == "__main__": main()