361 lines
13 KiB
Python
361 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Background scan runner for the web UI.
|
|
|
|
Runs scans in a daemon thread so HTTP requests never block, keeps an
|
|
in-memory job store, and serializes scans with a busy lock (WHOIS and
|
|
whitelist refresh must not run concurrently).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
import uuid
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any, Literal
|
|
|
|
from ..utils.constants import (
|
|
APP_REPORT_DATA_DIR,
|
|
COPYPARTY_REPORT_DIR,
|
|
MIKROTIK_REPORT_DIR,
|
|
NPM_REPORT_DIR,
|
|
SCAN_JOBS_MAX,
|
|
SCAN_JOBS_PATH,
|
|
)
|
|
from ..copyparty import CopypartyAnalyzer
|
|
from ..npm import NpmAnalyzer
|
|
from ..scan import ScanParams, ScanResult, run_scan
|
|
|
|
ScanKind = Literal["copyparty", "npm", "consolidated"]
|
|
|
|
#: serializes scans — only one runs at a time
|
|
_scan_lock = threading.Lock()
|
|
|
|
#: in-memory job store (job_id -> ScanJob)
|
|
_jobs: dict[str, "ScanJob"] = {}
|
|
_jobs_lock = threading.Lock()
|
|
|
|
#: True once persisted jobs have been loaded from disk
|
|
_loaded = False
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now().astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
@dataclass
|
|
class ScanJob:
|
|
"""One background scan job."""
|
|
|
|
job_id: str
|
|
kind: ScanKind
|
|
params: ScanParams
|
|
state: str = "queued" # queued | running | done | error
|
|
started: str | None = None
|
|
finished: str | None = None
|
|
result: ScanResult | None = None
|
|
error: str | None = None
|
|
busy: bool = False
|
|
details: dict = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
result = self.result
|
|
return {
|
|
"job_id": self.job_id,
|
|
"kind": self.kind,
|
|
"params": asdict(self.params),
|
|
"state": self.state,
|
|
"started": self.started,
|
|
"finished": self.finished,
|
|
"error": self.error,
|
|
"busy": self.busy,
|
|
"details": self.details,
|
|
"result": None if result is None else {
|
|
"ok": result.ok,
|
|
"cp_exists": result.cp_exists,
|
|
"npm_exists": result.npm_exists,
|
|
"cp_candidates": result.cp_candidates,
|
|
"npm_candidates": result.npm_candidates,
|
|
"blocked_ips": result.blocked_ips,
|
|
"blocked_targets": result.blocked_targets,
|
|
"blocked_ipv4": result.blocked_ipv4,
|
|
"blocked_ipv6": result.blocked_ipv6,
|
|
"blocklist_v4_path": result.blocklist_v4_path,
|
|
"blocklist_v6_path": result.blocklist_v6_path,
|
|
"firewall_connected": result.firewall_connected,
|
|
"firewall_changes": result.firewall_changes,
|
|
"errors": result.errors,
|
|
},
|
|
}
|
|
|
|
|
|
def start_scan(kind: ScanKind, params: ScanParams) -> dict[str, Any]:
|
|
"""Start a scan in the background and return its job dict.
|
|
|
|
If another scan is already running, returns a dict with an ``error`` key.
|
|
"""
|
|
if not _scan_lock.acquire(blocking=False):
|
|
return {"error": "a scan is already running; try again when it finishes"}
|
|
|
|
job = ScanJob(job_id=uuid.uuid4().hex[:12], kind=kind,
|
|
params=params, busy=True)
|
|
with _jobs_lock:
|
|
_jobs[job.job_id] = job
|
|
_save_jobs()
|
|
|
|
def _run() -> None:
|
|
job.started = _now()
|
|
job.state = "running"
|
|
try:
|
|
if kind == "consolidated":
|
|
job.result = run_scan(params)
|
|
elif kind == "copyparty":
|
|
info = CopypartyAnalyzer(
|
|
log_path=params.copyparty_log_path,
|
|
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)
|
|
job.details = {"report": info.get("report"),
|
|
"data_dir": info.get("data_dir")}
|
|
elif kind == "npm":
|
|
info = NpmAnalyzer(
|
|
log_path=params.npm_log_path,
|
|
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()
|
|
job.result = ScanResult(ok=True)
|
|
job.details = {"report": info.get("report"),
|
|
"data_dir": info.get("data_dir")}
|
|
else: # pragma: no cover - guarded by type hint
|
|
raise ValueError(f"unknown scan kind: {kind}")
|
|
job.state = "done"
|
|
except Exception as e:
|
|
job.error = str(e)
|
|
job.state = "error"
|
|
finally:
|
|
job.finished = _now()
|
|
job.busy = False
|
|
_scan_lock.release()
|
|
_save_jobs()
|
|
|
|
threading.Thread(target=_run, daemon=True,
|
|
name=f"scan-{job.job_id}").start()
|
|
return job.to_dict()
|
|
|
|
|
|
def get_scan(job_id: str) -> dict[str, Any] | None:
|
|
"""Return a single job as a dict, or None if unknown."""
|
|
load_persisted()
|
|
with _jobs_lock:
|
|
job = _jobs.get(job_id)
|
|
return job.to_dict() if job else None
|
|
|
|
|
|
def _iso_key(value: str | None) -> float:
|
|
"""ISO-8601 string -> epoch seconds (0.0 when unparseable/empty)."""
|
|
if not value:
|
|
return 0.0
|
|
try:
|
|
return datetime.fromisoformat(value).timestamp()
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def latest_file_scan() -> dict[str, Any] | None:
|
|
"""Build a ``done`` scan entry from the newest report files on disk.
|
|
|
|
Lets the dashboard show the latest completed scan even when it was not
|
|
run through the web UI (server restart, CLI-run scans). Returns None when
|
|
no scan outputs exist yet.
|
|
"""
|
|
|
|
def newest(dirpath: str, suffix: str = "") -> str | None:
|
|
"""Newest file in ``dirpath`` (optionally ending in ``suffix``)."""
|
|
try:
|
|
names = [n for n in os.listdir(dirpath)
|
|
if os.path.isfile(os.path.join(dirpath, n))
|
|
and (not suffix or n.endswith(suffix))]
|
|
except OSError:
|
|
return None
|
|
return os.path.join(dirpath, max(names)) if names else None
|
|
|
|
mikrotik = newest(MIKROTIK_REPORT_DIR, ".json")
|
|
cp_report = newest(COPYPARTY_REPORT_DIR, ".txt")
|
|
npm_report = newest(NPM_REPORT_DIR, ".txt")
|
|
v4_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv4.txt")
|
|
v6_path = os.path.join(APP_REPORT_DATA_DIR, "blocklist_ipv6.txt")
|
|
have_outputs = (mikrotik is not None or cp_report is not None
|
|
or npm_report is not None
|
|
or os.path.exists(v4_path) or os.path.exists(v6_path))
|
|
if not have_outputs:
|
|
return None
|
|
|
|
# Timestamp: prefer the newest snapshot's collected_at, else the newest
|
|
# report file's mtime, else now.
|
|
timestamp: str | None = None
|
|
if mikrotik:
|
|
try:
|
|
with open(mikrotik, "r", encoding="utf-8") as f:
|
|
timestamp = json.load(f).get("collected_at")
|
|
except (OSError, ValueError):
|
|
timestamp = None
|
|
if not timestamp:
|
|
candidates = [p for p in (mikrotik, cp_report, npm_report,
|
|
v4_path, v6_path)
|
|
if p and os.path.exists(p)]
|
|
if candidates:
|
|
tz = datetime.now().astimezone().tzinfo
|
|
timestamp = datetime.fromtimestamp(
|
|
max(os.path.getmtime(p) for p in candidates),
|
|
tz=tz).isoformat(timespec="seconds")
|
|
if not timestamp:
|
|
timestamp = _now()
|
|
|
|
blocked_targets = 0
|
|
for path in (v4_path, v6_path):
|
|
if os.path.exists(path):
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
blocked_targets += sum(1 for _ in f)
|
|
except OSError:
|
|
pass
|
|
|
|
return {
|
|
"job_id": "files-latest",
|
|
"kind": "consolidated",
|
|
"state": "done",
|
|
"started": timestamp,
|
|
"finished": timestamp,
|
|
"error": None,
|
|
"busy": False,
|
|
"details": {},
|
|
"result": {
|
|
"ok": True,
|
|
"cp_exists": cp_report is not None,
|
|
"npm_exists": npm_report is not None,
|
|
"cp_candidates": 0,
|
|
"npm_candidates": 0,
|
|
"blocked_ips": blocked_targets,
|
|
"blocked_targets": blocked_targets,
|
|
"blocked_ipv4": [],
|
|
"blocked_ipv6": [],
|
|
"blocklist_v4_path": v4_path,
|
|
"blocklist_v6_path": v6_path,
|
|
"firewall_connected": mikrotik is not None,
|
|
"firewall_changes": None,
|
|
"errors": [],
|
|
},
|
|
}
|
|
|
|
|
|
def latest_scans(limit: int = 10) -> list[dict[str, Any]]:
|
|
"""Return the most recently started jobs, newest first.
|
|
|
|
Appends a file-derived "latest scan" entry when the newest report files on
|
|
disk reflect a scan newer than any known job (e.g. after a restart or a
|
|
CLI run), so the dashboard always shows the latest completed scan.
|
|
"""
|
|
load_persisted()
|
|
with _jobs_lock:
|
|
jobs = sorted(_jobs.values(),
|
|
key=lambda j: j.started or "", reverse=True)
|
|
entries = [j.to_dict() for j in jobs]
|
|
file_entry = latest_file_scan()
|
|
if file_entry is not None:
|
|
newest = jobs[0] if jobs else None
|
|
newest_ts = _iso_key(
|
|
(newest.finished or newest.started) if newest else None)
|
|
# Only surface the file-derived entry when it is genuinely newer than
|
|
# every known job (tolerance avoids duplicating the run that produced
|
|
# it, where collected_at lands a moment after the job finished).
|
|
if newest_ts == 0 or _iso_key(file_entry["started"]) - newest_ts > 2.0:
|
|
entries.append(file_entry)
|
|
entries.sort(key=lambda e: _iso_key(e.get("started")), reverse=True)
|
|
return entries[:limit]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _save_jobs() -> None:
|
|
"""Persist the newest jobs to ``SCAN_JOBS_PATH`` (atomic tmp+rename)."""
|
|
with _jobs_lock:
|
|
jobs = sorted(_jobs.values(),
|
|
key=lambda j: j.started or "", reverse=True)
|
|
payload = [j.to_dict() for j in jobs[:SCAN_JOBS_MAX]]
|
|
os.makedirs(os.path.dirname(SCAN_JOBS_PATH), exist_ok=True)
|
|
tmp = SCAN_JOBS_PATH + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f, indent=2, sort_keys=True)
|
|
os.replace(tmp, SCAN_JOBS_PATH)
|
|
|
|
|
|
def _load_jobs() -> None:
|
|
"""Rebuild the in-memory job store from ``SCAN_JOBS_PATH``."""
|
|
try:
|
|
with open(SCAN_JOBS_PATH, "r", encoding="utf-8") as f:
|
|
payload = json.load(f)
|
|
except (OSError, ValueError, TypeError):
|
|
payload = []
|
|
if not isinstance(payload, list):
|
|
return
|
|
|
|
jobs: dict[str, ScanJob] = {}
|
|
for item in payload:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
try:
|
|
params = ScanParams(**item.get("params", {}))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
try:
|
|
job = ScanJob(
|
|
job_id=str(item["job_id"]),
|
|
kind=item.get("kind", "consolidated"),
|
|
params=params,
|
|
state=item.get("state", "done"),
|
|
started=item.get("started"),
|
|
finished=item.get("finished"),
|
|
error=item.get("error"),
|
|
busy=False,
|
|
details=item.get("details", {}) or {},
|
|
)
|
|
except (KeyError, TypeError, ValueError):
|
|
continue
|
|
result = item.get("result")
|
|
if isinstance(result, dict):
|
|
try:
|
|
job.result = ScanResult(**result)
|
|
except TypeError:
|
|
job.result = None
|
|
jobs[job.job_id] = job
|
|
|
|
with _jobs_lock:
|
|
_jobs.clear()
|
|
_jobs.update(jobs)
|
|
|
|
|
|
def load_persisted() -> None:
|
|
"""Load persisted jobs from disk once (idempotent)."""
|
|
global _loaded
|
|
if _loaded:
|
|
return
|
|
_load_jobs()
|
|
_loaded = True
|
|
|
|
|
|
def is_scan_running() -> bool:
|
|
"""True if any job currently holds the scan lock."""
|
|
return _scan_lock.locked()
|