feat(three-stage-scan): keep quick as stage-1 inventory

This commit is contained in:
Marius Ciepluch
2026-09-03 23:14:00 +02:00
parent 331b62f1c1
commit 66d985e9e6
6 changed files with 138 additions and 4 deletions
+2 -2
View File
@@ -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.
+4
View File
@@ -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")
+4 -1
View File
@@ -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")
+12
View File
@@ -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],
}
+94
View File
@@ -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",
+22 -1
View File
@@ -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": [],
}