142 lines
5.5 KiB
Python
142 lines
5.5 KiB
Python
#!/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
|
|
import os
|
|
from datetime import datetime
|
|
|
|
import utils
|
|
from copyparty import CopypartyAnalyzer, DEFAULT_LOG_PATH as CP_DEFAULT_LOG, DEFAULT_REPORT_DIR as CP_DEFAULT_REPORT
|
|
from npm import NpmAnalyzer, DEFAULT_LOG_PATH as NPM_DEFAULT_LOG, DEFAULT_REPORT_DIR as NPM_DEFAULT_REPORT
|
|
|
|
DEFAULT_CONSOLIDATED_REPORT_DIR = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)), "reports")
|
|
|
|
|
|
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=CP_DEFAULT_LOG,
|
|
help=f"copyparty log directory (default: {CP_DEFAULT_LOG})")
|
|
parser.add_argument("--copyparty-report-dir", default=CP_DEFAULT_REPORT,
|
|
help=f"copyparty report directory (default: {CP_DEFAULT_REPORT})")
|
|
parser.add_argument("--min-denied", type=int, default=1,
|
|
help="minimum copyparty denied hits before flagging (default: 1)")
|
|
|
|
# NPM arguments
|
|
parser.add_argument("--npm-log-path", default=NPM_DEFAULT_LOG,
|
|
help=f"NPM log directory (default: {NPM_DEFAULT_LOG})")
|
|
parser.add_argument("--npm-report-dir", default=NPM_DEFAULT_REPORT,
|
|
help=f"NPM report directory (default: {NPM_DEFAULT_REPORT})")
|
|
parser.add_argument("--min-errors", type=int, default=5,
|
|
help="minimum NPM error responses to flag (default: 5)")
|
|
parser.add_argument("--error-ratio", type=float, default=0.5,
|
|
help="minimum NPM error ratio to flag (default: 0.5)")
|
|
|
|
# Shared arguments
|
|
parser.add_argument("--report-dir", default=DEFAULT_CONSOLIDATED_REPORT_DIR,
|
|
help=f"consolidated reports directory (default: {DEFAULT_CONSOLIDATED_REPORT_DIR})")
|
|
parser.add_argument("--top", type=int, default=30,
|
|
help="rows in the top-IP tables (default: 30)")
|
|
parser.add_argument("--quiet", action="store_true",
|
|
help="write files, suppress console reports")
|
|
parser.add_argument("--no-rsc", action="store_true",
|
|
help="skip MikroTik consolidated .rsc generation")
|
|
|
|
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,
|
|
write_rsc=not args.no_rsc,
|
|
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,
|
|
write_rsc=not args.no_rsc,
|
|
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 utils.is_whitelisted(ip)}
|
|
|
|
# Group into subnets via WHOIS if multiple malicious IPs in same subnet (threshold = 3)
|
|
print("\nGrouping blocked IPs into subnets using WHOIS...")
|
|
final_blocked_targets = utils.consolidate_by_subnet(
|
|
final_blocked_ips, threshold=3)
|
|
|
|
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)}")
|
|
|
|
# 5. Write Consolidated MikroTik RouterOS Script
|
|
if not args.no_rsc:
|
|
rsc_path = os.path.join(args.report_dir, "mikrotik-block.rsc")
|
|
utils.ensure_dirs(args.report_dir)
|
|
path, n_blocked, n_whitelisted = utils.write_mikrotik(
|
|
rsc_path,
|
|
"ids",
|
|
final_blocked_targets
|
|
)
|
|
print(f"\nConsolidated MikroTik script: {path}")
|
|
print(
|
|
f" {n_blocked} targets blocked, {n_whitelisted} CIDR ranges whitelisted")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|