From 9627629a6763f33d8fcba06ce33fdb94687039d3 Mon Sep 17 00:00:00 2001 From: hardcoreerik <130813412+hardcoreerik@users.noreply.github.com> Date: Mon, 11 May 2026 18:40:44 -0700 Subject: [PATCH] Codex: expose the analyzer through a local machine interface --- .gitignore | 3 + README.md | 14 + core/analyze.py | 2 +- core/parser_pcap.py | 47 +- core/writers.py | 44 +- docs/ai-bridge.md | 56 ++ error_logger.py | 105 ++++ gui_step3d_scene.py | 1268 +++++++++++++++++++++++++++++++------------ launch.cmd | 58 ++ project_vault.py | 222 +++++++- run_step3d_scene.py | 23 +- wardrive_cli.py | 150 +++++ wardrive_service.py | 148 +++++ 13 files changed, 1741 insertions(+), 399 deletions(-) create mode 100644 docs/ai-bridge.md create mode 100644 error_logger.py create mode 100644 launch.cmd create mode 100644 wardrive_cli.py create mode 100644 wardrive_service.py diff --git a/.gitignore b/.gitignore index 9e22ffa..f742755 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,9 @@ Projects/ *.db-shm *.db-wal wardrive_run.log +error_reports/ +stdout_app.log +stderr_app.log # Evidence and exports (may contain sensitive location data) Projects/*/evidence/ diff --git a/README.md b/README.md index c821a4b..4c98a7a 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,18 @@ This repository is shared between Codex and Claude Code. Use pull requests as the visible handoff channel. Uncommitted local edits are invisible to the other agent. +## AI Bridge + +Local agents can inspect and operate the project through JSON commands without opening a network service. + +```powershell +python wardrive_cli.py project-summary "F:\WardriveAnalytics" +python wardrive_cli.py latest-run "F:\WardriveAnalytics" +python wardrive_cli.py analyze-project "F:\WardriveAnalytics" --events +``` + +See [docs/ai-bridge.md](docs/ai-bridge.md). + ## Verification Useful checks: @@ -90,6 +102,8 @@ Wardrive Analyzer/ |-- run_step3d_scene.py # App entrypoint |-- gui_step3d_scene.py # PySide6 Mission Control UI |-- project_vault.py # Project vault, SD scan, ingest, dedupe +|-- wardrive_service.py # Shared local service functions for GUI/CLI/agents +|-- wardrive_cli.py # JSON CLI and stdio bridge for local agents |-- core/ | |-- analyze.py # Report orchestration | |-- parser_logs.py # Wardrive log parsing diff --git a/core/analyze.py b/core/analyze.py index 6cac36c..c241695 100644 --- a/core/analyze.py +++ b/core/analyze.py @@ -30,7 +30,7 @@ def _emit_checklist(cb: Optional[Callable[[str], None]], run_dir: str, expected: _emit(cb, "=== Run Output Checklist ===") for rel in expected: p = os.path.join(run_dir, rel) - mark = "✅" if os.path.exists(p) else "❌" + mark = "[OK]" if os.path.exists(p) else "[MISSING]" _emit(cb, f"{mark} {rel}") _emit(cb, "=== End Checklist ===") diff --git a/core/parser_pcap.py b/core/parser_pcap.py index fe9ff4f..665b45c 100644 --- a/core/parser_pcap.py +++ b/core/parser_pcap.py @@ -29,6 +29,28 @@ except ImportError: _DPKT_OK = False +# Per-file packet cap — prevents multi-minute hangs on huge raw captures (e.g. 100MB+ files). +# 802.11 management frames are ~300 bytes each; 2M packets ≈ ~600MB of 802.11 data max. +_MAX_PACKETS_PER_FILE = 2_000_000 + +# Files larger than this (bytes) get a status warning before parsing begins. +_LARGE_FILE_WARN_BYTES = 50 * 1024 * 1024 # 50 MB + +_SSID_MAX_CHARS = 128 + + +def _clean_ssid_text(text: str) -> str: + cleaned = "".join( + ch if (ch.isprintable() and ch not in "\x00\r\n\t") else "?" + for ch in text + ) + cleaned = cleaned.replace("\ufffd", "?").strip() + if not cleaned: + return "" + if len(cleaned) > _SSID_MAX_CHARS: + return cleaned[:_SSID_MAX_CHARS] + "...[truncated]" + return cleaned + # --------------------------------------------------------------------------- # Frame-type constants (802.11) @@ -124,7 +146,7 @@ def _parse_ssid_from_ie(payload: bytes) -> str: if ie_id == 0: # SSID IE raw = payload[i + 2: i + 2 + ie_len] try: - return raw.decode("utf-8", errors="replace").strip() or "" + return _clean_ssid_text(raw.decode("utf-8", errors="replace")) except Exception: return "" i += 2 + ie_len @@ -178,6 +200,10 @@ def _parse_one_pcap(pcap_path: str) -> _PerFileResult: return result try: + file_size = os.path.getsize(pcap_path) + if file_size > _LARGE_FILE_WARN_BYTES: + result.error = f"large file ({file_size // (1024*1024)}MB) — parsing capped at {_MAX_PACKETS_PER_FILE:,} packets" + with open(pcap_path, "rb") as f: try: cap = dpkt.pcap.Reader(f) @@ -189,8 +215,14 @@ def _parse_one_pcap(pcap_path: str) -> _PerFileResult: result.error = str(e) return result + packet_count = 0 for _ts, buf in cap: _process_frame(buf, result) + packet_count += 1 + if packet_count >= _MAX_PACKETS_PER_FILE: + if not result.error: + result.error = f"truncated at {_MAX_PACKETS_PER_FILE:,} packets" + break except Exception as e: result.error = str(e) @@ -378,6 +410,19 @@ def load_pcaps( lock = threading.Lock() completed = [0] + # Warn up front about large files so the user sees them before the parser blocks. + if status_cb: + for p in pcap_files: + try: + sz = os.path.getsize(p) + if sz > _LARGE_FILE_WARN_BYTES: + status_cb( + f" [warn] large file ({sz // (1024*1024)}MB) — " + f"will cap at {_MAX_PACKETS_PER_FILE:,} packets: {os.path.basename(p)}" + ) + except OSError: + pass + def _work(path: str) -> _PerFileResult: return _parse_one_pcap(path) diff --git a/core/writers.py b/core/writers.py index 12f78c3..7ffc8aa 100644 --- a/core/writers.py +++ b/core/writers.py @@ -30,6 +30,34 @@ ) from .parser_logs import Sighting +_EXCEL_ILLEGAL_RE = re.compile(r"[\x00-\x08\x0B-\x0C\x0E-\x1F]") +_PRINTABLE_CONTROL_RE = re.compile(r"[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F-\x9F]") +_EXCEL_MAX_CELL_CHARS = 32767 + + +def clean_report_value(value, max_chars: int = _EXCEL_MAX_CELL_CHARS): + """Return a report-safe scalar for CSV/XLSX/HTML writers.""" + if value is None: + return NO_DATA + if isinstance(value, (int, float, bool)): + return value + + text = str(value) + if not text: + return NO_DATA + + text = _PRINTABLE_CONTROL_RE.sub("", text) + text = _EXCEL_ILLEGAL_RE.sub("", text) + text = text.replace("\ufffd", "?") + text = text.strip() + if len(text) > max_chars: + text = text[: max_chars - 20] + "... [truncated]" + return text or NO_DATA + + +def clean_report_row(row: dict) -> dict: + return {k: clean_report_value(v) for k, v in row.items()} + # --------------------------------------------------------------------------- # OUI lookup @@ -96,7 +124,7 @@ def category_guess(role: str, vendor: str, locally_admin: bool) -> str: def ssid_display(ssid_raw: str) -> str: - s = ssid_raw if ssid_raw not in (None, "", NO_DATA) else "" + s = clean_report_value(ssid_raw, max_chars=512) if ssid_raw not in (None, "", NO_DATA) else "" if not s: return "ASCII: " if re.fullmatch(r"[0-9A-Fa-f]+", s) and len(s) % 2 == 0 and len(s) >= 2: @@ -119,17 +147,19 @@ def ssid_display(ssid_raw: str) -> str: def write_csv(rows: List[dict], outdir: str, filename: str) -> str: path = os.path.join(outdir, filename) headers = list(rows[0].keys()) if rows else ["MAC"] + safe_rows = [clean_report_row(r) for r in rows] if rows else [] with open(path, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, headers) w.writeheader() - if rows: - w.writerows(rows) + if safe_rows: + w.writerows(safe_rows) return path def write_xlsx(rows: List[dict], outdir: str, filename: str) -> Optional[str]: try: from openpyxl import Workbook # type: ignore + from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE # type: ignore except Exception: return None path = os.path.join(outdir, filename) @@ -139,7 +169,13 @@ def write_xlsx(rows: List[dict], outdir: str, filename: str) -> Optional[str]: headers = list(rows[0].keys()) if rows else ["MAC"] ws.append(headers) for r in rows: - ws.append([r.get(h, NO_DATA) for h in headers]) + safe_values = [] + for h in headers: + value = clean_report_value(r.get(h, NO_DATA)) + if isinstance(value, str): + value = ILLEGAL_CHARACTERS_RE.sub("", value) + safe_values.append(value) + ws.append(safe_values) ws.freeze_panes = "A2" wb.save(path) return path diff --git a/docs/ai-bridge.md b/docs/ai-bridge.md new file mode 100644 index 0000000..a723ddf --- /dev/null +++ b/docs/ai-bridge.md @@ -0,0 +1,56 @@ +# AI Bridge + +Wardrive Analyzer exposes a local JSON bridge for Codex, Claude, and other local agents. It does not open a network port. + +## One-Shot CLI + +All commands print JSON to stdout. + +```powershell +python wardrive_cli.py project-summary "F:\WardriveAnalytics" +python wardrive_cli.py evidence-list "F:\WardriveAnalytics" --limit 25 +python wardrive_cli.py runs-list "F:\WardriveAnalytics" --limit 10 +python wardrive_cli.py latest-run "F:\WardriveAnalytics" +python wardrive_cli.py scan-folder "G:\" --limit 50 +python wardrive_cli.py analyze-project "F:\WardriveAnalytics" --events +``` + +`analyze-project` generates a new run folder under the project. Use it only when the operator wants a fresh report. + +## Stdio JSON Bridge + +Agents can keep one process open and send one JSON object per line. + +```powershell +python wardrive_cli.py serve-stdio +``` + +Request format: + +```json +{"id":"1","action":"project_summary","params":{"project":"F:\\WardriveAnalytics"}} +``` + +Response format: + +```json +{"status":"ok","id":"1","project_dir":"F:\\WardriveAnalytics"} +``` + +Supported actions: + +| Action | Params | Mutates | +| --- | --- | --- | +| `project_summary` | `project`, optional `import_limit` | Ensures project vault exists | +| `evidence_list` | `project`, optional `limit`, `offset` | No | +| `runs_list` | `project`, optional `limit` | No | +| `latest_run` | `project` | No | +| `scan_folder` | `folder`, optional `limit`, `include_hidden` | No | +| `analyze_project` | `project` | Yes, creates a new run | + +## Safety + +- The bridge is local-only. +- It returns paths and evidence metadata, not raw capture contents. +- Do not commit generated reports, evidence, local logs, or `error_reports/`. +- Prefer read-only commands before `analyze_project`. diff --git a/error_logger.py b/error_logger.py new file mode 100644 index 0000000..74362ed --- /dev/null +++ b/error_logger.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +import os +import platform +import sys +import threading +import traceback +from datetime import datetime +from pathlib import Path +from typing import Any, Optional + +_CONTROL_RE = __import__("re").compile(r"[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F-\x9F]") +_MAX_TEXT = 12000 + + +def safe_text(value: Any, limit: int = _MAX_TEXT) -> str: + text = str(value) if value is not None else "" + text = _CONTROL_RE.sub("", text).replace("\ufffd", "?") + if len(text) > limit: + return text[:limit] + "\n... [truncated]" + return text + + +def error_reports_dir(base_dir: Optional[str] = None) -> Path: + root = Path(base_dir or os.getcwd()) + path = root / "error_reports" + path.mkdir(parents=True, exist_ok=True) + return path + + +def _tail_file(path: Path, limit: int = 20000) -> str: + try: + if not path.exists() or path.stat().st_size <= 0: + return "" + with path.open("rb") as f: + if path.stat().st_size > limit: + f.seek(-limit, os.SEEK_END) + data = f.read() + return safe_text(data.decode("utf-8", errors="replace"), limit=limit) + except Exception as exc: + return f"(failed to read {path.name}: {exc})" + + +def write_error_report( + title: str, + exc: BaseException | None = None, + *, + context: Optional[dict[str, Any]] = None, + traceback_text: str = "", + base_dir: Optional[str] = None, +) -> str: + stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + safe_title = "".join(ch if ch.isalnum() else "_" for ch in title.lower()).strip("_")[:60] or "error" + path = error_reports_dir(base_dir) / f"{stamp}_{safe_title}.json" + + if exc is not None and not traceback_text: + traceback_text = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)) + + payload = { + "title": safe_text(title, 500), + "timestamp": datetime.now().isoformat(timespec="seconds"), + "cwd": os.getcwd(), + "executable": sys.executable, + "python": sys.version, + "platform": platform.platform(), + "thread": threading.current_thread().name, + "exception_type": type(exc).__name__ if exc else "", + "exception": safe_text(exc, 2000) if exc else "", + "traceback": safe_text(traceback_text), + "context": {str(k): safe_text(v, 3000) for k, v in (context or {}).items()}, + "wardrive_run_log_tail": _tail_file(Path(os.getcwd()) / "wardrive_run.log"), + } + + path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + return str(path) + + +def install_global_error_hooks(base_dir: Optional[str] = None) -> None: + def excepthook(exc_type, exc, tb): + report = write_error_report( + "unhandled_exception", + exc, + traceback_text="".join(traceback.format_exception(exc_type, exc, tb)), + base_dir=base_dir, + ) + print(f"Unhandled exception report: {report}", file=sys.stderr) + sys.__excepthook__(exc_type, exc, tb) + + sys.excepthook = excepthook + + if hasattr(threading, "excepthook"): + def thread_hook(args): + report = write_error_report( + "thread_exception", + args.exc_value, + traceback_text="".join(traceback.format_exception(args.exc_type, args.exc_value, args.exc_traceback)), + context={"thread_name": args.thread.name if args.thread else ""}, + base_dir=base_dir, + ) + print(f"Thread exception report: {report}", file=sys.stderr) + if getattr(threading, "__excepthook__", None): + threading.__excepthook__(args) + + threading.excepthook = thread_hook diff --git a/gui_step3d_scene.py b/gui_step3d_scene.py index 23aadca..aa9507c 100644 --- a/gui_step3d_scene.py +++ b/gui_step3d_scene.py @@ -2,6 +2,7 @@ import os import sys +import platform import random import subprocess import hashlib @@ -9,6 +10,7 @@ import re from dataclasses import dataclass from datetime import datetime +from error_logger import write_error_report from PySide6.QtCore import Qt, QThread, Signal, QTimer, QElapsedTimer, QRect, QRectF, QPointF, QUrl from PySide6.QtGui import QPainter, QLinearGradient, QColor, QFont, QPen, QBrush, QPixmap, QDesktopServices, QMovie @@ -99,235 +101,446 @@ def file_fingerprint(path: str) -> str: # ----------------------------- -# Procedural cyberpunk background (Qt painter) +# Procedural animated background (QPainter, full-window, per-frame) # ----------------------------- +import math + +_MATRIX_CHARS = ( + "ｦｧｨｩｪｫｬｭｮｯｰｱｲｳｴｵｶｷｸｹｺｻｼｽｾｿﾀﾁﾂﾃﾄﾅﾆﾇﾈﾉﾊﾋﾌﾍﾎﾏﾐﾑﾒﾓﾔﾕﾖﾗﾘﾙﾚﾛﾜﾝ" + "0123456789ABCDEF<>[]{}|\\/*+-=~@#$%^&" +) + + @dataclass -class BgSeed: - a: int - b: int - c: int - mode: str +class _MatrixDrop: + x: float + y: float # head y position (pixels) + speed: float # pixels per tick + trail: int # number of chars in trail + chars: list # pre-chosen char indices, len == trail + 1 + +@dataclass +class _CircuitNode: + x: float + y: float + phase_offset: float # per-node brightness phase + + +@dataclass +class _CircuitTrace: + n1: int # index into nodes list + n2: int + signal_t: float # 0..1, position of traveling signal along trace + signal_speed: float # units per tick (0.003 – 0.012) -class CyberBackground(QWidget): - MODES = ["pixel_demo_gif", "grid_horizon", "circuit_board", "matrix_rain", "nebula_noise"] + +@dataclass +class _NebulaBlob: + cx: float + cy: float + rx: float + ry: float + hue: int + vx: float # pixels per tick + vy: float + phase_offset: float + + +class ProceduralBackground(QWidget): + """Full-window animated background — pure QPainter, no static pixmap, no GIFs.""" + + MODES = ["grid_horizon", "circuit_board", "matrix_rain", "nebula_noise"] def __init__(self, parent: QWidget | None = None): super().__init__(parent) self.setAttribute(Qt.WA_TransparentForMouseEvents, True) self.setAutoFillBackground(False) - # Pixel demoscene GIF layer (behind UI) - self.gif_label = QLabel(self) - self.gif_label.setScaledContents(True) - self.gif_label.setAttribute(Qt.WA_TransparentForMouseEvents, True) - self.gif_label.hide() + # GIF fallback layer (shown only if gif_paths supplied and mode == gif) + self._gif_label = QLabel(self) + self._gif_label.setAttribute(Qt.WA_TransparentForMouseEvents, True) + self._gif_label.hide() self._movie: QMovie | None = None - self._gif_choice: str | None = None - self.seed = BgSeed(0, 0, 0, "grid_horizon") - self.phase = 0.0 - self._static = QPixmap() - self._drops: list[tuple[float, float, float]] = [] - - self.regenerate(initial=True) - - self._anim = QTimer(self) - self._anim.setInterval(16) - self._anim.timeout.connect(self._tick) - self._anim.start() - - def regenerate(self, initial: bool = False): - self.seed = BgSeed( - random.randint(0, 10_000_000), - random.randint(0, 10_000_000), - random.randint(0, 10_000_000), - random.choices(self.MODES, weights=[6,1,1,1,1], k=1)[0], - ) + self.mode: str = "grid_horizon" + self.phase: float = 0.0 + self.current_name: str = "grid_horizon" + + # Per-mode state (rebuilt on regenerate / resize) + self._drops: list[_MatrixDrop] = [] + self._circuit_nodes: list[_CircuitNode] = [] + self._circuit_traces: list[_CircuitTrace] = [] + self._nebula_blobs: list[_NebulaBlob] = [] + self._rng_seed: int = random.randint(0, 10_000_000) + self._hue_base: int = random.randint(0, 359) + self._last_w: int = 0 + self._last_h: int = 0 + + self._timer = QTimer(self) + self._timer.setInterval(16) # ~60 fps + self._timer.timeout.connect(self._tick) + self._timer.start() + + self.randomize() + + # ------------------------------------------------------------------ + # Public API (matches old DemoscenePanel for drop-in replacement) + # ------------------------------------------------------------------ + + def randomize(self) -> None: + self.set_mode(random.choice(self.MODES)) + + def set_mode(self, mode: str) -> None: + self.mode = mode + self.current_name = mode self.phase = 0.0 - self._static = QPixmap() + self._rng_seed = random.randint(0, 10_000_000) + self._hue_base = random.randint(0, 359) + self._rebuild_state() + self.update() + + # ------------------------------------------------------------------ + # State initialisation (called on set_mode + resize) + # ------------------------------------------------------------------ + + def _rebuild_state(self) -> None: + w, h = self.width(), self.height() + self._last_w, self._last_h = w, h + rng = random.Random(self._rng_seed) + + if self.mode == "matrix_rain": + self._build_matrix(w, h, rng) + elif self.mode == "circuit_board": + self._build_circuit(w, h, rng) + elif self.mode == "nebula_noise": + self._build_nebula(w, h, rng) + # grid_horizon: fully stateless, driven only by phase + + def _build_matrix(self, w: int, h: int, rng: random.Random) -> None: + if w <= 0 or h <= 0: + return + col_w = 18 + cols = max(1, w // col_w) self._drops = [] - self._apply_mode() - if not initial: - self.update() - - def _apply_mode(self): - # If pixel_demo_gif: load one of our baked GIFs and let QMovie animate it. - if self.seed.mode == "pixel_demo_gif": - choice = random.choice(ASSET_GIFS) - if choice != self._gif_choice: - self._gif_choice = choice - try: - path = resource_path(choice) - mv = QMovie(path) - self._movie = mv - self.gif_label.setMovie(mv) - mv.start() - except Exception: - self._movie = None - self.gif_label.show() - else: - self.gif_label.hide() - if self._movie: - try: - self._movie.stop() - except Exception: - pass - self._movie = None - self._gif_choice = None + for i in range(cols): + trail = rng.randint(8, 28) + self._drops.append(_MatrixDrop( + x=i * col_w + rng.randint(0, col_w // 2), + y=rng.random() * h, + speed=rng.uniform(1.5, 5.0), + trail=trail, + chars=[rng.randint(0, len(_MATRIX_CHARS) - 1) for _ in range(trail + 1)], + )) + + def _build_circuit(self, w: int, h: int, rng: random.Random) -> None: + if w <= 0 or h <= 0: + return + n_nodes = 60 + self._circuit_nodes = [ + _CircuitNode( + x=rng.random() * w, + y=rng.random() * h, + phase_offset=rng.random() * math.tau, + ) + for _ in range(n_nodes) + ] + # Connect each node to 1-3 neighbours picked by proximity + self._circuit_traces = [] + for i, nd in enumerate(self._circuit_nodes): + candidates = sorted( + (j for j in range(n_nodes) if j != i), + key=lambda j: (self._circuit_nodes[j].x - nd.x) ** 2 + (self._circuit_nodes[j].y - nd.y) ** 2, + ) + for j in candidates[: rng.randint(1, 3)]: + self._circuit_traces.append(_CircuitTrace( + n1=i, n2=j, + signal_t=rng.random(), + signal_speed=rng.uniform(0.002, 0.010), + )) + + def _build_nebula(self, w: int, h: int, rng: random.Random) -> None: + if w <= 0 or h <= 0: + return + self._nebula_blobs = [ + _NebulaBlob( + cx=rng.random() * w, + cy=rng.random() * h, + rx=(0.15 + rng.random() * 0.55) * w, + ry=(0.10 + rng.random() * 0.45) * h, + hue=rng.randint(0, 359), + vx=rng.uniform(-0.25, 0.25), + vy=rng.uniform(-0.15, 0.15), + phase_offset=rng.random() * math.tau, + ) + for _ in range(12) + ] - def resizeEvent(self, event): - super().resizeEvent(event) - self.gif_label.setGeometry(0, 0, self.width(), self.height()) + # ------------------------------------------------------------------ + # Animation tick + # ------------------------------------------------------------------ - def _tick(self): - self.phase += 0.04 - if self.phase > 9999: + def _tick(self) -> None: + self.phase += 0.05 + if self.phase > 99999: self.phase = 0.0 + + w, h = self.width(), self.height() + if w != self._last_w or h != self._last_h: + self._rebuild_state() + + if self.mode == "matrix_rain": + self._step_matrix(h) + elif self.mode == "circuit_board": + self._step_circuit() + elif self.mode == "nebula_noise": + self._step_nebula(w, h) + self.update() - def _ensure_static(self): + def _step_matrix(self, h: int) -> None: + rng = random.Random() + for drop in self._drops: + drop.y += drop.speed + if drop.y - drop.trail * 18 > h + 40: + drop.y = -rng.randint(10, 80) + drop.speed = rng.uniform(1.5, 5.0) + drop.trail = rng.randint(8, 28) + drop.chars = [rng.randint(0, len(_MATRIX_CHARS) - 1) for _ in range(drop.trail + 1)] + # Randomly mutate head char for flicker + if rng.random() < 0.15: + drop.chars[0] = rng.randint(0, len(_MATRIX_CHARS) - 1) + + def _step_circuit(self) -> None: + for trace in self._circuit_traces: + trace.signal_t += trace.signal_speed + if trace.signal_t > 1.0: + trace.signal_t = 0.0 + + def _step_nebula(self, w: int, h: int) -> None: + margin = 200 + for b in self._nebula_blobs: + b.cx += b.vx + b.cy += b.vy + if b.cx < -margin: + b.cx = w + margin + elif b.cx > w + margin: + b.cx = -margin + if b.cy < -margin: + b.cy = h + margin + elif b.cy > h + margin: + b.cy = -margin + + # ------------------------------------------------------------------ + # Painting + # ------------------------------------------------------------------ + + def resizeEvent(self, event): + super().resizeEvent(event) + self._gif_label.setGeometry(0, 0, self.width(), self.height()) + + def paintEvent(self, event): w, h = self.width(), self.height() if w <= 2 or h <= 2: return - if not self._static.isNull() and self._static.size().width() == w and self._static.size().height() == h: - return - pm = QPixmap(w, h) - pm.fill(Qt.transparent) - p = QPainter(pm) + p = QPainter(self) p.setRenderHint(QPainter.Antialiasing, True) - g = QLinearGradient(0, 0, w, h) - hue1 = (self.seed.a % 360) - hue2 = (self.seed.b % 360) - c1 = QColor.fromHsl(hue1, 180, 25) - c2 = QColor.fromHsl(hue2, 200, 12) - g.setColorAt(0.0, QColor(3, 5, 6)) - g.setColorAt(0.35, c2) - g.setColorAt(0.75, c1) - g.setColorAt(1.0, QColor(0, 0, 0)) - p.fillRect(pm.rect(), g) - - rng = random.Random(self.seed.c) - - if self.seed.mode == "grid_horizon": - self._draw_horizon_grid(p, w, h, rng) - elif self.seed.mode == "circuit_board": - self._draw_circuit(p, w, h, rng) - elif self.seed.mode == "matrix_rain": - p.fillRect(pm.rect(), QColor(0, 0, 0, 60)) - elif self.seed.mode == "nebula_noise": - self._draw_nebula(p, w, h, rng) - - vign = QLinearGradient(w * 0.5, h * 0.5, w * 0.5, h) - vign.setColorAt(0.0, QColor(0, 0, 0, 0)) - vign.setColorAt(1.0, QColor(0, 0, 0, 190)) - p.fillRect(pm.rect(), vign) + # Base gradient — deep black/midnight + bg = QLinearGradient(0, 0, 0, h) + bg.setColorAt(0.0, QColor(2, 4, 8)) + bg.setColorAt(1.0, QColor(0, 0, 0)) + p.fillRect(0, 0, w, h, bg) + + if self.mode == "matrix_rain": + self._paint_matrix(p, w, h) + elif self.mode == "circuit_board": + self._paint_circuit(p, w, h) + elif self.mode == "nebula_noise": + self._paint_nebula(p, w, h) + elif self.mode == "grid_horizon": + self._paint_horizon(p, w, h) + + # Scanline overlay (subtle) + p.setRenderHint(QPainter.Antialiasing, False) + for y in range(0, h, 3): + p.fillRect(0, y, w, 1, QColor(0, 0, 0, 18)) - p.setPen(QPen(QColor(0, 255, 204, 35), 1)) - font = QFont("Consolas") - font.setPointSize(9) - p.setFont(font) - glyphs = ["0xDEAD", "0xBEEF", "WIGLE", "PCAP", "RSSI", "BSSID", "SCAN", "NODE", "RUN"] - for _ in range(18): - txt = rng.choice(glyphs) - x = rng.randint(0, max(1, w - 60)) - y = rng.randint(20, max(21, h - 10)) - p.drawText(QPointF(x, y), txt) + # Bottom vignette + vign = QLinearGradient(0, h * 0.6, 0, h) + vign.setColorAt(0.0, QColor(0, 0, 0, 0)) + vign.setColorAt(1.0, QColor(0, 0, 0, 180)) + p.fillRect(0, 0, w, h, vign) p.end() - self._static = pm - - if self.seed.mode == "matrix_rain": - cols = max(10, w // 20) - self._drops = [(i * (w / cols), rng.random() * h, 30 + rng.random() * 120) for i in range(cols)] - - def _draw_horizon_grid(self, p: QPainter, w: int, h: int, rng: random.Random): - for _ in range(6): - x = rng.random() * w - y = rng.random() * h - rw = (0.25 + rng.random() * 0.6) * w - rh = (0.15 + rng.random() * 0.5) * h - hue = rng.randint(0, 359) - col = QColor.fromHsl(hue, 220, 55, 28) - p.setBrush(QBrush(col)) - p.setPen(Qt.NoPen) - p.drawEllipse(QRectF(x - rw / 2, y - rh / 2, rw, rh)) + # --- Matrix rain --- + def _paint_matrix(self, p: QPainter, w: int, h: int) -> None: p.setRenderHint(QPainter.Antialiasing, False) - grid_y = int(h * 0.70) - van_x = w * 0.5 - van_y = h * 0.55 - for i in range(-18, 19): - x = van_x + i * (w * 0.06) - p.setPen(QPen(QColor(0, 255, 170, 60), 1)) - p.drawLine(int(x), grid_y, int(van_x), int(van_y)) - for i in range(1, 26): - y = grid_y + int((i ** 1.35) * (h * 0.008)) - if y >= h: - break - alpha = max(10, 70 - i * 2) - p.setPen(QPen(QColor(0, 255, 170, alpha), 1)) - p.drawLine(0, y, w, y) + font = QFont("Consolas", 12) + p.setFont(font) + fm_h = 16 # approximate char cell height + + for drop in self._drops: + head_y = drop.y + for i, ci in enumerate(drop.chars): + char_y = head_y - i * fm_h + if char_y < -fm_h or char_y > h + fm_h: + continue + ch = _MATRIX_CHARS[ci] + if i == 0: + # Head: bright white-green + p.setPen(QColor(200, 255, 200, 255)) + else: + # Trail: exponential fade + fade = int(255 * (0.85 ** i)) + green = max(80, 255 - i * 12) + p.setPen(QColor(0, green, 50, fade)) + p.drawText(QPointF(drop.x, char_y), ch) + + # --- Circuit board --- + def _paint_circuit(self, p: QPainter, w: int, h: int) -> None: + nodes = self._circuit_nodes + if not nodes: + return - def _draw_circuit(self, p: QPainter, w: int, h: int, rng: random.Random): + # Draw traces p.setRenderHint(QPainter.Antialiasing, False) - for _ in range(120): - x1 = rng.randint(0, w) - y1 = rng.randint(0, h) - x2 = x1 + rng.randint(-120, 120) - y2 = y1 + rng.randint(-120, 120) - col = QColor(0, 255, 208, rng.randint(15, 60)) - p.setPen(QPen(col, 1)) - p.drawLine(x1, y1, x2, y1) - p.drawLine(x2, y1, x2, y2) - p.fillRect(max(0, x2 - 2), max(0, y2 - 2), 4, 4, QColor(0, 255, 208, rng.randint(30, 80))) - - def _draw_nebula(self, p: QPainter, w: int, h: int, rng: random.Random): + for trace in self._circuit_traces: + n1 = nodes[trace.n1] + n2 = nodes[trace.n2] + p.setPen(QPen(QColor(0, 200, 160, 35), 1)) + # L-shaped trace (one horizontal + one vertical segment) + mid_x = n2.x + mid_y = n1.y + p.drawLine(int(n1.x), int(n1.y), int(mid_x), int(mid_y)) + p.drawLine(int(mid_x), int(mid_y), int(n2.x), int(n2.y)) + + # Traveling signal dot + t = trace.signal_t + if t < 0.5: + sx = n1.x + (mid_x - n1.x) * (t * 2) + sy = n1.y + else: + sx = mid_x + sy = mid_y + (n2.y - mid_y) * ((t - 0.5) * 2) + sig_alpha = int(180 + 75 * math.sin(self.phase * 3)) + p.fillRect(int(sx) - 2, int(sy) - 2, 4, 4, QColor(0, 255, 200, sig_alpha)) + + # Draw nodes with pulsing brightness p.setRenderHint(QPainter.Antialiasing, True) - for _ in range(10): - x = rng.random() * w - y = rng.random() * h - rw = (0.2 + rng.random() * 0.8) * w - rh = (0.2 + rng.random() * 0.7) * h - hue = rng.randint(0, 359) - col = QColor.fromHsl(hue, 180, 40, 22) - p.setBrush(QBrush(col)) + for nd in nodes: + pulse = 0.5 + 0.5 * math.sin(self.phase * 2.5 + nd.phase_offset) + alpha = int(60 + 140 * pulse) + size = 3 + int(3 * pulse) + p.setBrush(QBrush(QColor(0, 255, 190, alpha))) p.setPen(Qt.NoPen) - p.drawEllipse(QRectF(x - rw / 2, y - rh / 2, rw, rh)) + p.drawEllipse(QPointF(nd.x, nd.y), size, size) - def paintEvent(self, event): - if self.seed.mode == "pixel_demo_gif": - # GIF handled by QLabel/QMovie. - return - self._ensure_static() - painter = QPainter(self) - painter.drawPixmap(0, 0, self._static) - - w, h = self.width(), self.height() - - painter.setRenderHint(QPainter.Antialiasing, False) - for y in range(0, h, 4): - a = 10 if (int(self.phase * 10) + y) % 12 else 20 - painter.fillRect(0, y, w, 1, QColor(0, 255, 150, a)) + # --- Nebula noise --- + def _paint_nebula(self, p: QPainter, w: int, h: int) -> None: + p.setRenderHint(QPainter.Antialiasing, True) + for b in self._nebula_blobs: + pulse = 0.5 + 0.5 * math.sin(self.phase * 1.2 + b.phase_offset) + alpha = int(12 + 18 * pulse) + hue = (b.hue + int(self.phase * 8)) % 360 + col = QColor.fromHsl(hue, 200, 55, alpha) + p.setBrush(QBrush(col)) + p.setPen(Qt.NoPen) + p.drawEllipse(QRectF(b.cx - b.rx, b.cy - b.ry, b.rx * 2, b.ry * 2)) - if self.seed.mode == "matrix_rain" and self._drops: - painter.setPen(QPen(QColor(0, 255, 120, 90), 1)) - font = QFont("Consolas") - font.setPointSize(10) - painter.setFont(font) - for i, (x, y, spd) in enumerate(self._drops): - y2 = (y + self.phase * spd) % (h + 100) - txt = chr(33 + ((int(x + y2) + i) % 60)) - painter.drawText(QPointF(x, y2), txt) + # Faint star field + p.setRenderHint(QPainter.Antialiasing, False) + rng = random.Random(self._rng_seed ^ 0xABCD) + for _ in range(80): + sx = rng.randint(0, w) + sy = rng.randint(0, h) + twinkle = int(60 + 60 * math.sin(self.phase * 2.1 + sx * 0.07 + sy * 0.05)) + p.fillRect(sx, sy, 1, 1, QColor(255, 255, 255, twinkle)) + + # --- Synthwave horizon grid --- + def _paint_horizon(self, p: QPainter, w: int, h: int) -> None: + p.setRenderHint(QPainter.Antialiasing, True) - painter.end() + # Dusk sky gradient (magenta → deep blue) + sky = QLinearGradient(0, 0, 0, h * 0.60) + sky.setColorAt(0.0, QColor(8, 2, 22)) + sky.setColorAt(0.6, QColor(80, 10, 60)) + sky.setColorAt(1.0, QColor(180, 30, 100)) + p.fillRect(0, 0, w, int(h * 0.60), sky) + + # Animated sun glow at horizon + sun_cx = w * 0.5 + sun_cy = h * 0.55 + sun_r = w * 0.18 + pulse = 0.5 + 0.5 * math.sin(self.phase * 0.6) + glow_r = sun_r * (1.0 + 0.15 * pulse) + glow = QLinearGradient(sun_cx - glow_r, sun_cy - glow_r, sun_cx + glow_r, sun_cy + glow_r) + glow.setColorAt(0.0, QColor(255, 100, 30, 90)) + glow.setColorAt(0.5, QColor(255, 60, 80, 45)) + glow.setColorAt(1.0, QColor(0, 0, 0, 0)) + p.setBrush(QBrush(glow)) + p.setPen(Qt.NoPen) + p.drawEllipse(QPointF(sun_cx, sun_cy), glow_r, glow_r * 0.65) + + # Sun disc with horizontal scan lines (retro style) + p.setBrush(QBrush(QColor(255, 120, 40, 220))) + p.drawEllipse(QPointF(sun_cx, sun_cy), sun_r, sun_r * 0.55) + p.setRenderHint(QPainter.Antialiasing, False) + p.setClipRect(QRect(int(sun_cx - sun_r), int(sun_cy - sun_r * 0.55), + int(sun_r * 2), int(sun_r * 1.1))) + for i in range(1, 10): + y = sun_cy + (sun_r * 0.55) * (i / 10.0) * 2 - sun_r * 0.55 + a = max(0, 200 - i * 18) + p.fillRect(0, int(y), w, 3, QColor(20, 0, 10, a)) + p.setClipping(False) + + # Floor + floor_top = int(h * 0.55) + floor = QLinearGradient(0, floor_top, 0, h) + floor.setColorAt(0.0, QColor(30, 0, 50)) + floor.setColorAt(1.0, QColor(5, 0, 15)) + p.fillRect(0, floor_top, w, h - floor_top, floor) + + # Vertical perspective lines (fan out from vanishing point) + van_x = w * 0.5 + van_y = h * 0.55 + p.setRenderHint(QPainter.Antialiasing, True) + for i in range(-14, 15): + x_bottom = van_x + i * (w * 0.075) + alpha = max(20, 90 - abs(i) * 4) + p.setPen(QPen(QColor(255, 0, 200, alpha), 1)) + p.drawLine(int(van_x), int(van_y), int(x_bottom), h) + + # Horizontal lines scrolling toward viewer (perspective spacing) + scroll = (self.phase * 12) % 1.0 # fractional scroll offset + n_lines = 24 + for i in range(n_lines): + t = ((i + scroll) / n_lines) ** 2.2 # perspective warp + y = int(van_y + (h - van_y) * t) + if y <= floor_top or y > h: + continue + alpha = int(15 + 65 * t) + p.setPen(QPen(QColor(255, 0, 200, alpha), 1)) + p.drawLine(0, y, w, y) # ----------------------------- # Analyzer worker thread # ----------------------------- +_GUI_TEXT_CONTROL_RE = re.compile(r"[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F-\x9F]") + + +def _safe_gui_text(value, limit: int = 4000) -> str: + text = str(value) if value is not None else "" + text = _GUI_TEXT_CONTROL_RE.sub("", text) + text = text.replace("\ufffd", "?") + if len(text) > limit: + text = text[:limit] + "\n... [truncated]" + return text + + class AnalyzeWorker(QThread): stage = Signal(str) done = Signal(dict) @@ -345,6 +558,7 @@ def __init__(self, wardrive_files: list[str], pcap_files: list[str], project_dir def run(self): try: def cb(msg: str): + msg = _safe_gui_text(msg, limit=1200) self.stage.emit(msg) # Heuristic progress extraction from core log lines. @@ -369,87 +583,103 @@ def cb(msg: str): self.stage.emit("Complete.") self.done.emit(results) except Exception as e: - self.failed.emit(str(e)) + import traceback as _tb + err = _safe_gui_text(f"{type(e).__name__}: {e}", limit=1200) + trace = _safe_gui_text(_tb.format_exc(), limit=4000) + report = write_error_report( + "analysis_worker_failure", + e, + context={ + "project_dir": self.project_dir, + "wardrive_files": len(self.wardrive_files), + "pcap_files": len(self.pcap_files), + }, + traceback_text=trace, + ) + self.stage.emit(f"[CRASH] {err}") + self.stage.emit(f"Error report written: {report}") + self.stage.emit(trace) + self.failed.emit(f"{err}\n\nReport: {report}") # ----------------------------- -# Main GUI +# Ingest worker thread # ----------------------------- +class ScanWorker(QThread): + """Runs scan_sd_folder on a background thread so the GUI stays live.""" + done = Signal(list) # list[CandidateFile] + failed = Signal(str) -class DemoscenePanel(QWidget): - """Right-side pixel demoscene GIF panel. + def __init__(self, folder: str): + super().__init__() + self.folder = folder - Renders the *right side* of the GIF as the focal point (scaled-to-fill, cropped, anchored right). - """ + def run(self): + try: + candidates = scan_sd_folder(self.folder) + self.done.emit(candidates) + except Exception as e: + self.failed.emit(str(e)) - def __init__(self, parent: QWidget | None = None, gif_paths: list[str] | None = None): - super().__init__(parent) - self.setObjectName("DemoscenePanel") - self._gif_paths = gif_paths or ASSET_GIFS - self._movie: QMovie | None = None - self.current_name = "(none)" - self._start_random() - def _start_random(self) -> None: - if not self._gif_paths: - return - path = random.choice(self._gif_paths) - self.set_gif(path) +class IngestWorker(QThread): + stage = Signal(str) # log line (throttled) + progress = Signal(int, int) # (files_done, files_total) + file_progress = Signal(str, str, int, int) # (filename, phase, bytes_done, bytes_total) + done = Signal(dict) + failed = Signal(str) - def randomize(self) -> None: - self._start_random() - - def set_gif(self, path: str) -> None: - abspath = resource_path(path) - if not os.path.exists(abspath): - # Try relative to current working dir as a fallback - if os.path.exists(path): - abspath = path - else: - return - if self._movie: - self._movie.stop() - self._movie = QMovie(abspath) - self._movie.setCacheMode(QMovie.CacheMode.CacheAll) - self._movie.frameChanged.connect(lambda _=None: self.update()) - self._movie.start() - self.current_name = os.path.basename(path) - - def paintEvent(self, event): # noqa: N802 - super().paintEvent(event) - if not self._movie: - return - pix = self._movie.currentPixmap() - if pix.isNull(): - return + # Emit a stage signal at most every N files to avoid flooding the event loop + _SIGNAL_EVERY = 5 - w = self.width() - h = self.height() - if w <= 0 or h <= 0: - return + def __init__(self, project_dir: str, sd_root: str, + candidates: list, label: str, skip_duplicates: bool): + super().__init__() + self.project_dir = project_dir + self.sd_root = sd_root + self.candidates = candidates + self.label = label + self.skip_duplicates = skip_duplicates + self._total = len(candidates) + self._done_count = 0 - # Scale-to-fill while preserving aspect ratio - pw = pix.width() - ph = pix.height() - if pw <= 0 or ph <= 0: - return + def run(self): + try: + every = max(1, self._SIGNAL_EVERY) - sx = w / pw - sy = h / ph - s = max(sx, sy) - sw = int(pw * s) - sh = int(ph * s) + def cb(msg: str): + self._done_count += 1 + is_notable = ( + msg.startswith("ERROR") + or msg.startswith("Duplicate") + or self._done_count % every == 0 + or self._done_count == self._total + ) + if is_notable: + self.stage.emit(msg) + self.progress.emit(self._done_count, self._total) - # Anchor RIGHT: place scaled pix so its right edge aligns with widget right edge - x = w - sw - y = (h - sh) // 2 + def file_cb(filename: str, phase: str, done: int, total: int): + self.file_progress.emit(filename, phase, done, total) - painter = QPainter(self) - painter.setRenderHint(QPainter.RenderHint.SmoothPixmapTransform, True) - painter.drawPixmap(QRect(x, y, sw, sh), pix, QRect(0, 0, pw, ph)) - painter.end() + stats = ingest_candidates_to_project( + project_dir=self.project_dir, + sd_root=self.sd_root, + candidates=self.candidates, + label=self.label, + skip_duplicates=self.skip_duplicates, + progress_cb=cb, + file_progress_cb=file_cb, + ) + self.done.emit(stats) + except Exception as e: + self.failed.emit(str(e)) +# ----------------------------- +# Main GUI +# ----------------------------- + class WardriveGUI(QWidget): def __init__(self): super().__init__() @@ -475,6 +705,13 @@ def __init__(self): self._jobs: list[dict] = [] self._active_job_id: int | None = None self._analysis_thread = None + self._ingest_worker: IngestWorker | None = None + self._ingest_job_id: int | None = None + self._ingest_total: int = 0 + self._ingest_file_active: bool = False + self._scan_worker: ScanWorker | None = None + self._scan_job_id: int | None = None + self._sd_scan_folder: str = "" # Run-state + progress / ETA tracking self._running: bool = False @@ -487,8 +724,8 @@ def __init__(self): # QSS first (so widgets created later inherit it) self._load_qss() - # Full-window background layer (pixel demoscene) - self.demoscene = DemoscenePanel(gif_paths=ASSET_GIFS) + # Full-window animated procedural background + self.demoscene = ProceduralBackground() self.demoscene.setObjectName("DemosceneBG") # Overlay layer (all interactive UI) @@ -498,7 +735,7 @@ def __init__(self): overlay.setStyleSheet( "QWidget#Overlay{background: transparent;}" - "QWidget#LeftPane{background-color: rgba(0,0,0,140); border: 1px solid rgba(0,255,220,80); border-radius: 14px;}" + "QWidget#LeftPane{background-color: rgba(0,0,0,110); border: 1px solid rgba(0,255,220,90); border-radius: 14px;}" "QWidget#RightPane{background: transparent;}" ) @@ -554,7 +791,11 @@ def __init__(self): except Exception: pass self.refresh_mission_control() - self._log(f"Demoscene: {self.demoscene.current_name}") + self._log(f"Background: {self.demoscene.current_name}") + + # Fire boot sequence after event loop starts (100ms delay) + QTimer.singleShot(100, self._boot_sequence) + def _load_qss(self) -> None: """Load and apply the QSS theme (if present). @@ -831,22 +1072,95 @@ def _build_console_tab(self) -> QWidget: panel.setObjectName("Panel") v = QVBoxLayout(panel) v.setContentsMargins(12, 12, 12, 12) - v.setSpacing(8) + v.setSpacing(6) - title = QLabel("Console / Operator Log") + hdr = QHBoxLayout() + title = QLabel("CONSOLE // OPERATOR LOG") title.setObjectName("PanelTitle") - v.addWidget(title) + hdr.addWidget(title, 1) + + btn_clear = QPushButton("Clear") + btn_clear.setFixedWidth(70) + btn_clear.clicked.connect(self._console_clear) + hdr.addWidget(btn_clear, 0) + v.addLayout(hdr) - # Big in-tab console (separate from the always-visible docked console) self.console_big = QTextEdit() self.console_big.setReadOnly(True) self.console_big.setObjectName("ConsoleBig") - self.console_big.setPlaceholderText( - "Output appears here while things run.\n" - "(If it looks idle, check STATUS + progress bar — no fake loading spinners in this basement.)" + # Tight monospace, no paragraph margins between lines + self.console_big.setStyleSheet( + "QTextEdit#ConsoleBig {" + " background-color: rgba(0,0,0,180);" + " color: #C8FFFA;" + " font-family: Consolas, 'Courier New', monospace;" + " font-size: 9pt;" + " border: 1px solid rgba(0,255,220,80);" + " border-radius: 10px;" + "}" + ) + self.console_big.document().setDefaultStyleSheet( + "p { margin:0; padding:0; line-height: 1.25; }" ) v.addWidget(self.console_big, 1) return panel + + def _console_clear(self): + for attr in ("console_dock", "console_big"): + w = getattr(self, attr, None) + if w is None: + continue + try: + w.clear() + except Exception: + pass + self._log("Console cleared.", "BOOT") + + def _boot_sequence(self): + """Emit a styled boot banner to the console after the UI is ready.""" + divider = "─" * 56 + + # Check dpkt + try: + import dpkt as _dpkt + dpkt_ver = getattr(_dpkt, "__version__", "ok") + dpkt_status = f"dpkt {dpkt_ver} — native 802.11 parser ACTIVE" + dpkt_ok = True + except ImportError: + dpkt_status = "dpkt NOT FOUND — PCAP parsing disabled" + dpkt_ok = False + + # Check openpyxl + try: + import openpyxl as _xl + xlsx_status = f"openpyxl {getattr(_xl, '__version__','ok')} — XLSX export ACTIVE" + except ImportError: + xlsx_status = "openpyxl NOT FOUND — XLSX export disabled" + + # Check leaflet assets + leaflet_path = resource_path("assets", "leaflet", "leaflet.js") + leaflet_status = "Leaflet offline assets OK" if os.path.exists(leaflet_path) else "Leaflet offline assets MISSING (CDN fallback)" + + seq = [ + (0, "BOOT", divider), + (40, "BOOT", " WARDRIVE MISSION CONTROL"), + (80, "BOOT", f" Core: {CORE_REVISION} | Python {sys.version.split()[0]}"), + (120, "BOOT", f" OS: {platform.system()} {platform.release()} ({platform.machine()})"), + (160, "BOOT", divider), + (220, "OK" if dpkt_ok else "ERROR", f" {dpkt_status}"), + (270, "OK", f" {xlsx_status}"), + (320, "OK" if "OK" in leaflet_status else "WARN", f" {leaflet_status}"), + (380, "BOOT", divider), + (440, "INFO", " Select a Project Folder, then scan your SD card to begin."), + (500, "INFO", " All events — scans, ingests, analysis — are logged here."), + (560, "BOOT", divider), + (620, "BOOT", " SYSTEM READY // awaiting operator input"), + (680, "BOOT", divider), + ] + + for delay_ms, level, msg in seq: + QTimer.singleShot(delay_ms, lambda lv=level, m=msg: self._log(m, lv)) + def _build_results_tab(self) -> QWidget: panel = QFrame() panel.setObjectName("Panel") @@ -1135,7 +1449,8 @@ def _build_settings_tab(self) -> QWidget: bg_row = QHBoxLayout() bg_row.addWidget(QLabel("Background mode:")) self.cmb_bg_mode = QComboBox() - self.cmb_bg_mode.addItems(["random_gif", "grid_horizon", "circuit_board", "matrix_rain", "nebula_noise"]) + self.cmb_bg_mode.addItems(["random", "grid_horizon", "circuit_board", "matrix_rain", "nebula_noise"]) + self.cmb_bg_mode.currentIndexChanged.connect(self._regen_bg) bg_row.addWidget(self.cmb_bg_mode) v.addLayout(bg_row) @@ -1569,32 +1884,86 @@ def _sync_bg_geometry(self): pass def _regen_bg(self): - # Randomly pick a new demoscene background (full-window). - self.demoscene.randomize() - self._log(f"Demoscene: {self.demoscene.current_name}") + mode = self.cmb_bg_mode.currentText() if hasattr(self, "cmb_bg_mode") else "random" + if mode == "random" or mode not in ProceduralBackground.MODES: + self.demoscene.randomize() + else: + self.demoscene.set_mode(mode) + self._log(f"Background: {self.demoscene.current_name}") - def _log(self, msg: str): + # ------------------------------------------------------------------ + # Console / logging + # ------------------------------------------------------------------ + _LOG_STYLES: dict = { + # level → (prefix, hex_color, plain_prefix) + "INFO": ("[*]", "#C8FFFA", "[*]"), + "OK": ("[+]", "#50FF90", "[+]"), + "WARN": ("[!]", "#FFD080", "[!]"), + "ERROR": ("[X]", "#FF5555", "[X]"), + "DEBUG": ("[~]", "#7A9090", "[~]"), + "BOOT": ("[>]", "#00FFCC", "[>]"), + "SCAN": ("[S]", "#A0CFFF", "[S]"), + } + + def _log(self, msg: str, level: str = "INFO"): stamp = datetime.now().strftime("%H:%M:%S") - line = f"[{stamp}] {msg}" + _, color, prefix = self._LOG_STYLES.get(level, self._LOG_STYLES["INFO"]) + plain = f"[{stamp}] {prefix} {msg}" + try: - logging.info(msg) + logging.info(plain) except Exception: pass - for attr in ("console_dock", "console_big"): - w = getattr(self, attr, None) - if w is None: - continue + + # Dock → always plain text (compact strip) + dock = getattr(self, "console_dock", None) + if dock is not None: try: - # QPlainTextEdit supports appendPlainText; QTextEdit supports append. - if hasattr(w, "appendPlainText"): - w.appendPlainText(line) - elif hasattr(w, "append"): - w.append(line) + dock.appendPlainText(plain) + dock.verticalScrollBar().setValue(dock.verticalScrollBar().maximum()) except Exception: pass + # Big console → colored HTML + big = getattr(self, "console_big", None) + if big is not None: + try: + safe = (msg + .replace("&", "&") + .replace("<", "<") + .replace(">", ">")) + html = ( + f'[{stamp}] ' + f'{prefix} ' + f'{safe}' + ) + big.append(html) + big.verticalScrollBar().setValue(big.verticalScrollBar().maximum()) + except Exception: + pass + + def _log_ok(self, msg: str): self._log(msg, "OK") + def _log_warn(self, msg: str): self._log(msg, "WARN") + def _log_err(self, msg: str): self._log(msg, "ERROR") + def _log_debug(self, msg: str): self._log(msg, "DEBUG") + def _log_scan(self, msg: str): self._log(msg, "SCAN") + + def _log_from_core(self, msg: str): + """Route core/worker messages to the right log level by inspecting prefix.""" + ml = msg.lstrip() + if ml.startswith("[warn]") or ml.startswith(" [warn]"): + self._log(msg, "WARN") + elif ml.startswith("ERROR") or ml.startswith("[X]"): + self._log(msg, "ERROR") + elif ml.startswith("[+]") or ml.startswith("Complete"): + self._log(msg, "OK") + elif ml.startswith("[*] P4R51NG") or ml.startswith("[*] WR1T1NG") or ml.startswith("[*] R3ND3R1NG"): + self._log(msg, "SCAN") + else: + self._log(msg, "INFO") + def _open_path(self, path: str | None): if not path: return @@ -1881,15 +2250,11 @@ def run_analysis(self): QMessageBox.warning(self, "Missing Project Folder", "Select a Project Folder first.") return - self._log("=== ANALYZE REQUESTED ===") - self._log("Logs = GPS evidence. PCAPs = radio evidence. Overlap = confidence.") + self._log(f"=== ANALYZE === {len(self.wardrive_files)} log(s) + {len(self.pcap_files)} PCAP(s)", "SCAN") + self._log("Wardrive logs = GPS evidence. PCAPs = radio/handshake evidence.", "INFO") job_id = self._new_job("Analysis", f"{len(self.wardrive_files)} logs, {len(self.pcap_files)} pcaps") self._set_running(True) - - # Busy banner (big console + dock) - self._log('=== LIVE CONSOLE ENGAGED ===') - self._log('[*] 1N1T… b00t s3qu3nc3 // generating reports…') - self._log('[*] Stand by… parsing evidence + rendering HTML…') + self._log("Engine initializing — parsing evidence, computing centroids, rendering HTML…", "BOOT") # Snap to console so it feels alive while it runs try: @@ -1899,7 +2264,7 @@ def run_analysis(self): self.worker = AnalyzeWorker(self.wardrive_files, self.pcap_files, self.project_dir) self.worker._mission_job_id = job_id - self.worker.stage.connect(self._log) + self.worker.stage.connect(self._log_from_core) self.worker.progress.connect(self._on_progress) self.worker.done.connect(self._on_done) self.worker.failed.connect(self._on_failed) @@ -1910,6 +2275,33 @@ def run_analysis(self): pass self.worker.start() + # Heartbeat timer: emit a console line every 5s while analysis is running + # so the user can see it has not frozen during long PCAP parsing. + self._analysis_heartbeat_secs = 0 + self._analysis_heartbeat_timer = QTimer(self) + self._analysis_heartbeat_timer.setInterval(5000) + self._analysis_heartbeat_timer.timeout.connect(self._on_analysis_heartbeat) + self._analysis_heartbeat_timer.start() + + def _on_analysis_heartbeat(self): + """Emit a periodic alive-signal to the console while analysis is running.""" + if not getattr(self, "_running", False): + self._stop_analysis_heartbeat() + return + self._analysis_heartbeat_secs = getattr(self, "_analysis_heartbeat_secs", 0) + 5 + elapsed = self._analysis_heartbeat_secs + mins, secs = divmod(elapsed, 60) + self._log(f"[*] Analysis running... {mins}m {secs:02d}s elapsed (parsing PCAPs / building reports)", "INFO") + + def _stop_analysis_heartbeat(self): + t = getattr(self, "_analysis_heartbeat_timer", None) + if t is not None: + try: + t.stop() + except Exception: + pass + self._analysis_heartbeat_timer = None + def _on_progress(self, cur: int, tot: int, label: str): """Update footer progress bar + ETA using a moving average of per-step times.""" self._update_job(getattr(self, "_active_job_id", None), progress=label) @@ -1973,6 +2365,7 @@ def _on_progress(self, cur: int, tot: int, label: str): pass def _on_done(self, results: dict): + self._stop_analysis_heartbeat() self._latest_summary = results.get("summary") self._latest_run_dir = results.get("run_dir") self._update_job( @@ -2008,7 +2401,13 @@ def _on_done(self, results: dict): QMessageBox.information(self, "Complete", msg) - self._log("Boot sequence complete. Awaiting next scan…") + self._log("=== ANALYSIS COMPLETE === all artifacts written.", "OK") + for k in ("csv", "xlsx", "map", "summary", "pcap_summary_html", "pcap_master_csv", "kml"): + if k in results: + try: + self._log(f" {k}: {os.path.basename(str(results[k]))}", "OK") + except Exception: + pass self.refresh_mission_control() # Auto-open: prefer summary.html if available, else open the run folder @@ -2048,8 +2447,18 @@ def _refresh_lists(self): pass def _on_failed(self, err: str): + err = _safe_gui_text(err, limit=1200) + try: + report = write_error_report( + "analysis_failure", + context={"project_dir": self.project_dir, "error": err}, + ) + err = f"{err}\n\nReport: {report}" + except Exception: + pass + self._stop_analysis_heartbeat() self._set_running(False) - self._log(f"ERROR: {err}") + self._log_err(f"ANALYSIS FAILED: {err}") self._update_job( getattr(self, "_active_job_id", None), status="Failed", @@ -2077,27 +2486,98 @@ def select_sd_folder(self): folder = QFileDialog.getExistingDirectory(self, "Select SD Folder") if not folder: return + if self._scan_worker is not None: + QMessageBox.information(self, "Busy", "SD scan already running.") + return + self._sd_root = folder - self.lbl_sd_root.setText(f"SD folder: {folder}") - self._log(f"Scanning SD folder: {folder}") + self._sd_scan_folder = folder + self.lbl_sd_root.setText(f"SD folder: {folder} [scanning…]") + self._log(f"=== SD SCAN === {folder}", "SCAN") + self._log("Walking directory tree… (UI stays live)", "INFO") + + # Disable ingest buttons while scanning + try: + self.btn_sd_ingest.setEnabled(False) + except Exception: + pass + job_id = self._new_job("SD Scan", folder) + self._scan_job_id = job_id + worker = ScanWorker(folder) + self._scan_worker = worker + worker.done.connect(self._on_scan_done) + worker.failed.connect(self._on_scan_failed) try: - self._sd_candidates = scan_sd_folder(folder) - except Exception as e: - self._update_job(job_id, status="Failed", ended=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), result=str(e)) - QMessageBox.critical(self, "SD Scan Error", str(e)) - return + worker.finished.connect(worker.deleteLater) + except Exception: + pass + worker.start() + + def _on_scan_done(self, candidates: list): + self._scan_worker = None + self._sd_candidates = candidates + folder = self._sd_scan_folder + + from collections import Counter + by_kind: Counter = Counter(c.kind for c in candidates) + by_app: Counter = Counter(c.source_app for c in candidates) + unknowns = [c for c in candidates if c.source_app == "unknown"] + rec_count = sum(1 for c in candidates if c.recommended) + total_mb = sum(c.size for c in candidates) / (1024 * 1024) + + self._log_scan(f"Found {len(candidates)} files ({total_mb:.1f} MB) | {rec_count} recommended") + for app, n in sorted(by_app.items()): + self._log(f" [{app}] {n} file(s)", "OK" if app != "unknown" else "WARN") + + self._log_scan("File types detected:") + for kind, n in sorted(by_kind.items()): + self._log(f" {n:3d} {kind}", "DEBUG") + + large = [c for c in candidates if c.ext in (".pcap", ".pcapng") and c.size > 50 * 1024 * 1024] + for c in large: + self._log_warn(f"Large PCAP ({c.size // (1024 * 1024)}MB) — parse capped at 2M packets: {c.rel_path}") + + if unknowns: + self._log_warn(f"{len(unknowns)} file(s) NOT recognized — will not be auto-selected:") + for c in unknowns: + self._log_warn(f" UNKNOWN {c.rel_path} [{c.ext}]") + else: + self._log_ok("All files recognized.") + self.lbl_sd_root.setText(f"SD folder: {folder}") self._populate_sd_lists() + + try: + self.btn_sd_ingest.setEnabled(True) + except Exception: + pass + self._update_job( - job_id, + self._scan_job_id, status="Complete", ended=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - progress=f"{len(self._sd_candidates)} candidate file(s)", + progress=f"{len(candidates)} candidate file(s)", result=folder, ) + def _on_scan_failed(self, err: str): + self._scan_worker = None + self._log_err(f"SD scan failed: {err}") + self.lbl_sd_root.setText("SD folder: (scan failed)") + try: + self.btn_sd_ingest.setEnabled(True) + except Exception: + pass + self._update_job( + self._scan_job_id, + status="Failed", + ended=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + result=err, + ) + QMessageBox.critical(self, "SD Scan Error", err) + def _populate_sd_lists(self): # Clear self.sd_list_all.clear() @@ -2127,7 +2607,6 @@ def _populate_sd_lists(self): self.lbl_sd_stats.setText( f"Found {total} file(s). Marauder: {counts['marauder']} | Porkchop: {counts['porkchop']} | Bruce: {counts['bruce']} | Nemo: {counts['nemo']} | Other: {counts['unknown']}." ) - self._log(self.lbl_sd_stats.text()) def _current_sd_list(self) -> QListWidget: # Single list mode (filters hide/show items) @@ -2180,6 +2659,9 @@ def sd_ingest_selected(self): if not self._sd_root: QMessageBox.warning(self, "Missing SD Folder", "Select an SD Folder first.") return + if getattr(self, "_ingest_worker", None) is not None: + QMessageBox.information(self, "Busy", "Ingest already running — wait for it to finish.") + return lw = self._current_sd_list() selected_idxs: List[int] = [] @@ -2190,7 +2672,6 @@ def sd_ingest_selected(self): if idx is not None: selected_idxs.append(int(idx)) - # De-duplicate indexes (same file appears in multiple tabs) selected_idxs = sorted(set(selected_idxs)) if not selected_idxs: QMessageBox.information(self, "Nothing Selected", "Select at least one file to attach.") @@ -2198,49 +2679,133 @@ def sd_ingest_selected(self): candidates = [self._sd_candidates[i] for i in selected_idxs] skip_dupes = self.chk_sd_skip_dupes.isChecked() + total_mb = sum(c.size for c in candidates) / (1024 * 1024) - self._log(f"=== SD INGEST ===") - self._log(f"Selected: {len(candidates)} file(s). Skip duplicates: {skip_dupes}") + self._log( + f"=== SD INGEST === {len(candidates)} file(s) ({total_mb:.1f} MB) skip_dupes={skip_dupes}", + "SCAN", + ) job_id = self._new_job("SD Ingest", f"{len(candidates)} selected") - def cb(msg: str): - self._log(msg) - self._update_job(job_id, progress=msg) + # Disable button so user can't double-fire; switch to console tab + self.btn_sd_ingest.setEnabled(False) + self.btn_sd_ingest.setText("Attaching…") + try: + self.tabs.setCurrentWidget(self.tab_console) + except Exception: + pass + # Progress bar try: - stats = ingest_candidates_to_project( - project_dir=self.project_dir, - sd_root=self._sd_root, - candidates=candidates, - label="SD", - skip_duplicates=skip_dupes, - progress_cb=cb, - ) - except Exception as e: - self._update_job(job_id, status="Failed", ended=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), result=str(e)) - QMessageBox.critical(self, "Ingest Error", str(e)) - return + self.progress.setValue(0) + self.progress.setFormat("Ingest 0%") + except Exception: + pass - self.lbl_sd_stats.setText( - f"Attached {len(candidates)} selected | Copied {stats['imported']} | Duplicates {stats['duplicates']} | Errors {stats['errors']}" + worker = IngestWorker( + project_dir=self.project_dir, + sd_root=self._sd_root, + candidates=candidates, + label="SD", + skip_duplicates=skip_dupes, ) + self._ingest_worker = worker + self._ingest_job_id = job_id + self._ingest_total = len(candidates) + + worker.stage.connect(self._on_ingest_stage) + worker.progress.connect(self._on_ingest_progress) + worker.file_progress.connect(self._on_ingest_file_progress) + worker.done.connect(self._on_ingest_done) + worker.failed.connect(self._on_ingest_failed) + try: + worker.finished.connect(worker.deleteLater) + except Exception: + pass + worker.start() + + def _on_ingest_stage(self, msg: str): + lvl = "ERROR" if msg.startswith("ERROR") else "WARN" if "Duplicate" in msg else "DEBUG" + self._log(msg, lvl) + self._update_job(getattr(self, "_ingest_job_id", None), progress=msg) + + def _on_ingest_progress(self, done: int, total: int): + if total > 0: + pct = int(done / total * 100) + try: + self.progress.setValue(pct) + # Only update format if no active per-file operation is being shown + if not getattr(self, "_ingest_file_active", False): + self.progress.setFormat(f"Ingest {pct}% ({done}/{total} files)") + except Exception: + pass + + def _on_ingest_file_progress(self, filename: str, phase: str, bdone: int, btotal: int): + """Per-file byte-level progress for large files during hash or copy phase.""" + try: + self._ingest_file_active = True + pct = int(bdone / btotal * 100) if btotal > 0 else 0 + mb_done = bdone / (1024 * 1024) + mb_total = btotal / (1024 * 1024) + verb = "Hashing" if phase == "hash" else "Copying" + file_total = getattr(self, "_ingest_total", 0) + file_done = getattr(self, "_ingest_worker", None) + # Get current file count from worker's internal counter (best effort) + n = getattr(file_done, "_done_count", 0) if file_done else 0 + counter = f" — File {n}/{file_total}" if file_total else "" + label = f"{verb} {filename} {pct}% ({mb_done:.0f} MB / {mb_total:.0f} MB){counter}" + self.progress.setValue(pct) + self.progress.setFormat(label) + # Clear the per-file lock once done so overall progress takes back over + if bdone >= btotal: + self._ingest_file_active = False + except Exception: + self._ingest_file_active = False + + def _on_ingest_done(self, stats: dict): + self._ingest_worker = None + self._ingest_file_active = False + candidates_count = getattr(self, "_ingest_total", stats.get("total", 0)) + summary = ( + f"Copied {stats['imported']} | " + f"Duplicates {stats['duplicates']} | " + f"Errors {stats['errors']} " + f"(of {candidates_count} selected)" + ) + lvl = "ERROR" if stats["errors"] else "OK" + self._log(f"=== INGEST COMPLETE === {summary}", lvl) + if stats["errors"]: + self._log_warn("Some files failed to copy — check console for details.") + + try: + self.lbl_sd_stats.setText(summary) + except Exception: + pass + try: + self.progress.setValue(100) + self.progress.setFormat("Ingest complete") + except Exception: + pass + + self.btn_sd_ingest.setEnabled(True) + self.btn_sd_ingest.setText("Attach Selected to Project") + + job_id = getattr(self, "_ingest_job_id", None) self._update_job( job_id, status="Complete" if stats["errors"] == 0 else "Complete with errors", ended=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - progress=self.lbl_sd_stats.text(), + progress=summary, result=self.project_dir, ) - QMessageBox.information(self, "Ingest Complete", self.lbl_sd_stats.text()) - self._log(self.lbl_sd_stats.text()) self.refresh_mission_control() + QMessageBox.information(self, "Ingest Complete", summary) - # Immediately offer to analyze after ingest (fast workflow). try: resp = QMessageBox.question( self, "Analyze Now?", - "Evidence has been attached to this project.\n\nAnalyze now and generate reports?", + "Evidence attached.\n\nAnalyze now and generate reports?", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes, ) @@ -2249,6 +2814,26 @@ def cb(msg: str): except Exception: pass + def _on_ingest_failed(self, err: str): + self._ingest_worker = None + self._ingest_file_active = False + self._log_err(f"Ingest failed: {err}") + self.btn_sd_ingest.setEnabled(True) + self.btn_sd_ingest.setText("Attach Selected to Project") + try: + self.progress.setValue(0) + self.progress.setFormat("Ingest failed") + except Exception: + pass + job_id = getattr(self, "_ingest_job_id", None) + self._update_job( + job_id, + status="Failed", + ended=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + result=err, + ) + QMessageBox.critical(self, "Ingest Error", err) + def sd_analyze_selected_now(self): """Analyze checked files from the SD scan WITHOUT requiring ingest/attach first.""" @@ -2326,41 +2911,6 @@ def _is_text(ext: str) -> bool: QMessageBox.information(self, "No Supported Evidence", "No supported evidence was detected in your selection. (Try Select Recommended.)") return - # NEW vs ALREADY SEEN detection (project DB sha256) - already = None - new = None - try: - dbp = os.path.join(self.project_dir, "project.db") - if os.path.exists(dbp): - already = 0 - new = 0 - # only hash selected evidence types (pcaps can be large, but this is the most accurate check) - for p in (pcaps + logs + structured): - try: - sha = sha256_file(p) - if sha256_exists(dbp, sha): - already += 1 - else: - new += 1 - except Exception: - pass - self._log(f"Project evidence match (by hash): already ingested {already} | new to project {new}") - except Exception: - pass - - # Allow re-analysis even if everything was already seen, but confirm first. - if already is not None and new is not None and already > 0 and new == 0: - resp = QMessageBox.question( - self, - "Analyze Anyway?", - "All selected evidence appears to have been ingested into this project already (hash match).\n\nAnalyze anyway and generate a fresh run folder?", - QMessageBox.Yes | QMessageBox.No, - QMessageBox.No, - ) - if resp != QMessageBox.Yes: - self._log("Analyze cancelled (operator declined re-analysis of already-seen evidence).") - return - # Feed analysis engine self.wardrive_files = logs + structured self.pcap_files = pcaps diff --git a/launch.cmd b/launch.cmd new file mode 100644 index 0000000..cc5c719 --- /dev/null +++ b/launch.cmd @@ -0,0 +1,58 @@ +@echo off +title Wardrive Mission Control +cd /d "%~dp0" + +echo. +echo ============================================================ +echo WARDRIVE MISSION CONTROL +echo ============================================================ +echo. + +:: Check Python is available +where python >nul 2>&1 +if errorlevel 1 ( + echo [X] Python not found in PATH. Install Python 3.11+ and retry. + pause + exit /b 1 +) + +:: Print Python version +for /f "tokens=*" %%v in ('python --version 2^>^&1') do echo [*] Runtime: %%v + +:: Auto-install required packages if missing +echo [*] Checking dependencies... +python -c "import dpkt" >nul 2>&1 +if errorlevel 1 ( + echo [!] dpkt not found -- installing... + python -m pip install dpkt --quiet + if errorlevel 1 ( + echo [X] Failed to install dpkt. PCAP parsing will be disabled. + ) else ( + echo [+] dpkt installed. + ) +) +python -c "import openpyxl" >nul 2>&1 +if errorlevel 1 ( + echo [!] openpyxl not found -- installing... + python -m pip install openpyxl --quiet +) + +:: Check entry point exists +if not exist "%~dp0run_step3d_scene.py" ( + echo [X] run_step3d_scene.py not found in %~dp0 + pause + exit /b 1 +) + +echo [*] Working dir: %~dp0 +echo [*] Launching... +echo. + +python run_step3d_scene.py + +if errorlevel 1 ( + echo. + echo [X] App exited with an error (code %errorlevel%). + echo Check the output above for details. + pause +) diff --git a/project_vault.py b/project_vault.py index 2a1b196..b62a4e2 100644 --- a/project_vault.py +++ b/project_vault.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import re import shutil import hashlib import sqlite3 @@ -28,7 +29,9 @@ class CandidateFile: size: int -# --- patterns / allow-lists (Option 1) --- +# --- patterns / allow-lists --- + +# Marauder numbered PCAP outputs (e.g. beacon_0.pcap, raw_1.pcap) _MARAUDER_PCAP_PREFIXES = ( "raw_", "beacon_", @@ -40,11 +43,16 @@ class CandidateFile: "multissid_", "deauth_", "sae_commit_", + "flock_", # Flock module captures + "yoink_", # Yoink module captures + "jammit_", # Jammit module root-level pcaps ) +# Marauder .log / .txt outputs (WiGLE CSV format or scan logs) _MARAUDER_LOG_PREFIXES = ( "wardrive_", "station_wardrive_", + "flock_wardrive_", "pingscan_", "arpscan_", "portscan_", @@ -53,8 +61,40 @@ class CandidateFile: "dns_", "http_", "https_", + "yoink_", # Yoink session logs (metadata, not WiGLE CSV) + "jammit_", # Jammit session logs (metadata) + "recon_", # Recon scan outputs + "flock_", # Flock session logs +) + +# Marauder scan outputs in CSV format (not WiGLE, not log extension) +_MARAUDER_CSV_PREFIXES = ( + "portscan_", + "arpscan_", + "pingscan_", + "recon_", + "sshscan_", + "telnetscan_", + "stats", # Jammit per-second stats.csv +) + +# Jammit per-target capture session: known bare filenames produced inside SSID_BSSID/ dirs +_JAMMIT_PCAP_NAMES = {"beacon.pcap", "handshakes.pcap", "raw.pcap", "jammit_deauths.pcap"} +_JAMMIT_META_NAMES = {"capture_summary.txt", "jammit_session.log", "stats.csv"} + +# Regex: a directory named _XX-XX-XX-XX-XX-XX (Jammit per-target folder) +_JAMMIT_TARGET_DIR_RE = re.compile( + r"(?:^|/)captures/[^/]+_[0-9a-f]{2}(?:[:-][0-9a-f]{2}){5}(?:/|$)", + re.IGNORECASE, ) +# Kinds that are NOT wardrive/GPS data — exclude from log analysis input +_NON_WARDRIVE_LOG_KINDS = { + "jammit_session", "yoink_session", "jammit_summary", + "jammit_stats", "jammit_other", "recon_csv", + "log_yoink", "log_jammit", "log_flock", +} + _BRUCE_FILES_RECOMMENDED = {"bruce_creds.csv"} _BRUCE_FILES_OPTIONAL = {"bruce.conf", "shortcuts.json", "boot.jpg", "boot.gif", "boot.wav"} @@ -351,17 +391,40 @@ def compare_run_masters(new_csv: str, old_csv: str) -> Dict[str, object]: } -def sha256_file(path: str, chunk_size: int = 1024 * 1024) -> str: +def sha256_file(path: str, chunk_size: int = 1024 * 1024, progress_cb=None) -> str: + """SHA-256 a file in chunks. progress_cb(bytes_done, bytes_total) called each chunk.""" h = hashlib.sha256() + total = os.path.getsize(path) + done = 0 with open(path, "rb") as f: while True: b = f.read(chunk_size) if not b: break h.update(b) + done += len(b) + if progress_cb: + progress_cb(done, total) return h.hexdigest() +def _copy_with_progress(src: str, dst: str, progress_cb=None, + chunk_size: int = 512 * 1024) -> None: + """Copy src → dst in chunks, calling progress_cb(bytes_done, bytes_total) each chunk.""" + total = os.path.getsize(src) + done = 0 + with open(src, "rb") as fsrc, open(dst, "wb") as fdst: + while True: + buf = fsrc.read(chunk_size) + if not buf: + break + fdst.write(buf) + done += len(buf) + if progress_cb: + progress_cb(done, total) + shutil.copystat(src, dst) + + def looks_like_project_dir(dir_path: str) -> bool: if os.path.isfile(os.path.join(dir_path, DB_NAME)): return True @@ -383,21 +446,31 @@ def should_skip_dir(dir_path: str) -> bool: return False +def _in_jammit_capture_dir(rel_lower: str) -> bool: + """True if this path is inside a Jammit per-target capture directory.""" + # Matches captures/ whether the dir uses dashes or not + return "/captures/" in rel_lower or rel_lower.startswith("captures/") + + def classify_candidate(rel_path: str, filename: str) -> Tuple[SourceApp, str, bool, str]: """Return (source_app, kind, recommended, reason).""" fn = filename.lower() rel_lower = rel_path.lower().replace("\\", "/") ext = os.path.splitext(fn)[1].lower() + # ---------------------------------------------------------------- # Nemo (M5Stick Nemo) - if 'nemo' in rel_lower or fn.startswith('nemo'): - if ext in ('.pcap', '.pcapng'): - return 'nemo', 'pcap_capture', True, 'Nemo capture/pcap' - if ext in ('.txt', '.log', '.csv', '.json', '.html'): - rec = fn.endswith('creds.txt') or 'creds' in fn or 'loot' in rel_lower - return 'nemo', 'nemo_output', True if rec else False, 'Nemo output/results' + # ---------------------------------------------------------------- + if "nemo" in rel_lower or fn.startswith("nemo"): + if ext in (".pcap", ".pcapng"): + return "nemo", "pcap_capture", True, "Nemo capture" + if ext in (".txt", ".log", ".csv", ".json", ".html"): + rec = "creds" in fn or "loot" in rel_lower + return "nemo", "nemo_output", rec, "Nemo output/results" + # ---------------------------------------------------------------- # Bruce + # ---------------------------------------------------------------- if fn in _BRUCE_FILES_RECOMMENDED: return "bruce", "bruce_creds", True, "Bruce creds/results" if fn in _BRUCE_FILES_OPTIONAL: @@ -405,29 +478,86 @@ def classify_candidate(rel_path: str, filename: str) -> Tuple[SourceApp, str, bo if "/scripts/" in rel_lower: return "bruce", "bruce_script", False, "Bruce scripts" - # Porkchop + # ---------------------------------------------------------------- + # Hashcat .22000 (from any tool — Jammit, Porkchop, manual) + # ---------------------------------------------------------------- if ext == ".22000": - return "porkchop", "handshake_22000", True, "Hashcat 22000 capture" + return "marauder", "handshake_22000", True, "Hashcat 22000 handshake" + + # ---------------------------------------------------------------- + # Porkchop loot directory + # ---------------------------------------------------------------- if "porkchop" in rel_lower or "/loot/" in rel_lower or rel_lower.endswith("/loot"): if ext in (".pcap", ".pcapng"): - return "porkchop", "pcap_capture", True, "Capture in Porkchop loot" + return "porkchop", "pcap_capture", True, "Porkchop loot capture" if ext in (".txt", ".log", ".csv", ".json"): return "porkchop", "loot_meta", False, "Porkchop loot metadata" - # Marauder + # ---------------------------------------------------------------- + # Jammit per-target capture directories: captures// + # These contain bare-named files (beacon.pcap, handshakes.pcap, etc.) + # ---------------------------------------------------------------- + if _in_jammit_capture_dir(rel_lower): + if ext in (".pcap", ".pcapng"): + if fn in _JAMMIT_PCAP_NAMES: + kind_tag = os.path.splitext(fn)[0] # beacon, handshakes, raw, jammit_deauths + return "marauder", f"pcap_{kind_tag}", True, f"Jammit {fn}" + # Named handshake pcaps inside Handshakes/ subdir: SSID_BSSID_N.pcap + if "/handshakes/" in rel_lower: + return "marauder", "pcap_handshakes", True, "Jammit named handshake PCAP" + return "marauder", "pcap_jammit", True, "Jammit capture" + if fn == "capture_summary.txt": + return "marauder", "jammit_summary", True, "Jammit capture summary" + if fn == "stats.csv": + return "marauder", "jammit_stats", True, "Jammit per-second stats" + if fn in ("jammit_session.log",): + return "marauder", "jammit_session", False, "Jammit session log (metadata)" + if ext in (".log", ".txt"): + return "marauder", "jammit_other", False, "Jammit session file" + if ext == ".csv": + for p in _MARAUDER_CSV_PREFIXES: + if fn.startswith(p): + return "marauder", "log_" + p.rstrip("_"), True, "Marauder scan CSV" + return "marauder", "jammit_other", False, "Jammit CSV file" + return "marauder", "jammit_other", False, "Jammit session file" + + # ---------------------------------------------------------------- + # Marauder — PCAP files + # ---------------------------------------------------------------- if ext in (".pcap", ".pcapng"): for p in _MARAUDER_PCAP_PREFIXES: if fn.startswith(p): return "marauder", "pcap_" + p.rstrip("_"), True, "Marauder capture" return "unknown", "pcap", False, "PCAP (unrecognized naming)" + + # ---------------------------------------------------------------- + # Marauder — CSV scan outputs (portscan_*.csv, recon_scan.csv, etc.) + # ---------------------------------------------------------------- + if ext == ".csv": + for p in _MARAUDER_CSV_PREFIXES: + if fn.startswith(p): + return "marauder", "recon_csv", True, "Marauder scan result (CSV)" + return "unknown", "other", False, "CSV (unrecognized)" + + # ---------------------------------------------------------------- + # Marauder — log / text outputs + # ---------------------------------------------------------------- if ext in (".log", ".txt"): for p in _MARAUDER_LOG_PREFIXES: if fn.startswith(p): - return "marauder", "log_" + p.rstrip("_"), True, "Marauder output log" + kind = "log_" + p.rstrip("_").rstrip("_wardrive").rstrip("_") + # Preserve full prefix for wardrive variants so analysis picks them up + kind = "log_" + p.rstrip("_") + return "marauder", kind, True, "Marauder output log" return "unknown", "log", False, "Log/text (unrecognized naming)" + # ---------------------------------------------------------------- + # Firmware / other + # ---------------------------------------------------------------- if ext in (".bin", ".uf2"): return "unknown", "firmware", False, "Firmware/update" + if ext == ".json": + return "unknown", "other", False, "JSON config/data" return "unknown", "other", False, "Not recognized" @@ -440,8 +570,19 @@ def scan_sd_folder(sd_root: str, include_hidden: bool = False) -> List[Candidate results: List[CandidateFile] = [] sd_root = os.path.abspath(sd_root) - for dirpath, dirnames, filenames in os.walk(sd_root): - dirnames[:] = [d for d in dirnames if not should_skip_dir(os.path.join(dirpath, d))] + for dirpath, dirnames, filenames in os.walk(sd_root, onerror=lambda _: None): + # Filter directories in-place: skip project output dirs and protected system folders + filtered = [] + for d in dirnames: + full = os.path.join(dirpath, d) + # Skip Windows system folders that cause permission hangs + if d in ("System Volume Information", "$RECYCLE.BIN", "$SysReset", + "System Recovery", "Recovery", "RECYCLER"): + continue + if should_skip_dir(full): + continue + filtered.append(d) + dirnames[:] = filtered for fn in filenames: if not include_hidden and fn.startswith("."): @@ -476,6 +617,9 @@ def scan_sd_folder(sd_root: str, include_hidden: bool = False) -> List[Candidate return results +_FILE_PROGRESS_MIN_BYTES = 1 * 1024 * 1024 # only emit byte progress for files >= 1 MB + + def ingest_candidates_to_project( project_dir: str, sd_root: str, @@ -483,8 +627,15 @@ def ingest_candidates_to_project( label: str = "", skip_duplicates: bool = True, progress_cb=None, + file_progress_cb=None, ) -> Dict[str, int]: - """Copy-ingest candidates into the project vault (Option A). Dedupe is SHA-256 content-based.""" + """ + Copy-ingest candidates into the project vault. Dedupe is SHA-256 content-based. + + file_progress_cb(filename, phase, bytes_done, bytes_total) + phase: "hash" | "copy" + Called each chunk for files >= _FILE_PROGRESS_MIN_BYTES. + """ db_path = ensure_project_vault(project_dir) conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row @@ -510,8 +661,15 @@ def ingest_candidates_to_project( for idx, cand in enumerate(candidates, start=1): if progress_cb: progress_cb(f"Ingest {idx}/{len(candidates)}: hashing {cand.filename} ...") + + large = cand.size >= _FILE_PROGRESS_MIN_BYTES + + def _hash_cb(done, total, _fn=cand.filename): + if file_progress_cb: + file_progress_cb(_fn, "hash", done, total) + try: - h = sha256_file(cand.abs_path) + h = sha256_file(cand.abs_path, progress_cb=(_hash_cb if large else None)) except Exception as e: stats["errors"] += 1 if progress_cb: @@ -527,8 +685,14 @@ def ingest_candidates_to_project( else: dest_path = os.path.join(dest_root, cand.rel_path) os.makedirs(os.path.dirname(dest_path), exist_ok=True) + + def _copy_cb(done, total, _fn=cand.filename): + if file_progress_cb: + file_progress_cb(_fn, "copy", done, total) + try: - shutil.copy2(cand.abs_path, dest_path) + _copy_with_progress(cand.abs_path, dest_path, + progress_cb=(_copy_cb if large else None)) except Exception as e: stats["errors"] += 1 if progress_cb: @@ -593,21 +757,27 @@ def list_project_evidence(project_dir: str) -> List[Dict[str, str]]: def gather_project_inputs_for_analysis(project_dir: str) -> Tuple[List[str], List[str]]: - """Build (logs, pcaps) lists from all evidence in project vault.""" + """Build (wardrive_logs, pcaps) from project vault evidence. + + Only WiGLE-format wardrive logs are included as logs; session/debug/scan + outputs are excluded even if their kind starts with 'log_'. + """ rows = list_project_evidence(project_dir) logs: List[str] = [] pcaps: List[str] = [] for r in rows: p = r.get("stored_path") or "" + if not p or not os.path.exists(p): + continue kind = (r.get("kind") or "").lower() ext = os.path.splitext(p)[1].lower() - if ext in (".pcap", ".pcapng") or kind.startswith("pcap"): - if os.path.exists(p): - pcaps.append(p) - elif ext in (".log", ".txt") or kind.startswith("log"): - if os.path.exists(p): - logs.append(p) - # stable sort and de-dupe by path + + if ext in (".pcap", ".pcapng") or kind.startswith("pcap_"): + pcaps.append(p) + elif ext in (".log", ".txt") and kind not in _NON_WARDRIVE_LOG_KINDS: + # Only accept WiGLE-format text log extensions; .csv is never a wardrive log + logs.append(p) + logs = sorted(dict.fromkeys(logs)) pcaps = sorted(dict.fromkeys(pcaps)) return logs, pcaps diff --git a/run_step3d_scene.py b/run_step3d_scene.py index fe5e622..6144e08 100644 --- a/run_step3d_scene.py +++ b/run_step3d_scene.py @@ -2,6 +2,7 @@ import os import logging from PySide6.QtWidgets import QApplication +from error_logger import install_global_error_hooks, write_error_report from gui_step3d_scene import WardriveGUI @@ -30,15 +31,21 @@ def load_stylesheet(): def main(): log_path = setup_logging() - app = QApplication(sys.argv) - app.setStyleSheet(load_stylesheet()) - window = WardriveGUI() try: - window._log(f"Log file: {log_path}" if log_path else "Log file: (failed to init)") - except Exception: - pass - window.show() - sys.exit(app.exec()) + install_global_error_hooks(os.getcwd()) + app = QApplication(sys.argv) + app.setStyleSheet(load_stylesheet()) + window = WardriveGUI() + try: + window._log(f"Log file: {log_path}" if log_path else "Log file: (failed to init)") + except Exception: + pass + window.show() + sys.exit(app.exec()) + except Exception as exc: + report = write_error_report("startup_or_event_loop_error", exc, base_dir=os.getcwd()) + logging.exception("Fatal startup/event-loop error. Report: %s", report) + raise if __name__ == "__main__": main() diff --git a/wardrive_cli.py b/wardrive_cli.py new file mode 100644 index 0000000..4962c99 --- /dev/null +++ b/wardrive_cli.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import argparse +import json +import sys +import traceback +from typing import Any, Callable + +import wardrive_service as svc +from error_logger import write_error_report + + +Action = Callable[[argparse.Namespace], dict[str, Any]] + + +def _print(data: dict[str, Any], pretty: bool = True) -> None: + print(svc.dumps_json(data, pretty=pretty)) + + +def _cmd_project_summary(args: argparse.Namespace) -> dict[str, Any]: + return svc.project_summary(args.project, import_limit=args.import_limit) + + +def _cmd_evidence_list(args: argparse.Namespace) -> dict[str, Any]: + return svc.evidence_list(args.project, limit=args.limit, offset=args.offset) + + +def _cmd_runs_list(args: argparse.Namespace) -> dict[str, Any]: + return svc.runs_list(args.project, limit=args.limit) + + +def _cmd_latest_run(args: argparse.Namespace) -> dict[str, Any]: + return svc.latest_run(args.project) + + +def _cmd_scan_folder(args: argparse.Namespace) -> dict[str, Any]: + return svc.scan_folder(args.folder, limit=args.limit, include_hidden=args.include_hidden) + + +def _cmd_analyze_project(args: argparse.Namespace) -> dict[str, Any]: + def event(message: str) -> None: + if args.events: + print(json.dumps({"event": message}, ensure_ascii=False), file=sys.stderr, flush=True) + + return svc.analyze_project(args.project, status_cb=event if args.events else None) + + +def _dispatch_action(action: str, params: dict[str, Any]) -> dict[str, Any]: + if action == "project_summary": + return svc.project_summary(str(params["project"]), import_limit=int(params.get("import_limit", 10))) + if action == "evidence_list": + return svc.evidence_list(str(params["project"]), limit=int(params.get("limit", 100)), offset=int(params.get("offset", 0))) + if action == "runs_list": + return svc.runs_list(str(params["project"]), limit=int(params.get("limit", 20))) + if action == "latest_run": + return svc.latest_run(str(params["project"])) + if action == "scan_folder": + return svc.scan_folder(str(params["folder"]), limit=int(params.get("limit", 250)), include_hidden=bool(params.get("include_hidden", False))) + if action == "analyze_project": + return svc.analyze_project(str(params["project"])) + return svc.result("error", error=f"Unknown action: {action}") + + +def _cmd_serve_stdio(args: argparse.Namespace) -> dict[str, Any]: + print(json.dumps({"status": "ready", "protocol": "wardrive-jsonl-v1"}), flush=True) + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + request_id = request.get("id") + action = str(request.get("action", "")) + params = request.get("params") or {} + response = _dispatch_action(action, params) + response["id"] = request_id + except Exception as exc: + report = write_error_report("cli_stdio_action_error", exc, traceback_text=traceback.format_exc()) + response = svc.result("error", error=f"{type(exc).__name__}: {exc}", report=report) + print(json.dumps(response, ensure_ascii=False), flush=True) + return svc.result("ok", message="stdio bridge stopped") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="wardrive_cli.py", + description="Local JSON command bridge for Wardrive Analyzer.", + ) + parser.add_argument("--compact", action="store_true", help="Print compact JSON.") + sub = parser.add_subparsers(dest="command", required=True) + + p = sub.add_parser("project-summary", help="Summarize a project vault.") + p.add_argument("--compact", action="store_true", help=argparse.SUPPRESS) + p.add_argument("project") + p.add_argument("--import-limit", type=int, default=10) + p.set_defaults(func=_cmd_project_summary) + + p = sub.add_parser("evidence-list", help="List project evidence rows.") + p.add_argument("--compact", action="store_true", help=argparse.SUPPRESS) + p.add_argument("project") + p.add_argument("--limit", type=int, default=100) + p.add_argument("--offset", type=int, default=0) + p.set_defaults(func=_cmd_evidence_list) + + p = sub.add_parser("runs-list", help="List discovered analysis runs.") + p.add_argument("--compact", action="store_true", help=argparse.SUPPRESS) + p.add_argument("project") + p.add_argument("--limit", type=int, default=20) + p.set_defaults(func=_cmd_runs_list) + + p = sub.add_parser("latest-run", help="Return the latest discovered run.") + p.add_argument("--compact", action="store_true", help=argparse.SUPPRESS) + p.add_argument("project") + p.set_defaults(func=_cmd_latest_run) + + p = sub.add_parser("scan-folder", help="Scan an SD/evidence folder without ingesting.") + p.add_argument("--compact", action="store_true", help=argparse.SUPPRESS) + p.add_argument("folder") + p.add_argument("--limit", type=int, default=250) + p.add_argument("--include-hidden", action="store_true") + p.set_defaults(func=_cmd_scan_folder) + + p = sub.add_parser("analyze-project", help="Analyze all attached project evidence and generate a new run.") + p.add_argument("--compact", action="store_true", help=argparse.SUPPRESS) + p.add_argument("project") + p.add_argument("--events", action="store_true", help="Stream progress events as JSON lines to stderr.") + p.set_defaults(func=_cmd_analyze_project) + + p = sub.add_parser("serve-stdio", help="Run a local JSONL stdio bridge for agents.") + p.add_argument("--compact", action="store_true", help=argparse.SUPPRESS) + p.set_defaults(func=_cmd_serve_stdio) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + data = args.func(args) + if args.command != "serve-stdio": + _print(data, pretty=not args.compact) + return 0 if data.get("status") == "ok" else 1 + except Exception as exc: + report = write_error_report("cli_command_error", exc, traceback_text=traceback.format_exc()) + _print(svc.result("error", error=f"{type(exc).__name__}: {exc}", report=report), pretty=not args.compact) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/wardrive_service.py b/wardrive_service.py new file mode 100644 index 0000000..72a7eda --- /dev/null +++ b/wardrive_service.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any, Callable, Optional + +from core.analyze import analyze +from project_vault import ( + discover_project_runs, + ensure_project_vault, + evidence_summary, + gather_project_inputs_for_analysis, + list_import_history, + list_project_evidence_detailed, + scan_sd_folder, +) + + +def _jsonable(value: Any) -> Any: + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, Path): + return str(value) + if isinstance(value, dict): + return {str(k): _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple, set)): + return [_jsonable(v) for v in value] + if hasattr(value, "__dict__"): + return _jsonable(vars(value)) + return str(value) + + +def result(status: str, **payload: Any) -> dict[str, Any]: + body = {"status": status} + body.update({k: _jsonable(v) for k, v in payload.items()}) + return body + + +def project_summary(project_dir: str, import_limit: int = 10) -> dict[str, Any]: + db_path = ensure_project_vault(project_dir) + logs, pcaps = gather_project_inputs_for_analysis(project_dir) + runs = discover_project_runs(project_dir) + latest_run = runs[0] if runs else None + return result( + "ok", + project_dir=os.path.abspath(project_dir), + db_path=db_path, + evidence=evidence_summary(project_dir), + analysis_inputs={"logs": len(logs), "pcaps": len(pcaps)}, + imports=list_import_history(project_dir, limit=import_limit), + runs_count=len(runs), + latest_run=latest_run, + ) + + +def evidence_list(project_dir: str, limit: int = 100, offset: int = 0) -> dict[str, Any]: + rows = list_project_evidence_detailed(project_dir) + total = len(rows) + offset = max(0, int(offset)) + limit = max(0, int(limit)) + page = rows[offset: offset + limit] if limit else rows[offset:] + return result( + "ok", + project_dir=os.path.abspath(project_dir), + total=total, + offset=offset, + limit=limit, + evidence=page, + ) + + +def runs_list(project_dir: str, limit: int = 20) -> dict[str, Any]: + runs = discover_project_runs(project_dir) + limit = max(0, int(limit)) + return result( + "ok", + project_dir=os.path.abspath(project_dir), + total=len(runs), + runs=runs[:limit] if limit else runs, + ) + + +def latest_run(project_dir: str) -> dict[str, Any]: + runs = discover_project_runs(project_dir) + return result( + "ok", + project_dir=os.path.abspath(project_dir), + latest_run=runs[0] if runs else None, + ) + + +def scan_folder(folder: str, limit: int = 250, include_hidden: bool = False) -> dict[str, Any]: + candidates = scan_sd_folder(folder, include_hidden=include_hidden) + by_kind: dict[str, int] = {} + by_source: dict[str, int] = {} + recommended = 0 + for cand in candidates: + by_kind[cand.kind] = by_kind.get(cand.kind, 0) + 1 + by_source[cand.source_app] = by_source.get(cand.source_app, 0) + 1 + if cand.recommended: + recommended += 1 + limit = max(0, int(limit)) + return result( + "ok", + folder=os.path.abspath(folder), + total=len(candidates), + recommended=recommended, + by_kind=by_kind, + by_source=by_source, + candidates=candidates[:limit] if limit else candidates, + ) + + +def analyze_project( + project_dir: str, + status_cb: Optional[Callable[[str], None]] = None, + status_tail_limit: int = 80, +) -> dict[str, Any]: + logs, pcaps = gather_project_inputs_for_analysis(project_dir) + events: list[str] = [] + + def capture_event(message: str) -> None: + if len(events) >= status_tail_limit: + events.pop(0) + events.append(str(message)) + if status_cb: + status_cb(str(message)) + + if not logs and not pcaps: + return result( + "error", + project_dir=os.path.abspath(project_dir), + error="No attached logs or PCAPs found for this project.", + ) + + outputs = analyze(logs, pcaps, project_dir, status_cb=capture_event) + return result( + "ok", + project_dir=os.path.abspath(project_dir), + inputs={"logs": len(logs), "pcaps": len(pcaps)}, + outputs=outputs, + status_tail=events, + ) + + +def dumps_json(data: dict[str, Any], pretty: bool = True) -> str: + return json.dumps(_jsonable(data), indent=2 if pretty else None, sort_keys=False)