From 76fb6fe3237a580bc736c61334e916c9b596d89e Mon Sep 17 00:00:00 2001 From: Ben Younes <2910651+ousamabenyounes@users.noreply.github.com> Date: Tue, 8 Sep 2026 03:06:02 +0000 Subject: [PATCH] fix(report): validate baseline runs Reject path-shaped, empty, and malformed baseline selections before CLI/TUI startup, including restored run state. Refs #900 --- strix/interface/cli_args.py | 24 +++- strix/report/state.py | 7 +- tests/test_cli_target_list.py | 223 +++++++++++++++++++++++++++++++++ tests/test_go_tui_runtime.py | 28 +++++ tests/test_list_reports.py | 39 ++++++ tests/test_report_writer.py | 22 ++++ tests/test_reporting_fields.py | 35 +++--- 7 files changed, 353 insertions(+), 25 deletions(-) diff --git a/strix/interface/cli_args.py b/strix/interface/cli_args.py index 58eb5a7f..5e927261 100644 --- a/strix/interface/cli_args.py +++ b/strix/interface/cli_args.py @@ -19,7 +19,9 @@ from strix.interface.utils import ( validate_config_file, ) + BASELINE_RUN_ARG = "--baseline-run" +RUN_NAME_PATH_SEPARATORS = ("/", "\\") def get_version() -> str: @@ -356,7 +358,7 @@ Strix Cloud: args.user_instruction = args.instruction or None if args.resume: - if args.target or args.target_list or args.baseline_run: + if args.target or args.target_list or args.baseline_run is not None: parser.error( "Cannot combine --resume with --target/--target-list/--baseline-run. " "--resume picks up where the prior run left off, including the " @@ -372,6 +374,9 @@ Strix Cloud: f"or remove --resume to start over with the same targets." ) else: + if args.baseline_run is not None: + _validate_baseline_run(args.baseline_run, parser) + if not args.target and not args.target_list: if args.non_interactive: parser.error( @@ -389,15 +394,22 @@ Strix Cloud: except ValueError as e: parser.error(str(e)) - if args.baseline_run: - _validate_baseline_run(args.baseline_run, parser) - return args -def _validate_baseline_run(baseline_run: str, parser: argparse.ArgumentParser) -> None: +def _validate_baseline_run(baseline_run: object, parser: argparse.ArgumentParser) -> None: from strix.report.writer import read_vulnerabilities + if not isinstance(baseline_run, str): + parser.error(f"{BASELINE_RUN_ARG} must be a run name") + if not baseline_run: + parser.error(f"{BASELINE_RUN_ARG} must be a non-empty run name") + if ( + Path(baseline_run).is_absolute() + or baseline_run in {".", ".."} + or any(separator in baseline_run for separator in RUN_NAME_PATH_SEPARATORS) + ): + parser.error(f"{BASELINE_RUN_ARG} must be a run name, not a path") baseline_run_dir = run_dir_for(baseline_run) try: read_vulnerabilities(baseline_run_dir) @@ -487,6 +499,8 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser if state.get("diff_scope"): args.diff_scope = state.get("diff_scope") args.baseline_run = state.get("baseline_run") + if args.baseline_run is not None: + _validate_baseline_run(args.baseline_run, parser) persisted_scan_mode = state.get("scan_mode") if persisted_scan_mode and args.scan_mode == "deep": args.scan_mode = persisted_scan_mode diff --git a/strix/report/state.py b/strix/report/state.py index 5530a73d..605e9fc1 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -16,6 +16,7 @@ from strix.report.coverage import write_coverage from strix.report.pricing import resolve_litellm_model from strix.report.sarif import write_sarif from strix.report.writer import ( + VULNERABILITIES_FILENAME, read_run_record, read_vulnerabilities, write_executive_report, @@ -193,7 +194,6 @@ class ReportState: self.vulnerability_reports: list[dict[str, Any]] = [] self.baseline_vulnerability_reports: list[dict[str, Any]] = [] - self.baseline_run_name: str | None = None self.final_scan_result: str | None = None self.scan_results: dict[str, Any] | None = None @@ -273,13 +273,13 @@ class ReportState: self._telemetry_llm_usage_baseline = self._build_llm_usage_record() logger.info("report state hydrated run.json from %s", run_dir) - json_path = run_dir / "vulnerabilities.json" + json_path = run_dir / VULNERABILITIES_FILENAME if json_path.exists(): try: self.vulnerability_reports = read_vulnerabilities(run_dir) except (RuntimeError, TypeError) as exc: raise RuntimeError( - f"vulnerabilities.json at {json_path} is corrupt ({exc}); " + f"{VULNERABILITIES_FILENAME} at {json_path} is corrupt ({exc}); " f"refusing to start fresh — that would overwrite prior " f"vulnerability MDs on disk. Inspect or delete the run dir.", ) from exc @@ -310,7 +310,6 @@ class ReportState: baseline_run_name: str, vulnerability_reports: list[dict[str, Any]], ) -> None: - self.baseline_run_name = baseline_run_name self.baseline_vulnerability_reports = list(vulnerability_reports) logger.info( "loaded %d baseline vulnerability report(s) from %s", diff --git a/tests/test_cli_target_list.py b/tests/test_cli_target_list.py index fe2afd14..70d92c5e 100644 --- a/tests/test_cli_target_list.py +++ b/tests/test_cli_target_list.py @@ -5,6 +5,7 @@ from __future__ import annotations import importlib import json import sys +from contextlib import nullcontext from types import SimpleNamespace from typing import TYPE_CHECKING, Any @@ -16,11 +17,14 @@ if TYPE_CHECKING: cli_main: Any = importlib.import_module("strix.interface.main") +cli_runtime: Any = importlib.import_module("strix.interface.cli") +cli_args: Any = importlib.import_module("strix.interface.cli_args") BASELINE_RUN_NAME = "baseline-alpha" RUNS_DIR_NAME = "strix_runs" VULNERABILITIES_FILENAME = "vulnerabilities.json" TARGET_URL = "https://test1.com/" +OUTSIDE_RUN_NAME = "outside-baseline" def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None: @@ -94,6 +98,110 @@ def test_parse_arguments_accepts_baseline_run( assert args.baseline_run == BASELINE_RUN_NAME +def test_parse_arguments_rejects_empty_baseline_run( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + _stub_settings(monkeypatch) + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + sys, + "argv", + ["strix", "--target", TARGET_URL, "--baseline-run", ""], + ) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "must be a non-empty run name" in capsys.readouterr().err + + +@pytest.mark.parametrize("path_kind", ["relative", "absolute"]) +def test_parse_arguments_rejects_baseline_paths_outside_runs_dir( + path_kind: str, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + work_dir = tmp_path / "work" + work_dir.mkdir() + (work_dir / RUNS_DIR_NAME).mkdir() + is_absolute = path_kind == "absolute" + outside_run_dir = tmp_path / OUTSIDE_RUN_NAME if is_absolute else work_dir / OUTSIDE_RUN_NAME + outside_run_dir.mkdir() + (outside_run_dir / VULNERABILITIES_FILENAME).write_text("[]", encoding="utf-8") + _stub_settings(monkeypatch) + monkeypatch.chdir(work_dir) + absolute_or_relative = str(outside_run_dir) if is_absolute else f"../{OUTSIDE_RUN_NAME}" + monkeypatch.setattr( + sys, + "argv", + ["strix", "--target", TARGET_URL, "--baseline-run", absolute_or_relative], + ) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "must be a run name, not a path" in capsys.readouterr().err + + +def test_parse_arguments_validates_baseline_before_interactive_setup( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + outside_run_dir = tmp_path / OUTSIDE_RUN_NAME + outside_run_dir.mkdir() + (outside_run_dir / VULNERABILITIES_FILENAME).write_text("[]", encoding="utf-8") + work_dir = tmp_path / "work" + work_dir.mkdir() + _stub_settings(monkeypatch) + monkeypatch.chdir(work_dir) + monkeypatch.setattr(sys, "argv", ["strix", "--baseline-run", f"../{OUTSIDE_RUN_NAME}"]) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "must be a run name, not a path" in capsys.readouterr().err + + +def test_parse_arguments_rejects_unreadable_baseline_run( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + _stub_settings(monkeypatch) + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + sys, + "argv", + ["strix", "--target", TARGET_URL, "--baseline-run", BASELINE_RUN_NAME], + ) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "vulnerabilities.json" in capsys.readouterr().err + + +def test_parse_arguments_reports_target_validation_error( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + def reject_target(_args: object) -> None: + raise ValueError("invalid test target") + + _stub_settings(monkeypatch) + monkeypatch.setattr(sys, "argv", ["strix", "--target", TARGET_URL]) + monkeypatch.setattr(cli_args, "build_targets_info", reject_target) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "invalid test target" in capsys.readouterr().err + + def test_parse_arguments_rejects_resume_with_target_list( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: @@ -121,6 +229,121 @@ def _write_run_record(runs_dir: Path, run_name: str, record: dict[str, Any]) -> (state_dir / "agents.json").write_text("{}", encoding="utf-8") +def test_resume_restores_and_validates_baseline_run( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runs_dir = tmp_path / RUNS_DIR_NAME + _write_run_record( + runs_dir, + "pentest-resume", + { + "run_name": "pentest-resume", + "targets_info": [{"type": "web_application", "original": TARGET_URL, "details": {}}], + "baseline_run": BASELINE_RUN_NAME, + }, + ) + baseline_dir = runs_dir / BASELINE_RUN_NAME + baseline_dir.mkdir() + (baseline_dir / VULNERABILITIES_FILENAME).write_text("[]", encoding="utf-8") + _stub_settings(monkeypatch) + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "argv", ["strix", "--resume", "pentest-resume"]) + + parsed = cli_main.parse_arguments() + + assert parsed.baseline_run == BASELINE_RUN_NAME + + +@pytest.mark.parametrize( + "invalid_baseline", + [[BASELINE_RUN_NAME], []], + ids=["truthy", "empty"], +) +def test_resume_rejects_non_string_baseline_run( + invalid_baseline: object, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + _write_run_record( + tmp_path / RUNS_DIR_NAME, + "pentest-resume", + { + "run_name": "pentest-resume", + "targets_info": [{"type": "web_application", "original": TARGET_URL, "details": {}}], + "baseline_run": invalid_baseline, + }, + ) + _stub_settings(monkeypatch) + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(sys, "argv", ["strix", "--resume", "pentest-resume"]) + + with pytest.raises(SystemExit): + cli_main.parse_arguments() + + assert "must be a run name" in capsys.readouterr().err + + +@pytest.mark.asyncio +async def test_cli_runtime_hydrates_selected_baseline(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[tuple[str, object]] = [] + + class ExpectedStopError(Exception): + pass + + class FakeReportState: + final_scan_result = None + + def __init__(self, run_name: str) -> None: + calls.append(("init", run_name)) + + def hydrate_from_run_dir(self) -> None: + calls.append(("hydrate", None)) + + def hydrate_baseline_run(self, run_name: str | None) -> None: + calls.append(("baseline", run_name)) + + def set_scan_config(self, config: dict[str, Any]) -> None: + calls.append(("config", config["baseline_run"])) + + def save_run_data(self) -> None: + calls.append(("save", None)) + + def cleanup(self, *, status: str | None = None) -> None: + calls.append(("cleanup", status)) + + async def stop_scan(**_kwargs: Any) -> None: + raise ExpectedStopError + + async def cleanup_session(_run_name: str) -> None: + return None + + runtime_args = SimpleNamespace( + run_name="current-run", + targets_info=[{"original": TARGET_URL}], + instruction=None, + baseline_run=BASELINE_RUN_NAME, + ) + monkeypatch.setattr(cli_runtime, "ReportState", FakeReportState) + monkeypatch.setattr(cli_runtime, "run_strix_scan", stop_scan) + monkeypatch.setattr(cli_runtime, "_resolve_sandbox_image", lambda: "test-image") + monkeypatch.setattr(cli_runtime, "has_model_response", lambda _state: False) + monkeypatch.setattr(cli_runtime, "build_live_stats_text", lambda _state: None) + monkeypatch.setattr( + cli_runtime, "Live", lambda *_args, **_kwargs: nullcontext(SimpleNamespace()) + ) + monkeypatch.setattr(cli_runtime.atexit, "register", lambda _callback: None) + monkeypatch.setattr(cli_runtime.signal, "signal", lambda *_args: None) + monkeypatch.setattr(cli_runtime.session_manager, "cleanup", cleanup_session) + + with pytest.raises(ExpectedStopError): + await cli_runtime.run_cli(runtime_args) + + assert ("baseline", BASELINE_RUN_NAME) in calls + assert ("config", BASELINE_RUN_NAME) in calls + + def test_resume_restores_a_target_less_workspace_mount( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_go_tui_runtime.py b/tests/test_go_tui_runtime.py index 05f81ece..01d579fa 100644 --- a/tests/test_go_tui_runtime.py +++ b/tests/test_go_tui_runtime.py @@ -38,6 +38,34 @@ def args() -> argparse.Namespace: ) +@pytest.mark.asyncio +async def test_init_run_state_hydrates_selected_baseline( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + baseline_name = "baseline-run" + baseline_report = { + "id": "vuln-0001", + "title": "Known SQL injection", + "severity": "high", + } + baseline_dir = tmp_path / "strix_runs" / baseline_name + baseline_dir.mkdir(parents=True) + (baseline_dir / "vulnerabilities.json").write_text( + json.dumps([baseline_report]), + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + runtime_args = args() + runtime_args.baseline_run = baseline_name + runtime = GoTuiRuntime(runtime_args) + + runtime.init_run_state() + + assert runtime.report_state.get_dedupe_vulnerabilities() == [baseline_report] + assert runtime.report_state.get_existing_vulnerabilities() == [] + + def test_binary_command_prefers_packaged_sidecar( monkeypatch: pytest.MonkeyPatch, tmp_path: Any, diff --git a/tests/test_list_reports.py b/tests/test_list_reports.py index b0f1519d..a0399674 100644 --- a/tests/test_list_reports.py +++ b/tests/test_list_reports.py @@ -217,6 +217,45 @@ def test_list_reports_and_get_report_exclude_baseline_findings( assert baseline_lookup["report"] is None +def test_hydrate_baseline_run_reads_dedupe_only_findings( + report_state: ReportState, +) -> None: + baseline_dir = report_state.get_run_dir().parent / BASELINE_RUN_NAME + baseline_dir.mkdir() + baseline_report = { + "id": BASELINE_REPORT_ID, + "title": BASELINE_REPORT_TITLE, + "severity": BASELINE_REPORT_SEVERITY, + } + (baseline_dir / "vulnerabilities.json").write_text( + json.dumps([baseline_report]), + encoding="utf-8", + ) + + report_state.hydrate_baseline_run(BASELINE_RUN_NAME) + + assert report_state.get_dedupe_vulnerabilities() == [baseline_report] + assert report_state.get_existing_vulnerabilities() == [] + + +def test_hydrate_baseline_run_ignores_missing_selection(report_state: ReportState) -> None: + report_state.hydrate_baseline_run(None) + + assert report_state.get_dedupe_vulnerabilities() == [] + + +def test_hydrate_from_run_dir_rejects_corrupt_vulnerabilities( + report_state: ReportState, +) -> None: + (report_state.get_run_dir() / "vulnerabilities.json").write_text( + "{not-json", + encoding="utf-8", + ) + + with pytest.raises(RuntimeError, match="refusing to start fresh"): + report_state.hydrate_from_run_dir() + + def test_list_reports_filter_severity(report_state: ReportState) -> None: _seed(report_state) result = _do_list_reports( diff --git a/tests/test_report_writer.py b/tests/test_report_writer.py index 9b849010..4cdaf776 100644 --- a/tests/test_report_writer.py +++ b/tests/test_report_writer.py @@ -11,6 +11,7 @@ import pytest from strix.report.writer import ( atomic_write_text, read_run_record, + read_vulnerabilities, render_vulnerability_md, write_executive_report, write_run_record, @@ -62,6 +63,27 @@ def test_write_and_read_run_record_round_trip(tmp_path: Path) -> None: assert read_run_record(tmp_path) == payload +def test_read_vulnerabilities_filters_non_object_entries(tmp_path: Path) -> None: + (tmp_path / "vulnerabilities.json").write_text( + json.dumps([_sample_report(), "invalid"]), + encoding="utf-8", + ) + + assert read_vulnerabilities(tmp_path) == [_sample_report()] + + +def test_read_vulnerabilities_missing_file_raises(tmp_path: Path) -> None: + with pytest.raises(RuntimeError, match="unreadable"): + read_vulnerabilities(tmp_path) + + +def test_read_vulnerabilities_non_list_raises(tmp_path: Path) -> None: + (tmp_path / "vulnerabilities.json").write_text("{}", encoding="utf-8") + + with pytest.raises(TypeError, match="not a list"): + read_vulnerabilities(tmp_path) + + def test_render_vulnerability_md_includes_core_sections() -> None: md = render_vulnerability_md( _sample_report( diff --git a/tests/test_reporting_fields.py b/tests/test_reporting_fields.py index 2684fb67..a9da8c22 100644 --- a/tests/test_reporting_fields.py +++ b/tests/test_reporting_fields.py @@ -589,22 +589,21 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata( return {"is_duplicate": False} monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate) - report_state.vulnerability_reports.append( - { - "id": "vuln-0001", - "title": "CVE-2024-0001 in other 1.0.0", - "severity": "low", - "timestamp": "2026-01-01 00:00:00 UTC", - "description": "Existing dependency finding.", - "target": "repo/package.json", - "cve": "CVE-2024-0001", - "dependency_metadata": { - "package_name": "other", - "installed_version": "1.0.0", - "package_ecosystem": "npm", - }, - } - ) + baseline_report = { + "id": "vuln-0001", + "title": "CVE-2024-0001 in other 1.0.0", + "severity": "low", + "timestamp": "2026-01-01 00:00:00 UTC", + "description": "Existing dependency finding.", + "target": "repo/package.json", + "cve": "CVE-2024-0001", + "dependency_metadata": { + "package_name": "other", + "installed_version": "1.0.0", + "package_ecosystem": "npm", + }, + } + report_state.load_baseline_vulnerabilities(BASELINE_RUN_NAME, [baseline_report]) result = await _do_create_dependency( title="CVE-2024-0001 in sample 1.0.0", @@ -630,6 +629,7 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata( ) assert result["success"] is True + assert captured["existing"] == [baseline_report] assert captured["candidate"] == { "title": "CVE-2024-0001 in sample 1.0.0", "description": "Published advisory affects the pinned version.", @@ -693,6 +693,9 @@ async def test_create_report_suppresses_duplicate_from_baseline_run( remediation_steps="Use parameterized queries.", evidence="The response contains authenticated data.", assumptions="Assumes the endpoint is reachable.", + counterevidence="No parameter binding is present.", + confidence="high", + severity_change_conditions="Parameterized queries would eliminate the issue.", fix_effort="low", cvss_breakdown=_CVSS, endpoint="/login",