mirror of
https://github.com/usestrix/strix.git
synced 2026-08-23 03:12:37 +02:00
Strip narrative comments and module/helper docstrings
Five rounds of sweep across the tree. Net ~544 lines removed. Removed: - Section-divider banners and one-line section labels (# Display utilities, # ----- list_requests -----, # CVSS breakdown, etc.). - Module-level prose docstrings on internal modules. Kept one-line summaries; trimmed multi-paragraph narration about SDK/Strix responsibility splits, cache strategies, three-source precedence. - Internal-helper docstrings that just restate the function name — caido_api helpers (caido_url, get_client, view_request, etc.), settings-class one-liners (LLMSettings, RuntimeSettings, ...), UI helper docstrings. - Args/Returns blocks on non-LLM-facing internal helpers (build_strix_agent, render_system_prompt, create_or_reuse, bootstrap_caido) — kept only the genuinely non-obvious params. - Internal-history phrasing — "Mirrors main-branch shape", "pre-SDK harness", "previous lookup matched no attribute". - Narrative comments inside function bodies that explained what the next line does, design rationale obvious from the surrounding code, or "we used to..." asides. - Trailing periods on every error-string literal across the tool tree. - Duplicated roundtripTime quirk comment (kept the LLM-facing copy in tools/proxy/tools.py). Kept (every one names an upstream bug, vendored-code provenance, or non-obvious data quirk): - core/runner.py: SDK replay-with-empty-initial-input + on_agent_end lifecycle gap. - runtime/docker_client.py: VERBATIM COPY block of the upstream _create_container body, pinned to SDK v0.14.6. - runtime/session_manager.py: NO_PROXY for agent-browser CDP loopback. - tools/proxy/caido_api.py: generated-pydantic Request.raw quirk, replay double-history pitfall. - tools/proxy/tools.py: Caido roundtripTime=0 quirk for proxy captures.
This commit is contained in:
@@ -93,9 +93,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
"local_sources": getattr(args, "local_sources", None) or [],
|
||||
"scope_mode": getattr(args, "scope_mode", "auto"),
|
||||
"diff_base": getattr(args, "diff_base", None),
|
||||
# Forward the new --instruction (if any) to the resume path so it
|
||||
# can deliver it as a fresh user message after SDK session replay.
|
||||
# Empty string when the user didn't pass one on resume — no-op.
|
||||
"resume_instruction": getattr(args, "user_explicit_instruction", None) or "",
|
||||
}
|
||||
|
||||
@@ -190,10 +187,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
finally:
|
||||
stop_updates.set()
|
||||
update_thread.join(timeout=1)
|
||||
# Best-effort: tear down the sandbox session even if the
|
||||
# run raised. ``run_strix_scan`` already does this in its
|
||||
# own ``finally``, but call here too in case the failure
|
||||
# was during early setup.
|
||||
with contextlib.suppress(Exception):
|
||||
await session_manager.cleanup(args.run_name)
|
||||
|
||||
|
||||
@@ -54,13 +54,6 @@ HOST_GATEWAY_HOSTNAME = "host.docker.internal"
|
||||
import logging # noqa: E402
|
||||
|
||||
|
||||
# Per-scan logging is set up by ``setup_scan_logging`` from inside
|
||||
# ``core.runner.run_strix_scan`` once the scan ``run_dir`` is
|
||||
# known — that's where ``strix.*`` levels and handlers are owned. Pre-scan
|
||||
# work (``main()``, env validation, image pull) emits via the module
|
||||
# logger; once setup_scan_logging runs, those records start landing in
|
||||
# the file too.
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -423,9 +416,6 @@ Examples:
|
||||
except Exception as e:
|
||||
parser.error(f"Failed to read instruction file '{instruction_path}': {e}")
|
||||
|
||||
# Capture before ``_load_resume_state`` overrides — used by the resume
|
||||
# path in ``run_strix_scan`` to decide whether to inject the new
|
||||
# instruction into the root's SDK session after replay.
|
||||
args.user_explicit_instruction = args.instruction if args.resume else None
|
||||
|
||||
if args.resume:
|
||||
@@ -472,7 +462,6 @@ Examples:
|
||||
|
||||
|
||||
def _persist_run_record(args: argparse.Namespace) -> None:
|
||||
"""Write the single public run descriptor used by resume and reporting."""
|
||||
run_dir = run_dir_for(args.run_name)
|
||||
run_dir.mkdir(parents=True, exist_ok=True)
|
||||
run_record = {
|
||||
@@ -511,10 +500,6 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
|
||||
if not args.targets_info:
|
||||
parser.error(f"--resume {args.resume}: run.json has no targets_info")
|
||||
|
||||
# Validate any persisted ``cloned_repo_path`` still exists on disk.
|
||||
# The resume path skips re-cloning, so a missing dir would mean the
|
||||
# container mounts an empty source tree and agents silently scan
|
||||
# nothing.
|
||||
for target in args.targets_info:
|
||||
if not isinstance(target, dict):
|
||||
continue
|
||||
@@ -539,11 +524,6 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
|
||||
args.diff_scope = state.get("diff_scope")
|
||||
persisted_scan_mode = state.get("scan_mode")
|
||||
if persisted_scan_mode and args.scan_mode == "deep":
|
||||
# Default scan_mode is "deep"; only override from disk if the user
|
||||
# didn't explicitly pass a different one. (Best-effort: argparse
|
||||
# can't tell "user passed 'deep'" from "default 'deep'"; if the
|
||||
# persisted run was "quick" and user re-runs with an explicit
|
||||
# ``-m deep``, we'll honor the persisted mode. Acceptable.)
|
||||
args.scan_mode = persisted_scan_mode
|
||||
|
||||
|
||||
@@ -730,9 +710,6 @@ def main() -> None:
|
||||
else:
|
||||
args.instruction = diff_scope.instruction_block
|
||||
|
||||
# Persist the fully-resolved run descriptor so a future
|
||||
# ``--resume <run_name>`` invocation can pick up without
|
||||
# re-supplying targets / instructions / scope.
|
||||
_persist_run_record(args)
|
||||
|
||||
_telemetry_start_kwargs = {
|
||||
|
||||
@@ -262,8 +262,6 @@ class StopAgentScreen(ModalScreen): # type: ignore[misc]
|
||||
|
||||
|
||||
class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
"""Modal screen to display vulnerability details."""
|
||||
|
||||
SEVERITY_COLORS: ClassVar[dict[str, str]] = {
|
||||
"critical": "#dc2626", # Red
|
||||
"high": "#ea580c", # Orange
|
||||
@@ -390,7 +388,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
text.append("CVE: ", style=self.FIELD_STYLE)
|
||||
text.append(cve)
|
||||
|
||||
# CVSS breakdown
|
||||
cvss_breakdown = vuln.get("cvss_breakdown", {})
|
||||
if cvss_breakdown:
|
||||
cvss_parts = []
|
||||
@@ -464,12 +461,10 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
vuln = self.vulnerability
|
||||
lines: list[str] = []
|
||||
|
||||
# Title
|
||||
title = vuln.get("title", "Untitled Vulnerability")
|
||||
lines.append(f"# {title}")
|
||||
lines.append("")
|
||||
|
||||
# Metadata
|
||||
if vuln.get("id"):
|
||||
lines.append(f"**ID:** {vuln['id']}")
|
||||
if vuln.get("severity"):
|
||||
@@ -489,7 +484,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
if vuln.get("cvss") is not None:
|
||||
lines.append(f"**CVSS:** {vuln['cvss']}")
|
||||
|
||||
# CVSS Vector
|
||||
cvss_breakdown = vuln.get("cvss_breakdown", {})
|
||||
if cvss_breakdown:
|
||||
abbrevs = {
|
||||
@@ -508,21 +502,17 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
if parts:
|
||||
lines.append(f"**CVSS Vector:** {'/'.join(parts)}")
|
||||
|
||||
# Description
|
||||
lines.append("")
|
||||
lines.append("## Description")
|
||||
lines.append("")
|
||||
lines.append(vuln.get("description") or "No description provided.")
|
||||
|
||||
# Impact
|
||||
if vuln.get("impact"):
|
||||
lines.extend(["", "## Impact", "", vuln["impact"]])
|
||||
|
||||
# Technical Analysis
|
||||
if vuln.get("technical_analysis"):
|
||||
lines.extend(["", "## Technical Analysis", "", vuln["technical_analysis"]])
|
||||
|
||||
# Proof of Concept
|
||||
if vuln.get("poc_description") or vuln.get("poc_script_code"):
|
||||
lines.extend(["", "## Proof of Concept", ""])
|
||||
if vuln.get("poc_description"):
|
||||
@@ -533,7 +523,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
lines.append(vuln["poc_script_code"])
|
||||
lines.append("```")
|
||||
|
||||
# Code Analysis
|
||||
if vuln.get("code_locations"):
|
||||
lines.extend(["", "## Code Analysis", ""])
|
||||
for i, loc in enumerate(vuln["code_locations"]):
|
||||
@@ -559,7 +548,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
lines.append("```")
|
||||
lines.append("")
|
||||
|
||||
# Remediation
|
||||
if vuln.get("remediation_steps"):
|
||||
lines.extend(["", "## Remediation", "", vuln["remediation_steps"]])
|
||||
|
||||
@@ -584,8 +572,6 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
|
||||
|
||||
|
||||
class VulnerabilityItem(Static): # type: ignore[misc]
|
||||
"""A clickable vulnerability item."""
|
||||
|
||||
def __init__(self, label: Text, vuln_data: dict[str, Any], **kwargs: Any) -> None:
|
||||
super().__init__(label, **kwargs)
|
||||
self.vuln_data = vuln_data
|
||||
@@ -596,8 +582,6 @@ class VulnerabilityItem(Static): # type: ignore[misc]
|
||||
|
||||
|
||||
class VulnerabilitiesPanel(VerticalScroll): # type: ignore[misc]
|
||||
"""A scrollable panel showing found vulnerabilities with severity-colored dots."""
|
||||
|
||||
SEVERITY_COLORS: ClassVar[dict[str, str]] = {
|
||||
"critical": "#dc2626", # Red
|
||||
"high": "#ea580c", # Orange
|
||||
@@ -716,8 +700,6 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
self.live_view.hydrate_from_run_dir(self.report_state.get_run_dir())
|
||||
self._agent_graph_sync_future: Any | None = None
|
||||
|
||||
# Pre-create the coordinator here so the TUI can route stop/chat
|
||||
# commands while the scan loop runs in a worker thread.
|
||||
from strix.core.agents import AgentCoordinator
|
||||
|
||||
self.coordinator = AgentCoordinator()
|
||||
@@ -731,13 +713,10 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
self._scan_loop: asyncio.AbstractEventLoop | None = None
|
||||
self._scan_stop_event = threading.Event()
|
||||
self._scan_completed = threading.Event()
|
||||
# Captured by ``scan_target`` when the scan thread crashes; read
|
||||
# by ``run_tui`` after ``run_async()`` returns so the user sees
|
||||
# the traceback on stderr instead of just a silent UI hang.
|
||||
self._scan_error: BaseException | None = None
|
||||
|
||||
self._spinner_frame_index: int = 0 # Current animation frame index
|
||||
self._sweep_num_squares: int = 6 # Number of squares in sweep animation
|
||||
self._spinner_frame_index: int = 0
|
||||
self._sweep_num_squares: int = 6
|
||||
self._sweep_colors: list[str] = [
|
||||
"#000000", # Dimmest (shows dot)
|
||||
"#031a09",
|
||||
@@ -764,8 +743,6 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
"local_sources": getattr(args, "local_sources", None) or [],
|
||||
"scope_mode": getattr(args, "scope_mode", "auto"),
|
||||
"diff_base": getattr(args, "diff_base", None),
|
||||
# Forward the new --instruction (if any) so the resume path
|
||||
# can deliver it as a fresh user message after session replay.
|
||||
"resume_instruction": getattr(args, "user_explicit_instruction", None) or "",
|
||||
}
|
||||
|
||||
@@ -1378,9 +1355,6 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
try:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
# Stash the loop so synchronous TUI handlers (stop /
|
||||
# chat) can submit coordinator coroutines onto it from the
|
||||
# main thread.
|
||||
self._scan_loop = loop
|
||||
|
||||
try:
|
||||
@@ -1410,8 +1384,6 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
logging.exception("Unexpected error during scan")
|
||||
self._scan_error = e
|
||||
finally:
|
||||
# Best-effort sandbox teardown if early setup failed
|
||||
# before run_strix_scan's own ``finally`` ran.
|
||||
with contextlib.suppress(Exception):
|
||||
loop.run_until_complete(
|
||||
session_manager.cleanup(self.scan_config["run_name"]),
|
||||
@@ -1722,11 +1694,6 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
return agent_name, False
|
||||
|
||||
def action_confirm_stop_agent(self, agent_id: str) -> None:
|
||||
# Graceful stop: each agent's current turn finishes (and is saved to
|
||||
# session) before the run loop honors the cancel. The interactive
|
||||
# outer loop parks with status="stopped".
|
||||
# The hard ``cancel_descendants`` path remains for KeyboardInterrupt
|
||||
# in entry.py where graceful isn't possible.
|
||||
if self._scan_loop is None or self._scan_loop.is_closed():
|
||||
logger.warning("No active scan loop; cannot stop agent %s", agent_id)
|
||||
return
|
||||
@@ -1869,12 +1836,7 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
|
||||
|
||||
async def run_tui(args: argparse.Namespace) -> None:
|
||||
"""Run strix in interactive TUI mode with textual."""
|
||||
app = StrixTUIApp(args)
|
||||
await app.run_async()
|
||||
# Propagate scan-thread failures: ``app.run_async`` returns normally
|
||||
# when the user quits (ctrl-q) regardless of whether the scan
|
||||
# crashed. Without this re-raise, ``main.py`` would treat a failed
|
||||
# scan as success and print the completion banner.
|
||||
if app._scan_error is not None:
|
||||
raise app._scan_error
|
||||
|
||||
@@ -15,8 +15,6 @@ from strix.interface.tui.history import load_session_history
|
||||
|
||||
|
||||
class TuiLiveView:
|
||||
"""UI projection of agent state plus SDK stream/session events."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.agents: dict[str, dict[str, Any]] = {}
|
||||
self.events: list[dict[str, Any]] = []
|
||||
|
||||
@@ -18,7 +18,6 @@ def send_user_message_to_agent(
|
||||
target_agent_id: str,
|
||||
message: str,
|
||||
) -> bool:
|
||||
"""Record a local user message and enqueue it into the target SDK session."""
|
||||
if loop is None or loop.is_closed():
|
||||
return False
|
||||
|
||||
|
||||
@@ -17,7 +17,6 @@ def _truncate(text: str, max_len: int = 80) -> str:
|
||||
|
||||
|
||||
def _sanitize(text: str, max_len: int = 150) -> str:
|
||||
"""Remove newlines and truncate text."""
|
||||
clean = text.replace("\n", " ").replace("\r", "").replace("\t", " ")
|
||||
return _truncate(clean, max_len)
|
||||
|
||||
|
||||
@@ -23,7 +23,6 @@ from rich.text import Text
|
||||
from strix.config import load_settings
|
||||
|
||||
|
||||
# Display utilities
|
||||
def get_severity_color(severity: str) -> str:
|
||||
severity_colors = {
|
||||
"critical": "#dc2626",
|
||||
@@ -57,7 +56,6 @@ def format_token_count(count: float | None) -> str:
|
||||
|
||||
|
||||
def format_vulnerability_report(report: dict[str, Any]) -> Text: # noqa: PLR0915
|
||||
"""Format a vulnerability report for CLI display with all rich fields."""
|
||||
field_style = "bold #4ade80"
|
||||
|
||||
text = Text()
|
||||
@@ -206,7 +204,6 @@ def format_vulnerability_report(report: dict[str, Any]) -> Text: # noqa: PLR091
|
||||
|
||||
|
||||
def _build_vulnerability_stats(stats_text: Text, report_state: Any) -> None:
|
||||
"""Build vulnerability section of stats text."""
|
||||
vuln_count = len(report_state.vulnerability_reports)
|
||||
|
||||
if vuln_count > 0:
|
||||
@@ -318,7 +315,6 @@ def _build_llm_usage_stats(
|
||||
|
||||
|
||||
def build_final_stats_text(report_state: Any) -> Text:
|
||||
"""Build final stats from Strix-owned scan artifacts."""
|
||||
stats_text = Text()
|
||||
if not report_state:
|
||||
return stats_text
|
||||
@@ -401,9 +397,6 @@ def build_tui_stats_text(report_state: Any) -> Text:
|
||||
return stats_text
|
||||
|
||||
|
||||
# Name generation utilities
|
||||
|
||||
|
||||
def _slugify_for_run_name(text: str, max_length: int = 32) -> str:
|
||||
text = text.lower().strip()
|
||||
text = re.sub(r"[^a-z0-9]+", "-", text)
|
||||
@@ -461,8 +454,6 @@ def generate_run_name(targets_info: list[dict[str, Any]] | None = None) -> str:
|
||||
return f"{slug}_{random_suffix}"
|
||||
|
||||
|
||||
# Target processing utilities
|
||||
|
||||
_SUPPORTED_SCOPE_MODES = {"auto", "diff", "full"}
|
||||
_MAX_FILES_PER_SECTION = 120
|
||||
|
||||
@@ -712,9 +703,6 @@ def _parse_name_status_z(raw_output: bytes) -> list[DiffEntry]:
|
||||
if len(status_raw) > 1 and status_raw[1:].isdigit():
|
||||
similarity = int(status_raw[1:])
|
||||
|
||||
# Git's -z output for --name-status is:
|
||||
# - non-rename/copy: <status>\0<path>\0
|
||||
# - rename/copy: <statusN>\0<old_path>\0<new_path>\0
|
||||
if status_code in {"R", "C"} and index + 2 < len(tokens):
|
||||
old_path = tokens[index + 1]
|
||||
new_path = tokens[index + 2]
|
||||
@@ -1264,7 +1252,6 @@ def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway:
|
||||
details["target_ip"] = host_gateway
|
||||
|
||||
|
||||
# Repository utilities
|
||||
def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None) -> str:
|
||||
console = Console()
|
||||
|
||||
@@ -1341,7 +1328,6 @@ def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# Docker utilities
|
||||
def check_docker_connection() -> Any:
|
||||
try:
|
||||
return docker.from_env()
|
||||
|
||||
Reference in New Issue
Block a user