diff --git a/app/db/repositories.py b/app/db/repositories.py index e5f353d..21375f2 100644 --- a/app/db/repositories.py +++ b/app/db/repositories.py @@ -11,7 +11,8 @@ from __future__ import annotations import json import sqlite3 from contextlib import contextmanager -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone +import ipaddress from typing import Any, Iterator, Sequence from ..utils.models import IPStats @@ -104,7 +105,8 @@ def _parse_ts(ts: str) -> tuple[str, float | None]: cleaned = ts.strip() # Accept a trailing timezone like +HH:MM or Z; fall back to naive parse. try: - epoch = datetime.fromisoformat(cleaned.replace("Z", "+00:00")).timestamp() + epoch = datetime.fromisoformat( + cleaned.replace("Z", "+00:00")).timestamp() except ValueError: try: epoch = datetime.fromisoformat(cleaned).timestamp() @@ -339,7 +341,7 @@ def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *, with transaction(conn): for cidr in to_add: _block_entry_inner(conn, cidr, added_by=added_by, last_seen=now, - reasons=reasons) + reasons=reasons) # Refresh last_seen on entries that are still desired so the ledger # reflects recency for existing rows too. conn.executemany( @@ -348,6 +350,52 @@ def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *, return {"added": to_add, "active": len(before) + len(to_add)} +def prune_old_single_ip_blocks(conn: sqlite3.Connection, *, + older_than_days: int = 30, + actor: str | None = "retention") -> list[str]: + """Remove active single-host entries older than ``older_than_days``. + + Keeps CIDR ranges/subnets intact. Host entries are detected via + ``ipaddress.ip_network(..., strict=False)`` and only removed when the + network is exactly one address (/32 for IPv4, /128 for IPv6). + """ + cutoff = datetime.now(timezone.utc) - \ + timedelta(days=max(0, older_than_days)) + rows = conn.execute( + "SELECT cidr, added_at, last_seen FROM blocklist_entries " + "WHERE status='active'").fetchall() + + to_remove: list[str] = [] + for row in rows: + cidr = str(row["cidr"]) + if not _is_single_host_cidr(cidr): + continue + seen = _parse_iso_ts( + row["last_seen"]) or _parse_iso_ts(row["added_at"]) + if seen is None: + continue + if seen < cutoff: + to_remove.append(cidr) + + if not to_remove: + return [] + + now = utcnow() + with transaction(conn): + conn.executemany( + "UPDATE blocklist_entries SET status='removed', removed_at=? " + "WHERE cidr=? AND status='active'", + [(now, cidr) for cidr in to_remove], + ) + conn.executemany( + "INSERT INTO blocklist_events (action, cidr, actor, fired_at, detail) " + "VALUES ('remove', ?, ?, ?, ?)", + [(cidr, actor, now, f"single-ip-retention>{older_than_days}d") + for cidr in to_remove], + ) + return sorted(to_remove) + + def sync_blocklist(conn: sqlite3.Connection, desired_cidrs: Sequence[str], *, added_by: str | None = None) -> dict[str, Any]: """Reconcile active blocklist to ``desired_cidrs``. @@ -402,6 +450,26 @@ def _cidr_version(cidr: str) -> int: return 4 if ":" not in str(cidr) else 6 +def _is_single_host_cidr(cidr: str) -> bool: + try: + net = ipaddress.ip_network(cidr, strict=False) + return net.num_addresses == 1 + except ValueError: + return False + + +def _parse_iso_ts(value: str | None) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + # --------------------------------------------------------------------------- # Whitelists # --------------------------------------------------------------------------- diff --git a/app/scan/__init__.py b/app/scan/__init__.py index b16b70d..d8b022c 100644 --- a/app/scan/__init__.py +++ b/app/scan/__init__.py @@ -1,424 +1,430 @@ -#!/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)`` as **sets** (same contract as - ``cumulative_blocklist``) 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 set(v4), set(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() +#!/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) -> dict | 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() + result = repo.record_blocklist( + conn, list(desired_v4) + list(desired_v6), added_by=added_by) + # Retention policy: expire stale single-host entries after 30 days + # while keeping CIDR ranges/subnets intact. + pruned = repo.prune_old_single_ip_blocks(conn, older_than_days=30) + result["pruned_single_ips"] = pruned + return result + 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)`` as **sets** (same contract as + ``cumulative_blocklist``) 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 set(v4), set(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() diff --git a/tests/test_repositories.py b/tests/test_repositories.py index 622314b..e4dc7a9 100644 --- a/tests/test_repositories.py +++ b/tests/test_repositories.py @@ -13,6 +13,7 @@ import os import sqlite3 import tempfile import unittest +from datetime import datetime, timedelta, timezone from app.db import connect from app.db import repositories as repo @@ -110,7 +111,8 @@ class BlocklistTest(unittest.TestCase): added_by="scan:1") self.assertEqual(r1["active"], 2) # a later scan only sees 5.5.5.5; 6.6.6.6 stays (cumulative). - r2 = repo.record_blocklist(self.conn, ["5.5.5.5/32"], added_by="scan:2") + r2 = repo.record_blocklist( + self.conn, ["5.5.5.5/32"], added_by="scan:2") active = [e["cidr"] for e in repo.active_blocklist(self.conn)] self.assertIn("6.6.6.6/32", active) self.assertEqual(r2["active"], 2) @@ -122,6 +124,38 @@ class BlocklistTest(unittest.TestCase): [e["cidr"] for e in repo.active_blocklist(self.conn)]) self.assertFalse(repo.unblock_entry(self.conn, "7.7.7.7/32")) + def test_prune_old_single_ip_blocks_keeps_ranges(self): + old = (datetime.now(timezone.utc) - timedelta(days=31)).isoformat( + timespec="seconds") + with repo.transaction(self.conn): + repo._block_entry_inner(self.conn, "1.2.3.4/32", last_seen=old) + repo._block_entry_inner( + self.conn, "2001:db8::1/128", last_seen=old) + repo._block_entry_inner(self.conn, "1.2.3.0/24", last_seen=old) + repo._block_entry_inner(self.conn, "2001:db8::/64", last_seen=old) + + removed = repo.prune_old_single_ip_blocks( + self.conn, older_than_days=30) + self.assertCountEqual(removed, ["1.2.3.4/32", "2001:db8::1/128"]) + + active = {e["cidr"] for e in repo.active_blocklist(self.conn)} + self.assertIn("1.2.3.0/24", active) + self.assertIn("2001:db8::/64", active) + self.assertNotIn("1.2.3.4/32", active) + self.assertNotIn("2001:db8::1/128", active) + + def test_prune_old_single_ip_blocks_respects_recency(self): + fresh = (datetime.now(timezone.utc) - timedelta(days=5)).isoformat( + timespec="seconds") + with repo.transaction(self.conn): + repo._block_entry_inner(self.conn, "9.9.9.9/32", last_seen=fresh) + + removed = repo.prune_old_single_ip_blocks( + self.conn, older_than_days=30) + self.assertEqual(removed, []) + self.assertIn("9.9.9.9/32", + {e["cidr"] for e in repo.active_blocklist(self.conn)}) + class WhitelistTest(unittest.TestCase): def setUp(self):