This commit is contained in:
Ben Younes 2026-09-18 07:20:35 +02:00 committed by GitHub
commit 98e51f94de
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 568 additions and 32 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. Target branch or commit to compare against (e.g., `origin/main`). Defaults to the repository's default branch.
</ParamField> </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"> <ParamField path="--non-interactive, -n" type="boolean">
Run in headless mode without TUI. Ideal for CI/CD. Run in headless mode without TUI. Ideal for CI/CD.
</ParamField> </ParamField>
@ -138,6 +145,9 @@ strix --target https://example.com --max-budget 25 --max-turns 300
# Force diff-scope against a specific base ref # Force diff-scope against a specific base ref
strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main 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 # Multi-target white-box testing
strix -t https://github.com/org/app -t https://staging.example.com 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"), "scope_mode": getattr(args, "scope_mode", "auto"),
"diff_base": getattr(args, "diff_base", None), "diff_base": getattr(args, "diff_base", None),
"resume_instruction": getattr(args, "user_explicit_instruction", None) or "", "resume_instruction": getattr(args, "user_explicit_instruction", None) or "",
"baseline_run": getattr(args, "baseline_run", None),
} }
report_state = ReportState(args.run_name) report_state = ReportState(args.run_name)
report_state.hydrate_from_run_dir() 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.set_scan_config(scan_config)
report_state.save_run_data() report_state.save_run_data()

View file

@ -20,6 +20,10 @@ from strix.interface.utils import (
) )
BASELINE_RUN_ARG = "--baseline-run"
RUN_NAME_PATH_SEPARATORS = ("/", "\\")
def get_version() -> str: def get_version() -> str:
try: try:
from importlib.metadata import version from importlib.metadata import version
@ -289,6 +293,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() args = parser.parse_args()
# Startup-resolved state lives alongside the parsed flags. The full schema # Startup-resolved state lives alongside the parsed flags. The full schema
# is established here so downstream code reads attributes directly. # is established here so downstream code reads attributes directly.
@ -345,11 +359,11 @@ Strix Cloud:
args.user_instruction = args.instruction or None args.user_instruction = args.instruction or None
if args.resume: if args.resume:
if args.target or args.target_list: if args.target or args.target_list or args.baseline_run is not None:
parser.error( 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 " "--resume picks up where the prior run left off, including the "
"original target list." "original target list and baseline."
) )
_load_resume_state(args, parser) _load_resume_state(args, parser)
agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json" agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json"
@ -361,6 +375,9 @@ Strix Cloud:
f"or remove --resume to start over with the same targets." f"or remove --resume to start over with the same targets."
) )
else: 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 not args.target and not args.target_list:
if args.non_interactive: if args.non_interactive:
parser.error( parser.error(
@ -381,6 +398,26 @@ Strix Cloud:
return args return args
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)
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: def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
"""Populate ``args.targets_info`` and friends from a prior run's run.json.""" """Populate ``args.targets_info`` and friends from a prior run's run.json."""
from strix.report.writer import read_run_record from strix.report.writer import read_run_record
@ -462,6 +499,9 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
attach_workspace_mount(args) attach_workspace_mount(args)
if state.get("diff_scope"): if state.get("diff_scope"):
args.diff_scope = 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") persisted_scan_mode = state.get("scan_mode")
if persisted_scan_mode and args.scan_mode == "deep": if persisted_scan_mode and args.scan_mode == "deep":
args.scan_mode = persisted_scan_mode 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}), "diff_scope": getattr(args, "diff_scope", {"active": False}),
"scope_mode": args.scope_mode, "scope_mode": args.scope_mode,
"diff_base": args.diff_base, "diff_base": args.diff_base,
"baseline_run": getattr(args, "baseline_run", None),
} }
write_run_record(run_dir, run_record) write_run_record(run_dir, run_record)

View file

@ -97,9 +97,11 @@ class GoTuiRuntime:
"resume_instruction": self.args.user_explicit_instruction or "", "resume_instruction": self.args.user_explicit_instruction or "",
"workspace_mount": getattr(self.args, "workspace_mount", None) or "", "workspace_mount": getattr(self.args, "workspace_mount", None) or "",
"workspace_subdir": getattr(self.args, "workspace_subdir", 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 = ReportState(self.scan_config["run_name"])
self.report_state.hydrate_from_run_dir() 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.set_scan_config(self.scan_config)
self.report_state.save_run_data() self.report_state.save_run_data()
set_global_report_state(self.report_state) set_global_report_state(self.report_state)

View file

@ -1,4 +1,3 @@
import json
import logging import logging
import re import re
import subprocess import subprocess
@ -17,7 +16,9 @@ from strix.report.coverage import write_coverage
from strix.report.pricing import resolve_litellm_model from strix.report.pricing import resolve_litellm_model
from strix.report.sarif import write_sarif from strix.report.sarif import write_sarif
from strix.report.writer import ( from strix.report.writer import (
VULNERABILITIES_FILENAME,
read_run_record, read_run_record,
read_vulnerabilities,
write_executive_report, write_executive_report,
write_run_record, write_run_record,
write_vulnerabilities, write_vulnerabilities,
@ -193,6 +194,7 @@ class ReportState:
self.end_time: str | None = None self.end_time: str | None = None
self.vulnerability_reports: list[dict[str, Any]] = [] self.vulnerability_reports: list[dict[str, Any]] = []
self.baseline_vulnerability_reports: list[dict[str, Any]] = []
self.final_scan_result: str | None = None self.final_scan_result: str | None = None
self.scan_results: dict[str, Any] | None = None self.scan_results: dict[str, Any] | None = None
@ -272,21 +274,16 @@ class ReportState:
self._telemetry_llm_usage_baseline = self._build_llm_usage_record() self._telemetry_llm_usage_baseline = self._build_llm_usage_record()
logger.info("report state hydrated run.json from %s", run_dir) 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(): if json_path.exists():
try: try:
data = json.loads(json_path.read_text(encoding="utf-8")) self.vulnerability_reports = read_vulnerabilities(run_dir)
except (OSError, json.JSONDecodeError) as exc: except (RuntimeError, TypeError) as exc:
raise RuntimeError( 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"refusing to start fresh — that would overwrite prior "
f"vulnerability MDs on disk. Inspect or delete the run dir.", f"vulnerability MDs on disk. Inspect or delete the run dir.",
) from exc ) 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: for r in self.vulnerability_reports:
# A finding written before the class was persisted still carries the # A finding written before the class was persisted still carries the
# metadata of its class, so name the class it always had. # metadata of its class, so name the class it always had.
@ -309,6 +306,27 @@ class ReportState:
len(self.vulnerability_reports), len(self.vulnerability_reports),
) )
def load_baseline_vulnerabilities(
self,
baseline_run_name: str,
vulnerability_reports: list[dict[str, Any]],
) -> None:
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( def add_vulnerability_report(
self, self,
title: str, title: str,
@ -519,6 +537,9 @@ class ReportState:
def get_existing_vulnerabilities(self) -> list[dict[str, Any]]: def get_existing_vulnerabilities(self) -> list[dict[str, Any]]:
return list(self.vulnerability_reports) return list(self.vulnerability_reports)
def get_dedupe_vulnerabilities(self) -> list[dict[str, Any]]:
return [*self.baseline_vulnerability_reports, *self.vulnerability_reports]
def record_sdk_usage( def record_sdk_usage(
self, self,
*, *,
@ -635,6 +656,7 @@ class ReportState:
"local_sources": config.get("local_sources", []), "local_sources": config.get("local_sources", []),
"scope_mode": config.get("scope_mode", "auto"), "scope_mode": config.get("scope_mode", "auto"),
"diff_base": config.get("diff_base"), "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__) logger = logging.getLogger(__name__)
_SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4} _SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
VULNERABILITIES_FILENAME = "vulnerabilities.json"
_CSV_FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r") _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: def write_executive_report(run_dir: Path, final_scan_result: str) -> None:
path = run_dir / "penetration_test_report.md" path = run_dir / "penetration_test_report.md"
with path.open("w", encoding="utf-8") as f: 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(csv_path, csv_buf.getvalue())
atomic_write_text( atomic_write_text(
run_dir / "vulnerabilities.json", run_dir / VULNERABILITIES_FILENAME,
json.dumps(vulnerability_reports, ensure_ascii=False, indent=2, default=str), json.dumps(vulnerability_reports, ensure_ascii=False, indent=2, default=str),
) )

View file

@ -791,7 +791,7 @@ async def _do_create(
from strix.report.dedupe import check_duplicate from strix.report.dedupe import check_duplicate
existing = report_state.get_existing_vulnerabilities() existing = report_state.get_dedupe_vulnerabilities()
candidate = { candidate = {
"title": title, "title": title,
"description": description, "description": description,
@ -1887,7 +1887,7 @@ async def _do_create_dependency( # noqa: PLR0912
from strix.report.dedupe import check_duplicate from strix.report.dedupe import check_duplicate
existing = report_state.get_existing_vulnerabilities() existing = report_state.get_dedupe_vulnerabilities()
candidate = { candidate = {
"title": title, "title": title,
"description": description, "description": description,

View file

@ -5,6 +5,7 @@ from __future__ import annotations
import importlib import importlib
import json import json
import sys import sys
from contextlib import nullcontext
from types import SimpleNamespace from types import SimpleNamespace
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
@ -16,6 +17,14 @@ if TYPE_CHECKING:
cli_main: Any = importlib.import_module("strix.interface.main") 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: def _stub_settings(monkeypatch: pytest.MonkeyPatch) -> None:
@ -69,6 +78,130 @@ 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_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( def test_parse_arguments_rejects_resume_with_target_list(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None: ) -> None:
@ -96,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") (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( def test_resume_restores_a_target_less_workspace_mount(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None: ) -> None:

View file

@ -39,6 +39,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( def test_binary_command_prefers_packaged_sidecar(
monkeypatch: pytest.MonkeyPatch, monkeypatch: pytest.MonkeyPatch,
tmp_path: Any, tmp_path: Any,

View file

@ -20,6 +20,15 @@ if TYPE_CHECKING:
from pathlib import Path from pathlib import Path
BASELINE_RUN_NAME = "baseline-run"
BASELINE_REPORT_ID = "vuln-0042"
BASELINE_REPORT_TITLE = "Baseline SQL injection"
BASELINE_REPORT_SEVERITY = "critical"
BASELINE_REPORT_TIMESTAMP = "2026-07-27 00:00:00 UTC"
BASELINE_REPORT_TARGET = "https://baseline.example.com"
CURRENT_REPORT_TITLE = "Reflected XSS in search"
@pytest.fixture @pytest.fixture
def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState: def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
monkeypatch.chdir(tmp_path) monkeypatch.chdir(tmp_path)
@ -174,6 +183,79 @@ def test_list_reports_metadata_first_and_sorted(report_state: ReportState) -> No
assert "evidence" not in first assert "evidence" not in first
def test_list_reports_and_get_report_exclude_baseline_findings(
report_state: ReportState,
) -> None:
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])
report_state.add_vulnerability_report(
title=CURRENT_REPORT_TITLE,
severity="medium",
description="q reflects unencoded input.",
target="https://app.example.com",
)
listed = _do_list_reports(
severity=None,
finding_class=None,
target=None,
search=None,
include_details=False,
)
baseline_lookup = _do_get_report(BASELINE_REPORT_ID)
assert listed["total_count"] == 1
assert listed["severity_counts"] == {"medium": 1}
assert [report["title"] for report in listed["reports"]] == [CURRENT_REPORT_TITLE]
assert baseline_lookup["success"] is False
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: def test_list_reports_filter_severity(report_state: ReportState) -> None:
_seed(report_state) _seed(report_state)
result = _do_list_reports( result = _do_list_reports(

View file

@ -11,6 +11,7 @@ import pytest
from strix.report.writer import ( from strix.report.writer import (
atomic_write_text, atomic_write_text,
read_run_record, read_run_record,
read_vulnerabilities,
render_vulnerability_md, render_vulnerability_md,
write_executive_report, write_executive_report,
write_run_record, 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 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: def test_render_vulnerability_md_includes_core_sections() -> None:
md = render_vulnerability_md( md = render_vulnerability_md(
_sample_report( _sample_report(

View file

@ -43,6 +43,14 @@ _CVSS = {
"availability": "H", "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 = { _DEP_CONTEXT = {
"attack_vector": "N", "attack_vector": "N",
@ -632,8 +640,7 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
return {"is_duplicate": False} return {"is_duplicate": False}
monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate) monkeypatch.setattr("strix.report.dedupe.check_duplicate", fake_check_duplicate)
report_state.vulnerability_reports.append( baseline_report = {
{
"id": "vuln-0001", "id": "vuln-0001",
"title": "CVE-2024-0001 in other 1.0.0", "title": "CVE-2024-0001 in other 1.0.0",
"severity": "low", "severity": "low",
@ -647,7 +654,7 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
"package_ecosystem": "npm", "package_ecosystem": "npm",
}, },
} }
) report_state.load_baseline_vulnerabilities(BASELINE_RUN_NAME, [baseline_report])
result = await _do_create_dependency( result = await _do_create_dependency(
title="CVE-2024-0001 in sample 1.0.0", title="CVE-2024-0001 in sample 1.0.0",
@ -673,6 +680,7 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
) )
assert result["success"] is True assert result["success"] is True
assert captured["existing"] == [baseline_report]
assert captured["candidate"] == { assert captured["candidate"] == {
"title": "CVE-2024-0001 in sample 1.0.0", "title": "CVE-2024-0001 in sample 1.0.0",
"description": "Published advisory affects the pinned version.", "description": "Published advisory affects the pinned version.",
@ -696,6 +704,65 @@ 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.",
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",
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: async def test_dependency_report_rejects_bad_cve(report_state: ReportState) -> None:
result = await _do_create_dependency( result = await _do_create_dependency(
title="bad", title="bad",