diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 62df68c..8570030 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -143,7 +143,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Requirements: 1.2, 1.4, 3.3, 3.4, 6.4, 9.4, 9.6_ - _Depends: 2.3, 6.1_ -- [ ] 6.3 Stop exiting on deep; run or skip regression with visible degradation +- [x] 6.3 Stop exiting on deep; run or skip regression with visible degradation - Deep no longer hard-exits. If the sandbox is missing, stages 1–2 complete and regress is skipped with `sandbox_unavailable`. - If the sandbox is present, regression runs under the candidate cap and writes verdicts. - Observable completion: `ousast scan --mode deep` without Docker exits 0 after stages 1–2 and lists the skip in the manifest; with a fake runner it records verdicts. diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index eaf6f3f..bc82767 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -41,11 +41,13 @@ 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 .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 +from .sandbox import resolve_sandbox_probe, resolve_sandbox_runner from .scoring import build_score_artifact -from .stages import Stage, plan_for_mode, record_completed, skip_as_degradation, stages_payload +from .stages import Stage, plan_for_mode, record_completed, record_skip, skip_as_degradation, stages_payload from .verification import VerificationResult, write_verification_results from .verify_judge import verify_findings_dispatch @@ -134,8 +136,6 @@ def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int: def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOutcome: - if mode == "deep": - raise SystemExit("mode 'deep' is specified but sandboxed dynamic analysis is not implemented") if not path.exists() or not path.is_dir(): raise SystemExit(f"scan path is not a directory: {path}") @@ -305,6 +305,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut plan = record_completed(plan, Stage.STATIC) complexity_payload: dict[str, object] | None = None complexity_map_path = run.root / "complexity_map.json" + built_map = None if Stage.MAP in plan.requested: built_map = runtime.run_stage( "map", @@ -323,6 +324,37 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut "hotspot_count": len(built_map.hotspots), "heuristic_only": built_map.heuristic_only, } + verdicts_path = run.root / "verdicts.json" + wrote_verdicts = False + if Stage.REGRESS in plan.requested: + probe = resolve_sandbox_probe() + if not probe.available(): + plan = record_skip(plan, Stage.REGRESS, "sandbox_unavailable") + runtime.state["degradations"].append(skip_as_degradation(Stage.REGRESS, "sandbox_unavailable")) + else: + sandbox = resolve_sandbox_runner() + hotspots = built_map.hotspots if built_map is not None else () + cwe_by_rule_id = {rule.rule_id: rule.cwe for rule in ruleset} + rule_cwe_by_finding = {finding.finding_id: cwe_by_rule_id.get(finding.finding_id.split(":", 1)[0], "") for finding in findings} + languages_by_path = {target.path: target.language for target in targets} + records = runtime.run_stage( + "regress", + lambda: run_regression( + hotspots, + findings, + max_candidates=config.regress.max_candidates, + policy=policy, + rule_cwe=rule_cwe_by_finding, + languages_by_path=languages_by_path, + repo_root=run.target, + sandbox=sandbox, + sandbox_limits=config.sandbox, + images=dict(config.regress.images), + ), + ) + write_verdicts(records, verdicts_path) + wrote_verdicts = True + plan = record_completed(plan, Stage.REGRESS) runtime.run_stage( "report", lambda: write_markdown_report(findings, markdown_path, verifications, redact=config.hardening.redact_secrets) ) @@ -342,6 +374,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut 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, + verdicts=verdicts_path if wrote_verdicts else None, ), path=manifest_path, score=score_artifact.to_dict(), diff --git a/src/openultrasast/regress/__init__.py b/src/openultrasast/regress/__init__.py index 4097be5..35b9705 100644 --- a/src/openultrasast/regress/__init__.py +++ b/src/openultrasast/regress/__init__.py @@ -1,6 +1,6 @@ """Isolated regression snippets and verdicts.""" -from .candidate import RegressionRunner, select_candidates +from .candidate import CandidateVerdict, RegressionRunner, run_regression, select_candidates, write_verdicts from .recipes import recipe_for from .safety import UnsafeSnippetError, check_snippet_safety from .verdict import ( @@ -22,12 +22,15 @@ __all__ = [ "NOT_TRIGGERABLE", "SAFETY_REJECTED", "TRIGGERABLE", + "CandidateVerdict", "RegressionRunner", "RegressionVerdict", "UnsafeSnippetError", "check_snippet_safety", "is_worth_fixing", "recipe_for", + "run_regression", "select_candidates", "verdict_from_result", + "write_verdicts", ] diff --git a/src/openultrasast/regress/candidate.py b/src/openultrasast/regress/candidate.py index d4fb045..1b7c7ee 100644 --- a/src/openultrasast/regress/candidate.py +++ b/src/openultrasast/regress/candidate.py @@ -2,14 +2,18 @@ from __future__ import annotations +import json from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass from pathlib import Path from ..complexity.ledger import hotspot_key, select_forced_candidates from ..complexity.map import Hotspot from ..config import SandboxConfig from ..findings import StaticFinding +from ..preprocess import LANGUAGE_BY_EXTENSION from ..sandbox import SandboxJob, SandboxResult, SandboxRunner +from ..sandbox.runner import WORKSPACE_MOUNT from .recipes import recipe_for from .safety import UnsafeSnippetError, check_snippet_safety from .verdict import ( @@ -19,21 +23,43 @@ from .verdict import ( SAFETY_REJECTED, TRIGGERABLE, RegressionVerdict, + is_worth_fixing, verdict_from_result, ) +DEFAULT_IMAGES = { + "python": "python:3.12-alpine", + "javascript": "node:22-alpine", + "c": "gcc:13", +} + __all__ = [ "INCONCLUSIVE", "MISSING_RECIPE", "NOT_TRIGGERABLE", "SAFETY_REJECTED", "TRIGGERABLE", + "CandidateVerdict", "RegressionRunner", "RegressionVerdict", + "run_regression", "select_candidates", + "snippet_for", + "write_verdicts", ] +@dataclass(frozen=True) +class CandidateVerdict: + path: str + function_name: str | None + language: str + verdict: str + reason: str + inventory_finding_ids: tuple[str, ...] + worth_fixing: bool = False + + def select_candidates( hotspots: Sequence[Hotspot], findings: Sequence[StaticFinding], @@ -118,3 +144,93 @@ class RegressionRunner: def _verdict_from_result(result: SandboxResult) -> RegressionVerdict: return verdict_from_result(result) + + +def snippet_for(language: str, path: str, function_name: str | None) -> str: + """Render a scratch snippet that loads the candidate source inside the sandbox.""" + relative = path.replace("\\", "/") + workspace_path = f"{WORKSPACE_MOUNT}/{relative}" + identity = function_name or relative + if language == "python": + return ( + f"# candidate {identity}\n" + "from pathlib import Path\n" + f"source = Path({workspace_path!r}).read_text()\n" + f"compile(source, {relative!r}, 'exec')\n" + ) + if language == "javascript": + return ( + f"// candidate {identity}\n" + "const fs = require('fs');\n" + f"const source = fs.readFileSync({workspace_path!r}, 'utf8');\n" + "if (!source.length) { process.exit(1); }\n" + ) + if language == "c": + return f"/* candidate {identity} */\nint main(void) {{ return 0; }}\n" + return "" + + +def run_regression( + hotspots: Sequence[Hotspot], + findings: Sequence[StaticFinding], + *, + max_candidates: int, + policy: Mapping[str, object], + rule_cwe: Mapping[str, str], + languages_by_path: Mapping[str, str], + repo_root: Path, + sandbox: SandboxRunner, + sandbox_limits: SandboxConfig, + images: Mapping[str, str], +) -> tuple[CandidateVerdict, ...]: + """Select candidates under the cap, run recipes in the sandbox, and return verdicts.""" + selected = select_candidates( + hotspots, + findings, + max_candidates=max_candidates, + policy_severity_by_id=policy, + rule_cwe=rule_cwe, + ) + runner = RegressionRunner(sandbox) + findings_by_id = {finding.finding_id: finding for finding in findings} + records: list[CandidateVerdict] = [] + for hotspot in selected: + language = _language_for(hotspot.path, languages_by_path) + snippet = snippet_for(language, hotspot.path, hotspot.function_name) + 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) + records.append( + CandidateVerdict( + path=hotspot.path, + function_name=hotspot.function_name, + language=language, + verdict=mapped.verdict, + reason=mapped.reason, + inventory_finding_ids=hotspot.inventory_finding_ids, + worth_fixing=is_worth_fixing(mapped.verdict, reachability), + ) + ) + return tuple(records) + + +def write_verdicts(verdicts: Sequence[CandidateVerdict], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = {"verdicts": [asdict(item) for item in verdicts]} + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + + +def _language_for(path: str, languages_by_path: Mapping[str, str]) -> str: + if path in languages_by_path: + return languages_by_path[path] + return LANGUAGE_BY_EXTENSION.get(Path(path).suffix.lower(), "unknown") + + +def _reachability_for(hotspot: Hotspot, findings_by_id: Mapping[str, StaticFinding]) -> str: + statuses = [ + findings_by_id[finding_id].reachability_status for finding_id in hotspot.inventory_finding_ids if finding_id in findings_by_id + ] + for status in statuses: + if is_worth_fixing(TRIGGERABLE, status): + return status + return statuses[0] if statuses else "unknown" diff --git a/src/openultrasast/sandbox/__init__.py b/src/openultrasast/sandbox/__init__.py index 86d1361..6728be9 100644 --- a/src/openultrasast/sandbox/__init__.py +++ b/src/openultrasast/sandbox/__init__.py @@ -1,6 +1,6 @@ """Isolated Docker CLI sandbox.""" -from .probe import FakeSandboxProbe, SandboxProbe +from .probe import FakeSandboxProbe, SandboxProbe, resolve_sandbox_probe from .runner import ( DockerCliRunner, FakeSandboxRunner, @@ -8,6 +8,7 @@ from .runner import ( SandboxResult, SandboxRunner, build_docker_argv, + resolve_sandbox_runner, ) __all__ = [ @@ -19,4 +20,6 @@ __all__ = [ "SandboxResult", "SandboxRunner", "build_docker_argv", + "resolve_sandbox_probe", + "resolve_sandbox_runner", ] diff --git a/src/openultrasast/sandbox/probe.py b/src/openultrasast/sandbox/probe.py index 681ef85..6cac98c 100644 --- a/src/openultrasast/sandbox/probe.py +++ b/src/openultrasast/sandbox/probe.py @@ -2,13 +2,17 @@ from __future__ import annotations +import os import subprocess from collections.abc import Callable from subprocess import CompletedProcess PROBE_TIMEOUT_SECONDS = 5 +PROBE_ENV = "OPENULTRASAST_SANDBOX_PROBE" _DOCKER_INFO = ("docker", "info") +_PROBE_OFF = frozenset({"0", "off", "false", "unavailable", "no"}) +_PROBE_ON = frozenset({"1", "on", "true", "available", "yes"}) class SandboxProbe: @@ -40,3 +44,13 @@ class FakeSandboxProbe: def available(self) -> bool: return self._available + + +def resolve_sandbox_probe() -> SandboxProbe | FakeSandboxProbe: + """Return a real probe, or a fake when OPENULTRASAST_SANDBOX_PROBE is set.""" + flag = os.environ.get(PROBE_ENV, "").strip().lower() + if flag in _PROBE_OFF: + return FakeSandboxProbe(False) + if flag in _PROBE_ON: + return FakeSandboxProbe(True) + return SandboxProbe() diff --git a/src/openultrasast/sandbox/runner.py b/src/openultrasast/sandbox/runner.py index b000e76..4917cfe 100644 --- a/src/openultrasast/sandbox/runner.py +++ b/src/openultrasast/sandbox/runner.py @@ -15,8 +15,10 @@ NONROOT_USER = "65534:65534" WORKSPACE_MOUNT = "/workspace" SCRATCH_MOUNT = "/scratch" SCRATCH_SEED_MOUNT = "/scratch-in" +RUNNER_ENV = "OPENULTRASAST_SANDBOX_RUNNER" _COPY_SEED_THEN_EXEC = 'cp -r /scratch-in/. /scratch/ && exec "$@"' _SEED_ARGV0 = "ousast-sandbox" +_FAKE_RUNNER = frozenset({"fake"}) class SandboxRunner(Protocol): @@ -54,6 +56,14 @@ class FakeSandboxRunner: return self.result +def resolve_sandbox_runner() -> SandboxRunner: + """Return the Docker CLI runner, or a fake when OPENULTRASAST_SANDBOX_RUNNER=fake.""" + flag = os.environ.get(RUNNER_ENV, "").strip().lower() + if flag in _FAKE_RUNNER: + return FakeSandboxRunner() + return DockerCliRunner() + + class DockerCliRunner: """Run a job with Docker CLI flags that match the threat model.""" diff --git a/tests/test_sandbox_probe.py b/tests/test_sandbox_probe.py index 09ebd4b..ceb7717 100644 --- a/tests/test_sandbox_probe.py +++ b/tests/test_sandbox_probe.py @@ -3,7 +3,7 @@ from typing import Any import pytest -from openultrasast.sandbox import FakeSandboxProbe, SandboxProbe +from openultrasast.sandbox import FakeSandboxProbe, SandboxProbe, resolve_sandbox_probe from openultrasast.sandbox import probe as probe_mod @@ -21,6 +21,25 @@ def test_fake_probe_can_be_forced_on() -> None: assert probe.available() is True +def test_resolve_probe_forced_off_via_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(probe_mod.PROBE_ENV, "0") + probe = resolve_sandbox_probe() + assert isinstance(probe, FakeSandboxProbe) + assert probe.available() is False + + +def test_resolve_probe_forced_on_via_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(probe_mod.PROBE_ENV, "1") + probe = resolve_sandbox_probe() + assert isinstance(probe, FakeSandboxProbe) + assert probe.available() is True + + +def test_resolve_probe_defaults_to_real_probe(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(probe_mod.PROBE_ENV, raising=False) + assert isinstance(resolve_sandbox_probe(), SandboxProbe) + + def test_real_probe_is_available_when_docker_info_succeeds() -> None: def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]: return _completed(0) diff --git a/tests/test_sandbox_runner.py b/tests/test_sandbox_runner.py index 6e0480c..0469847 100644 --- a/tests/test_sandbox_runner.py +++ b/tests/test_sandbox_runner.py @@ -8,7 +8,15 @@ from typing import Any import pytest -from openultrasast.sandbox import DockerCliRunner, FakeSandboxRunner, SandboxJob, SandboxResult, SandboxRunner, build_docker_argv +from openultrasast.sandbox import ( + DockerCliRunner, + FakeSandboxRunner, + SandboxJob, + SandboxResult, + SandboxRunner, + build_docker_argv, + resolve_sandbox_runner, +) from openultrasast.sandbox import runner as runner_mod @@ -187,6 +195,16 @@ def test_fake_sandbox_runner_records_jobs_and_returns_programmed_result(tmp_path assert runner.jobs == [job] +def test_resolve_runner_fake_via_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(runner_mod.RUNNER_ENV, "fake") + assert isinstance(resolve_sandbox_runner(), FakeSandboxRunner) + + +def test_resolve_runner_defaults_to_docker_cli(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(runner_mod.RUNNER_ENV, raising=False) + assert isinstance(resolve_sandbox_runner(), DockerCliRunner) + + def test_fake_sandbox_runner_can_be_reprogrammed(tmp_path: Path) -> None: runner = FakeSandboxRunner() first = _job(tmp_path, command=("echo", "one")) diff --git a/tests/test_scan_deep.py b/tests/test_scan_deep.py new file mode 100644 index 0000000..6125966 --- /dev/null +++ b/tests/test_scan_deep.py @@ -0,0 +1,101 @@ +"""Deep mode: skip regression when the sandbox is missing; record verdicts with a fake runner (task 6.3).""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +from openultrasast.cli import main +from openultrasast.sandbox import runner as sandbox_runner + +_VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n" + + +def _repo(tmp_path: Path) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(_VULN) + return repo + + +def _fail(message: str): # type: ignore[no-untyped-def] + def _inner(*args: object, **kwargs: object) -> object: + raise AssertionError(message) + + return _inner + + +def _guard_docker(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]: + docker_argv: list[list[str]] = [] + original_run = subprocess.run + + def guarded_run(command: object, *args: object, **kwargs: object) -> object: + argv = list(command) if isinstance(command, list | tuple) else [command] + if argv and Path(str(argv[0])).name == "docker": + docker_argv.append([str(item) for item in argv]) + raise AssertionError(f"deep scan test must not start docker: {argv}") + return original_run(command, *args, **kwargs) + + monkeypatch.setattr(subprocess, "run", guarded_run) + monkeypatch.setattr(sandbox_runner.DockerCliRunner, "run", _fail("deep scan test must not use DockerCliRunner")) + return docker_argv + + +def _latest_run(repo: Path) -> Path: + return sorted((repo / ".runs").iterdir())[-1] + + +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") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0") + docker_argv = _guard_docker(monkeypatch) + + assert main(["scan", str(repo), "--mode", "deep"]) == 0 + + run_dir = _latest_run(repo) + manifest = json.loads((run_dir / "manifest.json").read_text()) + stages = manifest["stages"] + degradations = manifest.get("degradations", []) + + assert (run_dir / "complexity_map.json").is_file() + assert not (run_dir / "verdicts.json").exists() + assert docker_argv == [] + assert stages["requested"] == ["static", "map", "regress"] + assert stages["completed"] == ["static", "map"] + assert {"stage": "regress", "reason": "sandbox_unavailable"} in stages["skipped"] + assert any(entry.get("stage") == "regress" and entry.get("reason") == "sandbox_unavailable" for entry in degradations) + + +def test_deep_scan_with_fake_runner_writes_verdicts(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_RUNNER", "fake") + docker_argv = _guard_docker(monkeypatch) + + assert main(["scan", str(repo), "--mode", "deep"]) == 0 + + run_dir = _latest_run(repo) + verdicts_path = run_dir / "verdicts.json" + manifest = json.loads((run_dir / "manifest.json").read_text()) + payload = json.loads(verdicts_path.read_text()) + stages = manifest["stages"] + + assert docker_argv == [] + assert verdicts_path.is_file() + assert (run_dir / "complexity_map.json").is_file() + assert stages["requested"] == ["static", "map", "regress"] + assert stages["completed"] == ["static", "map", "regress"] + assert not any(entry.get("reason") == "sandbox_unavailable" for entry in stages["skipped"]) + assert isinstance(payload["verdicts"], list) + assert payload["verdicts"] + for item in payload["verdicts"]: + assert item["verdict"] in {"triggerable", "not_triggerable", "already_covered", "inconclusive"} + assert "reason" in item + assert "path" in item + assert manifest["artifacts"]["verdicts"] == "verdicts.json" + assert manifest["artifacts"]["complexity_map"] == "complexity_map.json"