From 4e0d0f35d92c9ded354ae26eaf194107efedf6a8 Mon Sep 17 00:00:00 2001 From: 0xallam Date: Sat, 25 Apr 2026 08:03:00 -0700 Subject: [PATCH] =?UTF-8?q?feat(migration):=20phase=205b=20=E2=80=94=20STR?= =?UTF-8?q?IX=5FUSE=5FSDK=5FHARNESS=20dispatch=20flag?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the env-var gate that lets users opt into the SDK harness without disturbing the legacy default. Per PLAYBOOK §7.1, this is the cutover mechanism: STRIX_USE_SDK_HARNESS=1 routes scans through run_strix_scan (the Phase 5 entry point); anything else continues to use StrixAgent.execute_scan. - strix/interface/sdk_dispatch.py: - should_use_sdk_harness(): truthy-string parse of the env var. - _resolve_sandbox_image(): reads strix_image from Config; falls back to "strix-sandbox:latest" with a warning if unset. - _resolve_sources_path(): when --local-sources is given, mounts its parent so the agent walks down to the source tree; otherwise creates a per-run scratch dir under XDG_CACHE_HOME/strix/sources/. Phase 6 will replace this with the legacy clone-into-container flow once we port that. - run_scan_via_sdk(): the adapter — translates the legacy CLI (scan_config dict + argparse Namespace + Tracer) into the keyword arguments run_strix_scan expects. Returns the SDK RunResult; lets failures bubble up. - strix/interface/cli.py: adds the dispatch branch inside the existing Live/status loop. Legacy default unchanged; SDK path is reached only when STRIX_USE_SDK_HARNESS is truthy. Two pre-existing lazy imports hoisted to module level (cleanup_runtime + sdk_dispatch helpers) so ruff is happy. Pre-existing legacy lint/type issues surfaced when pre-commit checked the edited cli.py and chased imports — fixed or ignored in passing: - utils.py:1052 duplicate ``metadata`` annotation removed. - utils.py:1251 unused ``# type: ignore[import-not-found]`` for yarl. - main.py:456 ``panel_parts`` inferred type rejected later string entries — explicit ``list[Text | str]`` annotation. - utils.py:resolve_diff_scope_context PLR0912 (16 branches) per-file ignore — branches map 1:1 to scope-mode × target-type combinations. Tests: 18 new tests in tests/interface/test_sdk_dispatch.py — env flag parsing parametrized over truthy/falsy variants, image lookup with config hit + miss-with-warning, sources path resolution for local_sources / alternative key names / scratch-dir creation, and the adapter's kwarg handoff verified against a patched run_strix_scan (run_name from args + run_name from scan_config fallback + failure propagation). Refs: PLAYBOOK.md §7.1 (cutover), §7.2 (rollback). Co-Authored-By: Claude Opus 4.7 (1M context) --- pyproject.toml | 4 + strix/interface/cli.py | 37 +++-- strix/interface/main.py | 2 +- strix/interface/sdk_dispatch.py | 143 +++++++++++++++++++ strix/interface/utils.py | 10 +- tests/interface/test_sdk_dispatch.py | 199 +++++++++++++++++++++++++++ 6 files changed, 376 insertions(+), 19 deletions(-) create mode 100644 strix/interface/sdk_dispatch.py create mode 100644 tests/interface/test_sdk_dispatch.py diff --git a/pyproject.toml b/pyproject.toml index f31a8cb..e9cfa3d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -312,6 +312,10 @@ ignore = [ # walks every supported target type — splitting it into per-type # helpers would add indirection without simplifying anything. "strix/sdk_entry.py" = ["TC003", "PLR0912"] +# Legacy interface utility with intentionally many branches per supported +# scope-mode / target-type combination; refactor would obscure the +# decision tree without simplifying it. +"strix/interface/utils.py" = ["PLR0912"] # Sandbox dispatch helper has many short-circuit error returns (auth fail, # size cap, decode fail, etc). Each is a distinct, documented failure mode # the model needs to see verbatim — collapsing them harms readability. diff --git a/strix/interface/cli.py b/strix/interface/cli.py index ec853b3..6004abe 100644 --- a/strix/interface/cli.py +++ b/strix/interface/cli.py @@ -11,7 +11,9 @@ from rich.panel import Panel from rich.text import Text from strix.agents.StrixAgent import StrixAgent +from strix.interface.sdk_dispatch import run_scan_via_sdk, should_use_sdk_harness from strix.llm.config import LLMConfig +from strix.runtime import cleanup_runtime from strix.telemetry.tracer import Tracer, set_global_tracer from .utils import ( @@ -109,8 +111,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915 tracer.vulnerability_found_callback = display_vulnerability def cleanup_on_exit() -> None: - from strix.runtime import cleanup_runtime - tracer.cleanup() cleanup_runtime() @@ -163,18 +163,29 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915 update_thread.start() try: - agent = StrixAgent(agent_config) - result = await agent.execute_scan(scan_config) + if should_use_sdk_harness(): + # SDK harness opt-in (PLAYBOOK §7.1). Returns a + # RunResult, not the legacy success-dict shape, so + # we skip the legacy error-extraction block — + # failures inside run_strix_scan raise instead. + await run_scan_via_sdk( + scan_config=scan_config, + args=args, + tracer=tracer, + ) + else: + agent = StrixAgent(agent_config) + result = await agent.execute_scan(scan_config) - if isinstance(result, dict) and not result.get("success", True): - error_msg = result.get("error", "Unknown error") - error_details = result.get("details") - console.print() - console.print(f"[bold red]Penetration test failed:[/] {error_msg}") - if error_details: - console.print(f"[dim]{error_details}[/]") - console.print() - sys.exit(1) + if isinstance(result, dict) and not result.get("success", True): + error_msg = result.get("error", "Unknown error") + error_details = result.get("details") + console.print() + console.print(f"[bold red]Penetration test failed:[/] {error_msg}") + if error_details: + console.print(f"[dim]{error_details}[/]") + console.print() + sys.exit(1) finally: stop_updates.set() update_thread.join(timeout=1) diff --git a/strix/interface/main.py b/strix/interface/main.py index bc88da6..a2323aa 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -453,7 +453,7 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) -> stats_text = build_final_stats_text(tracer) - panel_parts = [completion_text, "\n\n", target_text] + panel_parts: list[Text | str] = [completion_text, "\n\n", target_text] if stats_text.plain: panel_parts.extend(["\n", stats_text]) diff --git a/strix/interface/sdk_dispatch.py b/strix/interface/sdk_dispatch.py new file mode 100644 index 0000000..3cbaf34 --- /dev/null +++ b/strix/interface/sdk_dispatch.py @@ -0,0 +1,143 @@ +"""STRIX_USE_SDK_HARNESS dispatch — selects legacy vs SDK harness at run-time. + +Phase 5b cutover gate. The legacy CLI (``strix.interface.cli``) calls +``StrixAgent(...).execute_scan(scan_config)`` directly. To roll out the +SDK migration safely we want a single env-var-gated branch: + + STRIX_USE_SDK_HARNESS=1 → await run_strix_scan(...) + STRIX_USE_SDK_HARNESS=0 → await StrixAgent(...).execute_scan(...) + +This module is a thin adapter: it reads the env var, and when the SDK +path is active, translates the legacy ``scan_config`` + ``args`` pair +into the keyword arguments :func:`run_strix_scan` expects. + +Per PLAYBOOK §7.1: the legacy default stays in place until end-to-end +validation against a stable target succeeds; the env flag is the +opt-in. Removal of the legacy branch happens one release after cutover. + +References: + - PLAYBOOK.md §7.1 (cutover strategy) + - PLAYBOOK.md §7.2 (rollback procedure) +""" + +from __future__ import annotations + +import logging +import os +from pathlib import Path +from typing import TYPE_CHECKING, Any + + +if TYPE_CHECKING: + from agents.result import RunResult + + +logger = logging.getLogger(__name__) + + +_ENV_FLAG = "STRIX_USE_SDK_HARNESS" + + +def should_use_sdk_harness() -> bool: + """Return True iff ``STRIX_USE_SDK_HARNESS`` is truthy in the env. + + Truthy values: ``"1"``, ``"true"``, ``"yes"`` (case-insensitive). + Anything else — including unset — returns False so the default + deployed posture stays the legacy harness. + """ + raw = os.environ.get(_ENV_FLAG, "") + return raw.strip().lower() in {"1", "true", "yes"} + + +def _resolve_sandbox_image() -> str: + """Read the sandbox image tag from Strix config. + + Falls back to ``"strix-sandbox:latest"`` if unset — same behavior + the legacy ``DockerRuntime`` would surface as a config error. + """ + from strix.config import Config + + image = Config.get("strix_image") + if not image: + logger.warning( + "strix_image not configured; falling back to strix-sandbox:latest. " + "Set this in ~/.strix/cli-config.json for production use.", + ) + return "strix-sandbox:latest" + return str(image) + + +def _resolve_sources_path(args: Any) -> Path: + """Pick the host directory to mount into ``/workspace/sources``. + + - When ``--local-sources`` was passed, use the parent of the first + source's ``host_path`` (the legacy harness then copies each + individual source under ``/workspace/``; we mount the + parent and let the agent walk down). + - Otherwise, use a per-run scratch directory under + ``$XDG_CACHE_HOME/strix`` (or ``~/.cache/strix``) — the legacy + flow eventually populates ``/workspace`` via post-create copies, + which the SDK session manager doesn't replicate yet (Phase 6 + will bring that in). + """ + local_sources: list[dict[str, str]] | None = getattr(args, "local_sources", None) + if local_sources: + first = local_sources[0] + host_path = first.get("host_path") or first.get("source_path") or first.get("path") + if host_path: + return Path(host_path).expanduser().resolve().parent + + cache_root = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache") + run_name = getattr(args, "run_name", "default") or "default" + sources = Path(cache_root) / "strix" / "sources" / str(run_name) + sources.mkdir(parents=True, exist_ok=True) + return sources + + +async def run_scan_via_sdk( + *, + scan_config: dict[str, Any], + args: Any, + tracer: Any, +) -> RunResult: + """Translate legacy CLI args into ``run_strix_scan`` kwargs. + + Args: + scan_config: The same dict the legacy ``StrixAgent.execute_scan`` + accepts. Forwarded verbatim to ``run_strix_scan``; the + entry point reads ``targets``, ``user_instructions``, + ``diff_scope``, ``scan_mode``, ``is_whitebox``, ``skills`` + from it. + args: argparse Namespace from ``strix.interface.cli``. We read + ``run_name``, ``local_sources``, ``scan_mode`` from it. + tracer: Live ``Tracer`` instance — flows through context so + tools (``create_vulnerability_report``, ``finish_scan``) + persist into the same on-disk run directory the legacy + path uses. + + Returns the SDK ``RunResult``. Raises whatever ``run_strix_scan`` + raises (sandbox bring-up failure, LLM error, etc.). + """ + from strix.sdk_entry import run_strix_scan + + run_name = getattr(args, "run_name", None) or scan_config.get("run_name") + image = _resolve_sandbox_image() + sources_path = _resolve_sources_path(args) + interactive = bool(getattr(args, "interactive", False)) + + logger.info( + "STRIX_USE_SDK_HARNESS active; dispatching scan %s via run_strix_scan " + "(image=%s, sources=%s)", + run_name, + image, + sources_path, + ) + + return await run_strix_scan( + scan_config=scan_config, + scan_id=run_name, + image=image, + sources_path=sources_path, + tracer=tracer, + interactive=interactive, + ) diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 3559fa9..aff33a0 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -55,7 +55,7 @@ def get_cvss_color(cvss_score: float) -> str: return "#6b7280" -def format_vulnerability_report(report: dict[str, Any]) -> Text: # noqa: PLR0912, PLR0915 +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" @@ -823,7 +823,7 @@ def _truncate_file_list( return files[:max_files], True -def build_diff_scope_instruction(scopes: list[RepoDiffScope]) -> str: # noqa: PLR0912 +def build_diff_scope_instruction(scopes: list[RepoDiffScope]) -> str: lines = [ "The user is requesting a review of a Pull Request.", "Instruction: Direct your analysis primarily at the changes in the listed files. " @@ -1049,7 +1049,7 @@ def resolve_diff_scope_context( ) instruction_block = build_diff_scope_instruction(repo_scopes) - metadata: dict[str, Any] = { + metadata = { "active": True, "mode": scope_mode, "repos": [scope.to_metadata() for scope in repo_scopes], @@ -1082,7 +1082,7 @@ def _is_http_git_repo(url: str) -> bool: return False -def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911, PLR0912 +def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911 if not target or not isinstance(target, str): raise ValueError("Target must be a non-empty string") @@ -1248,7 +1248,7 @@ def _is_localhost_host(host: str) -> bool: def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway: str) -> None: - from yarl import URL # type: ignore[import-not-found] + from yarl import URL for target_info in targets_info: target_type = target_info.get("type") diff --git a/tests/interface/test_sdk_dispatch.py b/tests/interface/test_sdk_dispatch.py new file mode 100644 index 0000000..5591202 --- /dev/null +++ b/tests/interface/test_sdk_dispatch.py @@ -0,0 +1,199 @@ +"""Phase 5b tests for the STRIX_USE_SDK_HARNESS dispatch. + +Covers the env-flag reader, source-path resolution, sandbox image +lookup, and the adapter that translates legacy CLI args into +``run_strix_scan`` kwargs. + +We never call ``run_strix_scan`` for real — that requires a live +Docker daemon + LLM. The tests patch it and verify the kwargs handoff. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from strix.interface.sdk_dispatch import ( + _resolve_sandbox_image, + _resolve_sources_path, + run_scan_via_sdk, + should_use_sdk_harness, +) + + +# --- env flag reader ---------------------------------------------------- + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("1", True), + ("true", True), + ("True", True), + ("YES", True), + ("0", False), + ("false", False), + ("no", False), + ("", False), + ("anything-else", False), + ], +) +def test_should_use_sdk_harness_parses_env( + monkeypatch: pytest.MonkeyPatch, + value: str, + expected: bool, +) -> None: + monkeypatch.setenv("STRIX_USE_SDK_HARNESS", value) + assert should_use_sdk_harness() is expected + + +def test_should_use_sdk_harness_defaults_false_when_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("STRIX_USE_SDK_HARNESS", raising=False) + assert should_use_sdk_harness() is False + + +# --- image lookup ------------------------------------------------------- + + +def test_resolve_sandbox_image_uses_config_value() -> None: + with patch( + "strix.config.Config.get", + return_value="strix-sandbox:0.1.13", + ): + assert _resolve_sandbox_image() == "strix-sandbox:0.1.13" + + +def test_resolve_sandbox_image_falls_back_when_unset( + caplog: pytest.LogCaptureFixture, +) -> None: + with ( + patch("strix.config.Config.get", return_value=None), + caplog.at_level(logging.WARNING, logger="strix.interface.sdk_dispatch"), + ): + out = _resolve_sandbox_image() + assert out == "strix-sandbox:latest" + assert any("strix_image not configured" in r.message for r in caplog.records) + + +# --- sources path ------------------------------------------------------- + + +def test_resolve_sources_path_uses_local_sources_parent(tmp_path: Path) -> None: + """When --local-sources is given, mount that path's parent so the + agent can walk down into the actual source directory tree.""" + src_dir = tmp_path / "my-project" + src_dir.mkdir() + args = SimpleNamespace( + local_sources=[{"host_path": str(src_dir)}], + run_name="run-1", + ) + assert _resolve_sources_path(args) == tmp_path + + +def test_resolve_sources_path_handles_alternative_keys(tmp_path: Path) -> None: + """Some legacy paths use 'source_path' or 'path' instead of + 'host_path' — we accept all three.""" + src_dir = tmp_path / "alt" + src_dir.mkdir() + args = SimpleNamespace( + local_sources=[{"path": str(src_dir)}], + run_name="run-2", + ) + assert _resolve_sources_path(args) == tmp_path + + +def test_resolve_sources_path_creates_scratch_dir_when_absent( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + args = SimpleNamespace(local_sources=None, run_name="scan-x") + out = _resolve_sources_path(args) + assert out == tmp_path / "strix" / "sources" / "scan-x" + assert out.exists() + assert out.is_dir() + + +# --- adapter ----------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_run_scan_via_sdk_translates_args_to_kwargs( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Verify every kwarg the entry point reads is forwarded correctly.""" + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + + scan_config = {"targets": [], "scan_mode": "deep"} + args = SimpleNamespace( + run_name="scan-42", + local_sources=None, + interactive=True, + ) + fake_tracer = MagicMock(name="tracer") + + fake_run = AsyncMock(return_value=MagicMock(name="run_result")) + with ( + patch("strix.config.Config.get", return_value="strix-sandbox:test"), + patch("strix.sdk_entry.run_strix_scan", new=fake_run), + ): + await run_scan_via_sdk(scan_config=scan_config, args=args, tracer=fake_tracer) + + fake_run.assert_awaited_once() + assert fake_run.await_args is not None + kwargs = fake_run.await_args.kwargs + assert kwargs["scan_config"] is scan_config + assert kwargs["scan_id"] == "scan-42" + assert kwargs["image"] == "strix-sandbox:test" + assert kwargs["sources_path"] == tmp_path / "strix" / "sources" / "scan-42" + assert kwargs["tracer"] is fake_tracer + assert kwargs["interactive"] is True + + +@pytest.mark.asyncio +async def test_run_scan_via_sdk_falls_back_to_scan_config_run_name( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """If args has no run_name, scan_config['run_name'] should be used.""" + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + scan_config = {"targets": [], "run_name": "from-config"} + args = SimpleNamespace(local_sources=None) + + fake_run = AsyncMock(return_value=MagicMock()) + with ( + patch("strix.config.Config.get", return_value="img:1"), + patch("strix.sdk_entry.run_strix_scan", new=fake_run), + ): + await run_scan_via_sdk(scan_config=scan_config, args=args, tracer=None) + + assert fake_run.await_args is not None + assert fake_run.await_args.kwargs["scan_id"] == "from-config" + + +@pytest.mark.asyncio +async def test_run_scan_via_sdk_propagates_run_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failure inside run_strix_scan should bubble up to the caller — + the legacy CLI relies on raised exceptions for the SDK path.""" + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + scan_config: dict[str, Any] = {"targets": []} + args = SimpleNamespace(run_name="r", local_sources=None) + + fake_run = AsyncMock(side_effect=RuntimeError("boom")) + with ( + patch("strix.config.Config.get", return_value="img"), + patch("strix.sdk_entry.run_strix_scan", new=fake_run), + pytest.raises(RuntimeError, match="boom"), + ): + await run_scan_via_sdk(scan_config=scan_config, args=args, tracer=None)