feat(report): suppress baseline run duplicates

This commit is contained in:
Ousama Ben Younes 2026-07-27 17:45:49 +00:00 committed by Ben Younes
parent 52b1923347
commit 191ed35da5
No known key found for this signature in database
9 changed files with 175 additions and 13 deletions

View file

@ -56,6 +56,13 @@ strix (--target <target> | --target-list <path>) [options]
Target branch or commit to compare against (e.g., `origin/main`). Defaults to the repository's default branch.
</ParamField>
<ParamField path="--baseline-run" type="string">
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.
</ParamField>
<ParamField path="--non-interactive, -n" type="boolean">
Run in headless mode without TUI. Ideal for CI/CD.
</ParamField>
@ -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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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