fix(report): validate baseline runs

Reject path-shaped, empty, and malformed baseline selections before CLI/TUI startup, including restored run state.

Refs #900
This commit is contained in:
Ben Younes 2026-09-08 03:06:02 +00:00
parent 17dc6130b1
commit 76fb6fe323
No known key found for this signature in database
7 changed files with 353 additions and 25 deletions

View file

@ -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

View file

@ -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",

View file

@ -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:

View file

@ -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,

View file

@ -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(

View file

@ -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(

View file

@ -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",