diff --git a/.gitignore b/.gitignore
index 9563132..7b38716 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,6 +7,9 @@
# python cache
__pycache__/
+# incremental-scan state (regenerated each run)
+data/scan_state.json
+
# reports data
reports/copyparty/
reports/mikrotik/
diff --git a/app/analyzer.py b/app/analyzer.py
index 29ad7b4..ddf9850 100644
--- a/app/analyzer.py
+++ b/app/analyzer.py
@@ -13,13 +13,14 @@ 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 .files import ensure_dirs, find_logs, read_lines, read_lines_from, validate_dir, write_csv, write_json, write_lines
from .flags import by_total, flag_malicious_patterns
from .ip_utils import is_whitelisted, split_versions
from .mikrotik import write_mikrotik
from .models import IPStats, Request, aggregate
from .reporting import Column, columns_header, columns_row, render_kv, render_table, timestamp_name
from .constants import DEFAULT_TOP, DEFAULT_TOP_PATHS
+from . import scan_state
from .whitelists import refresh_whitelists
@@ -44,13 +45,15 @@ class Analyzer:
flag_note = ""
def __init__(self, log_path: str, report_dir: str, top: int = DEFAULT_TOP,
- quiet: bool = False, write_rsc: bool = True):
+ quiet: bool = False, write_rsc: bool = True,
+ full: bool = False):
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.full = full
self.files: list[str] = []
self.requests: list[Request] = []
self.stats: dict[str, IPStats] = {}
@@ -89,14 +92,46 @@ class Analyzer:
return self.files
def load(self) -> list[Request]:
+ """Read log entries, resuming from the last processed offset.
+
+ Non-gzip logs are read incrementally (only new bytes since the last
+ scan) unless ``full`` is set; gzip archives are always fully read
+ because they cannot be seeked.
+ """
+ state: dict[str, int] = {}
+ if not self.full:
+ state = scan_state.load_state()
+ # Prune entries whose log file no longer exists (safe across
+ # analyzers: only removes dead keys, keeps live ones).
+ state = {k: v for k, v in state.items() if os.path.exists(k)}
for name in self.files:
+ full_path = os.path.join(self.log_path, name)
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
+ if self.full or full_path.endswith(".gz"):
+ for line in read_lines(full_path):
+ req = self.parse_line(line)
+ if req is not None:
+ self.requests.append(req)
+ count += 1
+ else:
+ try:
+ size = os.path.getsize(full_path)
+ except OSError:
+ size = 0
+ recorded = state.get(full_path, 0)
+ # Reset to the start if the file shrank or was truncated.
+ start = recorded if 0 < recorded <= size else 0
+ last_resume = start
+ for line, resume in read_lines_from(full_path, start):
+ req = self.parse_line(line)
+ if req is not None:
+ self.requests.append(req)
+ count += 1
+ last_resume = resume
+ state[full_path] = last_resume
self.per_file[name] = count
+ if not self.full:
+ scan_state.save_state(state)
return self.requests
def analyze(self) -> dict[str, IPStats]:
diff --git a/app/constants.py b/app/constants.py
index 82a4ed5..b4c1e91 100644
--- a/app/constants.py
+++ b/app/constants.py
@@ -53,6 +53,9 @@ MIKROTIK_DATA_DIR = os.path.join(DATA_DIR, "mikrotik")
WHOIS_CACHE_PATH = os.path.join(DATA_DIR, "whois_cache.json")
+# Incremental-scan state: absolute log path -> last processed byte offset
+SCAN_STATE_PATH = os.path.join(DATA_DIR, "scan_state.json")
+
# ---------------------------------------------------------------------------
# Log sources (configurable via environment)
# ---------------------------------------------------------------------------
diff --git a/app/copyparty.py b/app/copyparty.py
index e3b325f..704bbf3 100644
--- a/app/copyparty.py
+++ b/app/copyparty.py
@@ -164,6 +164,8 @@ def main() -> None:
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})")
+ parser.add_argument("--full", action="store_true",
+ help="rescan all logs from the start (ignore incremental state)")
args = parser.parse_args()
CopypartyAnalyzer(
@@ -171,6 +173,7 @@ def main() -> None:
report_dir=args.report_dir,
top=args.top,
quiet=args.quiet,
+ full=args.full,
min_denied=args.min_denied,
).run()
diff --git a/app/files.py b/app/files.py
index 5eb01dd..f2039a0 100644
--- a/app/files.py
+++ b/app/files.py
@@ -60,6 +60,26 @@ def read_lines(path: str) -> Iterator[str]:
yield ANSI_RE.sub("", line.rstrip("\r\n"))
+def read_lines_from(path: str, start: int = 0) -> Iterator[tuple[str, int]]:
+ """Yield ``(line, resume_offset)`` for a plain-text log from byte ``start``.
+
+ Only seekable (non-gzip) files can be resumed. ``resume_offset`` points
+ after the last complete line; a trailing partial line (no newline yet) is
+ yielded but resets the resume offset to its start so it is re-read once
+ completed. The last yielded ``resume_offset`` is the value to store.
+ """
+ with open(path, "rb") as f:
+ f.seek(start)
+ pos = start
+ for raw in f:
+ line_start = pos
+ pos += len(raw)
+ complete = raw.endswith(b"\n") or raw.endswith(b"\r")
+ text = ANSI_RE.sub(
+ "", raw.decode("utf-8", errors="replace").rstrip("\r\n"))
+ yield text, (pos if complete else line_start)
+
+
# -- writers ----------------------------------------------------------------
def write_csv(path: str, header: Sequence[str],
diff --git a/app/npm.py b/app/npm.py
index fd3be3d..e61e9b9 100644
--- a/app/npm.py
+++ b/app/npm.py
@@ -151,6 +151,8 @@ def main() -> None:
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})")
+ parser.add_argument("--full", action="store_true",
+ help="rescan all logs from the start (ignore incremental state)")
args = parser.parse_args()
NpmAnalyzer(
@@ -158,6 +160,7 @@ def main() -> None:
report_dir=args.report_dir,
top=args.top,
quiet=args.quiet,
+ full=args.full,
min_errors=args.min_errors,
error_ratio=args.error_ratio,
).run()
diff --git a/app/scan.py b/app/scan.py
index 785565c..d175001 100644
--- a/app/scan.py
+++ b/app/scan.py
@@ -48,6 +48,7 @@ class ScanParams:
report_dir: str = REPORT_DIR
top: int = DEFAULT_TOP
quiet: bool = False
+ full: bool = False
update_firewall: bool = False
dry_run: bool = False
@@ -105,6 +106,7 @@ def run_scan(params: ScanParams) -> ScanResult:
report_dir=params.copyparty_report_dir,
top=params.top,
quiet=params.quiet,
+ full=params.full,
min_denied=params.min_denied,
)
@@ -113,6 +115,7 @@ def run_scan(params: ScanParams) -> ScanResult:
report_dir=params.npm_report_dir,
top=params.top,
quiet=params.quiet,
+ full=params.full,
min_errors=params.min_errors,
error_ratio=params.error_ratio,
)
@@ -237,6 +240,8 @@ def _build_parser() -> argparse.ArgumentParser:
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("--full", action="store_true",
+ help="rescan all log files from the start (ignore incremental state)")
parser.add_argument("--update-firewall", action="store_true",
help="synchronize the MikroTik Blocked list after scanning")
parser.add_argument("--dry-run", action="store_true",
@@ -257,6 +262,7 @@ def main(argv: list[str] | None = None) -> None:
report_dir=args.report_dir,
top=args.top,
quiet=args.quiet,
+ full=args.full,
update_firewall=args.update_firewall,
dry_run=args.dry_run,
)
diff --git a/app/scan_state.py b/app/scan_state.py
new file mode 100644
index 0000000..5ca9000
--- /dev/null
+++ b/app/scan_state.py
@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+"""Persistent state for incremental log scanning.
+
+Tracks the last processed byte offset per absolute log file path so a scan
+only reads new entries instead of re-reading the whole file. The state file
+is written atomically (temp file + rename) to avoid corruption.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+
+from .constants import SCAN_STATE_PATH
+
+
+def load_state() -> dict[str, int]:
+ """Return ``{absolute log path: byte offset}`` from the scan-state file."""
+ try:
+ with open(SCAN_STATE_PATH, "r", encoding="utf-8") as f:
+ data = json.load(f)
+ return {k: int(v) for k, v in data.items()}
+ except (OSError, ValueError, TypeError):
+ return {}
+
+
+def save_state(state: dict[str, int]) -> None:
+ """Write the scan-state file atomically (temp file + rename)."""
+ os.makedirs(os.path.dirname(SCAN_STATE_PATH), exist_ok=True)
+ tmp = SCAN_STATE_PATH + ".tmp"
+ with open(tmp, "w", encoding="utf-8") as f:
+ json.dump(state, f, indent=2, sort_keys=True)
+ os.replace(tmp, SCAN_STATE_PATH)
diff --git a/app/templates/dashboard.html b/app/templates/dashboard.html
index f93d8b3..34b2a56 100644
--- a/app/templates/dashboard.html
+++ b/app/templates/dashboard.html
@@ -96,5 +96,7 @@ endblock %} {% block content %}
}
loadDashboard();
+ // Keep the latest-scans table fresh without hammering /api/scan/status.
+ setInterval(loadDashboard, 15000);
{% endblock %}
diff --git a/app/templates/scan.html b/app/templates/scan.html
index 4908c18..751a11f 100644
--- a/app/templates/scan.html
+++ b/app/templates/scan.html
@@ -42,6 +42,10 @@
> Dry-run (preview, no
changes)
+
@@ -86,6 +90,7 @@
parseFloat(document.getElementById("error_ratio").value) || 0.5,
update_firewall: document.getElementById("update_firewall").checked,
dry_run: document.getElementById("dry_run").checked,
+ full: document.getElementById("full").checked,
quiet: true,
};
runBtn.disabled = true;
@@ -173,6 +178,8 @@
runBtn.addEventListener("click", runScan);
refreshJobs();
- setInterval(refreshJobs, 5000);
+ // /api/scan/status is only polled every 15 s; the active job's own
+ // 2 s poll keeps live progress without hammering the status endpoint.
+ setInterval(refreshJobs, 15000);
{% endblock %}
diff --git a/app/web.py b/app/web.py
index b4dd56d..0eda3f2 100644
--- a/app/web.py
+++ b/app/web.py
@@ -102,6 +102,7 @@ def _params_from_request() -> ScanParams:
report_dir=str(_pick("report_dir", REPORT_DIR)),
top=int(_pick("top", DEFAULT_TOP)),
quiet=bool(data.get("quiet", False)),
+ full=bool(data.get("full", False)),
update_firewall=bool(data.get("update_firewall", False)),
dry_run=bool(data.get("dry_run", False)),
)
diff --git a/app/web_scan.py b/app/web_scan.py
index 804321d..b1221ca 100644
--- a/app/web_scan.py
+++ b/app/web_scan.py
@@ -100,6 +100,7 @@ def start_scan(kind: ScanKind, params: ScanParams) -> dict[str, Any]:
report_dir=params.copyparty_report_dir,
top=params.top,
quiet=params.quiet,
+ full=params.full,
min_denied=params.min_denied,
).run()
job.result = ScanResult(ok=True)
@@ -111,6 +112,7 @@ def start_scan(kind: ScanKind, params: ScanParams) -> dict[str, Any]:
report_dir=params.npm_report_dir,
top=params.top,
quiet=params.quiet,
+ full=params.full,
min_errors=params.min_errors,
error_ratio=params.error_ratio,
).run()
diff --git a/data/whois_cache.json b/data/whois_cache.json
index 8e03677..388a84c 100644
--- a/data/whois_cache.json
+++ b/data/whois_cache.json
@@ -966,5 +966,9 @@
"185.198.240.238": "185.198.240.0/24",
"34.125.213.146": "34.125.208.0/20",
"222.176.200.21": "222.176.0.0/13",
- "106.117.109.102": "106.117.0.0/16"
+ "106.117.109.102": "106.117.0.0/16",
+ "121.29.51.109": "121.28.0.0/15",
+ "74.248.97.169": "74.248.0.0/15",
+ "34.176.154.83": "34.176.144.0/20",
+ "35.196.132.85": "35.196.128.0/20"
}
\ No newline at end of file
diff --git a/reports/app/data/blocklist_ipv4.txt b/reports/app/data/blocklist_ipv4.txt
index a4de749..71ee960 100644
--- a/reports/app/data/blocklist_ipv4.txt
+++ b/reports/app/data/blocklist_ipv4.txt
@@ -38,6 +38,7 @@
118.212.120.78/32
118.81.196.142/32
118.81.207.202/32
+121.29.51.109/32
123.158.61.179/32
123.160.175.254/32
123.163.114.150/32
@@ -343,6 +344,7 @@
34.17.159.73/32
34.171.75.208/32
34.175.82.8/32
+34.176.154.83/32
34.18.38.137/32
34.181.134.189/32
34.186.83.134/32