diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 8da4253..62df68c 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -137,7 +137,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Boundary: StagePlan_ - _Depends: 1.1_ -- [ ] 6.2 Run the complexity map on standard and record heuristic-only when no model is set +- [x] 6.2 Run the complexity map on standard and record heuristic-only when no model is set - Standard writes the map artifact, includes it in the manifest, and records `hunter_model_unavailable` when the hunter is skipped. - Observable completion: `ousast scan --mode standard` on a fixture produces `complexity_map.json` and a completed map stage without Docker. - _Requirements: 1.2, 1.4, 3.3, 3.4, 6.4, 9.4, 9.6_ diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 5ce8100..eaf6f3f 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -26,6 +26,7 @@ from .calibration import ( write_false_positive_learnings, write_ranking_calibrations, ) +from .complexity import map as complexity_map from .config import load_config from .findings import StaticFinding, quick_scan_findings, write_findings from .fusion import FusionDecision, fuse_findings_dispatch @@ -44,7 +45,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 .stages import Stage, plan_for_mode, record_completed, skip_as_degradation, stages_payload from .verification import VerificationResult, write_verification_results from .verify_judge import verify_findings_dispatch @@ -302,6 +303,26 @@ 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) + complexity_payload: dict[str, object] | None = None + complexity_map_path = run.root / "complexity_map.json" + if Stage.MAP in plan.requested: + built_map = runtime.run_stage( + "map", + lambda: complexity_map.build_complexity_map( + targets, + findings, + complexity_map_path, + repo_files=[target.path for target in targets], + ledger_path=run.target / CALIBRATION_DIR / "complexity_ledger.json", + ), + ) + plan = record_completed(plan, Stage.MAP) + if not hunter_model: + runtime.state["degradations"].append(skip_as_degradation(Stage.MAP, "hunter_model_unavailable")) + complexity_payload = { + "hotspot_count": len(built_map.hotspots), + "heuristic_only": built_map.heuristic_only, + } runtime.run_stage( "report", lambda: write_markdown_report(findings, markdown_path, verifications, redact=config.hardening.redact_secrets) ) @@ -320,12 +341,14 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut score=score_path, trajectories=trajectories_path if trajectories else None, fusion=fusion_path if fusion_decisions else None, + complexity_map=complexity_map_path if complexity_payload is not None else None, ), path=manifest_path, 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), + complexity=complexity_payload, ), ) runtime.finish(status="succeeded") diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index d37da0a..c0e0d1d 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -83,6 +83,7 @@ def write_manifest( degradations: list[dict[str, object]] | None = None, fusion: list[dict[str, object]] | None = None, stages: dict[str, object] | None = None, + complexity: dict[str, object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -113,6 +114,8 @@ def write_manifest( payload["fusion"] = fusion if stages is not None: payload["stages"] = stages + if complexity is not None: + payload["complexity"] = complexity path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") diff --git a/tests/test_cli_hx_dispatch.py b/tests/test_cli_hx_dispatch.py index 1ac2a88..15286e0 100644 --- a/tests/test_cli_hx_dispatch.py +++ b/tests/test_cli_hx_dispatch.py @@ -45,9 +45,11 @@ def test_degradation_recorded_when_models_set_but_extra_absent(tmp_path: Path, m assert manifest["findings"] -def test_manifest_has_no_degradations_without_models(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_manifest_has_no_harnessx_degradations_without_models(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: manifest = _run(tmp_path, monkeypatch, config_body=None) - assert "degradations" not in manifest + degradations = manifest.get("degradations", []) + assert all(entry.get("reason") != "harnessx_extra_unavailable" for entry in degradations) + assert any(entry.get("reason") == "hunter_model_unavailable" for entry in degradations) def test_standard_scan_falls_back_to_run_hunter_pool(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/test_reports.py b/tests/test_reports.py index f34a761..f84012f 100644 --- a/tests/test_reports.py +++ b/tests/test_reports.py @@ -225,6 +225,116 @@ def test_quick_scan_report_is_inventory_without_map_or_docker(tmp_path: Path, mo assert manifest["stages"]["requested"] == ["static"] assert manifest["stages"]["completed"] == ["static"] assert manifest["stages"]["skipped"] == [] + assert "complexity" not in manifest + assert "complexity_map" not in manifest.get("artifacts", {}) + assert not any(entry.get("reason") == "hunter_model_unavailable" for entry in manifest.get("degradations", [])) + + +def test_standard_scan_writes_complexity_map_without_docker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + 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"standard 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("standard scan must not build docker argv")) + monkeypatch.setattr(sandbox_probe.SandboxProbe, "available", fail("standard scan must not probe docker")) + + assert main(["scan", str(repo), "--mode", "standard"]) == 0 + + run_dir = sorted((repo / ".runs").iterdir())[-1] + map_path = run_dir / "complexity_map.json" + manifest = json.loads((run_dir / "manifest.json").read_text()) + complexity_map = json.loads(map_path.read_text()) + + assert map_path.is_file() + assert complexity_map["heuristic_only"] is True + assert isinstance(complexity_map["hotspots"], list) + assert complexity_map["hotspots"] + assert docker_argv == [] + assert manifest["stages"]["requested"] == ["static", "map"] + assert manifest["stages"]["completed"] == ["static", "map"] + assert manifest["stages"]["skipped"] == [] + assert manifest["complexity"]["hotspot_count"] == len(complexity_map["hotspots"]) + assert manifest["complexity"]["heuristic_only"] is True + assert manifest["artifacts"]["complexity_map"] == "complexity_map.json" + + +def test_standard_scan_records_hunter_model_unavailable_when_model_unset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + 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") + + 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(sandbox_runner, "build_docker_argv", fail("standard scan must not build docker argv")) + monkeypatch.setattr(sandbox_probe.SandboxProbe, "available", fail("standard scan must not probe docker")) + + assert main(["scan", str(repo), "--mode", "standard"]) == 0 + + run_dir = sorted((repo / ".runs").iterdir())[-1] + manifest = json.loads((run_dir / "manifest.json").read_text()) + degradations = manifest.get("degradations", []) + + assert any(entry.get("reason") == "hunter_model_unavailable" for entry in degradations) + hunter_skip = next(entry for entry in degradations if entry.get("reason") == "hunter_model_unavailable") + assert hunter_skip["stage"] in {"map", "hunter_pool"} + assert manifest["complexity"]["heuristic_only"] is True + assert (run_dir / "complexity_map.json").is_file() + + +def test_manifest_includes_complexity_summary(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", + "complexity_map": run.root / "complexity_map.json", + } + output = run.root / "manifest.json" + + write_manifest( + run=run, + findings=[finding], + verifications=[verify_finding(finding)], + artifact_paths=artifacts, + path=output, + complexity={"hotspot_count": 3, "heuristic_only": True}, + ) + + payload = json.loads(output.read_text()) + assert payload["complexity"] == {"hotspot_count": 3, "heuristic_only": True} + assert payload["artifacts"]["complexity_map"] == "complexity_map.json" def _finding() -> StaticFinding: diff --git a/tests/test_scan_pipeline_integrity.py b/tests/test_scan_pipeline_integrity.py index 9cb0663..8a4d370 100644 --- a/tests/test_scan_pipeline_integrity.py +++ b/tests/test_scan_pipeline_integrity.py @@ -65,7 +65,8 @@ def test_standard_scan_extra_absent_matches_deterministic_baseline(tmp_path: Pat stages = {entry["stage"] for entry in requested_manifest.get("degradations", [])} assert stages == {"hunter_pool", "verify"} assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in requested_manifest["degradations"]) - assert "degradations" not in baseline_manifest + assert all(entry["reason"] != "hunter_model_unavailable" for entry in requested_manifest["degradations"]) + assert [entry["reason"] for entry in baseline_manifest.get("degradations", [])] == ["hunter_model_unavailable"] def test_quick_scan_is_unaffected_by_the_extra(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: