diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 891942f..436954a 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -127,7 +127,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 +- [x] 6. Run stages from the existing scan command - [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. diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 84a6af1..28ccd05 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -28,6 +28,7 @@ from .calibration import ( write_ranking_calibrations, ) from .complexity import map as complexity_map +from .complexity.ledger import persist_verdicts from .config import load_config from .findings import StaticFinding, quick_scan_findings, write_findings from .fusion import FusionDecision, fuse_findings_dispatch @@ -42,7 +43,7 @@ from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sar from .policy import assert_rules_resolve, load_policy from .preprocess import preprocess_repository, write_preprocess_artifact from .rank import rank_targets, write_rankings -from .regress import run_regression, write_verdicts +from .regress import CandidateVerdict, run_regression, write_verdicts from .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report from .ruleset import DEFAULT_RULESET_DIR, load_ruleset from .run import ScanRun, create_scan_run @@ -341,6 +342,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut } verdicts_path = run.root / "verdicts.json" wrote_verdicts = False + verdict_records: tuple[CandidateVerdict, ...] = () if Stage.REGRESS in plan.requested: probe = resolve_sandbox_probe() if not probe.available(): @@ -368,10 +370,21 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut ), ) write_verdicts(records, verdicts_path) + persist_verdicts(run.target / CALIBRATION_DIR / "complexity_ledger.json", records) + verdict_records = records wrote_verdicts = True plan = record_completed(plan, Stage.REGRESS) + worth_fixing_payload = _worth_fixing_payload(verdict_records) if wrote_verdicts else None runtime.run_stage( - "report", lambda: write_markdown_report(findings, markdown_path, verifications, redact=config.hardening.redact_secrets) + "report", + lambda: write_markdown_report( + findings, + markdown_path, + verifications, + redact=config.hardening.redact_secrets, + complexity_map=built_map, + verdicts=verdict_records if wrote_verdicts else None, + ), ) runtime.run_stage("sarif", lambda: write_sarif_report(findings, verifications, sarif_path)) runtime.run_stage( @@ -397,6 +410,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut fusion=[_fusion_summary(decision) for decision in fusion_decisions] or None, stages=stages_payload(plan), complexity=complexity_payload, + worth_fixing=worth_fixing_payload, ), ) runtime.finish(status="succeeded") @@ -407,7 +421,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut ranked_target_count=len(rankings), finding_count=len(findings), calibrations_applied=len(applied_calibrations), - exit_code=scan_exit_code(findings, verifications, fail_on), + exit_code=scan_exit_code(findings, verifications, fail_on, worth_fixing_verdicts=verdict_records), ) @@ -493,6 +507,23 @@ def _load_static_hints(sarif_paths: tuple[str, ...]) -> list[object]: return hints +def _worth_fixing_payload(verdicts: tuple[CandidateVerdict, ...]) -> dict[str, object]: + worth = [record for record in verdicts if record.worth_fixing] + return { + "count": len(worth), + "verdicts": [ + { + "path": record.path, + "function_name": record.function_name, + "verdict": record.verdict, + "reason": record.reason, + "worth_fixing": True, + } + for record in worth + ], + } + + def _artifact_paths(**paths: Path | None) -> dict[str, Path]: return {name: path for name, path in paths.items() if path is not None} diff --git a/src/openultrasast/complexity/__init__.py b/src/openultrasast/complexity/__init__.py index c21894b..3a92c16 100644 --- a/src/openultrasast/complexity/__init__.py +++ b/src/openultrasast/complexity/__init__.py @@ -7,6 +7,7 @@ from .ledger import ( hotspot_key, load_ledger, must_keep_as_candidate, + persist_verdicts, record_verdict, select_forced_candidates, write_ledger, @@ -27,6 +28,7 @@ __all__ = [ "hotspot_key", "load_ledger", "must_keep_as_candidate", + "persist_verdicts", "record_verdict", "select_forced_candidates", "write_complexity_map", diff --git a/src/openultrasast/complexity/hints.py b/src/openultrasast/complexity/hints.py index f7d93ce..3dbe480 100644 --- a/src/openultrasast/complexity/hints.py +++ b/src/openultrasast/complexity/hints.py @@ -35,6 +35,7 @@ class TestHint: gap: str # no_adjacent_test | no_function_reference | no_fuzz_entry | covered test_kind: str | None reason: str + __test__ = False def __post_init__(self) -> None: if self.gap not in GAPS: diff --git a/src/openultrasast/complexity/ledger.py b/src/openultrasast/complexity/ledger.py index 4ad823b..59911ce 100644 --- a/src/openultrasast/complexity/ledger.py +++ b/src/openultrasast/complexity/ledger.py @@ -57,6 +57,27 @@ def write_ledger(path: Path, ledger: Mapping[str, LedgerEntry | Mapping[str, obj path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") +def persist_verdicts(path: Path, verdicts: Sequence[object]) -> dict[str, LedgerEntry]: + """Record triggerable / not_triggerable outcomes into the loop-owned complexity ledger.""" + ledger = load_ledger(path) + changed = False + for item in verdicts: + verdict = str(getattr(item, "verdict", "")) + if verdict not in VERDICT_DELTAS: + continue + raw_function = getattr(item, "function_name", None) + ledger = record_verdict( + ledger, + path=str(getattr(item, "path", "")), + function_name=raw_function if isinstance(raw_function, str) else None, + verdict=verdict, + ) + changed = True + if changed: + write_ledger(path, ledger) + return ledger + + def record_verdict( ledger: Mapping[str, LedgerEntry | Mapping[str, object]], *, diff --git a/src/openultrasast/regress/candidate.py b/src/openultrasast/regress/candidate.py index 0fdf55a..966a857 100644 --- a/src/openultrasast/regress/candidate.py +++ b/src/openultrasast/regress/candidate.py @@ -4,7 +4,7 @@ from __future__ import annotations import json from collections.abc import Mapping, Sequence -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, replace from pathlib import Path from ..complexity.ledger import hotspot_key, select_forced_candidates @@ -200,6 +200,7 @@ def run_regression( image = images.get(language) or DEFAULT_IMAGES.get(language, "ousast-missing-image") mapped = runner.run_recipe(language, snippet, image, sandbox_limits, repo_root=repo_root) reachability = _reachability_for(hotspot, findings_by_id) + mapped = replace(mapped, worth_fixing=is_worth_fixing(mapped.verdict, reachability)) records.append( CandidateVerdict( path=hotspot.path, @@ -208,7 +209,7 @@ def run_regression( verdict=mapped.verdict, reason=mapped.reason, inventory_finding_ids=hotspot.inventory_finding_ids, - worth_fixing=is_worth_fixing(mapped.verdict, reachability), + worth_fixing=mapped.worth_fixing, ) ) return tuple(records) diff --git a/src/openultrasast/regress/verdict.py b/src/openultrasast/regress/verdict.py index 4ec569d..60442ac 100644 --- a/src/openultrasast/regress/verdict.py +++ b/src/openultrasast/regress/verdict.py @@ -50,6 +50,7 @@ _CRASH_MARKERS = ( class RegressionVerdict: verdict: str reason: str = "" + worth_fixing: bool = False def is_worth_fixing(verdict: str, reachability: str) -> bool: diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index c0e0d1d..7bcd6de 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -11,13 +11,20 @@ from .verification import VerificationResult def write_markdown_report( - findings: list[StaticFinding], path: Path, verifications: list[VerificationResult] | None = None, *, redact: bool = True + findings: list[StaticFinding], + path: Path, + verifications: list[VerificationResult] | None = None, + *, + redact: bool = True, + complexity_map: object | None = None, + verdicts: Sequence[object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications or []) lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""] if not findings: lines.append("No quick-mode findings were emitted.") + lines.append("") for finding in findings: verification = verification_by_id.get(finding.finding_id) lines.extend( @@ -41,6 +48,10 @@ def write_markdown_report( "", ] ) + if complexity_map is not None: + _append_complexity_map(lines, complexity_map) + if verdicts is not None: + _append_worth_fixing(lines, verdicts) report = "\n".join(lines).rstrip() + "\n" if redact: from .redaction import redact_secrets @@ -49,6 +60,68 @@ def write_markdown_report( path.write_text(report) +def _append_complexity_map(lines: list[str], complexity_map: object) -> None: + hotspots = tuple(getattr(complexity_map, "hotspots", ()) or ()) + heuristic_only = getattr(complexity_map, "heuristic_only", None) + summary = f"Hotspots: {len(hotspots)}" + if heuristic_only is not None: + summary += f" (heuristic_only={str(bool(heuristic_only)).lower()})" + lines.extend(["## Complexity map", "", summary, ""]) + if not hotspots: + lines.extend(["No complexity hotspots were emitted.", ""]) + return + for hotspot in hotspots: + identity = _report_identity(getattr(hotspot, "path", ""), getattr(hotspot, "function_name", None)) + lines.extend( + [ + f"### `{identity}`", + "", + f"- Score: `{getattr(hotspot, 'score', '')}`", + f"- Band: `{getattr(hotspot, 'band', '')}`", + ] + ) + hint = getattr(hotspot, "test_hint", None) + if hint is not None: + lines.append(f"- Gap: `{getattr(hint, 'gap', '')}`") + kind = getattr(hint, "test_kind", None) + if kind: + lines.append(f"- Recommended test: `{kind}`") + reason = getattr(hint, "reason", "") + if reason: + lines.append(f"- Hint: {reason}") + inventory_ids = tuple(getattr(hotspot, "inventory_finding_ids", ()) or ()) + if inventory_ids: + lines.append(f"- Inventory: `{', '.join(str(item) for item in inventory_ids)}`") + rationale = str(getattr(hotspot, "rationale", "") or "") + lines.extend(["", rationale, ""] if rationale else [""]) + + +def _append_worth_fixing(lines: list[str], verdicts: Sequence[object]) -> None: + worth = [item for item in verdicts if getattr(item, "worth_fixing", False)] + lines.extend(["## Worth fixing", "", f"Count: {len(worth)}", ""]) + if not worth: + lines.extend(["No worth-fixing verdicts were produced.", ""]) + return + for item in worth: + identity = _report_identity(getattr(item, "path", ""), getattr(item, "function_name", None)) + lines.extend( + [ + f"### `{identity}`", + "", + f"- Verdict: `{getattr(item, 'verdict', '')}`", + f"- Reason: `{getattr(item, 'reason', '')}`", + "", + ] + ) + + +def _report_identity(path: object, function_name: object) -> str: + path_text = str(path or "") + if function_name: + return f"{path_text}::{function_name}" + return path_text + + def write_sarif_report(findings: list[StaticFinding], verifications: list[VerificationResult], path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -84,6 +157,7 @@ def write_manifest( fusion: list[dict[str, object]] | None = None, stages: dict[str, object] | None = None, complexity: dict[str, object] | None = None, + worth_fixing: dict[str, object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -116,6 +190,8 @@ def write_manifest( payload["stages"] = stages if complexity is not None: payload["complexity"] = complexity + if worth_fixing is not None: + payload["worth_fixing"] = worth_fixing path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") diff --git a/tests/test_complexity_ledger.py b/tests/test_complexity_ledger.py index 584bdc3..36ac18d 100644 --- a/tests/test_complexity_ledger.py +++ b/tests/test_complexity_ledger.py @@ -1,5 +1,6 @@ import json from pathlib import Path +from types import SimpleNamespace from openultrasast.complexity.ledger import ( NOT_TRIGGERABLE_DELTA, @@ -9,6 +10,7 @@ from openultrasast.complexity.ledger import ( hotspot_key, load_ledger, must_keep_as_candidate, + persist_verdicts, record_verdict, select_forced_candidates, write_ledger, @@ -248,6 +250,25 @@ def test_overlay_reorders_and_rebands_without_dropping_hits() -> None: assert overlaid[0].inventory_finding_ids == ("low:1",) +def test_persist_verdicts_records_triggerable_and_skips_inconclusive(tmp_path: Path) -> None: + path = tmp_path / ".openultrasast" / "calibration" / "complexity_ledger.json" + persist_verdicts( + path, + [ + SimpleNamespace(path="app.py", function_name="admin", verdict="triggerable"), + SimpleNamespace(path="safe.py", function_name=None, verdict="not_triggerable"), + SimpleNamespace(path="other.py", function_name="skip", verdict="inconclusive"), + ], + ) + loaded = load_ledger(path) + + assert loaded[hotspot_key("app.py", "admin")].last_verdict == "triggerable" + assert loaded[hotspot_key("app.py", "admin")].score_delta == TRIGGERABLE_DELTA + assert loaded[hotspot_key("safe.py", None)].last_verdict == "not_triggerable" + assert loaded[hotspot_key("safe.py", None)].score_delta == NOT_TRIGGERABLE_DELTA + assert hotspot_key("other.py", "skip") not in loaded + + def test_build_complexity_map_without_ledger_path_skips_overlay(tmp_path: Path) -> None: target = _target("src/app.py", loc=20) finding = _hit("src/app.py", "rule:src/app.py:1") diff --git a/tests/test_reports.py b/tests/test_reports.py index f84012f..94afdbc 100644 --- a/tests/test_reports.py +++ b/tests/test_reports.py @@ -6,7 +6,10 @@ from types import SimpleNamespace import pytest from openultrasast.cli import main +from openultrasast.complexity.hints import TestHint +from openultrasast.complexity.map import ComplexityMap, Hotspot from openultrasast.findings import StaticFinding +from openultrasast.regress import CandidateVerdict 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 @@ -96,6 +99,69 @@ def test_markdown_report_includes_verification_status(tmp_path: Path) -> None: assert "worth_fixing" not in text.lower() +def test_markdown_report_includes_complexity_map_and_worth_fixing_sections(tmp_path: Path) -> None: + finding = _finding() + output = tmp_path / "report.md" + hint = TestHint( + path="app.py", + function_name="admin", + gap="no_adjacent_test", + test_kind="http-contract", + reason="no adjacent test file for app.py::admin; recommend http-contract.", + ) + hotspot = Hotspot( + path="app.py", + function_name="admin", + score=6.5, + band="high", + signals={"loc": 12, "has_adjacent_test": False}, + rationale="high band: nested public route", + test_hint=hint, + inventory_finding_ids=(finding.finding_id,), + ) + verdict = CandidateVerdict( + path="app.py", + function_name="admin", + language="python", + verdict="triggerable", + reason="crash", + inventory_finding_ids=(finding.finding_id,), + worth_fixing=True, + ) + skipped = CandidateVerdict( + path="safe.py", + function_name=None, + language="python", + verdict="not_triggerable", + reason="exit_zero", + inventory_finding_ids=(), + worth_fixing=False, + ) + + write_markdown_report( + [finding], + output, + [verify_finding(finding)], + complexity_map=ComplexityMap(hotspots=(hotspot,), heuristic_only=True), + verdicts=(verdict, skipped), + ) + + text = output.read_text() + inventory_at = text.index("## Inventory") + map_at = text.index("## Complexity map") + worth_at = text.index("## Worth fixing") + assert inventory_at < map_at < worth_at + map_section = text[map_at:worth_at] + worth_section = text[worth_at:] + assert "app.py::admin" in map_section + assert "Gap: `no_adjacent_test`" in map_section + assert "Recommended test: `http-contract`" in map_section + assert "triggerable" in worth_section + assert "crash" in worth_section + assert "safe.py" not in worth_section + assert "## Inventory" in text + + def test_markdown_report_labels_hits_as_inventory_not_worth_fixing(tmp_path: Path) -> None: finding = _finding() output = tmp_path / "report.md" @@ -277,6 +343,11 @@ def test_standard_scan_writes_complexity_map_without_docker(tmp_path: Path, monk assert manifest["complexity"]["hotspot_count"] == len(complexity_map["hotspots"]) assert manifest["complexity"]["heuristic_only"] is True assert manifest["artifacts"]["complexity_map"] == "complexity_map.json" + report = (run_dir / "report.md").read_text() + assert "## Inventory" in report + assert "## Complexity map" in report + assert "Gap:" in report + assert "## Worth fixing" not in report def test_standard_scan_records_hunter_model_unavailable_when_model_unset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -337,6 +408,38 @@ def test_manifest_includes_complexity_summary(tmp_path: Path) -> None: assert payload["artifacts"]["complexity_map"] == "complexity_map.json" +def test_manifest_includes_worth_fixing_count(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", + "verdicts": run.root / "verdicts.json", + } + output = run.root / "manifest.json" + + write_manifest( + run=run, + findings=[finding], + verifications=[verify_finding(finding)], + artifact_paths=artifacts, + path=output, + stages={"requested": ["static", "map", "regress"], "completed": ["static", "map", "regress"], "skipped": []}, + worth_fixing={ + "count": 1, + "verdicts": [{"path": "app.py", "function_name": "admin", "verdict": "triggerable", "reason": "crash"}], + }, + ) + + payload = json.loads(output.read_text()) + assert payload["stages"]["completed"] == ["static", "map", "regress"] + assert payload["worth_fixing"]["count"] == 1 + assert payload["worth_fixing"]["verdicts"][0]["verdict"] == "triggerable" + + def _finding() -> StaticFinding: return StaticFinding( finding_id="python-unsafe-eval:app.py:3", diff --git a/tests/test_scan_deep.py b/tests/test_scan_deep.py index b2f2405..eb462cc 100644 --- a/tests/test_scan_deep.py +++ b/tests/test_scan_deep.py @@ -9,6 +9,7 @@ from pathlib import Path import pytest from openultrasast.cli import main +from openultrasast.sandbox import FakeSandboxRunner, SandboxResult from openultrasast.sandbox import runner as sandbox_runner _VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n" @@ -48,6 +49,20 @@ def _latest_run(repo: Path) -> Path: return sorted((repo / ".runs").iterdir())[-1] +def _crash_runner() -> FakeSandboxRunner: + return FakeSandboxRunner(SandboxResult(exit_code=139, stdout="", stderr="Segmentation fault\n", timed_out=False)) + + +def _enable_crash_sandbox(monkeypatch: pytest.MonkeyPatch) -> FakeSandboxRunner: + from openultrasast import cli + + fake = _crash_runner() + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1") + monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake) + return fake + + def test_deep_scan_skips_regress_when_sandbox_unavailable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: repo = _repo(tmp_path) monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") @@ -99,6 +114,20 @@ def test_deep_scan_with_fake_runner_writes_verdicts(tmp_path: Path, monkeypatch: assert "path" in item assert manifest["artifacts"]["verdicts"] == "verdicts.json" assert manifest["artifacts"]["complexity_map"] == "complexity_map.json" + assert manifest["worth_fixing"]["count"] == sum(1 for item in payload["verdicts"] if item.get("worth_fixing")) + report = (run_dir / "report.md").read_text() + inventory_at = report.index("## Inventory") + map_at = report.index("## Complexity map") + worth_at = report.index("## Worth fixing") + assert inventory_at < map_at < worth_at + map_section = report[map_at:worth_at] + assert "app.py" in map_section + assert "Gap:" in map_section + ledger_path = repo / ".openultrasast" / "calibration" / "complexity_ledger.json" + assert ledger_path.is_file() + ledger = json.loads(ledger_path.read_text()) + assert ledger + assert any(entry.get("last_verdict") in {"triggerable", "not_triggerable"} for entry in ledger.values()) def test_deep_scan_unsafe_hunter_snippet_is_safety_rejected_without_docker_job(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: @@ -132,3 +161,40 @@ def test_deep_scan_unsafe_hunter_snippet_is_safety_rejected_without_docker_job(t for item in payload["verdicts"]: if item["reason"] == "safety_rejected": assert item["verdict"] == "inconclusive" + + +def test_deep_scan_fail_on_worth_fixing_exits_one_when_triggerable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path) + _enable_crash_sandbox(monkeypatch) + docker_argv = _guard_docker(monkeypatch) + + assert main(["scan", str(repo), "--mode", "deep", "--fail-on", "worth-fixing"]) == 1 + assert docker_argv == [] + + run_dir = _latest_run(repo) + payload = json.loads((run_dir / "verdicts.json").read_text()) + manifest = json.loads((run_dir / "manifest.json").read_text()) + report = (run_dir / "report.md").read_text() + worth_fixing = [item for item in payload["verdicts"] if item.get("worth_fixing")] + + assert worth_fixing + assert len(worth_fixing) >= 1 + assert any(item["verdict"] == "triggerable" for item in worth_fixing) + assert manifest["worth_fixing"]["count"] == len(worth_fixing) + assert "## Complexity map" in report + assert "## Worth fixing" in report + assert "triggerable" in report + ledger = json.loads((repo / ".openultrasast" / "calibration" / "complexity_ledger.json").read_text()) + assert any(entry.get("last_verdict") == "triggerable" for entry in ledger.values()) + + +def test_deep_scan_fail_on_never_exits_zero_with_worth_fixing_verdict(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path) + _enable_crash_sandbox(monkeypatch) + docker_argv = _guard_docker(monkeypatch) + + assert main(["scan", str(repo), "--mode", "deep", "--fail-on", "never"]) == 0 + assert docker_argv == [] + + payload = json.loads((_latest_run(repo) / "verdicts.json").read_text()) + assert any(item.get("worth_fixing") for item in payload["verdicts"])