Refactor copyparty and npm analyzers to use updated imports and remove utils module

Update whois cache with additional IP ranges
This commit is contained in:
zwitschi committed 2026-08-04 21:56:55 +02:00
1 parent 0e2c69579a
commit 40212d84f7
4 files changed
+36 -258

No files matched your search

+16 -11
View File
@@ -12,8 +12,13 @@ from __future__ import annotations
import os
import re
import utils
from utils import Request
from analyzer import Analyzer
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(
@@ -37,7 +42,7 @@ CLIENT_BAN_RE = re.compile(r"client banned")
BAN_SECS_RE = re.compile(r"banned for (\d+) sec")
class CopypartyAnalyzer(utils.Analyzer):
class CopypartyAnalyzer(Analyzer):
name = "copyparty"
title = "copyparty log analysis report"
log_pattern = re.compile(r"\d{4}-\d{2}-\d{2}\.txt(\.\d+)?$")
@@ -115,7 +120,7 @@ class CopypartyAnalyzer(utils.Analyzer):
# --- flagging ----------------------------------------------------------
def flag(self, stats):
return utils.flag_denied_only(stats, min_denied=self.min_denied)
return flag_denied_only(stats, min_denied=self.min_denied)
def flag_comment(self, s):
return (f"denied={s.denied} total={s.total} "
@@ -127,7 +132,7 @@ class CopypartyAnalyzer(utils.Analyzer):
# --- extras ------------------------------------------------------------
def banned_ips(self) -> dict[str, utils.IPStats]:
def banned_ips(self) -> dict[str, IPStats]:
return {ip: s for ip, s in self.stats.items() if s.banned}
def authed_users(self) -> dict[str, int]:
@@ -144,17 +149,17 @@ class CopypartyAnalyzer(utils.Analyzer):
("authenticated users", self.authed_users() or "{}")]
def write_extra(self, data_dir: str) -> list[str]:
v4, v6 = utils.split_versions(self.banned_ips())
v4, v6 = split_versions(self.banned_ips())
return [
utils.write_lines(os.path.join(data_dir, "banned_ips.txt"), v4),
utils.write_lines(os.path.join(data_dir, "banned_ipv6.txt"), v6),
utils.write_json(os.path.join(data_dir, "authed_users.json"),
self.authed_users()),
write_lines(os.path.join(data_dir, "banned_ips.txt"), v4),
write_lines(os.path.join(data_dir, "banned_ipv6.txt"), v6),
write_json(os.path.join(data_dir, "authed_users.json"),
self.authed_users()),
]
def main() -> None:
parser = utils.arg_parser(
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,