From 40212d84f7c02104edd91a91d0e825ff554d48ab Mon Sep 17 00:00:00 2001 From: zwitschi Date: Tue, 4 Aug 2026 21:56:55 +0200 Subject: [PATCH] Refactor copyparty and npm analyzers to use updated imports and remove utils module Update whois cache with additional IP ranges --- copyparty.py | 27 +++-- data/whois_cache.json | 7 +- npm.py | 24 +++-- utils.py | 236 ------------------------------------------ 4 files changed, 36 insertions(+), 258 deletions(-) diff --git a/copyparty.py b/copyparty.py index e2b35f4..bae91aa 100644 --- a/copyparty.py +++ b/copyparty.py @@ -12,8 +12,13 @@ from __future__ import annotations import os import re -import utils -from utils import Request +from analyzer import Analyzer +from ip_utils import split_versions +from files import write_json, write_lines +from flags import flag_denied_only +from models import IPStats, Request +from utils import arg_parser + DEFAULT_LOG_PATH = os.environ.get("COPYPARTY_LOG_PATH", r"P:\copyparty") DEFAULT_REPORT_DIR = os.path.join( @@ -37,7 +42,7 @@ CLIENT_BAN_RE = re.compile(r"client banned") BAN_SECS_RE = re.compile(r"banned for (\d+) sec") -class CopypartyAnalyzer(utils.Analyzer): +class CopypartyAnalyzer(Analyzer): name = "copyparty" title = "copyparty log analysis report" log_pattern = re.compile(r"\d{4}-\d{2}-\d{2}\.txt(\.\d+)?$") @@ -115,7 +120,7 @@ class CopypartyAnalyzer(utils.Analyzer): # --- flagging ---------------------------------------------------------- def flag(self, stats): - return utils.flag_denied_only(stats, min_denied=self.min_denied) + return flag_denied_only(stats, min_denied=self.min_denied) def flag_comment(self, s): return (f"denied={s.denied} total={s.total} " @@ -127,7 +132,7 @@ class CopypartyAnalyzer(utils.Analyzer): # --- extras ------------------------------------------------------------ - def banned_ips(self) -> dict[str, utils.IPStats]: + def banned_ips(self) -> dict[str, IPStats]: return {ip: s for ip, s in self.stats.items() if s.banned} def authed_users(self) -> dict[str, int]: @@ -144,17 +149,17 @@ class CopypartyAnalyzer(utils.Analyzer): ("authenticated users", self.authed_users() or "{}")] def write_extra(self, data_dir: str) -> list[str]: - v4, v6 = utils.split_versions(self.banned_ips()) + v4, v6 = split_versions(self.banned_ips()) return [ - utils.write_lines(os.path.join(data_dir, "banned_ips.txt"), v4), - utils.write_lines(os.path.join(data_dir, "banned_ipv6.txt"), v6), - utils.write_json(os.path.join(data_dir, "authed_users.json"), - self.authed_users()), + write_lines(os.path.join(data_dir, "banned_ips.txt"), v4), + write_lines(os.path.join(data_dir, "banned_ipv6.txt"), v6), + write_json(os.path.join(data_dir, "authed_users.json"), + self.authed_users()), ] def main() -> None: - parser = utils.arg_parser( + parser = arg_parser( "analyze copyparty logs, flag denied-only IPs, export MikroTik rules", DEFAULT_LOG_PATH, DEFAULT_REPORT_DIR) parser.add_argument("--min-denied", type=int, default=1, diff --git a/data/whois_cache.json b/data/whois_cache.json index 6925c0e..42e9ba3 100644 --- a/data/whois_cache.json +++ b/data/whois_cache.json @@ -910,5 +910,10 @@ "163.7.13.111": "163.7.13.0/24", "222.176.200.251": "222.176.0.0/13", "123.163.114.150": "123.160.0.0/14", - "136.64.132.60": "136.64.0.0/15" + "136.64.132.60": "136.64.0.0/15", + "20.215.218.105": "20.192.0.0/10", + "20.203.208.191": "20.192.0.0/10", + "2400:8901::2000:beff:fea3:5a44": "2400:8901::/32", + "163.7.12.95": "163.7.12.0/24", + "209.87.169.76": "209.87.169.0/24" } \ No newline at end of file diff --git a/npm.py b/npm.py index b4c548c..1c6aec1 100644 --- a/npm.py +++ b/npm.py @@ -12,8 +12,12 @@ import os import re from collections import Counter -import utils -from utils import Request +from analyzer import Analyzer +from files import write_csv +from flags import flag_scanners +from models import Request +from utils import arg_parser + DEFAULT_LOG_PATH = os.environ.get("NPM_LOG_PATH", r"Y:\npm\logs") DEFAULT_REPORT_DIR = os.path.join( @@ -40,7 +44,7 @@ LINE_RE = re.compile( ) -class NpmAnalyzer(utils.Analyzer): +class NpmAnalyzer(Analyzer): name = "npm" title = "Nginx Proxy Manager log analysis report" log_suffixes = (".log", ".gz") @@ -112,8 +116,8 @@ class NpmAnalyzer(utils.Analyzer): # --- flagging ---------------------------------------------------------- def flag(self, stats): - return utils.flag_scanners(stats, min_errors=self.min_errors, - min_ratio=self.min_ratio) + return flag_scanners(stats, min_errors=self.min_errors, + min_ratio=self.min_ratio) # --- extras ------------------------------------------------------------ @@ -131,15 +135,15 @@ class NpmAnalyzer(utils.Analyzer): uas = Counter(r.ua for r in self.requests if r.ua and r.ua != "-") hosts = Counter(r.host for r in self.requests if r.host) return [ - utils.write_csv(os.path.join(data_dir, "ua_summary.csv"), - ["user_agent", "count"], uas.most_common()), - utils.write_csv(os.path.join(data_dir, "host_summary.csv"), - ["host", "count"], hosts.most_common()), + write_csv(os.path.join(data_dir, "ua_summary.csv"), + ["user_agent", "count"], uas.most_common()), + write_csv(os.path.join(data_dir, "host_summary.csv"), + ["host", "count"], hosts.most_common()), ] def main() -> None: - parser = utils.arg_parser( + parser = arg_parser( "analyze NPM proxy logs, flag scanner IPs, export MikroTik rules", DEFAULT_LOG_PATH, DEFAULT_REPORT_DIR) parser.add_argument("--min-errors", type=int, default=5, diff --git a/utils.py b/utils.py index fc4895e..fa0a19a 100644 --- a/utils.py +++ b/utils.py @@ -649,239 +649,3 @@ def arg_parser(description: str, default_log_path: str, p.add_argument("--no-rsc", action="store_true", help="skip MikroTik .rsc generation") return p - - -# --------------------------------------------------------------------------- -# analyzer -# --------------------------------------------------------------------------- - -class Analyzer: - """Log-source-agnostic analysis pipeline. - - Subclasses must set the class attributes below and implement - `parse_line()` and `flag()`. - """ - - name = "ids" # short id, used in file names / tags - title = "IDS log analysis report" - log_pattern: re.Pattern | None = None - log_suffixes: tuple[str, ...] = () - exclude_prefixes: tuple[str, ...] = () - ip_columns: Sequence[Column] = () - request_columns: Sequence[Column] = () - paths_file = "error_paths.csv" - paths_label = "error paths" - flag_files = ("flagged_ips.txt", "flagged_ipv6.txt") - flag_label = "Flagged IPs" - flag_note = "" - - def __init__(self, log_path: str, report_dir: str, top: int = 30, - quiet: bool = False, write_rsc: bool = True): - self.log_path = log_path - self.report_dir = report_dir - self.data_dir = os.path.join(report_dir, "data") - self.top = top - self.quiet = quiet - self.write_rsc = write_rsc - self.files: list[str] = [] - self.requests: list[Request] = [] - self.stats: dict[str, IPStats] = {} - self.flagged: dict[str, IPStats] = {} - self.per_file: dict[str, int] = {} - - # --- hooks ------------------------------------------------------------- - - def parse_line(self, line: str) -> Request | None: - raise NotImplementedError - - def flag(self, stats: dict[str, IPStats]) -> dict[str, IPStats]: - raise NotImplementedError - - def flag_comment(self, s: IPStats) -> str: - return (f"errors={s.errors} total={s.total} ratio={s.error_ratio} " - f"first={s.first} last={s.last}") - - def block_candidates(self) -> set[str]: - malicious = flag_malicious_patterns(self.stats) - return set(self.flagged) | set(malicious) - - def extra_summary_rows(self) -> list[tuple[str, Any]]: - return [] - - def write_extra(self, data_dir: str) -> list[str]: - return [] - - # --- pipeline ---------------------------------------------------------- - - def discover(self) -> list[str]: - self.files = find_logs(self.log_path, self.log_pattern, - self.log_suffixes, self.exclude_prefixes) - if not self.files: - raise FileNotFoundError(f"no log files found in: {self.log_path}") - return self.files - - def load(self) -> list[Request]: - for name in self.files: - count = 0 - for line in read_lines(os.path.join(self.log_path, name)): - req = self.parse_line(line) - if req is not None: - self.requests.append(req) - count += 1 - self.per_file[name] = count - return self.requests - - def analyze(self) -> dict[str, IPStats]: - self.stats = aggregate(self.requests) - self.flagged = self.flag(self.stats) - return self.stats - - # --- derived data ------------------------------------------------------ - - def error_paths(self) -> Counter: - return Counter(r.path for r in self.requests if r.is_error and r.path) - - def flagged_split(self) -> tuple[list[IPStats], list[IPStats]]: - v4 = [s for s in self.flagged.values() if s.version == 4] - v6 = [s for s in self.flagged.values() if s.version == 6] - def key(s): return (-s.errors, s.ip) - return sorted(v4, key=key), sorted(v6, key=key) - - def summary_rows(self) -> list[tuple[str, Any]]: - v4, v6 = split_versions(self.stats) - rows = [ - ("log path", self.log_path), - ("log files", len(self.files)), - ("total requests", len(self.requests)), - ("unique IPs", - f"{len(self.stats)} (IPv4 {len(v4)} / IPv6 {len(v6)})"), - ("error requests", sum(1 for r in self.requests if r.is_error)), - (f"unique {self.paths_label}", len(self.error_paths())), - ] - rows += self.extra_summary_rows() - rows += [ - (self.flag_label.lower(), len(self.flagged)), - ("whitelisted IPs seen", - sum(1 for ip in self.stats if is_whitelisted(ip))), - ("block candidates", len(self.block_candidates())), - ] - return rows - - # --- reporting (one text body, used for console AND file) -------------- - - def text_report(self) -> str: - out = [self.title, - f"generated: {datetime.now().isoformat(timespec='seconds')}", - "", - render_kv(self.summary_rows()), ""] - - out.append(f"== top {self.top} IPs ==") - rows = [columns_row(self.ip_columns, s) - for s in by_total(self.stats)[:self.top]] - out.append(render_table(columns_header(self.ip_columns), rows)) - if len(self.stats) > self.top: - out.append(f"... {len(self.stats) - self.top} more IPs omitted") - out.append("") - - out.append(f"== top 20 {self.paths_label} ==") - for path, n in self.error_paths().most_common(20): - out.append(f" {n:>5} {path}") - out.append("") - - v4, v6 = self.flagged_split() - out.append(f"== {self.flag_label} ({len(self.flagged)}) ==") - if self.flag_note: - out.append(self.flag_note) - for label, group in (("IPv4", v4), ("IPv6", v6)): - if not group: - continue - out.append(f"-- {label} ({len(group)}) --") - rows = [columns_row(self.ip_columns, s) for s in group] - out.append(render_table(columns_header(self.ip_columns), rows)) - out.append("") - return "\n".join(out) - - # --- writing ----------------------------------------------------------- - - def write_data(self) -> list[str]: - ensure_dirs(self.report_dir, self.data_dir) - written: list[str] = [] - d = self.data_dir - header = columns_header(self.ip_columns) - ordered = by_total(self.stats) - - for version, fname in ((4, "ips_ipv4.csv"), (6, "ips_ipv6.csv")): - rows = [columns_row(self.ip_columns, s) - for s in ordered if s.version == version] - written.append(write_csv(os.path.join(d, fname), header, rows)) - - written.append(write_json( - os.path.join(d, "ips.json"), - {s.ip: s.to_dict() for s in ordered})) - - written.append(write_csv( - os.path.join(d, self.paths_file), ["path", "count"], - self.error_paths().most_common())) - - v4, v6 = self.flagged_split() - for group, fname in zip((v4, v6), self.flag_files): - written.append(write_lines( - os.path.join(d, fname), - [f"{s.ip} # {self.flag_comment(s)}" for s in group])) - - blocked_v4, blocked_v6 = split_versions( - ip for ip in self.block_candidates() if not is_whitelisted(ip)) - written.append(write_lines( - os.path.join(d, "blocked_ipv4.txt"), blocked_v4)) - written.append(write_lines( - os.path.join(d, "blocked_ipv6.txt"), blocked_v6)) - - written.append(write_csv( - os.path.join(d, "requests.csv"), - columns_header(self.request_columns), - (columns_row(self.request_columns, r) for r in self.requests))) - - written += self.write_extra(d) - return written - - def write_report(self) -> str: - ensure_dirs(self.report_dir) - path = os.path.join(self.report_dir, f"{timestamp_name()}.txt") - with open(path, "w", encoding="utf-8") as f: - f.write(self.text_report()) - return path - - def write_rsc_file(self) -> tuple[str, int, int]: - ensure_dirs(self.data_dir) - return write_mikrotik( - os.path.join(self.data_dir, "mikrotik-block.rsc"), - f"{self.name}-ids", self.block_candidates()) - - # --- entrypoint -------------------------------------------------------- - - def run(self) -> dict[str, Any]: - validate_dir(self.log_path) - ensure_dirs(self.report_dir, self.data_dir) - refresh_whitelists() - self.discover() - self.load() - self.analyze() - - if not self.quiet: - print(self.text_report()) - - self.write_data() - report_path = self.write_report() - print(f"data files: {self.data_dir}") - print(f"summary report: {report_path}") - - result: dict[str, Any] = {"report": report_path, - "data_dir": self.data_dir} - if self.write_rsc: - rsc, n_blocked, n_whitelisted = self.write_rsc_file() - print(f"mikrotik script: {rsc}") - print(f" {n_blocked} IPs blocked, " - f"{n_whitelisted} CIDR ranges whitelisted") - result.update(rsc=rsc, blocked=n_blocked, - whitelisted=n_whitelisted) - return result