From 191ed35da598e86ea6468f24ecc5edff78e35799 Mon Sep 17 00:00:00 2001 From: Ousama Ben Younes Date: Mon, 27 Jul 2026 17:45:49 +0000 Subject: [PATCH] feat(report): suppress baseline run duplicates --- docs/usage/cli.mdx | 10 ++++++ strix/interface/cli.py | 2 ++ strix/interface/cli_args.py | 32 +++++++++++++++-- strix/interface/scan_setup.py | 1 + strix/interface/tui/runtime.py | 2 ++ strix/report/state.py | 38 +++++++++++++++----- strix/report/writer.py | 14 +++++++- tests/test_cli_target_list.py | 25 +++++++++++++ tests/test_reporting_fields.py | 64 ++++++++++++++++++++++++++++++++++ 9 files changed, 175 insertions(+), 13 deletions(-) diff --git a/docs/usage/cli.mdx b/docs/usage/cli.mdx index 699fb1cb..b12b4c34 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -56,6 +56,13 @@ strix (--target | --target-list ) [options] Target branch or commit to compare against (e.g., `origin/main`). Defaults to the repository's default branch. + + Run name under `./strix_runs/` whose `vulnerabilities.json` should be used as + a known-findings baseline. Fresh findings that dedupe against the baseline are + rejected like same-run duplicates, so scheduled or post-fix scans can spend + budget on new issues instead of re-reporting known ones. + + Run in headless mode without TUI. Ideal for CI/CD. @@ -138,6 +145,9 @@ strix --target https://example.com --max-budget 25 --max-turns 300 # Force diff-scope against a specific base ref strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main +# Suppress findings already reported by a prior run +strix -n --target ./ --baseline-run scan-20260727-120000 + # Multi-target white-box testing strix -t https://github.com/org/app -t https://staging.example.com diff --git a/strix/interface/cli.py b/strix/interface/cli.py index 684805d0..f71a1583 100644 --- a/strix/interface/cli.py +++ b/strix/interface/cli.py @@ -98,10 +98,12 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915 "scope_mode": getattr(args, "scope_mode", "auto"), "diff_base": getattr(args, "diff_base", None), "resume_instruction": getattr(args, "user_explicit_instruction", None) or "", + "baseline_run": getattr(args, "baseline_run", None), } report_state = ReportState(args.run_name) report_state.hydrate_from_run_dir() + report_state.hydrate_baseline_run(scan_config.get("baseline_run")) report_state.set_scan_config(scan_config) report_state.save_run_data() diff --git a/strix/interface/cli_args.py b/strix/interface/cli_args.py index fc53e15f..58eb5a7f 100644 --- a/strix/interface/cli_args.py +++ b/strix/interface/cli_args.py @@ -19,6 +19,8 @@ from strix.interface.utils import ( validate_config_file, ) +BASELINE_RUN_ARG = "--baseline-run" + def get_version() -> str: try: @@ -288,6 +290,16 @@ Strix Cloud: ), ) + parser.add_argument( + BASELINE_RUN_ARG, + type=str, + metavar="RUN_NAME", + help=( + "Use vulnerabilities.json from a prior run under ./strix_runs/ as a " + "known-findings baseline for cross-run duplicate suppression." + ), + ) + args = parser.parse_args() # Startup-resolved state lives alongside the parsed flags. The full schema # is established here so downstream code reads attributes directly. @@ -344,11 +356,11 @@ Strix Cloud: args.user_instruction = args.instruction or None if args.resume: - if args.target or args.target_list: + if args.target or args.target_list or args.baseline_run: parser.error( - "Cannot combine --resume with --target/--target-list. " + "Cannot combine --resume with --target/--target-list/--baseline-run. " "--resume picks up where the prior run left off, including the " - "original target list." + "original target list and baseline." ) _load_resume_state(args, parser) agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json" @@ -377,9 +389,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: + from strix.report.writer import read_vulnerabilities + + baseline_run_dir = run_dir_for(baseline_run) + try: + read_vulnerabilities(baseline_run_dir) + except (RuntimeError, TypeError) as exc: + parser.error(f"{BASELINE_RUN_ARG} {baseline_run}: {exc}") + + def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: """Populate ``args.targets_info`` and friends from a prior run's run.json.""" from strix.report.writer import read_run_record @@ -461,6 +486,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser attach_workspace_mount(args) if state.get("diff_scope"): args.diff_scope = state.get("diff_scope") + args.baseline_run = state.get("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/interface/scan_setup.py b/strix/interface/scan_setup.py index ae7caf2f..1694e44e 100644 --- a/strix/interface/scan_setup.py +++ b/strix/interface/scan_setup.py @@ -264,5 +264,6 @@ def _persist_run_record(args: argparse.Namespace) -> None: "diff_scope": getattr(args, "diff_scope", {"active": False}), "scope_mode": args.scope_mode, "diff_base": args.diff_base, + "baseline_run": getattr(args, "baseline_run", None), } write_run_record(run_dir, run_record) diff --git a/strix/interface/tui/runtime.py b/strix/interface/tui/runtime.py index d398e5ab..8c66fc2c 100644 --- a/strix/interface/tui/runtime.py +++ b/strix/interface/tui/runtime.py @@ -92,9 +92,11 @@ class GoTuiRuntime: "resume_instruction": self.args.user_explicit_instruction or "", "workspace_mount": getattr(self.args, "workspace_mount", None) or "", "workspace_subdir": getattr(self.args, "workspace_subdir", None) or "", + "baseline_run": getattr(self.args, "baseline_run", None), } self.report_state = ReportState(self.scan_config["run_name"]) self.report_state.hydrate_from_run_dir() + self.report_state.hydrate_baseline_run(self.scan_config.get("baseline_run")) self.report_state.set_scan_config(self.scan_config) self.report_state.save_run_data() set_global_report_state(self.report_state) diff --git a/strix/report/state.py b/strix/report/state.py index 6a082371..d3b96b70 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -1,4 +1,3 @@ -import json import logging import re import subprocess @@ -18,6 +17,7 @@ from strix.report.pricing import resolve_litellm_model from strix.report.sarif import write_sarif from strix.report.writer import ( read_run_record, + read_vulnerabilities, write_executive_report, write_run_record, write_vulnerabilities, @@ -192,6 +192,8 @@ class ReportState: self.end_time: str | None = None 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 @@ -274,18 +276,13 @@ class ReportState: json_path = run_dir / "vulnerabilities.json" if json_path.exists(): try: - data = json.loads(json_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError) as exc: + self.vulnerability_reports = read_vulnerabilities(run_dir) + except (RuntimeError, TypeError) as exc: raise RuntimeError( f"vulnerabilities.json 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 - if not isinstance(data, list): - raise RuntimeError( - f"vulnerabilities.json at {json_path} is not a list", - ) - self.vulnerability_reports = [r for r in data if isinstance(r, dict)] for r in self.vulnerability_reports: # A finding written before the class was persisted still carries the # metadata of its class, so name the class it always had. @@ -308,6 +305,28 @@ class ReportState: len(self.vulnerability_reports), ) + def load_baseline_vulnerabilities( + self, + 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", + len(self.baseline_vulnerability_reports), + baseline_run_name, + ) + + def hydrate_baseline_run(self, baseline_run_name: str | None) -> None: + if not baseline_run_name: + return + baseline_run_dir = run_dir_for(baseline_run_name) + self.load_baseline_vulnerabilities( + baseline_run_name, + read_vulnerabilities(baseline_run_dir), + ) + def add_vulnerability_report( self, title: str, @@ -509,7 +528,7 @@ class ReportState: return report def get_existing_vulnerabilities(self) -> list[dict[str, Any]]: - return list(self.vulnerability_reports) + return [*self.baseline_vulnerability_reports, *self.vulnerability_reports] def record_sdk_usage( self, @@ -627,6 +646,7 @@ class ReportState: "local_sources": config.get("local_sources", []), "scope_mode": config.get("scope_mode", "auto"), "diff_base": config.get("diff_base"), + "baseline_run": config.get("baseline_run"), } ) diff --git a/strix/report/writer.py b/strix/report/writer.py index cf858f7d..265004fc 100644 --- a/strix/report/writer.py +++ b/strix/report/writer.py @@ -25,6 +25,7 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) _SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4} +VULNERABILITIES_FILENAME = "vulnerabilities.json" _CSV_FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r") @@ -136,6 +137,17 @@ def write_run_record(run_dir: Path, run_record: dict[str, Any]) -> None: ) +def read_vulnerabilities(run_dir: Path) -> list[dict[str, Any]]: + path = run_dir / VULNERABILITIES_FILENAME + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise RuntimeError(f"{VULNERABILITIES_FILENAME} at {path} is unreadable: {exc}") from exc + if not isinstance(data, list): + raise TypeError(f"{VULNERABILITIES_FILENAME} at {path} is not a list") + return [report for report in data if isinstance(report, dict)] + + def write_executive_report(run_dir: Path, final_scan_result: str) -> None: path = run_dir / "penetration_test_report.md" with path.open("w", encoding="utf-8") as f: @@ -184,7 +196,7 @@ def write_vulnerabilities( atomic_write_text(csv_path, csv_buf.getvalue()) atomic_write_text( - run_dir / "vulnerabilities.json", + run_dir / VULNERABILITIES_FILENAME, json.dumps(vulnerability_reports, ensure_ascii=False, indent=2, default=str), ) diff --git a/tests/test_cli_target_list.py b/tests/test_cli_target_list.py index d20230b1..fe2afd14 100644 --- a/tests/test_cli_target_list.py +++ b/tests/test_cli_target_list.py @@ -17,6 +17,11 @@ if TYPE_CHECKING: cli_main: Any = importlib.import_module("strix.interface.main") +BASELINE_RUN_NAME = "baseline-alpha" +RUNS_DIR_NAME = "strix_runs" +VULNERABILITIES_FILENAME = "vulnerabilities.json" +TARGET_URL = "https://test1.com/" + def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( @@ -69,6 +74,26 @@ def test_parse_arguments_combines_target_and_target_list( ] +def test_parse_arguments_accepts_baseline_run( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + baseline_run_dir = tmp_path / RUNS_DIR_NAME / BASELINE_RUN_NAME + baseline_run_dir.mkdir(parents=True) + (baseline_run_dir / VULNERABILITIES_FILENAME).write_text(json.dumps([]), encoding="utf-8") + _stub_settings(monkeypatch) + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + sys, + "argv", + ["strix", "--target", TARGET_URL, "--baseline-run", BASELINE_RUN_NAME], + ) + + args = cli_main.parse_arguments() + + assert args.baseline_run == BASELINE_RUN_NAME + + 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_reporting_fields.py b/tests/test_reporting_fields.py index edb393ea..2684fb67 100644 --- a/tests/test_reporting_fields.py +++ b/tests/test_reporting_fields.py @@ -38,6 +38,14 @@ _CVSS = { "availability": "H", } +BASELINE_RUN_NAME = "baseline-run" +BASELINE_REPORT_ID = "vuln-0042" +BASELINE_REPORT_TITLE = "Baseline SQL injection" +BASELINE_REPORT_SEVERITY = "high" +BASELINE_REPORT_TIMESTAMP = "2026-07-27 00:00:00 UTC" +BASELINE_REPORT_TARGET = "https://app.example.com" +NEW_REPORT_TITLE = "SQL injection in login" + _DEP_CONTEXT = { "attack_vector": "N", @@ -645,6 +653,62 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata( } +async def test_create_report_suppresses_duplicate_from_baseline_run( + report_state: ReportState, + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, object] = {} + + async def fake_check_duplicate( + candidate: dict[str, object], + existing: list[dict[str, object]], + ) -> dict[str, object]: + captured["candidate"] = candidate + captured["existing"] = existing + return { + "is_duplicate": True, + "duplicate_id": BASELINE_REPORT_ID, + "confidence": 1.0, + "reason": "Same endpoint and root cause.", + } + + monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate) + baseline_report = { + "id": BASELINE_REPORT_ID, + "title": BASELINE_REPORT_TITLE, + "severity": BASELINE_REPORT_SEVERITY, + "timestamp": BASELINE_REPORT_TIMESTAMP, + "target": BASELINE_REPORT_TARGET, + } + report_state.load_baseline_vulnerabilities(BASELINE_RUN_NAME, [baseline_report]) + + result = await _do_create( + title=NEW_REPORT_TITLE, + description="Unsanitized input reaches a SQL query.", + impact="Database read access.", + target=BASELINE_REPORT_TARGET, + technical_analysis="The login query interpolates the username.", + poc_description="Submit a tautology payload.", + poc_script_code="curl https://app.example.com/login", + remediation_steps="Use parameterized queries.", + evidence="The response contains authenticated data.", + assumptions="Assumes the endpoint is reachable.", + fix_effort="low", + cvss_breakdown=_CVSS, + endpoint="/login", + method="POST", + cve=None, + cwe=None, + code_locations=None, + ) + + assert result["success"] is False + assert result["duplicate_of"] == BASELINE_REPORT_ID + assert result["duplicate_title"] == BASELINE_REPORT_TITLE + assert captured["existing"] == [baseline_report] + assert not report_state.vulnerability_reports + + async def test_dependency_report_rejects_bad_cve(report_state: ReportState) -> None: result = await _do_create_dependency( title="bad",