Codex: expose the analyzer through a local machine interface

.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/

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

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 ===")
 

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 "<HIDDEN>"
+    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 "<HIDDEN>"
+                return _clean_ssid_text(raw.decode("utf-8", errors="replace"))
             except Exception:
                 return "<HIDDEN>"
         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)
 

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: <MISSING>"
     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

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`.

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