zwitschi/intrusion-detection
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add centralized constants module and .env configuration support

Browse Source
This commit is contained in:
zwitschi committed 2026-08-05 19:33:12 +02:00
1 parent 94be1da199
commit eea5ef6885
14 files changed
+334 -126
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/analyzer.py
+4 -3
View File
@@ -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("")
app/constants.py
+120
View File
@@ -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")
app/copyparty.py
+9 -10
View File
@@ -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 <ip> <port> 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()
app/firewall.py
+10 -15
View File
@@ -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:
"""
app/flags.py
+4 -3
View File
@@ -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()
app/mikrotik.py
+4 -3
View File
@@ -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("")
app/npm.py
+13 -13
View File
@@ -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()
app/scan.py
+34 -30
View File
@@ -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 ===")
app/utils.py
+4 -2
View File
@@ -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",
app/whitelists.py
+12 -19
View File
@@ -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"(?<![0-9A-Fa-f:.])([0-9A-Fa-f:.]+/\d{1,3})(?![0-9A-Fa-f:.])")
# ---------------------------------------------------------------------------
# Provider URLs
# ---------------------------------------------------------------------------
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"
# ---------------------------------------------------------------------------
# Local file paths (checked first, 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")
# (paths live in app/constants.py as OPENFACTORY_FILE, CLOUDFLARE_V4_FILE,
# CLOUDFLARE_V6_FILE, PALOALTO_FILE; URLs as OPENFACTORY_URL, ...)
# ---------------------------------------------------------------------------
# Built-in fallback CIDR lists
app/whois.py
+15 -17
View File
@@ -9,22 +9,23 @@ 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.dirname(os.path.abspath(__file__))
), "data", "whois_cache.json"
from .constants import (
SUBNET_THRESHOLD,
WHOIS_CACHE_PATH,
WHOIS_PORT,
WHOIS_SERVER,
WHOIS_TIMEOUT,
)
from .ip_utils import ip_version, parse_ip
_whois_cache: dict[str, str] = {}
def load_whois_cache() -> 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