feat(three-stage-scan): add docker sandbox availability probe

This commit is contained in:
Marius Ciepluch
2026-09-03 21:29:38 +02:00
parent b34a75a821
commit 5b7e58ea3a
4 changed files with 148 additions and 2 deletions
+3 -2
View File
@@ -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:`.
+5
View File
@@ -0,0 +1,5 @@
"""Isolated Docker CLI sandbox."""
from .probe import FakeSandboxProbe, SandboxProbe
__all__ = ["FakeSandboxProbe", "SandboxProbe"]
+42
View File
@@ -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
+98
View File
@@ -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