diff --git a/strix/interface/cli_args.py b/strix/interface/cli_args.py index caa70a8e..e82579f4 100644 --- a/strix/interface/cli_args.py +++ b/strix/interface/cli_args.py @@ -5,7 +5,7 @@ from __future__ import annotations import argparse import os import sys -from pathlib import Path +from pathlib import Path, PureWindowsPath from strix.config import apply_config_override from strix.config.settings import DEFAULT_MAX_TURNS @@ -447,6 +447,7 @@ def _validate_baseline_run(baseline_run: object) -> None: raise ValueError(f"{BASELINE_RUN_ARG} must be a non-empty run name") if ( Path(baseline_run).is_absolute() + or PureWindowsPath(baseline_run).drive or baseline_run in {".", ".."} or any(separator in baseline_run for separator in RUN_NAME_PATH_SEPARATORS) ): @@ -570,12 +571,13 @@ def load_resume_state(args: argparse.Namespace) -> None: attach_workspace_mount(args) 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: + baseline_run = state.get("baseline_run") + if baseline_run is not None: try: - _validate_baseline_run(args.baseline_run) + _validate_baseline_run(baseline_run) except (TypeError, ValueError) as exc: raise ResumeError(f"--resume {args.resume}: {exc}") from exc + args.baseline_run = baseline_run 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 49ace501..0efef0db 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -62,6 +62,11 @@ def _strix_version() -> str | None: return None +# Baseline findings join the dedupe corpus under this id prefix. Every run +# numbers its findings from vuln-0001, so a prior run's raw ids would collide +# with this run's and a duplicate verdict could point at the wrong finding. +BASELINE_REPORT_ID_PREFIX = "baseline-" + # Content a revision may replace. The identity of the finding (id, timestamp, # finding_class) and its original author stay put. dependency_metadata is # replaced whole, so a caller carries the package identity over itself. @@ -212,6 +217,7 @@ 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 @@ -329,7 +335,11 @@ class ReportState: baseline_run_name: str, vulnerability_reports: list[dict[str, Any]], ) -> None: - self.baseline_vulnerability_reports = list(vulnerability_reports) + self.baseline_run_name = baseline_run_name + self.baseline_vulnerability_reports = [ + {**report, "id": f"{BASELINE_REPORT_ID_PREFIX}{report.get('id', '')}"} + for report in vulnerability_reports + ] logger.info( "loaded %d baseline vulnerability report(s) from %s", len(self.baseline_vulnerability_reports), @@ -658,6 +668,13 @@ class ReportState: def get_dedupe_vulnerabilities(self) -> list[dict[str, Any]]: return [*self.baseline_vulnerability_reports, *self.vulnerability_reports] + def get_baseline_report(self, report_id: str) -> dict[str, Any] | None: + """The prior-run finding a dedupe verdict matched, or ``None`` for this run's own.""" + return next( + (r for r in self.baseline_vulnerability_reports if r.get("id") == report_id), + None, + ) + def record_sdk_usage( self, *, diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index 0af5becb..dde6889c 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -271,6 +271,35 @@ def _with_warning(result: dict[str, Any], warning: str | None) -> dict[str, Any] return result +def _baseline_duplicate_response( + report_state: ReportState, dedupe: dict[str, Any] +) -> dict[str, Any] | None: + """The rejection for a finding a ``--baseline-run`` already holds, else ``None``. + + The matched finding belongs to a prior scan, so the agent must neither + re-file it nor revise it as if it were one of this run's reports. + """ + duplicate_id = str(dedupe.get("duplicate_id") or "") + baseline_report = report_state.get_baseline_report(duplicate_id) + if baseline_report is None: + return None + baseline_run = report_state.baseline_run_name + duplicate_title = str(baseline_report.get("title") or "") + return { + "success": False, + "error": ( + f"Already reported in baseline run '{baseline_run}' as '{duplicate_title}' " + "- a known finding from a prior scan. Do not re-report it, and do not " + "revise or delete it: it is not part of this run." + ), + "duplicate_of": duplicate_id, + "duplicate_title": duplicate_title, + "baseline_run": baseline_run, + "confidence": dedupe.get("confidence", 0.0), + "reason": dedupe.get("reason", ""), + } + + def _validate_cvss_breakdown(breakdown: Any) -> list[str]: """Check the 8 CVSS metrics are all present with legal values.""" if not isinstance(breakdown, dict) or not breakdown: @@ -920,7 +949,7 @@ async def _do_create( (r.get("title", "Unknown") for r in existing if r.get("id") == duplicate_id), "", ) - return { + return _baseline_duplicate_response(report_state, dedupe) or { "success": False, "error": ( f"Potential duplicate of '{duplicate_title}' " @@ -2036,7 +2065,7 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915 dedupe = await check_duplicate(candidate, existing) if dedupe.get("is_duplicate"): duplicate_id = dedupe.get("duplicate_id", "") - return { + return _baseline_duplicate_response(report_state, dedupe) or { "success": False, "error": ( f"Potential duplicate (id={duplicate_id[:8]}...) — " diff --git a/tests/test_cli_target_list.py b/tests/test_cli_target_list.py index b2d2b3c4..3fc135dc 100644 --- a/tests/test_cli_target_list.py +++ b/tests/test_cli_target_list.py @@ -146,6 +146,25 @@ def test_parse_arguments_rejects_baseline_paths_outside_runs_dir( assert "must be a run name, not a path" in capsys.readouterr().err +def test_parse_arguments_rejects_windows_drive_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", f"C:{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_validates_baseline_before_interactive_setup( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -185,23 +204,6 @@ def test_parse_arguments_rejects_unreadable_baseline_run( 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: diff --git a/tests/test_go_tui_runtime.py b/tests/test_go_tui_runtime.py index 240a1ab2..324ffb32 100644 --- a/tests/test_go_tui_runtime.py +++ b/tests/test_go_tui_runtime.py @@ -19,7 +19,7 @@ from strix.config.settings import DEFAULT_MAX_TURNS from strix.interface.tui import runtime as go_tui from strix.interface.tui import sidecar from strix.interface.tui.runtime import GoTuiRuntime -from strix.report.state import ReportState +from strix.report.state import BASELINE_REPORT_ID_PREFIX, ReportState def args() -> argparse.Namespace: @@ -63,7 +63,9 @@ async def test_init_run_state_hydrates_selected_baseline( runtime.init_run_state() - assert runtime.report_state.get_dedupe_vulnerabilities() == [baseline_report] + assert runtime.report_state.get_dedupe_vulnerabilities() == [ + {**baseline_report, "id": f"{BASELINE_REPORT_ID_PREFIX}{baseline_report['id']}"} + ] assert runtime.report_state.get_existing_vulnerabilities() == [] diff --git a/tests/test_list_reports.py b/tests/test_list_reports.py index a0399674..bdd845f7 100644 --- a/tests/test_list_reports.py +++ b/tests/test_list_reports.py @@ -7,7 +7,7 @@ from typing import TYPE_CHECKING import pytest -from strix.report.state import ReportState, set_global_report_state +from strix.report.state import BASELINE_REPORT_ID_PREFIX, ReportState, set_global_report_state from strix.tools.reporting.tool import ( _do_get_report, _do_list_reports, @@ -234,7 +234,9 @@ def test_hydrate_baseline_run_reads_dedupe_only_findings( report_state.hydrate_baseline_run(BASELINE_RUN_NAME) - assert report_state.get_dedupe_vulnerabilities() == [baseline_report] + assert report_state.get_dedupe_vulnerabilities() == [ + {**baseline_report, "id": f"{BASELINE_REPORT_ID_PREFIX}{BASELINE_REPORT_ID}"} + ] assert report_state.get_existing_vulnerabilities() == [] diff --git a/tests/test_reporting_fields.py b/tests/test_reporting_fields.py index 251ea6a1..1f7b8a82 100644 --- a/tests/test_reporting_fields.py +++ b/tests/test_reporting_fields.py @@ -14,7 +14,12 @@ from strix.report.dedupe import ( _prepare_report_for_comparison, check_duplicate, ) -from strix.report.state import ReportRollbackError, ReportState, set_global_report_state +from strix.report.state import ( + BASELINE_REPORT_ID_PREFIX, + ReportRollbackError, + ReportState, + set_global_report_state, +) from strix.report.writer import write_run_record from strix.tools.finish.tool import finish_scan from strix.tools.reporting import tool as reporting_tool @@ -49,6 +54,8 @@ _CVSS = { BASELINE_RUN_NAME = "baseline-run" BASELINE_REPORT_ID = "vuln-0042" +BASELINE_DEDUPE_ID = f"{BASELINE_REPORT_ID_PREFIX}{BASELINE_REPORT_ID}" +CURRENT_REPORT_TITLE = "Stored XSS in profile" BASELINE_REPORT_TITLE = "Baseline SQL injection" BASELINE_REPORT_SEVERITY = "high" BASELINE_REPORT_TIMESTAMP = "2026-07-27 00:00:00 UTC" @@ -842,7 +849,9 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata( ) assert result["success"] is True - assert captured["existing"] == [baseline_report] + assert captured["existing"] == [ + {**baseline_report, "id": f"{BASELINE_REPORT_ID_PREFIX}{baseline_report['id']}"} + ] assert captured["candidate"] == { "title": "CVE-2024-0001 in sample 1.0.0", "description": "Published advisory affects the pinned version.", @@ -880,7 +889,7 @@ async def test_create_report_suppresses_duplicate_from_baseline_run( captured["existing"] = existing return { "is_duplicate": True, - "duplicate_id": BASELINE_REPORT_ID, + "duplicate_id": BASELINE_DEDUPE_ID, "confidence": 1.0, "reason": "Same endpoint and root cause.", } @@ -894,6 +903,10 @@ async def test_create_report_suppresses_duplicate_from_baseline_run( "target": BASELINE_REPORT_TARGET, } report_state.load_baseline_vulnerabilities(BASELINE_RUN_NAME, [baseline_report]) + # This run numbers its findings from vuln-0001 too, so it can hold an + # unrelated finding under the baseline finding's raw id. + current_report = {"id": BASELINE_REPORT_ID, "title": CURRENT_REPORT_TITLE} + report_state.vulnerability_reports = [current_report] result = await _do_create( title=NEW_REPORT_TITLE, @@ -919,9 +932,62 @@ async def test_create_report_suppresses_duplicate_from_baseline_run( ) assert result["success"] is False - assert result["duplicate_of"] == BASELINE_REPORT_ID + assert result["duplicate_of"] == BASELINE_DEDUPE_ID assert result["duplicate_title"] == BASELINE_REPORT_TITLE - assert captured["existing"] == [baseline_report] + assert result["baseline_run"] == BASELINE_RUN_NAME + assert f"baseline run '{BASELINE_RUN_NAME}'" in result["error"] + assert "do not revise" in result["error"] + assert captured["existing"] == [{**baseline_report, "id": BASELINE_DEDUPE_ID}, current_report] + assert report_state.vulnerability_reports == [current_report] + + +async def test_dependency_report_suppresses_duplicate_from_baseline_run( + report_state: ReportState, + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def fake_check_duplicate( + _candidate: dict[str, object], + _existing: list[dict[str, object]], + ) -> dict[str, object]: + return { + "is_duplicate": True, + "duplicate_id": BASELINE_DEDUPE_ID, + "confidence": 1.0, + "reason": "Same dependency CVE/package identity", + } + + monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate) + report_state.load_baseline_vulnerabilities( + BASELINE_RUN_NAME, + [{"id": BASELINE_REPORT_ID, "title": BASELINE_REPORT_TITLE}], + ) + + result = await _do_create_dependency( + title="CVE-2024-0001 in sample 1.0.0", + description="Published advisory affects the pinned version.", + target="repo/package.json", + cve="CVE-2024-0001", + package_name="sample", + installed_version="1.0.0", + impact="Low-impact dependency advisory.", + remediation_steps="Upgrade to 1.0.1.", + assumptions="Assumes the package is included in deployed builds.", + package_ecosystem="npm", + manifest_path="package-lock.json", + fixed_version="1.0.1", + cwe=None, + advisory_cvss=0.0, + technical_analysis=None, + fix_effort="low", + reachability="imported", + reachability_evidence=_DEP_EVIDENCE, + contextual_cvss_breakdown=_DEP_CONTEXT, + contextual_cvss_reasoning=_DEP_REASONING, + ) + + assert result["success"] is False + assert result["duplicate_of"] == BASELINE_DEDUPE_ID + assert result["baseline_run"] == BASELINE_RUN_NAME assert not report_state.vulnerability_reports