This commit is contained in:
Ben Younes 2026-10-20 18:04:05 +00:00 • committed by GitHub
commit 6b9b0e3806
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 723 additions and 38 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. When no terminal is
attached (CI, `nohup`, pipes) and a target is given, Strix runs headless
@ -151,6 +158,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

@ -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
@ -25,6 +25,10 @@ from strix.report.runs import list_run_summaries
FAIL_ON_SEVERITIES = ("critical", "high", "medium", "low", "info")
BASELINE_RUN_ARG = "--baseline-run"
RUN_NAME_PATH_SEPARATORS = ("/", "\\")
def get_version() -> str:
try:
from importlib.metadata import version
@ -313,6 +317,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.
@ -375,11 +389,11 @@ Strix Cloud:
args.user_instruction = args.instruction or None
if args.resume is not None:
if args.target or args.target_list:
if args.target or args.target_list or args.baseline_run is not None:
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."
)
if not args.resume.strip():
# A bare --resume: main() opens the inline picker of prior runs
@ -397,6 +411,12 @@ Strix Cloud:
except ResumeError as exc:
parser.error(str(exc))
else:
if args.baseline_run is not None:
try:
_validate_baseline_run(args.baseline_run)
except (TypeError, ValueError) as exc:
parser.error(str(exc))
if not args.target and not args.target_list:
if args.non_interactive:
parser.error(
@ -417,6 +437,28 @@ Strix Cloud:
return args
def _validate_baseline_run(baseline_run: object) -> None:
"""Raise ``TypeError``/``ValueError`` unless ``baseline_run`` names a readable prior run."""
from strix.report.writer import read_vulnerabilities
if not isinstance(baseline_run, str):
raise TypeError(f"{BASELINE_RUN_ARG} must be a run name")
if not baseline_run:
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)
):
raise ValueError(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:
raise ValueError(f"{BASELINE_RUN_ARG} {baseline_run}: {exc}") from exc
def terminal_attached() -> bool:
"""Whether the TUI can take over the terminal: a tty on both ends, not dumb."""
if os.environ.get("TERM", "").strip().lower() == "dumb":
@ -529,6 +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")
baseline_run = state.get("baseline_run")
if baseline_run is not None:
try:
_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

View file

@ -339,5 +339,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

@ -102,9 +102,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
@ -17,7 +16,9 @@ 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,
write_run_record,
write_vulnerabilities,
@ -63,6 +64,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 +218,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
@ -292,21 +300,16 @@ 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:
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"{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
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.
@ -329,6 +332,31 @@ 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 = [
{**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),
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,
@ -639,6 +667,16 @@ class ReportState:
def get_existing_vulnerabilities(self) -> list[dict[str, Any]]:
return list(self.vulnerability_reports)
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,
*,
@ -784,6 +822,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

@ -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:
@ -872,7 +901,7 @@ async def _do_create(
from strix.report.dedupe import check_duplicate
existing = report_state.get_existing_vulnerabilities()
existing = report_state.get_dedupe_vulnerabilities()
candidate = {
"title": title,
"description": description,
@ -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}' "
@ -2024,7 +2053,7 @@ async def _do_create_dependency( # noqa: PLR0912, PLR0915
from strix.report.dedupe import check_duplicate
existing = report_state.get_existing_vulnerabilities()
existing = report_state.get_dedupe_vulnerabilities()
candidate = {
"title": title,
"description": description,
@ -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]}...) — "

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,6 +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:
@ -69,6 +78,132 @@ 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_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,
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_rejects_resume_with_target_list(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
@ -96,6 +231,143 @@ 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
def test_load_resume_state_reports_invalid_baseline_as_resume_error(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
# The bare --resume picker loads the chosen run outside parse_arguments, so
# a bad persisted baseline must surface as a ResumeError it can print.
_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": f"../{OUTSIDE_RUN_NAME}",
},
)
monkeypatch.chdir(tmp_path)
args = SimpleNamespace(resume="pentest-resume", instruction=None, scan_mode="deep")
with pytest.raises(cli_args.ResumeError, match="must be a run name, not a path"):
cli_args.load_resume_state(args)
@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

@ -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:
@ -39,6 +39,36 @@ 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, "id": f"{BASELINE_REPORT_ID_PREFIX}{baseline_report['id']}"}
]
assert runtime.report_state.get_existing_vulnerabilities() == []
def test_binary_command_prefers_packaged_sidecar(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Any,

View file

@ -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,
@ -20,6 +20,15 @@ if TYPE_CHECKING:
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
def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState:
monkeypatch.chdir(tmp_path)
@ -174,6 +183,81 @@ def test_list_reports_metadata_first_and_sorted(report_state: ReportState) -> No
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, "id": f"{BASELINE_REPORT_ID_PREFIX}{BASELINE_REPORT_ID}"}
]
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

@ -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
@ -47,6 +52,16 @@ _CVSS = {
"availability": "H",
}
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"
BASELINE_REPORT_TARGET = "https://app.example.com"
NEW_REPORT_TITLE = "SQL injection in login"
_DEP_CONTEXT = {
"attack_vector": "N",
@ -794,22 +809,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",
@ -835,6 +849,9 @@ async def test_dependency_report_dedupe_candidate_includes_dependency_metadata(
)
assert result["success"] is True
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.",
@ -858,6 +875,122 @@ 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_DEDUPE_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])
# 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,
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_DEDUPE_ID
assert result["duplicate_title"] == BASELINE_REPORT_TITLE
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
async def test_dependency_report_rejects_bad_cve(report_state: ReportState) -> None:
result = await _do_create_dependency(
title="bad",