From bbf07dec5a143d75919cd0b93b18c0584b52b257 Mon Sep 17 00:00:00 2001 From: zwitschi Date: Mon, 3 Aug 2026 17:01:36 +0200 Subject: [PATCH] Refactor monolith utils into domain modules and update docs --- README.md | 127 +++++++++++-------- analyzer.py | 254 +++++++++++++++++++++++++++++++++++++ data/whois_cache.json | 3 +- files.py | 84 ++++++++++++ flags.py | 46 +++++++ ip_utils.py | 63 +++++++++ mikrotik.py | 110 ++++++++++++++++ models.py | 150 ++++++++++++++++++++++ reporting.py | 55 ++++++++ reports/mikrotik-block.rsc | 5 +- whois.py | 129 +++++++++++++++++++ 11 files changed, 972 insertions(+), 54 deletions(-) create mode 100644 analyzer.py create mode 100644 files.py create mode 100644 flags.py create mode 100644 ip_utils.py create mode 100644 mikrotik.py create mode 100644 models.py create mode 100644 reporting.py create mode 100644 whois.py diff --git a/README.md b/README.md index 4ece0da..7a81b3a 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,23 @@ -# allucanget-ids +# All You Can GET Intrusion Detection System Local-development intrusion detection / log analysis for two log sources: -| Script | Source | Detection rule | -| --- | --- | --- | -| `copyparty.py` | [copyparty](https://github.com/9001/copyparty) server logs | denied-only IPs + copyparty auto-bans | -| `npm.py` | [Nginx Proxy Manager](https://nginxproxymanager.com) proxy logs | high error-rate IPs (scanners) | -| `scan.py` | both of the above | consolidated scan + unified MikroTik export | -| `whitelists.py` | whitelist module | provider CIDR loading, refresh, validation | -| `bad_patterns.py` | pattern detection | malicious URL/file regex classification | -| `utils.py` | shared core (no CLI) | aggregation, reports, rsc, pipeline | +| Script | Source | Detection rule | +| ----------------- | --------------------------------------------------------------- | -------------------------------------------------------------------------- | +| `scan.py` | both sources | consolidated scan + unified MikroTik export | +| `copyparty.py` | [copyparty](https://github.com/9001/copyparty) server logs | denied-only IPs + copyparty auto-bans | +| `npm.py` | [Nginx Proxy Manager](https://nginxproxymanager.com) proxy logs | high error-rate IPs (scanners) | +| `utils.py` | re-export layer | thin barrel — imports from sub-modules below | +| `models.py` | data model | `Request`, `IPStats`, `aggregate()` | +| `ip_utils.py` | IP helpers | `parse_ip`, `whitelist_of`, `is_whitelisted`, `sort_ips`, `split_versions` | +| `flags.py` | detection rules | `flag_scanners`, `flag_denied_only`, `flag_malicious_patterns`, `by_total` | +| `analyzer.py` | pipeline | `Analyzer` — discover → load → analyze → report → write | +| `whitelists.py` | whitelist module | provider CIDR constants, local/remote load, CIDR extraction | +| `bad_patterns.py` | pattern detection | malicious URL/file regex classification | +| `whois.py` | WHOIS query | `query_whois_subnet`, `consolidate_by_subnet` | +| `mikrotik.py` | RouterOS export | `mikrotik_script`, `write_mikrotik`, `split_cidr_versions` | +| `files.py` | I/O helpers | `read_lines`, `find_logs`, `write_csv`, `write_json`, etc. | +| `reporting.py` | text rendering | `Column`, `render_table`, `render_kv`, `timestamp_name` | Both per-source scripts parse logs, aggregate per-IP behaviour, split IPv4/IPv6, skip whitelisted provider ranges, detect malicious path patterns, @@ -25,14 +33,22 @@ consolidated `.rsc`. ## Architecture -``` -allucanget-ids/ +```text +intrusion-detection/ ├── scan.py consolidated scanner entry point ├── copyparty.py copyparty log adapter (~150 lines) ├── npm.py NPM proxy log adapter (~150 lines) -├── whitelists.py whitelist constants, download, CIDR extraction +├── utils.py thin re-export barrel (~90 lines) +├── models.py Request, IPStats, aggregate() +├── ip_utils.py IP parsing, whitelist lookups, sorting +├── flags.py flag_scanners, flag_denied_only, flag_malicious_patterns +├── analyzer.py Analyzer pipeline base class +├── whitelists.py provider CIDR constants, local/remote load ├── bad_patterns.py malicious URL/file regex patterns -├── utils.py shared core: aggregation, reporting, MikroTik export +├── whois.py query_whois_subnet, consolidate_by_subnet +├── mikrotik.py RouterOS .rsc script generation +├── files.py read_lines, find_logs, write_csv/json/lines +├── reporting.py Column, render_table, render_kv, timestamp_name ├── data/ whitelist source files (cached locally) │ ├── openfactory/ │ │ └── geofeed.csv @@ -49,18 +65,24 @@ allucanget-ids/ ``` `copyparty.py` and `npm.py` are thin adapters. Each supplies a log matcher, -a line parser, a flagging rule, and column definitions. Everything else lives in -`utils.py`: +a line parser, a flagging rule, and column definitions. `utils.py` is a barrel +module that re-exports everything from the sub-modules so callers only need +`import utils`. -- `Request` — normalised request record -- `IPStats` + `aggregate()` — per-IP counters, malicious-hit tracking -- `flag_scanners()` / `flag_denied_only()` / `flag_malicious_patterns()` — detection rules -- `whitelist_of()` / `is_whitelisted()` — whitelist lookups (backed by `whitelists.py`) -- `is_malicious()` / `classify_path()` — pattern classification (backed by `bad_patterns.py`) -- `query_whois_subnet()` / `consolidate_by_subnet()` — WHOIS subnet grouping -- IP version helpers, log discovery, transparent `.gz`, ANSI stripping -- CSV/JSON/table writers, `mikrotik_script()` — RouterOS export -- `Analyzer` — pipeline: discover → load → analyze → report → write +Core responsibilities by module: + +| Module | Owns | +| ----------------- | -------------------------------------------------------------------------------- | +| `models.py` | `Request`, `IPStats`, `aggregate()` | +| `ip_utils.py` | `parse_ip`, `whitelist_of`, `is_whitelisted`, `sort_ips`, `split_versions` | +| `flags.py` | `flag_scanners`, `flag_denied_only`, `flag_malicious_patterns`, `by_total` | +| `analyzer.py` | `Analyzer` — discover → load → analyze → report → write | +| `whitelists.py` | provider CIDR constants, local-file / remote / built-in loading | +| `bad_patterns.py` | malicious URL/file regex classification | +| `whois.py` | `query_whois_subnet`, `consolidate_by_subnet` | +| `mikrotik.py` | `mikrotik_script`, `write_mikrotik`, `split_cidr_versions` | +| `files.py` | `read_lines`, `find_logs`, `validate_dir`, `ensure_dirs`, `write_csv/json/lines` | +| `reporting.py` | `Column`, `render_table`, `render_kv`, `timestamp_name` | Console output and saved reports are the **same string** (`Analyzer.text_report()`). @@ -82,16 +104,16 @@ python scan.py All scripts accept CLI overrides: -| Flag | Default | Meaning | -| --- | --- | --- | -| `--log-path` | see below | log directory | -| `--report-dir` | `reports/` | output directory | -| `--top N` | `30` | rows in the top-IP table | -| `--quiet` | off | write files, print only paths | -| `--no-rsc` | off | skip MikroTik export | -| `--min-errors N` | `5` (`npm.py`) | error count threshold | -| `--error-ratio F` | `0.5` (`npm.py`) | error ratio threshold | -| `--min-denied N` | `1` (`copyparty.py`) | denied-hit threshold | +| Flag | Default | Meaning | +| ----------------- | -------------------- | ----------------------------- | +| `--log-path` | see below | log directory | +| `--report-dir` | `reports/` | output directory | +| `--top N` | `30` | rows in the top-IP table | +| `--quiet` | off | write files, print only paths | +| `--no-rsc` | off | skip MikroTik export | +| `--min-errors N` | `5` (`npm.py`) | error count threshold | +| `--error-ratio F` | `0.5` (`npm.py`) | error ratio threshold | +| `--min-denied N` | `1` (`copyparty.py`) | denied-hit threshold | `scan.py` accepts all flags from both scanners plus `--copyparty-log-path`, `--copyparty-report-dir`, `--npm-log-path`, `--npm-report-dir`, and @@ -99,10 +121,10 @@ All scripts accept CLI overrides: Default log paths (overridable by `.env`): -| Script | Env var | Default | -| --- | --- | --- | +| Script | Env var | Default | +| -------------- | -------------------- | -------------- | | `copyparty.py` | `COPYPARTY_LOG_PATH` | `P:\copyparty` | -| `npm.py` | `NPM_LOG_PATH` | `Y:\npm\logs` | +| `npm.py` | `NPM_LOG_PATH` | `Y:\npm\logs` | Examples: @@ -175,15 +197,15 @@ Blocklist candidates come from three independent sources, merged at scan time: ### Pattern categories (`bad_patterns.py`) -| Pattern | Severity | Matches | -| --- | --- | --- | -| `env_file` | 3 (critical) | `.env`, `.env.production`, `.aws/credentials`, `config.json` | -| `vcs_probe` | 3 (critical) | `.git/config`, `.svn/` | -| `sensitive_file` | 3 (critical) | `id_rsa`, `id_dsa`, `.pem`, `.key`, `.ppk` | -| `webshell_probe` | 3 (critical) | Known webshell filenames (xstelth, adminfuns, wp_filemanager, etc.) | -| `wp_probe` | 2 (high) | `wp-admin`, `wp-login`, `wp-content`, `wp-includes` | -| `generic_php_probe` | 1 (moderate) | Single/double-char/numeric `.php` files | -| `well_known_probe` | 1 (moderate) | `.well-known/` paths | +| Pattern | Severity | Matches | +| ------------------- | ------------ | ------------------------------------------------------------------- | +| `env_file` | 3 (critical) | `.env`, `.env.production`, `.aws/credentials`, `config.json` | +| `vcs_probe` | 3 (critical) | `.git/config`, `.svn/` | +| `sensitive_file` | 3 (critical) | `id_rsa`, `id_dsa`, `.pem`, `.key`, `.ppk` | +| `webshell_probe` | 3 (critical) | Known webshell filenames (xstelth, adminfuns, wp_filemanager, etc.) | +| `wp_probe` | 2 (high) | `wp-admin`, `wp-login`, `wp-content`, `wp-includes` | +| `generic_php_probe` | 1 (moderate) | Single/double-char/numeric `.php` files | +| `well_known_probe` | 1 (moderate) | `.well-known/` paths | Flagging metadata fields: @@ -236,11 +258,14 @@ Fields: `[datetime]`, upstream status, proxy status, method, scheme, host, ## Extending -- **new whitelist provider** → add URLs, files, and fallback CIDRs to - `whitelists.py` -- **new detection rule** → add a `BadPattern` to `bad_patterns.py` or a - `flag_*()` function to `utils.py`, return it from the analyzer's `flag()` -- **new log source** → subclass `utils.Analyzer`, implement `parse_line()` + +- **new whitelist provider** → add URLs, local file paths, and fallback CIDRs + to `whitelists.py` +- **new detection pattern** → add a `BadPattern` to `bad_patterns.py` +- **new detection rule** → add a `flag_*()` function to `flags.py`, return it + from the analyzer's `flag()` method +- **new log source** → subclass `analyzer.Analyzer`, implement `parse_line()` + `flag()`, declare `ip_columns` / `request_columns` - **new output file** → override `write_extra(data_dir)` and return the paths written +- **new I/O format** → add writer to `files.py` +- **new report section** → add rendering logic to `reporting.py` diff --git a/analyzer.py b/analyzer.py new file mode 100644 index 0000000..04791ec --- /dev/null +++ b/analyzer.py @@ -0,0 +1,254 @@ +#!/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 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 = 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 --------------------------------------------------------- + + 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 \ No newline at end of file diff --git a/data/whois_cache.json b/data/whois_cache.json index d0d1d2b..7b0186c 100644 --- a/data/whois_cache.json +++ b/data/whois_cache.json @@ -863,5 +863,6 @@ "145.132.103.19": "145.132.0.0/15", "209.99.190.99": "209.99.184.0/21", "20.104.18.253": "20.64.0.0/10", - "178.128.176.132": "178.128.176.0/20" + "178.128.176.132": "178.128.176.0/20", + "81.171.74.60": "81.171.74.0/24" } \ No newline at end of file diff --git a/files.py b/files.py new file mode 100644 index 0000000..5eb01dd --- /dev/null +++ b/files.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Filesystem and I/O helpers for the intrusion-detection system.""" + +from __future__ import annotations + +import csv +import gzip +import json +import os +import re +from typing import Any, Iterable, Iterator, Sequence + +ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") + + +# -- directory -------------------------------------------------------------- + +def validate_dir(path: str, need_write: bool = False) -> str: + if not os.path.exists(path): + raise FileNotFoundError(f"path does not exist: {path}") + if not os.path.isdir(path): + raise NotADirectoryError(f"path is not a directory: {path}") + if not os.access(path, os.R_OK): + raise PermissionError(f"path not readable: {path}") + if need_write and not os.access(path, os.W_OK): + raise PermissionError(f"path not writable: {path}") + return path + + +def ensure_dirs(*paths: str) -> None: + for p in paths: + os.makedirs(p, exist_ok=True) + + +# -- log discovery ---------------------------------------------------------- + +def find_logs(path: str, pattern: re.Pattern | None = None, + suffixes: Sequence[str] = (), + exclude_prefixes: Sequence[str] = ()) -> list[str]: + """List matching log file names (not full paths), sorted.""" + out = [] + for name in os.listdir(path): + if not os.path.isfile(os.path.join(path, name)): + continue + if any(name.startswith(p) for p in exclude_prefixes): + continue + if pattern is not None and not pattern.match(name): + continue + if suffixes and not name.endswith(tuple(suffixes)): + continue + out.append(name) + return sorted(out) + + +def read_lines(path: str) -> Iterator[str]: + """Yield ANSI-stripped, newline-stripped lines. Transparently gunzips.""" + opener = gzip.open if path.endswith(".gz") else open + with opener(path, "rt", encoding="utf-8", errors="replace") as f: + for line in f: + yield ANSI_RE.sub("", line.rstrip("\r\n")) + + +# -- writers ---------------------------------------------------------------- + +def write_csv(path: str, header: Sequence[str], + rows: Iterable[Sequence[Any]]) -> str: + with open(path, "w", newline="", encoding="utf-8") as f: + w = csv.writer(f) + w.writerow(header) + w.writerows(rows) + return path + + +def write_json(path: str, obj: Any) -> str: + with open(path, "w", encoding="utf-8") as f: + json.dump(obj, f, indent=2, default=str) + return path + + +def write_lines(path: str, lines: Iterable[str]) -> str: + with open(path, "w", encoding="utf-8") as f: + for line in lines: + f.write(line + "\n") + return path diff --git a/flags.py b/flags.py new file mode 100644 index 0000000..321b629 --- /dev/null +++ b/flags.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Flagging rules for the intrusion-detection system. + +Detection functions that identify hostile IPs based on aggregated stats. +""" + +from __future__ import annotations + +import bad_patterns +from models import IPStats + + +def flag_scanners(stats: dict[str, IPStats], min_errors: int = 5, + min_ratio: float = 0.5) -> dict[str, IPStats]: + """IPs with many errors AND a high error ratio.""" + return { + ip: s for ip, s in stats.items() + if s.errors >= min_errors and s.error_ratio >= min_ratio + } + + +def flag_denied_only(stats: dict[str, IPStats], + min_denied: int = 1) -> dict[str, IPStats]: + """IPs that only ever hit denied paths and never authenticated.""" + return { + ip: s for ip, s in stats.items() + if s.denied >= min_denied and not s.authenticated + } + + +def flag_malicious_patterns(stats: dict[str, IPStats], + min_severity: int = 2, + min_hits: int = 1) -> dict[str, IPStats]: + """IPs with hits against malicious path patterns at or above severity threshold.""" + return { + ip: s for ip, s in stats.items() + if s.malicious_hits >= min_hits and any( + bad_patterns.PATTERNS_BY_NAME[name].severity >= min_severity + for name in s.malicious_patterns + ) + } + + +def by_total(stats: dict[str, IPStats]) -> list[IPStats]: + """Stats sorted by request count, descending.""" + return sorted(stats.values(), key=lambda s: (-s.total, s.ip)) diff --git a/ip_utils.py b/ip_utils.py new file mode 100644 index 0000000..9698a95 --- /dev/null +++ b/ip_utils.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""IP helper functions for the intrusion-detection system.""" + +from __future__ import annotations + +import ipaddress +from typing import Iterable + +from whitelists import WHITELIST_NETS as _WHITELIST_NETS + + +def parse_ip(value: str): + """Return an ip_address object, or None if `value` is not an IP.""" + try: + return ipaddress.ip_address(value) + except (ValueError, TypeError): + return None + + +def ip_version(value: str) -> int | None: + addr = parse_ip(value) + return addr.version if addr else None + + +def is_ipv6(value: str) -> bool: + return ip_version(value) == 6 + + +def is_ipv4(value: str) -> bool: + return ip_version(value) == 4 + + +def whitelist_of(value: str) -> str | None: + """Return the whitelist name an IP belongs to, else None.""" + addr = parse_ip(value) + if addr is None: + return None + for name, nets in _WHITELIST_NETS.items(): + for net in nets: + if addr.version == net.version and addr in net: + return name + return None + + +def is_whitelisted(value: str) -> bool: + return whitelist_of(value) is not None + + +def sort_ips(ips: Iterable[str]) -> list[str]: + """Numeric sort, IPv4 before IPv6, unparseable last.""" + def key(ip: str): + addr = parse_ip(ip) + if addr is None: + return (9, b"", ip) + return (addr.version, addr.packed, "") + return sorted(ips, key=key) + + +def split_versions(ips: Iterable[str]) -> tuple[list[str], list[str]]: + """Split an IP iterable into (sorted IPv4, sorted IPv6).""" + ips = list(ips) + return (sort_ips(ip for ip in ips if is_ipv4(ip)), + sort_ips(ip for ip in ips if is_ipv6(ip))) diff --git a/mikrotik.py b/mikrotik.py new file mode 100644 index 0000000..5721411 --- /dev/null +++ b/mikrotik.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +"""MikroTik RouterOS script generation.""" + +from __future__ import annotations + +import ipaddress +from datetime import datetime +from typing import Iterable + +from ip_utils import is_ipv4, is_ipv6, is_whitelisted, parse_ip +from whitelists import WHITELISTS + + +def split_cidr_versions(targets: Iterable[str]) -> tuple[list[str], list[str]]: + """Split target CIDRs/IPs into (sorted IPv4 list, sorted IPv6 list).""" + v4 = [] + v6 = [] + for t in targets: + ip = t.split("/")[0] + if is_ipv4(ip): + v4.append(t) + elif is_ipv6(ip): + v6.append(t) + + def key_cidr(cidr: str): + ip = cidr.split("/")[0] + addr = parse_ip(ip) + if addr is None: + return (9, b"", cidr) + return (addr.version, addr.packed, cidr) + + return sorted(v4, key=key_cidr), sorted(v6, key=key_cidr) + + +def mikrotik_script(tag: str, blocked_ips_or_subnets: Iterable[str], + whitelists: dict[str, list[str]] | None = None, + generated: str | None = None) -> tuple[str, int, int]: + """Build a RouterOS .rsc script. + + Whitelisted ranges become `accept` rules placed above the `drop` rules. + Blocked targets can be individual IPs (e.g. 1.2.3.4/32) or CIDR blocks + (e.g. 1.2.3.0/24). + + Returns (script_text, blocked_count, whitelisted_cidr_count). + """ + whitelists = WHITELISTS if whitelists is None else whitelists + generated = generated or datetime.now().isoformat(timespec="seconds") + + def clean_blocked(targets: Iterable[str]) -> list[str]: + cleaned = [] + for t in targets: + ip = t.split("/")[0] + if not is_whitelisted(ip): + cleaned.append(t) + return cleaned + + blocked = clean_blocked(set(blocked_ips_or_subnets)) + block_v4, block_v6 = split_cidr_versions(blocked) + + lines: list[str] = [f"# {tag} firewall export - generated {generated}", ""] + wl_count = 0 + + for name, cidrs in whitelists.items(): + nets = [ipaddress.ip_network(c) for c in cidrs] + wl_count += len(nets) + for version, cmd in ((4, "/ip firewall address-list"), + (6, "/ipv6 firewall address-list")): + chunk = [n for n in nets if n.version == version] + if not chunk: + continue + lines.append(f"# {name} IPv{version} ranges (whitelisted)") + lines.append(cmd) + lines += [f"add list={name} address={n}" for n in chunk] + lines.append("") + + if block_v4: + lines.append(f"# Blocked IPv4 ({len(block_v4)}, non-whitelisted)") + lines.append("/ip firewall address-list") + lines += [f"add list=Blocked address={t}" for t in block_v4] + lines.append("") + if block_v6: + lines.append(f"# Blocked IPv6 ({len(block_v6)}, non-whitelisted)") + lines.append("/ipv6 firewall address-list") + lines += [f"add list=Blocked address={t}" for t in block_v6] + lines.append("") + + for version, cmd, blocks in ((4, "/ip firewall filter", block_v4), + (6, "/ipv6 firewall filter", block_v6)): + lines.append(cmd) + for name in whitelists: + lines.append( + f"add chain=input action=accept src-address-list={name} " + f'comment="{tag} {name.lower()} whitelist (v{version})"') + if blocks: + for chain in ("input", "forward"): + lines.append( + f"add chain={chain} action=drop src-address-list=Blocked " + f'comment="{tag} blocked (v{version})"') + lines.append("") + + return "\n".join(lines), len(blocked), wl_count + + +def write_mikrotik(path: str, tag: str, + blocked_ips_or_subnets: Iterable[str]) -> tuple[str, int, int]: + text, n_blocked, n_whitelisted = mikrotik_script( + tag, blocked_ips_or_subnets) + with open(path, "w", encoding="utf-8") as f: + f.write(text) + return path, n_blocked, n_whitelisted diff --git a/models.py b/models.py new file mode 100644 index 0000000..76f109a --- /dev/null +++ b/models.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +"""Data models for the intrusion-detection system. + +`Request` and `IPStats` plus the `aggregate()` fold function. +""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass, field +from typing import Iterable + +import bad_patterns +from ip_utils import ip_version, whitelist_of + + +@dataclass(slots=True) +class Request: + """One normalised request line, produced by any log parser.""" + + ts: str = "" + ip: str = "" + method: str | None = None + path: str | None = None + status: int | None = None + upstream_status: int | None = None + host: str | None = None + user: str | None = None + ua: str | None = None + referer: str | None = None + denied: bool = False + banned: bool = False + extra: dict = field(default_factory=dict) + + @property + def is_client_error(self) -> bool: + return self.status is not None and 400 <= self.status < 500 + + @property + def is_server_error(self) -> bool: + return self.status is not None and 500 <= self.status < 600 + + @property + def is_error(self) -> bool: + return self.denied or self.is_client_error or self.is_server_error + + +@dataclass +class IPStats: + """Aggregated per-IP counters.""" + + ip: str + version: int = 0 + total: int = 0 + errors: int = 0 + c4xx: int = 0 + c5xx: int = 0 + denied: int = 0 + banned: int = 0 + first: str = "" + last: str = "" + paths: Counter = field(default_factory=Counter) + methods: Counter = field(default_factory=Counter) + hosts: Counter = field(default_factory=Counter) + uas: Counter = field(default_factory=Counter) + users: set[str] = field(default_factory=set) + malicious_hits: int = 0 + malicious_patterns: dict[str, int] = field(default_factory=dict) + + @property + def error_ratio(self) -> float: + return round(self.errors / self.total, 3) if self.total else 0.0 + + @property + def authenticated(self) -> bool: + """True if the IP ever presented a real (non-anonymous) user.""" + return any(u and u != "*" for u in self.users) + + @property + def whitelist(self) -> str: + return whitelist_of(self.ip) or "" + + @property + def users_str(self) -> str: + return ",".join(sorted(self.users)) or "-" + + @property + def methods_str(self) -> str: + return ",".join(sorted(self.methods)) or "-" + + def to_dict(self) -> dict: + return { + "ip": self.ip, + "version": self.version, + "total": self.total, + "errors": self.errors, + "error_ratio": self.error_ratio, + "status_4xx": self.c4xx, + "status_5xx": self.c5xx, + "denied": self.denied, + "banned": self.banned, + "first": self.first, + "last": self.last, + "whitelist": self.whitelist or None, + "users": sorted(self.users), + "methods": dict(self.methods), + "hosts": dict(self.hosts), + "paths": dict(self.paths.most_common()), + "uas": dict(self.uas.most_common(10)), + "malicious_hits": self.malicious_hits, + "malicious_patterns": dict(self.malicious_patterns), + } + + +def aggregate(requests: Iterable[Request]) -> dict[str, IPStats]: + """Fold request records into per-IP statistics.""" + stats: dict[str, IPStats] = {} + for r in requests: + s = stats.get(r.ip) + if s is None: + s = stats[r.ip] = IPStats( + ip=r.ip, version=ip_version(r.ip) or 0, first=r.ts) + s.total += 1 + s.last = r.ts + if r.denied: + s.denied += 1 + if r.banned: + s.banned += 1 + if r.is_client_error: + s.c4xx += 1 + if r.is_server_error: + s.c5xx += 1 + if r.is_error: + s.errors += 1 + if r.path: + s.paths[r.path] += 1 + if r.method: + s.methods[r.method] += 1 + if r.host: + s.hosts[r.host] += 1 + if r.ua and r.ua != "-": + s.uas[r.ua] += 1 + if r.user: + s.users.add(r.user) + if r.path and bad_patterns.is_malicious(r.path): + s.malicious_hits += 1 + for p in bad_patterns.classify_path(r.path): + s.malicious_patterns[p.name] = s.malicious_patterns.get( + p.name, 0) + 1 + return stats diff --git a/reporting.py b/reporting.py new file mode 100644 index 0000000..6b24184 --- /dev/null +++ b/reporting.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +"""Text rendering helpers for the intrusion-detection system.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Callable, Sequence + +#: (header, getter) pair used for CSV + console tables +Column = tuple[str, Callable[[Any], Any]] + + +def columns_header(columns: Sequence[Column]) -> list[str]: + return [c[0] for c in columns] + + +def columns_row(columns: Sequence[Column], obj: Any) -> list[Any]: + return [fn(obj) for _, fn in columns] + + +def render_table(headers: Sequence[str], rows: Sequence[Sequence[Any]], + max_width: int = 48) -> str: + """Aligned fixed-width text table. Numeric columns right-aligned.""" + def cell(v: Any) -> str: + s = "" if v is None else str(v) + return s if len(s) <= max_width else s[:max_width - 1] + "~" + + body = [[cell(v) for v in row] for row in rows] + aligns = [] + for i in range(len(headers)): + numeric = bool(rows) and all( + isinstance(row[i], (int, float)) and not isinstance(row[i], bool) + for row in rows) + aligns.append(">" if numeric else "<") + widths = [len(h) for h in headers] + for row in body: + for i, v in enumerate(row): + widths[i] = max(widths[i], len(v)) + + def line(values: Sequence[str]) -> str: + return " ".join( + f"{v:{a}{w}}" for v, a, w in zip(values, aligns, widths)).rstrip() + + out = [line(list(headers)), "-" * (sum(widths) + 2 * (len(widths) - 1))] + out += [line(row) for row in body] + return "\n".join(out) + + +def render_kv(rows: Sequence[tuple[str, Any]]) -> str: + width = max((len(k) for k, _ in rows), default=0) + 1 + return "\n".join(f"{k + ':':<{width + 1}} {v}" for k, v in rows) + + +def timestamp_name() -> str: + return datetime.now().strftime("%Y-%m-%d-%H-%M-%S") diff --git a/reports/mikrotik-block.rsc b/reports/mikrotik-block.rsc index e08c2c1..e2c346f 100644 --- a/reports/mikrotik-block.rsc +++ b/reports/mikrotik-block.rsc @@ -1,4 +1,4 @@ -# ids firewall export - generated 2026-08-03T16:15:01 +# ids firewall export - generated 2026-08-03T16:51:42 # Openfactory IPv4 ranges (whitelisted) /ip firewall address-list @@ -50,7 +50,7 @@ add list=PaloAlto address=2604:a940:300:5b6::/64 add list=PaloAlto address=2604:a940:301:225::/64 add list=PaloAlto address=2604:a940:302:118::/64 -# Blocked IPv4 (508, non-whitelisted) +# Blocked IPv4 (509, non-whitelisted) /ip firewall address-list add list=Blocked address=2.57.122.173/32 add list=Blocked address=2.58.56.101/32 @@ -271,6 +271,7 @@ add list=Blocked address=77.90.185.230/32 add list=Blocked address=80.91.86.169/32 add list=Blocked address=81.171.72.93/32 add list=Blocked address=81.171.72.135/32 +add list=Blocked address=81.171.74.60/32 add list=Blocked address=82.39.206.154/32 add list=Blocked address=82.102.18.0/24 add list=Blocked address=82.102.18.190/32 diff --git a/whois.py b/whois.py new file mode 100644 index 0000000..9805bb6 --- /dev/null +++ b/whois.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""WHOIS subnet querying and IP consolidation.""" + +from __future__ import annotations + +import ipaddress +import json +import os +import re +from typing import Iterable + +from ip_utils import ip_version, parse_ip + +# WHOIS Cache file path +_WHOIS_CACHE_PATH = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "data", "whois_cache.json" +) + +_whois_cache: dict[str, str] = {} + + +def load_whois_cache() -> None: + global _whois_cache + if not _whois_cache and os.path.exists(_WHOIS_CACHE_PATH): + try: + with open(_WHOIS_CACHE_PATH, "r", encoding="utf-8") as f: + _whois_cache = json.load(f) + except Exception: + _whois_cache = {} + + +def save_whois_cache() -> None: + if _whois_cache: + os.makedirs(os.path.dirname(_WHOIS_CACHE_PATH), exist_ok=True) + try: + with open(_WHOIS_CACHE_PATH, "w", encoding="utf-8") as f: + json.dump(_whois_cache, f, indent=2) + except Exception: + pass + + +def query_whois_subnet(ip: str) -> str | None: + """Query whois.radb.net for the route/subnet of an IP address.""" + addr = parse_ip(ip) + if addr is None: + return None + + load_whois_cache() + if ip in _whois_cache: + return _whois_cache[ip] or None + + import socket + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(3.0) + s.connect(("whois.radb.net", 43)) + s.sendall(f"{ip}\r\n".encode()) + res = "" + while True: + chunk = s.recv(4096).decode(errors="ignore") + if not chunk: + break + res += chunk + s.close() + + routes = re.findall(r"^(?:route|route6):\s*(\S+)", + res, re.IGNORECASE | re.MULTILINE) + if routes: + routes_parsed = [] + for r in routes: + try: + routes_parsed.append(ipaddress.ip_network(r)) + except ValueError: + continue + if routes_parsed: + best_route = max(routes_parsed, key=lambda n: n.prefixlen) + subnet_str = str(best_route) + _whois_cache[ip] = subnet_str + save_whois_cache() + return subnet_str + + except Exception: + pass + + subnet_str = f"{ip}/32" if addr.version == 4 else f"{ip}/128" + _whois_cache[ip] = subnet_str + save_whois_cache() + return subnet_str + + +def consolidate_by_subnet(ips: Iterable[str], threshold: int = 3) -> set[str]: + """Group IPs by WHOIS subnet; if >= threshold IPs in a subnet, block the subnet. + + Otherwise, block individual IPs. + """ + ip_list = list(ips) + subnet_to_ips: dict[str, set[str]] = {} + ip_to_subnet: dict[str, str] = {} + + for ip in ip_list: + subnet = query_whois_subnet(ip) + if subnet: + try: + net = ipaddress.ip_network(subnet, strict=False) + addr = ipaddress.ip_address(ip) + if addr in net: + subnet_to_ips.setdefault(subnet, set()).add(ip) + ip_to_subnet[ip] = subnet + continue + except ValueError: + pass + fallback_net = f"{ip}/32" if ip_version(ip) == 4 else f"{ip}/128" + subnet_to_ips.setdefault(fallback_net, set()).add(ip) + ip_to_subnet[ip] = fallback_net + + blocked_targets: set[str] = set() + for subnet, mapped_ips in subnet_to_ips.items(): + if len(mapped_ips) >= threshold: + blocked_targets.add(subnet) + else: + for ip in mapped_ips: + blocked_targets.add( + f"{ip}/32" if ip_version(ip) == 4 else f"{ip}/128") + + return blocked_targets + + +# Re-export path for backward compat +WHOIS_CACHE_PATH = _WHOIS_CACHE_PATH