256 lines
9.5 KiB
Python
256 lines
9.5 KiB
Python
#!/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
|