From 9b31e9fd29148a939f40963a67f329c215ba6a01 Mon Sep 17 00:00:00 2001 From: 0xallam Date: Sat, 25 Apr 2026 13:47:37 -0700 Subject: [PATCH] refactor: nuke ``events.jsonl`` pipeline and the unused PII sanitizer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The JSONL trace sink was never read — TUI consumes ``Tracer`` state directly (chat_messages, agents, tool_executions, vulnerability_reports, LLM stats), and SQLiteSession owns the conversation history. The whole ``StrixTracingProcessor`` → ``_emit_event`` → ``append_jsonl_record`` pipeline was producing files nothing opens. Deleted: - ``strix/telemetry/strix_processor.py`` (the SDK ``TracingProcessor``). - ``strix/telemetry/utils.py`` — ``TelemetrySanitizer`` (no remaining callers), ``append_jsonl_record``, ``get_events_write_lock``, ``reset_events_write_locks``. - ``strix/telemetry/flags.py`` — ``is_telemetry_enabled`` / ``is_posthog_enabled`` collapsed into a 4-line check inside ``posthog._is_enabled`` (its only caller). - ``Tracer._emit_event`` and every event-emit call inside the tracer (``run.started``, ``run.configured``, ``run.completed``, ``finding.created``, ``finding.reviewed``, ``chat.message``). - ``Tracer._enrich_actor`` (only used by ``_emit_event``). - ``Tracer._sanitize_data`` + ``_sanitizer`` field (PII scrub only ran on JSONL events). - ``Tracer.events_file_path`` property and the ``_events_file_path`` / ``_telemetry_enabled`` / ``_run_completed_emitted`` / ``_next_execution_id`` fields. - ``Tracer._calculate_duration`` (one caller in posthog — inlined). - ``add_trace_processor(StrixTracingProcessor(run_dir))`` from ``entry.py``. The ``Tracer`` class is now ~275 LoC of pure runtime state for the TUI + vulnerability artifact writer (markdown / CSV / pentest report). Conversation history goes to ``SQLiteSession``; SDK trace events are not persisted. Co-Authored-By: Claude Opus 4.7 (1M context) --- strix/entry.py | 14 --- strix/telemetry/flags.py | 21 ---- strix/telemetry/posthog.py | 23 +++- strix/telemetry/strix_processor.py | 169 ---------------------------- strix/telemetry/tracer.py | 173 ++++------------------------- strix/telemetry/utils.py | 102 ----------------- 6 files changed, 39 insertions(+), 463 deletions(-) delete mode 100644 strix/telemetry/flags.py delete mode 100644 strix/telemetry/strix_processor.py delete mode 100644 strix/telemetry/utils.py diff --git a/strix/entry.py b/strix/entry.py index 8874d10..903d131 100644 --- a/strix/entry.py +++ b/strix/entry.py @@ -23,7 +23,6 @@ from typing import TYPE_CHECKING, Any, Literal from agents import Runner from agents.memory import SQLiteSession -from agents.tracing import add_trace_processor from strix.agents.factory import build_strix_agent, make_child_factory from strix.config.config import Config @@ -36,7 +35,6 @@ from strix.run_config_factory import ( ) from strix.sandbox import session_manager from strix.sandbox.healthcheck import wait_for_http_ready, wait_for_tcp_ready -from strix.telemetry.strix_processor import StrixTracingProcessor if TYPE_CHECKING: @@ -203,18 +201,6 @@ async def run_strix_scan( bus = AgentMessageBus() root_id = uuid.uuid4().hex[:8] - # Wire SDK tracing into the scan's run-directory ``events.jsonl``. - # ``add_trace_processor`` is idempotent at the provider level — if - # the user runs multiple scans in one process they each get their - # own processor, all writing to their respective run dirs. - if tracer is not None: - try: - run_dir = tracer.get_run_dir() if hasattr(tracer, "get_run_dir") else None - if run_dir is not None: - add_trace_processor(StrixTracingProcessor(run_dir)) - except Exception: - logger.exception("Failed to register StrixTracingProcessor") - bundle = await session_manager.create_or_reuse( scan_id, image=image, diff --git a/strix/telemetry/flags.py b/strix/telemetry/flags.py deleted file mode 100644 index c605d90..0000000 --- a/strix/telemetry/flags.py +++ /dev/null @@ -1,21 +0,0 @@ -from strix.config import Config - - -_DISABLED_VALUES = {"0", "false", "no", "off"} - - -def _is_enabled(raw_value: str | None, default: str = "1") -> bool: - value = (raw_value if raw_value is not None else default).strip().lower() - return value not in _DISABLED_VALUES - - -def is_telemetry_enabled() -> bool: - """Master gate — controls JSONL event emission and posthog.""" - return _is_enabled(Config.get("strix_telemetry"), default="1") - - -def is_posthog_enabled() -> bool: - explicit = Config.get("strix_posthog_telemetry") - if explicit is not None: - return _is_enabled(explicit) - return is_telemetry_enabled() diff --git a/strix/telemetry/posthog.py b/strix/telemetry/posthog.py index aa534d2..6fbbf30 100644 --- a/strix/telemetry/posthog.py +++ b/strix/telemetry/posthog.py @@ -6,7 +6,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Any from uuid import uuid4 -from strix.telemetry.flags import is_posthog_enabled +from strix.config import Config if TYPE_CHECKING: @@ -15,11 +15,18 @@ if TYPE_CHECKING: _POSTHOG_PUBLIC_API_KEY = "phc_7rO3XRuNT5sgSKAl6HDIrWdSGh1COzxw0vxVIAR6vVZ" _POSTHOG_HOST = "https://us.i.posthog.com" +_DISABLED_VALUES = {"0", "false", "no", "off"} + _SESSION_ID = uuid4().hex[:16] def _is_enabled() -> bool: - return is_posthog_enabled() + """Master telemetry gate. ``STRIX_POSTHOG_TELEMETRY`` overrides ``STRIX_TELEMETRY``.""" + explicit = Config.get("strix_posthog_telemetry") + if explicit is not None: + return explicit.strip().lower() not in _DISABLED_VALUES + fallback = Config.get("strix_telemetry") or "1" + return fallback.strip().lower() not in _DISABLED_VALUES def _is_first_run() -> bool: @@ -114,12 +121,22 @@ def end(tracer: "Tracer", exit_reason: str = "completed") -> None: llm = tracer.get_total_llm_stats() total = llm.get("total", {}) + duration = 0.0 + try: + from datetime import datetime + + start = datetime.fromisoformat(tracer.start_time.replace("Z", "+00:00")) + end_iso = tracer.end_time or datetime.now(start.tzinfo).isoformat() + duration = (datetime.fromisoformat(end_iso.replace("Z", "+00:00")) - start).total_seconds() + except (ValueError, TypeError, AttributeError): + pass + _send( "scan_ended", { **_base_props(), "exit_reason": exit_reason, - "duration_seconds": round(tracer._calculate_duration()), + "duration_seconds": round(duration), "vulnerabilities_total": len(tracer.vulnerability_reports), **{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()}, "agent_count": len(tracer.agents), diff --git a/strix/telemetry/strix_processor.py b/strix/telemetry/strix_processor.py deleted file mode 100644 index 8ec2040..0000000 --- a/strix/telemetry/strix_processor.py +++ /dev/null @@ -1,169 +0,0 @@ -"""``StrixTracingProcessor`` — SDK trace processor that writes ``events.jsonl``. - -Hooks into the SDK's tracing pipeline and writes events to -``strix_runs//events.jsonl``. PII scrubbing runs through -:class:`TelemetrySanitizer`. -""" - -from __future__ import annotations - -import json -import logging -import threading -from pathlib import Path -from typing import TYPE_CHECKING, Any - -from agents.tracing.processor_interface import TracingProcessor - - -if TYPE_CHECKING: - from agents.tracing.spans import Span - from agents.tracing.traces import Trace - - from strix.telemetry.utils import TelemetrySanitizer - - -logger = logging.getLogger(__name__) - - -# Module-level lock registry — one per JSONL file so two processors writing -# different run-dirs don't serialize unnecessarily, but two processors -# writing the *same* run-dir do. -_FILE_LOCKS: dict[Path, threading.Lock] = {} -_GUARD = threading.Lock() - - -def _lock_for(path: Path) -> threading.Lock: - with _GUARD: - return _FILE_LOCKS.setdefault(path, threading.Lock()) - - -class StrixTracingProcessor(TracingProcessor): - """Append trace + span events as JSONL into ``run_dir/events.jsonl``. - - Every hook is synchronous — required by the ``TracingProcessor`` - ABC. Each write is protected by a per-path ``threading.Lock`` so - concurrent spans can't interleave bytes mid-line. ``OSError`` is - swallowed so a full disk or permission error doesn't tear the run - down. PII scrubbing runs on every event before it hits the file. - """ - - def __init__( - self, - run_dir: Path, - sanitizer: TelemetrySanitizer | None = None, - ) -> None: - run_dir = Path(run_dir) - run_dir.mkdir(parents=True, exist_ok=True) - self.run_dir: Path = run_dir - self.events_path: Path = run_dir / "events.jsonl" - if sanitizer is None: - from strix.telemetry.utils import TelemetrySanitizer - - sanitizer = TelemetrySanitizer() - self.sanitizer: TelemetrySanitizer = sanitizer - - # --- Internal helpers ------------------------------------------------- - - def _emit(self, event: dict[str, Any]) -> None: - """Sanitize ``event`` and append it as one JSONL line. - - Failures are swallowed — we'd rather lose a trace event than - fail the run. - """ - try: - clean = self.sanitizer.sanitize(event) - except Exception: - logger.exception("Trace event sanitization failed; dropping event") - return - try: - with ( - _lock_for(self.events_path), - self.events_path.open( - "a", - encoding="utf-8", - ) as f, - ): - f.write(json.dumps(clean, ensure_ascii=True) + "\n") - except OSError: - logger.exception("Failed to append trace event to %s", self.events_path) - - @staticmethod - def _span_kind(span: Span[Any]) -> str: - """Map ``SomethingSpanData`` → ``"something"`` for the event_type.""" - name = type(span.span_data).__name__ - if name.endswith("SpanData"): - name = name[: -len("SpanData")] - return name.lower() or "span" - - @staticmethod - def _trace_metadata(trace: Trace) -> dict[str, Any]: - meta: dict[str, Any] = {"name": getattr(trace, "name", None)} - # ``Trace.export()`` includes metadata + group_id when set. - try: - exported = trace.export() - if isinstance(exported, dict): - for key in ("metadata", "group_id", "workflow_name"): - if key in exported and exported[key] is not None: - meta[key] = exported[key] - except Exception: - logger.debug("trace.export failed", exc_info=True) - return meta - - # --- TracingProcessor ABC -------------------------------------------- - - def on_trace_start(self, trace: Trace) -> None: - self._emit( - { - "event_type": "run.started", - "trace_id": trace.trace_id, - "metadata": self._trace_metadata(trace), - } - ) - - def on_trace_end(self, trace: Trace) -> None: - self._emit( - { - "event_type": "run.completed", - "trace_id": trace.trace_id, - } - ) - - def on_span_start(self, span: Span[Any]) -> None: - kind = self._span_kind(span) - self._emit( - { - "event_type": f"{kind}.started", - "span_id": span.span_id, - "trace_id": span.trace_id, - "data": self._safe_export(span), - } - ) - - def on_span_end(self, span: Span[Any]) -> None: - kind = self._span_kind(span) - self._emit( - { - "event_type": f"{kind}.completed", - "span_id": span.span_id, - "trace_id": span.trace_id, - "data": self._safe_export(span), - } - ) - - def force_flush(self) -> None: - """All writes are synchronous; nothing to flush.""" - - def shutdown(self) -> None: - """No-op; nothing to release.""" - - # --- helpers ---------------------------------------------------------- - - @staticmethod - def _safe_export(span: Span[Any]) -> dict[str, Any] | None: - try: - data = span.span_data.export() - return data if isinstance(data, dict) else None - except Exception: - logger.debug("span_data.export failed for %s", span.span_id, exc_info=True) - return None diff --git a/strix/telemetry/tracer.py b/strix/telemetry/tracer.py index 902bea4..d69433e 100644 --- a/strix/telemetry/tracer.py +++ b/strix/telemetry/tracer.py @@ -6,8 +6,6 @@ from typing import Any, Optional from uuid import uuid4 from strix.telemetry import posthog -from strix.telemetry.flags import is_telemetry_enabled -from strix.telemetry.utils import TelemetrySanitizer, append_jsonl_record logger = logging.getLogger(__name__) @@ -25,6 +23,16 @@ def set_global_tracer(tracer: "Tracer") -> None: class Tracer: + """Per-scan in-memory state the TUI renders + per-scan artifact writer. + + Holds live state the TUI reads (chat messages, agent tree, tool + executions, vulnerability reports, LLM usage). Writes vulnerability + markdown + CSV + final pentest report to ``strix_runs//``. + + Conversation history goes to the SDK's ``SQLiteSession`` instead; + SDK trace events are not persisted here. + """ + def __init__(self, run_name: str | None = None): self.run_name = run_name self.run_id = run_name or f"run-{uuid4().hex[:8]}" @@ -58,111 +66,18 @@ class Tracer: "status": "running", } self._run_dir: Path | None = None - self._events_file_path: Path | None = None - self._next_execution_id = 1 self._next_message_id = 1 self._saved_vuln_ids: set[str] = set() - self._run_completed_emitted = False - self._telemetry_enabled = is_telemetry_enabled() - self._sanitizer = TelemetrySanitizer() self.caido_url: str | None = None self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None - self._emit_run_started_event() - - @property - def events_file_path(self) -> Path: - if self._events_file_path is None: - self._events_file_path = self.get_run_dir() / "events.jsonl" - return self._events_file_path - - def _sanitize_data(self, data: Any, key_hint: str | None = None) -> Any: - return self._sanitizer.sanitize(data, key_hint=key_hint) - - def _append_event_record(self, record: dict[str, Any]) -> None: - try: - append_jsonl_record(self.events_file_path, record) - except OSError: - logger.exception("Failed to append JSONL event record") - - def _enrich_actor(self, actor: dict[str, Any] | None) -> dict[str, Any] | None: - if not actor: - return None - - enriched = dict(actor) - if "agent_name" in enriched: - return enriched - - agent_id = enriched.get("agent_id") - if not isinstance(agent_id, str): - return enriched - - agent_data = self.agents.get(agent_id, {}) - agent_name = agent_data.get("name") - if isinstance(agent_name, str) and agent_name: - enriched["agent_name"] = agent_name - - return enriched - - def _emit_event( - self, - event_type: str, - actor: dict[str, Any] | None = None, - payload: Any | None = None, - status: str | None = None, - error: Any | None = None, - source: str = "strix.tracer", - include_run_metadata: bool = False, - ) -> None: - if not self._telemetry_enabled: - return - - enriched_actor = self._enrich_actor(actor) - sanitized_actor = self._sanitize_data(enriched_actor) if enriched_actor else None - sanitized_payload = self._sanitize_data(payload) if payload is not None else None - sanitized_error = self._sanitize_data(error) if error is not None else None - - record = { - "timestamp": datetime.now(UTC).isoformat(), - "event_type": event_type, - "run_id": self.run_id, - "trace_id": uuid4().hex, - "span_id": uuid4().hex[:16], - "actor": sanitized_actor, - "payload": sanitized_payload, - "status": status, - "error": sanitized_error, - "source": source, - } - if include_run_metadata: - record["run_metadata"] = self._sanitize_data(self.run_metadata) - self._append_event_record(record) - def set_run_name(self, run_name: str) -> None: self.run_name = run_name self.run_id = run_name self.run_metadata["run_name"] = run_name self.run_metadata["run_id"] = run_name self._run_dir = None - self._events_file_path = None - self._run_completed_emitted = False - self._emit_run_started_event() - - def _emit_run_started_event(self) -> None: - if not self._telemetry_enabled: - return - - self._emit_event( - "run.started", - payload={ - "run_name": self.run_name, - "start_time": self.start_time, - "local_jsonl_path": str(self.events_file_path), - }, - status="running", - include_run_metadata=True, - ) def get_run_dir(self) -> Path: if self._run_dir is None: @@ -235,12 +150,6 @@ class Tracer: self.vulnerability_reports.append(report) logger.info(f"Added vulnerability report: {report_id} - {title}") posthog.finding(severity) - self._emit_event( - "finding.created", - payload={"report": report}, - status=report["severity"], - source="strix.findings", - ) if self.vulnerability_found_callback: self.vulnerability_found_callback(report) @@ -285,15 +194,6 @@ class Tracer: """ logger.info("Updated scan final fields") - self._emit_event( - "finding.reviewed", - payload={ - "scan_completed": True, - "vulnerability_count": len(self.vulnerability_reports), - }, - status="completed", - source="strix.findings", - ) self.save_run_data(mark_complete=True) posthog.end(self, exit_reason="finished_by_tool") @@ -307,22 +207,15 @@ class Tracer: message_id = self._next_message_id self._next_message_id += 1 - message_data = { - "message_id": message_id, - "content": content, - "role": role, - "agent_id": agent_id, - "timestamp": datetime.now(UTC).isoformat(), - "metadata": metadata or {}, - } - - self.chat_messages.append(message_data) - self._emit_event( - "chat.message", - actor={"agent_id": agent_id, "role": role}, - payload={"message_id": message_id, "content": content, "metadata": metadata or {}}, - status="logged", - source="strix.chat", + self.chat_messages.append( + { + "message_id": message_id, + "content": content, + "role": role, + "agent_id": agent_id, + "timestamp": datetime.now(UTC).isoformat(), + "metadata": metadata or {}, + } ) return message_id @@ -335,12 +228,6 @@ class Tracer: "max_iterations": config.get("max_iterations", 200), } ) - self._emit_event( - "run.configured", - payload={"scan_config": config}, - status="configured", - source="strix.run", - ) def save_run_data(self, mark_complete: bool = False) -> None: try: @@ -489,32 +376,10 @@ class Tracer: logger.info("Updated vulnerability index: %s", vuln_csv_file) logger.info("📊 Essential scan data saved to: %s", run_dir) - if mark_complete and not self._run_completed_emitted: - self._emit_event( - "run.completed", - payload={ - "duration_seconds": self._calculate_duration(), - "vulnerability_count": len(self.vulnerability_reports), - }, - status="completed", - source="strix.run", - include_run_metadata=True, - ) - self._run_completed_emitted = True except (OSError, RuntimeError): logger.exception("Failed to save scan data") - def _calculate_duration(self) -> float: - try: - start = datetime.fromisoformat(self.start_time.replace("Z", "+00:00")) - if self.end_time: - end = datetime.fromisoformat(self.end_time.replace("Z", "+00:00")) - return (end - start).total_seconds() - except (ValueError, TypeError): - pass - return 0.0 - def get_real_tool_count(self) -> int: return sum( 1 diff --git a/strix/telemetry/utils.py b/strix/telemetry/utils.py deleted file mode 100644 index a5d7ea4..0000000 --- a/strix/telemetry/utils.py +++ /dev/null @@ -1,102 +0,0 @@ -import json -import logging -import re -import threading -from pathlib import Path -from typing import Any - -from scrubadub import Scrubber -from scrubadub.detectors import RegexDetector -from scrubadub.filth import Filth - - -logger = logging.getLogger(__name__) - -_REDACTED = "[REDACTED]" -_SCREENSHOT_OMITTED = "[SCREENSHOT_OMITTED]" -_SCREENSHOT_KEY_PATTERN = re.compile(r"screenshot", re.IGNORECASE) -_SENSITIVE_KEY_PATTERN = re.compile( - r"(api[_-]?key|token|secret|password|authorization|cookie|session|credential|private[_-]?key)", - re.IGNORECASE, -) -_SENSITIVE_TOKEN_PATTERN = re.compile( - r"(?i)\b(" - r"bearer\s+[a-z0-9._-]+|" - r"sk-[a-z0-9_-]{8,}|" - r"gh[pousr]_[a-z0-9_-]{12,}|" - r"xox[baprs]-[a-z0-9-]{12,}" - r")\b" -) -_SCRUBADUB_PLACEHOLDER_PATTERN = re.compile(r"\{\{[^}]+\}\}") -_EVENTS_FILE_LOCKS_LOCK = threading.Lock() -_EVENTS_FILE_LOCKS: dict[str, threading.Lock] = {} - - -class _SecretFilth(Filth): # type: ignore[misc] - type = "secret" - - -class _SecretTokenDetector(RegexDetector): # type: ignore[misc] - name = "strix_secret_token_detector" - filth_cls = _SecretFilth - regex = _SENSITIVE_TOKEN_PATTERN - - -class TelemetrySanitizer: - def __init__(self) -> None: - self._scrubber = Scrubber(detector_list=[_SecretTokenDetector]) - - def sanitize(self, data: Any, key_hint: str | None = None) -> Any: # noqa: PLR0911 - if data is None: - return None - - if isinstance(data, dict): - sanitized: dict[str, Any] = {} - for key, value in data.items(): - key_str = str(key) - if _SCREENSHOT_KEY_PATTERN.search(key_str): - sanitized[key_str] = _SCREENSHOT_OMITTED - elif _SENSITIVE_KEY_PATTERN.search(key_str): - sanitized[key_str] = _REDACTED - else: - sanitized[key_str] = self.sanitize(value, key_hint=key_str) - return sanitized - - if isinstance(data, list): - return [self.sanitize(item, key_hint=key_hint) for item in data] - - if isinstance(data, tuple): - return [self.sanitize(item, key_hint=key_hint) for item in data] - - if isinstance(data, str): - if key_hint and _SENSITIVE_KEY_PATTERN.search(key_hint): - return _REDACTED - - cleaned = self._scrubber.clean(data) - return _SCRUBADUB_PLACEHOLDER_PATTERN.sub(_REDACTED, cleaned) - - if isinstance(data, int | float | bool): - return data - - return str(data) - - -def get_events_write_lock(output_path: Path) -> threading.Lock: - path_key = str(output_path.resolve(strict=False)) - with _EVENTS_FILE_LOCKS_LOCK: - lock = _EVENTS_FILE_LOCKS.get(path_key) - if lock is None: - lock = threading.Lock() - _EVENTS_FILE_LOCKS[path_key] = lock - return lock - - -def reset_events_write_locks() -> None: - with _EVENTS_FILE_LOCKS_LOCK: - _EVENTS_FILE_LOCKS.clear() - - -def append_jsonl_record(output_path: Path, record: dict[str, Any]) -> None: - output_path.parent.mkdir(parents=True, exist_ok=True) - with get_events_write_lock(output_path), output_path.open("a", encoding="utf-8") as f: - f.write(json.dumps(record, ensure_ascii=False) + "\n")