intrusion-detection/app/scan.py

#!/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 script.
"""

from __future__ import annotations

import argparse

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
from .copyparty import CopypartyAnalyzer
from .npm import NpmAnalyzer
from .ip_utils import is_whitelisted, split_versions
from .whois import consolidate_by_subnet


def main() -> None:
    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")

    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.
    cp_analyzer = CopypartyAnalyzer(
        log_path=args.copyparty_log_path,
        report_dir=args.copyparty_report_dir,
        top=args.top,
        quiet=args.quiet,
        min_denied=args.min_denied,
    )

    npm_analyzer = NpmAnalyzer(
        log_path=args.npm_log_path,
        report_dir=args.npm_report_dir,
        top=args.top,
        quiet=args.quiet,
        min_errors=args.min_errors,
        error_ratio=args.error_ratio,
    )

    # 2. Run copyparty analysis
    print("=== Running copyparty scan ===")
    cp_exists = True
    try:
        cp_analyzer.run()
    except Exception as e:
        print(f"Error running copyparty scan: {e}")
        cp_exists = False

    # 3. Run NPM analysis
    print("\n=== Running NPM scan ===")
    npm_exists = True
    try:
        npm_analyzer.run()
    except Exception as e:
        print(f"Error running NPM scan: {e}")
        npm_exists = False

    if not cp_exists and not npm_exists:
        print("\nBoth scans failed. Unable to generate consolidated blocklist.")
        return

    # 4. 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 (threshold = SUBNET_THRESHOLD)
    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)

    print("\n=== Consolidated Summary ===")
    print(
        f"copyparty block candidates: {len(cp_analyzer.block_candidates()) if cp_exists else 0}")
    print(
        f"NPM block candidates:       {len(npm_analyzer.block_candidates()) if npm_exists else 0}")
    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()
        try:
            changes = firewall.sync_blocklist(
                blocked_ipv4, blocked_ipv6, dry_run=args.dry_run,
                report_dir=args.report_dir)
        finally:
            firewall.disconnect()
        action = "Would" if args.dry_run else ""
        print(f"{action} add {len(changes['added_ipv4'])} IPv4 and "
              f"{len(changes['added_ipv6'])} IPv6 addresses; "
              f"remove {len(changes['removed_ipv4'])} IPv4 and "
              f"{len(changes['removed_ipv6'])} IPv6 addresses.")


if __name__ == "__main__":
    main()