feat(three-stage-scan): report map and honor fail-on worth-fixing

This commit is contained in:
Marius Ciepluch
2026-09-04 00:18:33 +02:00
parent a916e2d173
commit 818a0b10d3
11 changed files with 330 additions and 7 deletions
+1 -1
View File
@@ -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 ## 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 - [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. - Quick still loads policy, ruleset, preprocess, rank, pattern inventory, structural verify, and score.
- The report labels those hits as inventory, not as worth-fixing. - The report labels those hits as inventory, not as worth-fixing.
+34 -3
View File
@@ -28,6 +28,7 @@ from .calibration import (
write_ranking_calibrations, write_ranking_calibrations,
) )
from .complexity import map as complexity_map from .complexity import map as complexity_map
from .complexity.ledger import persist_verdicts
from .config import load_config from .config import load_config
from .findings import StaticFinding, quick_scan_findings, write_findings from .findings import StaticFinding, quick_scan_findings, write_findings
from .fusion import FusionDecision, fuse_findings_dispatch 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 .policy import assert_rules_resolve, load_policy
from .preprocess import preprocess_repository, write_preprocess_artifact from .preprocess import preprocess_repository, write_preprocess_artifact
from .rank import rank_targets, write_rankings 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 .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report
from .ruleset import DEFAULT_RULESET_DIR, load_ruleset from .ruleset import DEFAULT_RULESET_DIR, load_ruleset
from .run import ScanRun, create_scan_run 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" verdicts_path = run.root / "verdicts.json"
wrote_verdicts = False wrote_verdicts = False
verdict_records: tuple[CandidateVerdict, ...] = ()
if Stage.REGRESS in plan.requested: if Stage.REGRESS in plan.requested:
probe = resolve_sandbox_probe() probe = resolve_sandbox_probe()
if not probe.available(): 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) write_verdicts(records, verdicts_path)
persist_verdicts(run.target / CALIBRATION_DIR / "complexity_ledger.json", records)
verdict_records = records
wrote_verdicts = True wrote_verdicts = True
plan = record_completed(plan, Stage.REGRESS) plan = record_completed(plan, Stage.REGRESS)
worth_fixing_payload = _worth_fixing_payload(verdict_records) if wrote_verdicts else None
runtime.run_stage( 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("sarif", lambda: write_sarif_report(findings, verifications, sarif_path))
runtime.run_stage( 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, fusion=[_fusion_summary(decision) for decision in fusion_decisions] or None,
stages=stages_payload(plan), stages=stages_payload(plan),
complexity=complexity_payload, complexity=complexity_payload,
worth_fixing=worth_fixing_payload,
), ),
) )
runtime.finish(status="succeeded") 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), ranked_target_count=len(rankings),
finding_count=len(findings), finding_count=len(findings),
calibrations_applied=len(applied_calibrations), 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 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]: def _artifact_paths(**paths: Path | None) -> dict[str, Path]:
return {name: path for name, path in paths.items() if path is not None} return {name: path for name, path in paths.items() if path is not None}
+2
View File
@@ -7,6 +7,7 @@ from .ledger import (
hotspot_key, hotspot_key,
load_ledger, load_ledger,
must_keep_as_candidate, must_keep_as_candidate,
persist_verdicts,
record_verdict, record_verdict,
select_forced_candidates, select_forced_candidates,
write_ledger, write_ledger,
@@ -27,6 +28,7 @@ __all__ = [
"hotspot_key", "hotspot_key",
"load_ledger", "load_ledger",
"must_keep_as_candidate", "must_keep_as_candidate",
"persist_verdicts",
"record_verdict", "record_verdict",
"select_forced_candidates", "select_forced_candidates",
"write_complexity_map", "write_complexity_map",
+1
View File
@@ -35,6 +35,7 @@ class TestHint:
gap: str # no_adjacent_test | no_function_reference | no_fuzz_entry | covered gap: str # no_adjacent_test | no_function_reference | no_fuzz_entry | covered
test_kind: str | None test_kind: str | None
reason: str reason: str
__test__ = False
def __post_init__(self) -> None: def __post_init__(self) -> None:
if self.gap not in GAPS: if self.gap not in GAPS:
+21
View File
@@ -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") 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( def record_verdict(
ledger: Mapping[str, LedgerEntry | Mapping[str, object]], ledger: Mapping[str, LedgerEntry | Mapping[str, object]],
*, *,
+3 -2
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
import json import json
from collections.abc import Mapping, Sequence from collections.abc import Mapping, Sequence
from dataclasses import asdict, dataclass from dataclasses import asdict, dataclass, replace
from pathlib import Path from pathlib import Path
from ..complexity.ledger import hotspot_key, select_forced_candidates 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") 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) mapped = runner.run_recipe(language, snippet, image, sandbox_limits, repo_root=repo_root)
reachability = _reachability_for(hotspot, findings_by_id) reachability = _reachability_for(hotspot, findings_by_id)
mapped = replace(mapped, worth_fixing=is_worth_fixing(mapped.verdict, reachability))
records.append( records.append(
CandidateVerdict( CandidateVerdict(
path=hotspot.path, path=hotspot.path,
@@ -208,7 +209,7 @@ def run_regression(
verdict=mapped.verdict, verdict=mapped.verdict,
reason=mapped.reason, reason=mapped.reason,
inventory_finding_ids=hotspot.inventory_finding_ids, inventory_finding_ids=hotspot.inventory_finding_ids,
worth_fixing=is_worth_fixing(mapped.verdict, reachability), worth_fixing=mapped.worth_fixing,
) )
) )
return tuple(records) return tuple(records)
+1
View File
@@ -50,6 +50,7 @@ _CRASH_MARKERS = (
class RegressionVerdict: class RegressionVerdict:
verdict: str verdict: str
reason: str = "" reason: str = ""
worth_fixing: bool = False
def is_worth_fixing(verdict: str, reachability: str) -> bool: def is_worth_fixing(verdict: str, reachability: str) -> bool:
+77 -1
View File
@@ -11,13 +11,20 @@ from .verification import VerificationResult
def write_markdown_report( 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: ) -> None:
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
verification_by_id = _verification_by_id(verifications or []) verification_by_id = _verification_by_id(verifications or [])
lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""] lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""]
if not findings: if not findings:
lines.append("No quick-mode findings were emitted.") lines.append("No quick-mode findings were emitted.")
lines.append("")
for finding in findings: for finding in findings:
verification = verification_by_id.get(finding.finding_id) verification = verification_by_id.get(finding.finding_id)
lines.extend( 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" report = "\n".join(lines).rstrip() + "\n"
if redact: if redact:
from .redaction import redact_secrets from .redaction import redact_secrets
@@ -49,6 +60,68 @@ def write_markdown_report(
path.write_text(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: def write_sarif_report(findings: list[StaticFinding], verifications: list[VerificationResult], path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
verification_by_id = _verification_by_id(verifications) verification_by_id = _verification_by_id(verifications)
@@ -84,6 +157,7 @@ def write_manifest(
fusion: list[dict[str, object]] | None = None, fusion: list[dict[str, object]] | None = None,
stages: dict[str, object] | None = None, stages: dict[str, object] | None = None,
complexity: dict[str, object] | None = None, complexity: dict[str, object] | None = None,
worth_fixing: dict[str, object] | None = None,
) -> None: ) -> None:
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
verification_by_id = _verification_by_id(verifications) verification_by_id = _verification_by_id(verifications)
@@ -116,6 +190,8 @@ def write_manifest(
payload["stages"] = stages payload["stages"] = stages
if complexity is not None: if complexity is not None:
payload["complexity"] = complexity 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") path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+21
View File
@@ -1,5 +1,6 @@
import json import json
from pathlib import Path from pathlib import Path
from types import SimpleNamespace
from openultrasast.complexity.ledger import ( from openultrasast.complexity.ledger import (
NOT_TRIGGERABLE_DELTA, NOT_TRIGGERABLE_DELTA,
@@ -9,6 +10,7 @@ from openultrasast.complexity.ledger import (
hotspot_key, hotspot_key,
load_ledger, load_ledger,
must_keep_as_candidate, must_keep_as_candidate,
persist_verdicts,
record_verdict, record_verdict,
select_forced_candidates, select_forced_candidates,
write_ledger, write_ledger,
@@ -248,6 +250,25 @@ def test_overlay_reorders_and_rebands_without_dropping_hits() -> None:
assert overlaid[0].inventory_finding_ids == ("low:1",) 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: def test_build_complexity_map_without_ledger_path_skips_overlay(tmp_path: Path) -> None:
target = _target("src/app.py", loc=20) target = _target("src/app.py", loc=20)
finding = _hit("src/app.py", "rule:src/app.py:1") finding = _hit("src/app.py", "rule:src/app.py:1")
+103
View File
@@ -6,7 +6,10 @@ from types import SimpleNamespace
import pytest import pytest
from openultrasast.cli import main 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.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.reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report
from openultrasast.run import ScanRun from openultrasast.run import ScanRun
from openultrasast.stages import Stage, plan_for_mode, record_completed, stages_payload 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() 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: def test_markdown_report_labels_hits_as_inventory_not_worth_fixing(tmp_path: Path) -> None:
finding = _finding() finding = _finding()
output = tmp_path / "report.md" 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"]["hotspot_count"] == len(complexity_map["hotspots"])
assert manifest["complexity"]["heuristic_only"] is True assert manifest["complexity"]["heuristic_only"] is True
assert manifest["artifacts"]["complexity_map"] == "complexity_map.json" 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: 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" 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: def _finding() -> StaticFinding:
return StaticFinding( return StaticFinding(
finding_id="python-unsafe-eval:app.py:3", finding_id="python-unsafe-eval:app.py:3",
+66
View File
@@ -9,6 +9,7 @@ from pathlib import Path
import pytest import pytest
from openultrasast.cli import main from openultrasast.cli import main
from openultrasast.sandbox import FakeSandboxRunner, SandboxResult
from openultrasast.sandbox import runner as sandbox_runner from openultrasast.sandbox import runner as sandbox_runner
_VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n" _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] 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: def test_deep_scan_skips_regress_when_sandbox_unavailable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
repo = _repo(tmp_path) repo = _repo(tmp_path)
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") 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 "path" in item
assert manifest["artifacts"]["verdicts"] == "verdicts.json" assert manifest["artifacts"]["verdicts"] == "verdicts.json"
assert manifest["artifacts"]["complexity_map"] == "complexity_map.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: 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"]: for item in payload["verdicts"]:
if item["reason"] == "safety_rejected": if item["reason"] == "safety_rejected":
assert item["verdict"] == "inconclusive" 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"])