#!/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 from dataclasses import dataclass, field from .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, SUBNET_THRESHOLD, ) from .firewall import MikrotikFirewall, write_blocklist_files from .copyparty import CopypartyAnalyzer from .npm import NpmAnalyzer from .ip_utils import is_whitelisted from .mikrotik import split_cidr_versions from .whois import consolidate_by_subnet @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 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 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 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, 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, min_errors=params.min_errors, error_ratio=params.error_ratio, ) print("=== Running copyparty scan ===") cp_exists = True try: cp_analyzer.run() 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() 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() 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 ---------- blocklist_v4_path, 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 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.") 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, 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("--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, update_firewall=args.update_firewall, dry_run=args.dry_run, ) run_scan(params) if __name__ == "__main__": main()