#!/usr/bin/env python3 """Analysis pipeline for the intrusion-detection system. `Analyzer` is the log-source-agnostic base class. Subclasses implement `parse_line()` and `flag()`. """ from __future__ import annotations import os import re from collections import Counter from datetime import datetime from typing import Any, Sequence from .files import ensure_dirs, find_logs, read_lines, validate_dir, write_csv, write_json, write_lines from .flags import by_total, flag_malicious_patterns from .ip_utils import is_whitelisted, split_versions from .mikrotik import write_mikrotik from .models import IPStats, Request, aggregate from .reporting import Column, columns_header, columns_row, render_kv, render_table, timestamp_name from .constants import DEFAULT_TOP, DEFAULT_TOP_PATHS from .whitelists import refresh_whitelists 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 = DEFAULT_TOP, 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 --------------------------------------------------------- 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 {DEFAULT_TOP_PATHS} {self.paths_label} ==") for path, n in self.error_paths().most_common(DEFAULT_TOP_PATHS): 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