From 5b7e58ea3a332ce655733827a7a027b78a9b698d Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 3 Sep 2026 21:29:38 +0200 Subject: [PATCH] feat(three-stage-scan): add docker sandbox availability probe --- .kiro/specs/three-stage-scan/tasks.md | 5 +- src/openultrasast/sandbox/__init__.py | 5 ++ src/openultrasast/sandbox/probe.py | 42 ++++++++++++ tests/test_sandbox_probe.py | 98 +++++++++++++++++++++++++++ 4 files changed, 148 insertions(+), 2 deletions(-) create mode 100644 src/openultrasast/sandbox/__init__.py create mode 100644 src/openultrasast/sandbox/probe.py create mode 100644 tests/test_sandbox_probe.py diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index e8559e7..a175299 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -70,7 +70,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Requirements: 6.1, 6.2_ - _Boundary: HunterTools_ -- [ ] 3.2 Run a bounded tool-using hunter that records suspicion only +- [x] 3.2 Run a bounded tool-using hunter that records suspicion only - When a hunter model is configured, the loop may call the three tools against map hotspots and must not stamp results as static corroboration. - When no model is configured, the loop does not run and a degradation is produced by the orchestrator later. - Observable completion: a scripted client that never calls a tool fails the test; a client that greps then returns a finding yields evidence level `suspicion` and id prefix `tool-hunter:`. @@ -79,7 +79,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Depends: 3.1, 2.2_ - [ ] 4. Stand up the isolated sandbox runner -- [ ] 4.1 (P) Probe whether the sandbox runtime is usable +- [x] 4.1 (P) Probe whether the sandbox runtime is usable - A short `docker info` style probe returns available or unavailable without throwing into the scan. - Observable completion: the fake probe can be forced off; the real probe is skippable in unit tests. - _Requirements: 9.3_ @@ -198,3 +198,4 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - `scan_exit_code` duck-types `worth_fixing` on verdict objects; `_run_scan` still passes no verdicts until task 6.5. Dict-shaped verdicts would not fail the gate. - Map `top_k` default is 20 (not named in design.md); `max_hunter_hotspots=8` and `max_candidates=5` follow design. - `collect_signals` adjacent-test detection is false unless the caller also passes `repo_files`; task 2.2 must pass enumerated repo paths. +- ToolHunter rejects JSON dumps unless a repo tool was invoked; findings are always `suspicion` with prefix `tool-hunter:`. diff --git a/src/openultrasast/sandbox/__init__.py b/src/openultrasast/sandbox/__init__.py new file mode 100644 index 0000000..a855dfd --- /dev/null +++ b/src/openultrasast/sandbox/__init__.py @@ -0,0 +1,5 @@ +"""Isolated Docker CLI sandbox.""" + +from .probe import FakeSandboxProbe, SandboxProbe + +__all__ = ["FakeSandboxProbe", "SandboxProbe"] diff --git a/src/openultrasast/sandbox/probe.py b/src/openultrasast/sandbox/probe.py new file mode 100644 index 0000000..681ef85 --- /dev/null +++ b/src/openultrasast/sandbox/probe.py @@ -0,0 +1,42 @@ +"""Docker CLI presence and usability probe.""" + +from __future__ import annotations + +import subprocess +from collections.abc import Callable +from subprocess import CompletedProcess + +PROBE_TIMEOUT_SECONDS = 5 + +_DOCKER_INFO = ("docker", "info") + + +class SandboxProbe: + """Report whether `docker info` succeeds without raising into the scan.""" + + def __init__(self, *, runner: Callable[..., CompletedProcess[str]] | None = None) -> None: + self._runner = runner + + def available(self) -> bool: + run = self._runner if self._runner is not None else subprocess.run + try: + result = run( + list(_DOCKER_INFO), + capture_output=True, + text=True, + timeout=PROBE_TIMEOUT_SECONDS, + check=False, + ) + return result.returncode == 0 + except Exception: + return False + + +class FakeSandboxProbe: + """Test double that reports a programmed availability without touching Docker.""" + + def __init__(self, available: bool) -> None: + self._available = available + + def available(self) -> bool: + return self._available diff --git a/tests/test_sandbox_probe.py b/tests/test_sandbox_probe.py new file mode 100644 index 0000000..09ebd4b --- /dev/null +++ b/tests/test_sandbox_probe.py @@ -0,0 +1,98 @@ +import subprocess +from typing import Any + +import pytest + +from openultrasast.sandbox import FakeSandboxProbe, SandboxProbe +from openultrasast.sandbox import probe as probe_mod + + +def _completed(returncode: int) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(["docker", "info"], returncode, stdout="", stderr="") + + +def test_fake_probe_can_be_forced_off() -> None: + probe = FakeSandboxProbe(available=False) + assert probe.available() is False + + +def test_fake_probe_can_be_forced_on() -> None: + probe = FakeSandboxProbe(available=True) + assert probe.available() is True + + +def test_real_probe_is_available_when_docker_info_succeeds() -> None: + def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]: + return _completed(0) + + assert SandboxProbe(runner=runner).available() is True + + +def test_real_probe_is_unavailable_when_docker_info_fails() -> None: + def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]: + return _completed(1) + + assert SandboxProbe(runner=runner).available() is False + + +@pytest.mark.parametrize( + "error", + [ + FileNotFoundError("docker"), + PermissionError("denied"), + OSError("cannot talk to daemon"), + subprocess.TimeoutExpired(cmd=["docker", "info"], timeout=5), + RuntimeError("unexpected probe failure"), + ], +) +def test_real_probe_returns_unavailable_without_raising(error: Exception) -> None: + def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]: + raise error + + assert SandboxProbe(runner=runner).available() is False + + +def test_real_probe_runs_docker_info_with_short_timeout() -> None: + seen: dict[str, Any] = {} + + def runner(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + seen["command"] = list(command) + seen["timeout"] = kwargs.get("timeout") + seen["check"] = kwargs.get("check") + return _completed(0) + + assert SandboxProbe(runner=runner).available() is True + assert seen["command"] == ["docker", "info"] + assert seen["timeout"] == probe_mod.PROBE_TIMEOUT_SECONDS + assert probe_mod.PROBE_TIMEOUT_SECONDS <= 10 + assert seen["check"] is False + + +def test_real_probe_is_skippable_in_unit_tests(monkeypatch: pytest.MonkeyPatch) -> None: + def boom(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]: + raise AssertionError("unit tests must not invoke the docker daemon") + + monkeypatch.setattr(probe_mod.subprocess, "run", boom) + + def runner(*_args: object, **_kwargs: object) -> subprocess.CompletedProcess[str]: + return _completed(0) + + assert SandboxProbe(runner=runner).available() is True + + +def test_real_probe_default_path_uses_subprocess_run(monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, Any] = {} + + def fake_run(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + seen["command"] = list(command) + seen["timeout"] = kwargs.get("timeout") + seen["capture_output"] = kwargs.get("capture_output") + seen["text"] = kwargs.get("text") + return _completed(0) + + monkeypatch.setattr(probe_mod.subprocess, "run", fake_run) + assert SandboxProbe().available() is True + assert seen["command"] == ["docker", "info"] + assert seen["timeout"] == probe_mod.PROBE_TIMEOUT_SECONDS + assert seen["capture_output"] is True + assert seen["text"] is True