From 66d985e9e6e536776e90735a0784693c513817d7 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 3 Sep 2026 23:14:00 +0200 Subject: [PATCH] feat(three-stage-scan): keep quick as stage-1 inventory --- .kiro/specs/three-stage-scan/tasks.md | 4 +- src/openultrasast/cli.py | 4 ++ src/openultrasast/reports.py | 5 +- src/openultrasast/stages.py | 12 ++++ tests/test_reports.py | 94 +++++++++++++++++++++++++++ tests/test_stages.py | 23 ++++++- 6 files changed, 138 insertions(+), 4 deletions(-) diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index f1100ab..8da4253 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -117,7 +117,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Boundary: RegressionRunner_ - _Depends: 5.1_ -- [ ] 5.3 Map sandbox outcomes to the four verdicts and the worth-fixing gate +- [x] 5.3 Map sandbox outcomes to the four verdicts and the worth-fixing gate - Outcomes map to triggerable, not_triggerable, already_covered, or inconclusive; worth-fixing requires triggerable plus reachable or inferred-file-surface. - Not-triggerable and already-covered do not fail the build; inconclusive records a reason and does not fail the build. - Observable completion: a table-driven test covers the four verdicts and the worth-fixing boolean. @@ -128,7 +128,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex ## Phase 4 — Integration: Wire stages into the scan - [ ] 6. Run stages from the existing scan command -- [ ] 6.1 Keep quick as stage-1 inventory with no model and no target execution +- [x] 6.1 Keep quick as stage-1 inventory with no model and no target execution - Quick still loads policy, ruleset, preprocess, rank, pattern inventory, structural verify, and score. - The report labels those hits as inventory, not as worth-fixing. - Policy still fails loud on an unmapped enabled CWE. diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 637d590..5ce8100 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -44,6 +44,7 @@ from .reports import scan_exit_code, write_manifest, write_markdown_report, writ from .ruleset import DEFAULT_RULESET_DIR, load_ruleset from .run import ScanRun, create_scan_run from .scoring import build_score_artifact +from .stages import Stage, plan_for_mode, record_completed, stages_payload from .verification import VerificationResult, write_verification_results from .verify_judge import verify_findings_dispatch @@ -137,6 +138,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut if not path.exists() or not path.is_dir(): raise SystemExit(f"scan path is not a directory: {path}") + plan = plan_for_mode(mode) config = load_config(config_path if config_path.exists() else None) # Capability gates for the optional HarnessX agentic plane. When the extra is # absent (default/CI), both stay False and the deterministic path runs unchanged. @@ -299,6 +301,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut ) score_path = run.root / "score.json" score_path.write_text(json.dumps(score_artifact.to_dict(), indent=2, sort_keys=True) + "\n") + plan = record_completed(plan, Stage.STATIC) runtime.run_stage( "report", lambda: write_markdown_report(findings, markdown_path, verifications, redact=config.hardening.redact_secrets) ) @@ -322,6 +325,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut score=score_artifact.to_dict(), degradations=runtime.state["degradations"] or None, fusion=[_fusion_summary(decision) for decision in fusion_decisions] or None, + stages=stages_payload(plan), ), ) runtime.finish(status="succeeded") diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 1cb6b35..d37da0a 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -15,7 +15,7 @@ def write_markdown_report( ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications or []) - lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", ""] + lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""] if not findings: lines.append("No quick-mode findings were emitted.") for finding in findings: @@ -82,6 +82,7 @@ def write_manifest( score: dict[str, object] | None = None, degradations: list[dict[str, object]] | None = None, fusion: list[dict[str, object]] | None = None, + stages: dict[str, object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -110,6 +111,8 @@ def write_manifest( payload["degradations"] = degradations if fusion is not None: payload["fusion"] = fusion + if stages is not None: + payload["stages"] = stages path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") diff --git a/src/openultrasast/stages.py b/src/openultrasast/stages.py index 633ac32..4708beb 100644 --- a/src/openultrasast/stages.py +++ b/src/openultrasast/stages.py @@ -35,3 +35,15 @@ def skip_as_degradation(stage: Stage, reason: str) -> dict[str, str]: def record_skip(plan: StagePlan, stage: Stage, reason: str) -> StagePlan: return replace(plan, skipped=plan.skipped + (skip_as_degradation(stage, reason),)) + + +def record_completed(plan: StagePlan, stage: Stage) -> StagePlan: + return replace(plan, completed=plan.completed + (stage,)) + + +def stages_payload(plan: StagePlan) -> dict[str, object]: + return { + "requested": [str(stage) for stage in plan.requested], + "completed": [str(stage) for stage in plan.completed], + "skipped": [dict(item) for item in plan.skipped], + } diff --git a/tests/test_reports.py b/tests/test_reports.py index cad6e3b..f34a761 100644 --- a/tests/test_reports.py +++ b/tests/test_reports.py @@ -1,4 +1,5 @@ import json +import subprocess from pathlib import Path from types import SimpleNamespace @@ -8,6 +9,7 @@ from openultrasast.cli import main from openultrasast.findings import StaticFinding from openultrasast.reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report from openultrasast.run import ScanRun +from openultrasast.stages import Stage, plan_for_mode, record_completed, stages_payload from openultrasast.verification import verify_finding @@ -50,6 +52,34 @@ def test_manifest_links_shared_artifacts_by_finding_id(tmp_path: Path) -> None: assert manifest_finding["finding_id"] == finding.finding_id assert manifest_finding["verification_status"] == "accepted" assert manifest_finding["artifact_refs"]["sarif"] == "report.sarif" + assert "stages" not in payload + + +def test_manifest_includes_requested_completed_skipped_stages(tmp_path: Path) -> None: + run = ScanRun(scan_id="scan-1", root=tmp_path / "run", target=tmp_path / "repo") + run.root.mkdir() + finding = _finding() + artifacts = { + "findings": run.root / "findings.json", + "verification": run.root / "verification.json", + "markdown": run.root / "report.md", + "sarif": run.root / "report.sarif", + } + output = run.root / "manifest.json" + plan = record_completed(plan_for_mode("quick"), Stage.STATIC) + + write_manifest( + run=run, + findings=[finding], + verifications=[verify_finding(finding)], + artifact_paths=artifacts, + path=output, + stages=stages_payload(plan), + ) + + payload = json.loads(output.read_text()) + assert payload["stages"] == {"requested": ["static"], "completed": ["static"], "skipped": []} + assert payload["findings"][0]["finding_id"] == finding.finding_id def test_markdown_report_includes_verification_status(tmp_path: Path) -> None: @@ -59,8 +89,24 @@ def test_markdown_report_includes_verification_status(tmp_path: Path) -> None: write_markdown_report([finding], output, [verify_finding(finding)]) text = output.read_text() + assert "## Inventory" in text assert "Verification: `accepted`" in text assert finding.finding_id in text + assert "worth-fixing" not in text.lower() + assert "worth_fixing" not in text.lower() + + +def test_markdown_report_labels_hits_as_inventory_not_worth_fixing(tmp_path: Path) -> None: + finding = _finding() + output = tmp_path / "report.md" + + write_markdown_report([finding], output, [verify_finding(finding)]) + + text = output.read_text() + inventory_at = text.index("## Inventory") + finding_at = text.index(f"## {finding.title}") + assert inventory_at < finding_at + assert "worth-fixing" not in text.lower() def test_scan_exit_code_policy() -> None: @@ -133,6 +179,54 @@ def test_cli_scan_fail_on_worth_fixing_exits_zero_without_verdicts(tmp_path: Pat assert main(["scan", str(repo), "--mode", "quick", "--fail-on", "worth-fixing"]) == 0 +def test_quick_scan_report_is_inventory_without_map_or_docker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast import tool_hunter + from openultrasast.complexity import map as complexity_map + from openultrasast.sandbox import probe as sandbox_probe + from openultrasast.sandbox import runner as sandbox_runner + + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n") + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + + docker_argv: list[list[str]] = [] + original_run = subprocess.run + + def guarded_run(command, *args, **kwargs): # type: ignore[no-untyped-def] + argv = list(command) if isinstance(command, list | tuple) else [command] + if argv and Path(str(argv[0])).name == "docker": + docker_argv.append(argv) + raise AssertionError(f"quick scan must not start docker: {argv}") + return original_run(command, *args, **kwargs) + + def fail(message: str): # type: ignore[no-untyped-def] + def _fail(*args, **kwargs): # type: ignore[no-untyped-def] + raise AssertionError(message) + + return _fail + + monkeypatch.setattr(subprocess, "run", guarded_run) + monkeypatch.setattr(sandbox_runner, "build_docker_argv", fail("quick scan must not build docker argv")) + monkeypatch.setattr(sandbox_probe.SandboxProbe, "available", fail("quick scan must not probe docker")) + monkeypatch.setattr(complexity_map, "build_complexity_map", fail("quick scan must not build a complexity map")) + monkeypatch.setattr(tool_hunter, "run_tool_hunter", fail("quick scan must not start a hunter model")) + + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + + run_dir = sorted((repo / ".runs").iterdir())[-1] + report = (run_dir / "report.md").read_text() + manifest = json.loads((run_dir / "manifest.json").read_text()) + + assert "## Inventory" in report + assert "worth-fixing" not in report.lower() + assert not (run_dir / "complexity_map.json").exists() + assert docker_argv == [] + assert manifest["stages"]["requested"] == ["static"] + assert manifest["stages"]["completed"] == ["static"] + assert manifest["stages"]["skipped"] == [] + + def _finding() -> StaticFinding: return StaticFinding( finding_id="python-unsafe-eval:app.py:3", diff --git a/tests/test_stages.py b/tests/test_stages.py index fa1a7aa..207e9b8 100644 --- a/tests/test_stages.py +++ b/tests/test_stages.py @@ -1,6 +1,6 @@ import json -from openultrasast.stages import MODE_STAGES, Stage, StagePlan, plan_for_mode, record_skip +from openultrasast.stages import MODE_STAGES, Stage, StagePlan, plan_for_mode, record_completed, record_skip, stages_payload def test_mode_stages_maps_existing_depths() -> None: @@ -45,3 +45,24 @@ def test_skipped_stage_is_structured_degradation_not_exception() -> None: assert degradations[0]["stage"] == "regress" assert degradations[0]["reason"] == "sandbox_unavailable" assert json.dumps(degradations) == json.dumps([{"stage": "regress", "reason": "sandbox_unavailable"}]) + + +def test_record_completed_static_for_quick() -> None: + plan = record_completed(plan_for_mode("quick"), Stage.STATIC) + + assert plan.requested == (Stage.STATIC,) + assert plan.completed == (Stage.STATIC,) + assert plan.skipped == () + assert Stage.MAP not in plan.completed + assert Stage.REGRESS not in plan.completed + + +def test_stages_payload_is_manifest_shaped() -> None: + plan = record_completed(plan_for_mode("quick"), Stage.STATIC) + payload = stages_payload(plan) + + assert payload == { + "requested": ["static"], + "completed": ["static"], + "skipped": [], + }