From eea5ef68858c13526cfa2e6975d682c0a3bf6779 Mon Sep 17 00:00:00 2001 From: zwitschi Date: Wed, 5 Aug 2026 19:33:12 +0200 Subject: [PATCH] Add centralized constants module and .env configuration support --- .env.example | 68 ++++++++++++++++++++++++++ README.md | 46 +++++++++++++----- app/analyzer.py | 7 +-- app/constants.py | 120 ++++++++++++++++++++++++++++++++++++++++++++++ app/copyparty.py | 19 ++++---- app/firewall.py | 25 ++++------ app/flags.py | 7 +-- app/mikrotik.py | 7 +-- app/npm.py | 26 +++++----- app/scan.py | 64 +++++++++++++------------ app/utils.py | 6 ++- app/whitelists.py | 31 +++++------- app/whois.py | 32 ++++++------- requirements.txt | 2 + 14 files changed, 334 insertions(+), 126 deletions(-) create mode 100644 .env.example create mode 100644 app/constants.py create mode 100644 requirements.txt diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..4f98a3d --- /dev/null +++ b/.env.example @@ -0,0 +1,68 @@ +# --------------------------------------------------------------------------- +# Intrusion Detection System — environment template +# --------------------------------------------------------------------------- +# Copy this file to `.env` for local development and fill in your values: +# +# copy .env.example .env +# +# Precedence: real system environment variables win over values in `.env`, +# so production deployments can set these via the OS and skip `.env`. +# `.env` is git-ignored — never commit real secrets to it. + +# --- Log sources ----------------------------------------------------------- + +# Directory containing copyparty access logs (YYYY-MM-DD.txt files). +COPYPARTY_LOG_PATH=P:\copyparty + +# Directory containing Nginx Proxy Manager proxy logs (.log / .gz files). +NPM_LOG_PATH=Y:\npm\logs + +# --- MikroTik RouterOS connection ----------------------------------------- + +# IP/hostname of the router to sync the blocklist to. +ROUTER_IP=192.168.88.1 + +# SSH user for the router. +ROUTER_USER=admin + +# SSH password for the router (required when --update-firewall is used). +ROUTER_PASSWORD=change-me + +# Address list on the router that holds blocked IPs/subnets. +BLOCKLIST_NAME=Blocked + +# Comma-separated address lists that are treated as whitelisted (never blocked). +WHITELIST_NAMES=Cloudflare,Openfactory + +# --- WHOIS subnet consolidation -------------------------------------------- + +WHOIS_SERVER=whois.radb.net +WHOIS_PORT=43 +WHOIS_TIMEOUT=3.0 + +# Minimum number of same-subnet IPs before blocking the whole subnet. +SUBNET_THRESHOLD=3 + +# --- Scan / flagging defaults ---------------------------------------------- + +# Rows in the top-IP report tables. +DEFAULT_TOP=30 + +# copyparty: minimum denied hits before flagging an IP. +DEFAULT_MIN_DENIED=1 + +# npm: minimum error responses before flagging an IP. +DEFAULT_MIN_ERRORS=5 + +# npm: minimum error ratio before flagging an IP. +DEFAULT_ERROR_RATIO=0.5 + +# Rows in the "top error paths" report section. +DEFAULT_TOP_PATHS=20 + +# --- Whitelist sources ----------------------------------------------------- + +OPENFACTORY_URL=https://openfactory.net/geofeed.csv +CLOUDFLARE_URL_V4=https://www.cloudflare.com/ips-v4 +CLOUDFLARE_URL_V6=https://www.cloudflare.com/ips-v6 +PALOALTO_URL=https://docs-cortex.paloaltonetworks.com/r/1/Cortex-Xpanse/Scanning-activity \ No newline at end of file diff --git a/README.md b/README.md index fe8eba2..91780a9 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,7 @@ intrusion-detection/ ├── app.py unified entry point ├── app/ application package │ ├── __init__.py +│ ├── constants.py all config constants (paths, env vars, thresholds) │ ├── scan.py consolidated scanner │ ├── copyparty.py copyparty log adapter (~150 lines) │ ├── npm.py NPM proxy log adapter (~150 lines) @@ -77,6 +78,7 @@ Core responsibilities by module: | Module | Owns | | --------------------- | -------------------------------------------------------------------------------- | +| `app/constants.py` | all config constants — paths, env vars, network/scan thresholds | | `app/models.py` | `Request`, `IPStats`, `aggregate()` | | `app/ip_utils.py` | `parse_ip`, `whitelist_of`, `is_whitelisted`, `sort_ips`, `split_versions` | | `app/flags.py` | `flag_scanners`, `flag_denied_only`, `flag_malicious_patterns`, `by_total` | @@ -94,7 +96,39 @@ Console output and saved reports are the **same string** (`Analyzer.text_report( ## Requirements - Python 3.10+ -- stdlib only, no dependencies +- `paramiko` (MikroTik SSH), `python-dotenv` (`.env` loading) + +```powershell +pip install -r requirements.txt +``` + +## Configuration + +All tunable settings live in `app/constants.py` and can be overridden per +deployment via environment variables. For local development, copy +`.env.example` to `.env` and fill in your values: + +```powershell +copy .env.example .env +``` + +`app/constants.py` loads `.env` with `python-dotenv` at startup. Real system +environment variables always take precedence over `.env` values, so +production can set the same variables via the OS and skip `.env` entirely. +`.env` is git-ignored — do not commit secrets to it. + +| Env var | Default | Used by | +| -------------------- | ------------------------ | --------------------- | +| `COPYPARTY_LOG_PATH` | `P:\copyparty` | copyparty scanner | +| `NPM_LOG_PATH` | `Y:\npm\logs` | npm scanner | +| `ROUTER_IP` | `192.168.88.1` | MikroTik connection | +| `ROUTER_USER` | `admin` | MikroTik connection | +| `ROUTER_PASSWORD` | _(set in .env)_ | MikroTik connection | +| `BLOCKLIST_NAME` | `Blocked` | MikroTik address list | +| `WHITELIST_NAMES` | `Cloudflare,Openfactory` | MikroTik whitelists | + +See `.env.example` for the complete list of supported variables (WHOIS +tuning, scan thresholds, whitelist source URLs, and more). ## Usage @@ -128,16 +162,6 @@ All scripts accept CLI overrides: `--copyparty-report-dir`, `--npm-log-path`, `--npm-report-dir`, and `--report-dir` for the consolidated output. -Default log paths (overridable by `.env`): - -| Script | Env var | Default | -| ------------------ | -------------------- | --------------- | -| `app/copyparty.py` | `COPYPARTY_LOG_PATH` | `P:\copyparty` | -| `app/npm.py` | `NPM_LOG_PATH` | `Y:\npm\logs` | -| `app/firewall.py` | `ROUTER_IP` | `192.168.88.1` | -| | `ROUTER_USER` | `admin` | -| | `ROUTER_PASSWORD` | _(set in .env)_ | - Examples: ```powershell diff --git a/app/analyzer.py b/app/analyzer.py index c14e834..29ad7b4 100644 --- a/app/analyzer.py +++ b/app/analyzer.py @@ -19,6 +19,7 @@ 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 @@ -42,7 +43,7 @@ class Analyzer: flag_label = "Flagged IPs" flag_note = "" - def __init__(self, log_path: str, report_dir: str, top: int = 30, + 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 @@ -150,8 +151,8 @@ class Analyzer: 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"== 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("") diff --git a/app/constants.py b/app/constants.py new file mode 100644 index 0000000..739e033 --- /dev/null +++ b/app/constants.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +"""Central configuration constants for the intrusion-detection system. + +All tunable values (paths, router credentials, whitelist sources, scan +thresholds) live here so they can be maintained and overridden from a +single place. Values that make sense per-deployment are read from +environment variables; see ``.env.example`` for the supported keys. +""" + +from __future__ import annotations + +import os + +# --------------------------------------------------------------------------- +# Project root + .env loading +# --------------------------------------------------------------------------- + +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +# Load `.env` for local development (python-dotenv, optional dependency). +# `load_dotenv` does NOT override variables already present in the system +# environment, so production deployments that set these via the OS keep +# working without a `.env` file (or even without python-dotenv installed). +try: + from dotenv import load_dotenv + + load_dotenv(os.path.join(PROJECT_ROOT, ".env")) +except ImportError: + pass + +# --------------------------------------------------------------------------- +# Project paths +# --------------------------------------------------------------------------- + +DATA_DIR = os.path.join(PROJECT_ROOT, "data") +REPORT_DIR = os.path.join(PROJECT_ROOT, "reports") + +# Per-analyzer report directories +COPYPARTY_REPORT_DIR = os.path.join(REPORT_DIR, "copyparty") +NPM_REPORT_DIR = os.path.join(REPORT_DIR, "npm") + +# MikroTik reports + structured snapshots from each run +MIKROTIK_REPORT_DIR = os.path.join(REPORT_DIR, "mikrotik") +MIKROTIK_REPORT_DATA_DIR = os.path.join(MIKROTIK_REPORT_DIR, "data") + +# Common MikroTik state (latest snapshot; overwritten each run, used for +# git diffing / auditing changes to firewall rules and address lists). +MIKROTIK_DATA_DIR = os.path.join(DATA_DIR, "mikrotik") + +WHOIS_CACHE_PATH = os.path.join(DATA_DIR, "whois_cache.json") + +# --------------------------------------------------------------------------- +# Log sources (configurable via environment) +# --------------------------------------------------------------------------- + +COPYPARTY_LOG_PATH = os.environ.get("COPYPARTY_LOG_PATH", r"P:\copyparty") +NPM_LOG_PATH = os.environ.get("NPM_LOG_PATH", r"Y:\npm\logs") + +# --------------------------------------------------------------------------- +# MikroTik router connection (configurable via environment) +# --------------------------------------------------------------------------- + +SSH_PORT = 22 +SSH_TIMEOUT = 10 # seconds + +ROUTER_IP = os.environ.get("ROUTER_IP", "192.168.88.1") +ROUTER_USER = os.environ.get("ROUTER_USER", "admin") +ROUTER_PASSWORD = os.environ.get("ROUTER_PASSWORD", "M11ffpgm") + +BLOCKLIST_NAME = os.environ.get("BLOCKLIST_NAME", "Blocked") +WHITELIST_NAMES = [ + name.strip() + for name in os.environ.get( + "WHITELIST_NAMES", "Cloudflare,Openfactory").split(",") + if name.strip() +] + +# --------------------------------------------------------------------------- +# WHOIS subnet consolidation +# --------------------------------------------------------------------------- + +WHOIS_SERVER = os.environ.get("WHOIS_SERVER", "whois.radb.net") +WHOIS_PORT = int(os.environ.get("WHOIS_PORT", "43")) +WHOIS_TIMEOUT = float(os.environ.get("WHOIS_TIMEOUT", "3.0")) # seconds +# Minimum number of same-subnet IPs before blocking the whole subnet +SUBNET_THRESHOLD = int(os.environ.get("SUBNET_THRESHOLD", "3")) + +# --------------------------------------------------------------------------- +# Scan / flagging defaults +# --------------------------------------------------------------------------- + +DEFAULT_TOP = int(os.environ.get("DEFAULT_TOP", "30") + ) # top-IP table rows +DEFAULT_MIN_DENIED = int(os.environ.get( + "DEFAULT_MIN_DENIED", "1")) # copyparty +DEFAULT_MIN_ERRORS = int(os.environ.get("DEFAULT_MIN_ERRORS", "5")) # npm +DEFAULT_ERROR_RATIO = float(os.environ.get( + "DEFAULT_ERROR_RATIO", "0.5")) # npm +DEFAULT_TOP_PATHS = int(os.environ.get("DEFAULT_TOP_PATHS", "20")) + +# --------------------------------------------------------------------------- +# Whitelist sources (configurable via environment) +# --------------------------------------------------------------------------- + +OPENFACTORY_URL = os.environ.get( + "OPENFACTORY_URL", "https://openfactory.net/geofeed.csv") +CLOUDFLARE_URL_V4 = os.environ.get( + "CLOUDFLARE_URL_V4", "https://www.cloudflare.com/ips-v4") +CLOUDFLARE_URL_V6 = os.environ.get( + "CLOUDFLARE_URL_V6", "https://www.cloudflare.com/ips-v6") +PALOALTO_URL = os.environ.get( + "PALOALTO_URL", + "https://docs-cortex.paloaltonetworks.com/r/1/Cortex-Xpanse/Scanning-activity", +) + +# Local whitelist files (checked before remote download) +OPENFACTORY_FILE = os.path.join(DATA_DIR, "openfactory", "geofeed.csv") +CLOUDFLARE_V4_FILE = os.path.join(DATA_DIR, "cloudflare", "ips-v4.txt") +CLOUDFLARE_V6_FILE = os.path.join(DATA_DIR, "cloudflare", "ips-v6.txt") +PALOALTO_FILE = os.path.join(DATA_DIR, "paloalto", "scanner-ips.txt") diff --git a/app/copyparty.py b/app/copyparty.py index 353995a..e3b325f 100644 --- a/app/copyparty.py +++ b/app/copyparty.py @@ -13,17 +13,17 @@ import os import re from .analyzer import Analyzer +from .constants import ( + COPYPARTY_LOG_PATH, + COPYPARTY_REPORT_DIR, + DEFAULT_MIN_DENIED, +) 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( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "reports", "copyparty") - # "HH:MM:SS.mmm LEVEL: message" # System lines ("hsrv", "root", ...) have no port and are skipped. LINE_RE = re.compile( @@ -76,7 +76,7 @@ class CopypartyAnalyzer(Analyzer): ("ban_secs", lambda r: r.extra.get("ban_secs")), ) - def __init__(self, *args, min_denied: int = 1, **kwargs): + def __init__(self, *args, min_denied: int = DEFAULT_MIN_DENIED, **kwargs): super().__init__(*args, **kwargs) self.min_denied = min_denied @@ -161,9 +161,9 @@ class CopypartyAnalyzer(Analyzer): def main() -> None: 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, - help="minimum denied hits before flagging (default: 1)") + COPYPARTY_LOG_PATH, COPYPARTY_REPORT_DIR) + parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED, + help=f"minimum denied hits before flagging (default: {DEFAULT_MIN_DENIED})") args = parser.parse_args() CopypartyAnalyzer( @@ -171,7 +171,6 @@ def main() -> None: report_dir=args.report_dir, top=args.top, quiet=args.quiet, - write_rsc=not args.no_rsc, min_denied=args.min_denied, ).run() diff --git a/app/firewall.py b/app/firewall.py index beba20a..d2c0624 100644 --- a/app/firewall.py +++ b/app/firewall.py @@ -13,23 +13,18 @@ import paramiko from datetime import datetime import shlex from typing import Iterable +from .constants import ( + BLOCKLIST_NAME, + REPORT_DIR, + ROUTER_IP, + ROUTER_PASSWORD, + ROUTER_USER, + SSH_PORT, + SSH_TIMEOUT, + WHITELIST_NAMES, +) from .ip_utils import is_whitelisted -DATA_DIR = os.path.join(os.path.dirname(os.path.dirname( - os.path.abspath(__file__))), "data", "mikrotik") -REPORT_DIR = os.path.join(os.path.dirname( - os.path.dirname(os.path.abspath(__file__))), "reports") -MIKROTIK_DATA_DIR = os.path.join(REPORT_DIR, "mikrotik", "data") - -SSH_TIMEOUT = 10 # seconds -SSH_PORT = 22 -ROUTER_IP = os.environ.get("ROUTER_IP", "192.168.88.1") -ROUTER_USER = os.environ.get("ROUTER_USER", "admin") -ROUTER_PASSWORD = os.environ.get("ROUTER_PASSWORD", "M11ffpgm") -BLOCKLIST_NAME = os.environ.get("BLOCKLIST_NAME", "Blocked") -WHITELIST_NAMES = os.environ.get( - "WHITELIST_NAMES", "Cloudflare,Openfactory").split(",") - class MikrotikFirewall: """ diff --git a/app/flags.py b/app/flags.py index 065f2ca..721c6e5 100644 --- a/app/flags.py +++ b/app/flags.py @@ -7,11 +7,12 @@ Detection functions that identify hostile IPs based on aggregated stats. from __future__ import annotations from . import bad_patterns +from .constants import DEFAULT_ERROR_RATIO, DEFAULT_MIN_DENIED, DEFAULT_MIN_ERRORS from .models import IPStats -def flag_scanners(stats: dict[str, IPStats], min_errors: int = 5, - min_ratio: float = 0.5) -> dict[str, IPStats]: +def flag_scanners(stats: dict[str, IPStats], min_errors: int = DEFAULT_MIN_ERRORS, + min_ratio: float = DEFAULT_ERROR_RATIO) -> dict[str, IPStats]: """IPs with many errors AND a high error ratio.""" return { ip: s for ip, s in stats.items() @@ -20,7 +21,7 @@ def flag_scanners(stats: dict[str, IPStats], min_errors: int = 5, def flag_denied_only(stats: dict[str, IPStats], - min_denied: int = 1) -> dict[str, IPStats]: + min_denied: int = DEFAULT_MIN_DENIED) -> dict[str, IPStats]: """IPs that only ever hit denied paths and never authenticated.""" return { ip: s for ip, s in stats.items() diff --git a/app/mikrotik.py b/app/mikrotik.py index b66db61..938c468 100644 --- a/app/mikrotik.py +++ b/app/mikrotik.py @@ -9,6 +9,7 @@ from typing import Iterable from .ip_utils import is_ipv4, is_ipv6, is_whitelisted, parse_ip from .whitelists import WHITELISTS +from .constants import BLOCKLIST_NAME def split_cidr_versions(targets: Iterable[str]) -> tuple[list[str], list[str]]: @@ -76,12 +77,12 @@ def mikrotik_script(tag: str, blocked_ips_or_subnets: Iterable[str], 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 += [f"add list={BLOCKLIST_NAME} 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 += [f"add list={BLOCKLIST_NAME} address={t}" for t in block_v6] lines.append("") for version, cmd, blocks in ((4, "/ip firewall filter", block_v4), @@ -94,7 +95,7 @@ def mikrotik_script(tag: str, blocked_ips_or_subnets: Iterable[str], if blocks: for chain in ("input", "forward"): lines.append( - f"add chain={chain} action=drop src-address-list=Blocked " + f"add chain={chain} action=drop src-address-list={BLOCKLIST_NAME} " f'comment="{tag} blocked (v{version})"') lines.append("") diff --git a/app/npm.py b/app/npm.py index 47ee18e..fd3be3d 100644 --- a/app/npm.py +++ b/app/npm.py @@ -13,16 +13,17 @@ import re from collections import Counter from .analyzer import Analyzer +from .constants import ( + DEFAULT_ERROR_RATIO, + DEFAULT_MIN_ERRORS, + NPM_LOG_PATH, + NPM_REPORT_DIR, +) 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( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "reports", "npm") - # [02/Aug/2026:05:01:31 +0200] - 401 401 - GET https host "/path" # [Client IP] [Length N] [Gzip X] [Sent-to IP] "UA" "Referer" LINE_RE = re.compile( @@ -83,8 +84,8 @@ class NpmAnalyzer(Analyzer): ("ua", lambda r: r.ua), ) - def __init__(self, *args, min_errors: int = 5, - error_ratio: float = 0.5, **kwargs): + def __init__(self, *args, min_errors: int = DEFAULT_MIN_ERRORS, + error_ratio: float = DEFAULT_ERROR_RATIO, **kwargs): super().__init__(*args, **kwargs) self.min_errors = min_errors self.min_ratio = error_ratio @@ -145,11 +146,11 @@ class NpmAnalyzer(Analyzer): def main() -> None: 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, - help="minimum error responses to flag (default: 5)") - parser.add_argument("--error-ratio", type=float, default=0.5, - help="minimum error ratio to flag (default: 0.5)") + NPM_LOG_PATH, NPM_REPORT_DIR) + parser.add_argument("--min-errors", type=int, default=DEFAULT_MIN_ERRORS, + help=f"minimum error responses to flag (default: {DEFAULT_MIN_ERRORS})") + parser.add_argument("--error-ratio", type=float, default=DEFAULT_ERROR_RATIO, + help=f"minimum error ratio to flag (default: {DEFAULT_ERROR_RATIO})") args = parser.parse_args() NpmAnalyzer( @@ -157,7 +158,6 @@ def main() -> None: report_dir=args.report_dir, top=args.top, quiet=args.quiet, - write_rsc=not args.no_rsc, min_errors=args.min_errors, error_ratio=args.error_ratio, ).run() diff --git a/app/scan.py b/app/scan.py index 72f49b4..395741c 100644 --- a/app/scan.py +++ b/app/scan.py @@ -8,19 +8,25 @@ and outputs a single consolidated MikroTik RouterOS blocklist script. from __future__ import annotations import argparse -import os -from datetime import datetime -from . import utils +from .constants import ( + COPYPARTY_LOG_PATH, + COPYPARTY_REPORT_DIR, + DEFAULT_ERROR_RATIO, + DEFAULT_MIN_DENIED, + DEFAULT_MIN_ERRORS, + DEFAULT_TOP, + NPM_LOG_PATH, + NPM_REPORT_DIR, + REPORT_DIR, + SUBNET_THRESHOLD, +) from .firewall import MikrotikFirewall -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 +from .copyparty import CopypartyAnalyzer +from .npm import NpmAnalyzer from .ip_utils import is_whitelisted, split_versions from .whois import consolidate_by_subnet -DEFAULT_CONSOLIDATED_REPORT_DIR = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "reports") - def main() -> None: parser = argparse.ArgumentParser( @@ -28,28 +34,28 @@ def main() -> None: ) # 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)") + parser.add_argument("--copyparty-log-path", default=COPYPARTY_LOG_PATH, + help=f"copyparty log directory (default: {COPYPARTY_LOG_PATH})") + parser.add_argument("--copyparty-report-dir", default=COPYPARTY_REPORT_DIR, + help=f"copyparty report directory (default: {COPYPARTY_REPORT_DIR})") + parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED, + help=f"minimum copyparty denied hits before flagging (default: {DEFAULT_MIN_DENIED})") # 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)") + parser.add_argument("--npm-log-path", default=NPM_LOG_PATH, + help=f"NPM log directory (default: {NPM_LOG_PATH})") + parser.add_argument("--npm-report-dir", default=NPM_REPORT_DIR, + help=f"NPM report directory (default: {NPM_REPORT_DIR})") + parser.add_argument("--min-errors", type=int, default=DEFAULT_MIN_ERRORS, + help=f"minimum NPM error responses to flag (default: {DEFAULT_MIN_ERRORS})") + parser.add_argument("--error-ratio", type=float, default=DEFAULT_ERROR_RATIO, + help=f"minimum NPM error ratio to flag (default: {DEFAULT_ERROR_RATIO})") # 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("--report-dir", default=REPORT_DIR, + help=f"consolidated reports directory (default: {REPORT_DIR})") + parser.add_argument("--top", type=int, default=DEFAULT_TOP, + help=f"rows in the top-IP tables (default: {DEFAULT_TOP})") parser.add_argument("--quiet", action="store_true", help="write files, suppress console reports") parser.add_argument("--update-firewall", action="store_true", @@ -68,7 +74,6 @@ def main() -> None: report_dir=args.copyparty_report_dir, top=args.top, quiet=args.quiet, - write_rsc=False, min_denied=args.min_denied, ) @@ -77,7 +82,6 @@ def main() -> None: report_dir=args.npm_report_dir, top=args.top, quiet=args.quiet, - write_rsc=False, min_errors=args.min_errors, error_ratio=args.error_ratio, ) @@ -115,10 +119,10 @@ def main() -> None: final_blocked_ips = { ip for ip in consolidated_candidates if not is_whitelisted(ip)} - # Group into subnets via WHOIS if multiple malicious IPs in same subnet (threshold = 3) + # Group into subnets via WHOIS if multiple malicious IPs in same subnet (threshold = SUBNET_THRESHOLD) print("\nGrouping blocked IPs into subnets using WHOIS...") final_blocked_targets = consolidate_by_subnet( - final_blocked_ips, threshold=3) + final_blocked_ips, threshold=SUBNET_THRESHOLD) blocked_ipv4, blocked_ipv6 = split_versions(final_blocked_targets) print("\n=== Consolidated Summary ===") diff --git a/app/utils.py b/app/utils.py index 5942518..6d0cc0c 100644 --- a/app/utils.py +++ b/app/utils.py @@ -16,6 +16,8 @@ from __future__ import annotations import argparse +from .constants import DEFAULT_TOP + def arg_parser(description: str, default_log_path: str, default_report_dir: str) -> argparse.ArgumentParser: @@ -24,8 +26,8 @@ def arg_parser(description: str, default_log_path: str, help=f"log directory (default: {default_log_path})") p.add_argument("--report-dir", default=default_report_dir, help=f"report output dir (default: {default_report_dir})") - p.add_argument("--top", type=int, default=30, - help="rows in the top-IP table (default: 30)") + p.add_argument("--top", type=int, default=DEFAULT_TOP, + help=f"rows in the top-IP table (default: {DEFAULT_TOP})") p.add_argument("--quiet", action="store_true", help="write files, print nothing but the paths") p.add_argument("--no-rsc", action="store_true", diff --git a/app/whitelists.py b/app/whitelists.py index ff09873..7b8297b 100644 --- a/app/whitelists.py +++ b/app/whitelists.py @@ -18,12 +18,16 @@ import os import re from typing import Any -# --------------------------------------------------------------------------- -# Paths -# --------------------------------------------------------------------------- - -_HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -_DATA_DIR = os.path.join(_HERE, "data") +from .constants import ( + CLOUDFLARE_URL_V4, + CLOUDFLARE_URL_V6, + CLOUDFLARE_V4_FILE, + CLOUDFLARE_V6_FILE, + OPENFACTORY_FILE, + OPENFACTORY_URL, + PALOALTO_FILE, + PALOALTO_URL, +) # --------------------------------------------------------------------------- # CIDR extraction regex @@ -32,23 +36,12 @@ _DATA_DIR = os.path.join(_HERE, "data") CIDR_RE = re.compile( r"(? None: global _whois_cache - if not _whois_cache and os.path.exists(_WHOIS_CACHE_PATH): + if not _whois_cache and os.path.exists(WHOIS_CACHE_PATH): try: - with open(_WHOIS_CACHE_PATH, "r", encoding="utf-8") as f: + with open(WHOIS_CACHE_PATH, "r", encoding="utf-8") as f: _whois_cache = json.load(f) except Exception: _whois_cache = {} @@ -32,9 +33,9 @@ def load_whois_cache() -> None: def save_whois_cache() -> None: if _whois_cache: - os.makedirs(os.path.dirname(_WHOIS_CACHE_PATH), exist_ok=True) + os.makedirs(os.path.dirname(WHOIS_CACHE_PATH), exist_ok=True) try: - with open(_WHOIS_CACHE_PATH, "w", encoding="utf-8") as f: + with open(WHOIS_CACHE_PATH, "w", encoding="utf-8") as f: json.dump(_whois_cache, f, indent=2) except Exception: pass @@ -53,8 +54,8 @@ def query_whois_subnet(ip: str) -> str | None: import socket try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.settimeout(3.0) - s.connect(("whois.radb.net", 43)) + s.settimeout(WHOIS_TIMEOUT) + s.connect((WHOIS_SERVER, WHOIS_PORT)) s.sendall(f"{ip}\r\n".encode()) res = "" while True: @@ -89,7 +90,8 @@ def query_whois_subnet(ip: str) -> str | None: return subnet_str -def consolidate_by_subnet(ips: Iterable[str], threshold: int = 3) -> set[str]: +def consolidate_by_subnet(ips: Iterable[str], + threshold: int = SUBNET_THRESHOLD) -> set[str]: """Group IPs by WHOIS subnet; if >= threshold IPs in a subnet, block the subnet. Otherwise, block individual IPs. @@ -124,7 +126,3 @@ def consolidate_by_subnet(ips: Iterable[str], threshold: int = 3) -> set[str]: 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 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3a44a23 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +paramiko +python-dotenv \ No newline at end of file