#!/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 .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()