From eb5f2462ddeb2e28a8e222796956cd0e1ba34a34 Mon Sep 17 00:00:00 2001 From: David Fitzsimmons Date: Sat, 5 Sep 2026 09:51:22 -0400 Subject: [PATCH] security: isolate generated Python example in mandatory pytest container --- examples/01_code_pytest.py | 39 ++--- examples/README.md | 29 ++++ examples/_sandbox.py | 261 ++++++++++++++++++++++++++++++++++ tests/test_example_sandbox.py | 253 ++++++++++++++++++++++++++++++++ 4 files changed, 565 insertions(+), 17 deletions(-) create mode 100644 examples/_sandbox.py create mode 100644 tests/test_example_sandbox.py diff --git a/examples/01_code_pytest.py b/examples/01_code_pytest.py index 8f24f67..836d63c 100644 --- a/examples/01_code_pytest.py +++ b/examples/01_code_pytest.py @@ -17,6 +17,7 @@ Loop type: verify_revise. Run: + # First prepare the trusted pytest image described in examples/README.md. pip install 'loopgain[examples]' python examples/01_code_pytest.py """ @@ -24,12 +25,11 @@ from __future__ import annotations import re -import subprocess -import sys import tempfile from pathlib import Path from loopgain import LoopGain +from _sandbox import SandboxExecutionError, ensure_available, execute_pytest from _common import ( call_claude, @@ -99,14 +99,16 @@ def strip_code_fences(text: str) -> str: def run_pytest(workdir: Path) -> tuple[int, str]: - proc = subprocess.run( - [sys.executable, "-m", "pytest", "-q", "--tb=short", "--no-header"], - cwd=workdir, capture_output=True, text=True, timeout=30, - ) - out = proc.stdout + proc.stderr + try: + returncode, out = execute_pytest( + (workdir / "solution.py").read_text(), + (workdir / "test_solution.py").read_text(), + ) + except SandboxExecutionError as exc: + return 15, f"Sandbox candidate failure: {exc}" m = re.search(r"(\d+)\s+failed", out) failures = int(m.group(1)) if m else 0 - if proc.returncode != 0 and failures == 0: + if returncode != 0 and failures == 0: failures = 15 # collection/import error → worst-case signal return failures, out @@ -121,6 +123,7 @@ def one_iteration(client, workdir: Path, prev_code: str, prev_failures: str): f"\n\npytest reported these failures:\n```\n{prev_failures[-1500:]}\n```" f"\n\nReturn a fully corrected `solution.py`. Code only, no prose." ) + ensure_available() # Fail before spending on a candidate that cannot be isolated. code = strip_code_fences(call_claude(client, prompt)) if not code: return 15, "", "" @@ -154,17 +157,19 @@ def loopgain_run(client, workdir: Path): def main() -> None: + ensure_available() client = get_client() - workdir = Path(tempfile.mkdtemp(prefix="loopgain-ex01-")) - (workdir / "test_solution.py").write_text(TESTS) - print(f"Workdir: {workdir}") - print(f"Spec: format_duration — 15 parametrized test cases.\n") - - baseline_err, baseline_iters = baseline_run(client, workdir) + with tempfile.TemporaryDirectory(prefix="loopgain-ex01-") as temporary: + workdir = Path(temporary) + (workdir / "test_solution.py").write_text(TESTS) + print(f"Workdir: {workdir}") + print("Spec: format_duration — 15 parametrized test cases.\n") + baseline_err, baseline_iters = baseline_run(client, workdir) # Fresh workdir for the LoopGain run so the comparison is apples-to-apples. - workdir = Path(tempfile.mkdtemp(prefix="loopgain-ex01-lg-")) - (workdir / "test_solution.py").write_text(TESTS) - lg = loopgain_run(client, workdir) + with tempfile.TemporaryDirectory(prefix="loopgain-ex01-lg-") as temporary: + workdir = Path(temporary) + (workdir / "test_solution.py").write_text(TESTS) + lg = loopgain_run(client, workdir) print_comparison(baseline_iters, baseline_err, lg) send_telemetry(lg, workload_id=WORKLOAD_ID, loop_type="verify_revise") diff --git a/examples/README.md b/examples/README.md index 66caeac..ce47aba 100644 --- a/examples/README.md +++ b/examples/README.md @@ -42,6 +42,35 @@ Override the model via `LOOPGAIN_EXAMPLE_MODEL` (default `claude-haiku-4-5`). --- +## Example 01 sandbox prerequisite + +Example 01 requires Docker on a POSIX host and a trusted local Python 3.12 image +containing pytest. The default image is `loopgain-python-pytest:local`; select a +different trusted image with `LOOPGAIN_SANDBOX_IMAGE`. Image preparation is a +separate operator action. For example, use a Dockerfile with a trusted Python +base and installed pytest, then tag it with the default name. Prepare or update +that image before starting the example; the runtime never builds or pulls one. + +The example resolves the image to an immutable local ID and probes pytest with +the actual isolation settings before creating a model client and before every +model request. Missing Docker, image, pytest or failed cleanup stops the loop; +there is no host execution fallback. Candidate/test files are staged read-only +in a disposable non-root container with networking disabled, a read-only root, +no host credentials or Docker socket, dropped capabilities, no-new-privileges, +128 MiB memory, one CPU, 32 PIDs, 16 MiB `/tmp`, and bounded input/output and +wall time. Containers and descendants are forcibly removed on every exit. +External pytest plugins are disabled. Keep the image and Docker/kernel patched; +containers are not a separate VM boundary. Python/runtime limits differ from +older host execution, so do not compare historical measurements unchanged. + +Verify only the synthetic sandbox tests without provider/model calls: + +```bash +LOOPGAIN_SANDBOX_INTEGRATION=1 python -m pytest tests/test_example_sandbox.py +``` + +The core `loopgain` package gains no runtime dependencies from this helper. + ## Run ```bash diff --git a/examples/_sandbox.py b/examples/_sandbox.py new file mode 100644 index 0000000..7e42074 --- /dev/null +++ b/examples/_sandbox.py @@ -0,0 +1,261 @@ +"""Fail-closed Docker execution for untrusted candidates (stdlib only). + +Images must already exist locally; this module never pulls or builds one. +The image ID is resolved before execution so a mutable tag cannot change mid-run. +""" + +from __future__ import annotations + +import json +import os +import re +import selectors +import shutil +import signal +import subprocess +import tempfile +import time +import uuid +from pathlib import Path + +MAX_INPUT_BYTES = 2 * 1024 * 1024 +MAX_OUTPUT_BYTES = 64 * 1024 + + +class SandboxUnavailable(RuntimeError): + """Infrastructure failure: stop rather than spend on an unexecutable trial.""" + + +class SandboxExecutionError(RuntimeError): + """Candidate exceeded a limit, crashed, or returned invalid output.""" + + +def _docker() -> str: + executable = shutil.which("docker") + if os.name != "posix" or not executable: + raise SandboxUnavailable( + "Candidate execution requires Docker on a POSIX host; no host fallback." + ) + return executable + + +def _docker_env() -> dict[str, str]: + # These configure the CLI only; the candidate environment is cleared by env -i. + return { + key: os.environ[key] + for key in ("PATH", "HOME", "DOCKER_CONFIG") + if key in os.environ + } + + +def _local_image(docker: str) -> str: + image = os.environ.get("LOOPGAIN_SANDBOX_IMAGE", "loopgain-python-pytest:local") + try: + result = subprocess.run( + [docker, "image", "inspect", "--format", "{{.Id}}", image], + capture_output=True, + text=True, + timeout=10, + env=_docker_env(), + check=True, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise SandboxUnavailable( + "Docker or local Python sandbox image unavailable; no automatic pull." + ) from exc + image_id = result.stdout.strip() + if not re.fullmatch(r"sha256:[0-9a-f]{64}", image_id): + raise SandboxUnavailable("Docker did not return an immutable local image ID.") + return image_id + + +def _command(docker: str, image: str, name: str, directory: Path) -> list[str]: + return [ + docker, + "create", + "--pull=never", + "--name", + name, + "--network=none", + "--read-only", + "--user=65534:65534", + "--cap-drop=ALL", + "--security-opt=no-new-privileges", + "--pids-limit=32", + "--memory=128m", + "--memory-swap=128m", + "--cpus=1", + "--ulimit=cpu=20:20", + "--ulimit=nofile=64:64", + "--ulimit=core=0:0", + "--tmpfs=/tmp:rw,noexec,nosuid,nodev,size=16m,mode=1777", + "--mount", + f"type=bind,src={directory},dst=/input,readonly", + "--workdir=/tmp", + "--log-driver=none", + "--entrypoint=/usr/bin/env", + image, + "-i", + "PATH=/usr/local/bin:/usr/bin:/bin", + "HOME=/tmp", + "PYTEST_DISABLE_PLUGIN_AUTOLOAD=1", + "python3", + "-I", + "-B", + "-c", + ( + "import runpy,sys; sys.stdin=open('/input/request.json'); " + "runpy.run_path('/input/worker.py',run_name='__main__')" + ), + ] + + +def _run(image: str, worker: str, payload: dict, timeout: float, files=None) -> str: + encoded = json.dumps(payload).encode() + files = files or {} + if any(name not in {"solution.py", "test_solution.py"} for name in files): + raise ValueError("Unexpected sandbox fixture name") + if ( + len(encoded) + sum(len(text.encode()) for text in files.values()) + > MAX_INPUT_BYTES + ): + raise SandboxExecutionError("Sandbox input exceeds 2 MiB") + docker = _docker() + name = f"loopgain-example-candidate-{uuid.uuid4().hex}" + with tempfile.TemporaryDirectory(prefix="loopgain-example-candidate-") as temporary: + directory = Path(temporary) + # Only staged worker/request/fixtures are mounted; never repo, home or Docker socket. + directory.chmod(0o755) + for filename, contents in ( + ("worker.py", worker.encode()), + ("request.json", encoded), + ): + path = directory / filename + path.write_bytes(contents) + path.chmod(0o444) + for filename, contents in files.items(): + path = directory / filename + path.write_text(contents) + path.chmod(0o444) + proc = None + try: + subprocess.run( + _command(docker, image, name, directory), + capture_output=True, + timeout=15, + env=_docker_env(), + check=True, + ) + proc = subprocess.Popen( + [docker, "start", "--attach", name], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=_docker_env(), + start_new_session=True, + ) + output = bytearray() + error = bytearray() + deadline = time.monotonic() + timeout + with selectors.DefaultSelector() as selector: + selector.register(proc.stdout, selectors.EVENT_READ, output) + selector.register(proc.stderr, selectors.EVENT_READ, error) + while selector.get_map(): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise SandboxExecutionError("Sandbox wall-clock timeout") + for key, _ in selector.select(min(remaining, 0.1)): + chunk = os.read(key.fileobj.fileno(), 4096) + if not chunk: + selector.unregister(key.fileobj) + continue + key.data.extend(chunk) + if len(output) + len(error) > MAX_OUTPUT_BYTES: + raise SandboxExecutionError("Sandbox output limit exceeded") + proc.wait(timeout=max(0.01, deadline - time.monotonic())) + if proc.returncode: + raise SandboxExecutionError( + f"Sandbox worker exited {proc.returncode}: {error[:200]!r}" + ) + try: + return output.decode("utf8") + except UnicodeError as exc: + raise SandboxExecutionError("Invalid sandbox output encoding") from exc + except (OSError, subprocess.SubprocessError) as exc: + raise SandboxUnavailable("Docker sandbox process failed") from exc + finally: + # Remove the container explicitly, including every descendant process. + # Do not rely on terminating the attached docker CLI to stop a container. + if proc is not None and proc.poll() is None: + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + try: + cleanup = subprocess.run( + [docker, "rm", "--force", name], + capture_output=True, + timeout=10, + env=_docker_env(), + check=False, + ) + if cleanup.returncode and b"No such container" not in cleanup.stderr: + raise SandboxUnavailable( + "Could not confirm sandbox container cleanup; stop the run." + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise SandboxUnavailable( + "Sandbox cleanup unavailable; stop the run." + ) from exc + finally: + if proc is not None: + if proc.poll() is None: + os.killpg(proc.pid, signal.SIGKILL) + proc.wait(timeout=5) + proc.stdout.close() + proc.stderr.close() + + +def ensure_available() -> str: + """Exercise the actual isolation flags before any model request is made.""" + image = _local_image(_docker()) + try: + result = _run(image, 'import pytest; print("sandbox-ready")', {}, 15) + except SandboxExecutionError as exc: + raise SandboxUnavailable( + "Sandbox preflight failed; do not start model requests." + ) from exc + if result.strip() != "sandbox-ready": + raise SandboxUnavailable("Unexpected sandbox preflight result") + return image + + +_PYTEST_WORKER = """import contextlib,io,json,pytest +output=io.StringIO() +with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output): + result=pytest.main(['/input/test_solution.py','-q','--tb=short','--no-header','-p','no:cacheprovider']) +print(json.dumps({'returncode':int(result),'output':output.getvalue()})) +""" + + +def execute_pytest(solution: str, tests: str) -> tuple[int, str]: + """Run the fixed pytest fixture and candidate exclusively inside Docker.""" + image = _local_image(_docker()) + raw = _run( + image, + _PYTEST_WORKER, + {}, + 30, + {"solution.py": solution, "test_solution.py": tests}, + ) + try: + result = json.loads(raw) + if ( + not isinstance(result, dict) + or not isinstance(result.get("returncode"), int) + or not isinstance(result.get("output"), str) + ): + raise ValueError("Invalid pytest result") + return result["returncode"], result["output"] + except (TypeError, ValueError) as exc: + raise SandboxExecutionError("Invalid sandbox result") from exc diff --git a/tests/test_example_sandbox.py b/tests/test_example_sandbox.py new file mode 100644 index 0000000..23908b6 --- /dev/null +++ b/tests/test_example_sandbox.py @@ -0,0 +1,253 @@ +"""Synthetic sandbox checks; integration requires an existing local image.""" + +import json +import os +from pathlib import Path +from unittest.mock import Mock + +import pytest + +import importlib.util + +EXAMPLES = Path(__file__).resolve().parents[1] / "examples" +SPEC = importlib.util.spec_from_file_location( + "example_sandbox", EXAMPLES / "_sandbox.py" +) +sandbox = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(sandbox) + +ROOT = Path(__file__).resolve().parents[1] +INTEGRATION = pytest.mark.skipif( + os.environ.get("LOOPGAIN_SANDBOX_INTEGRATION") != "1", + reason="opt in with LOOPGAIN_SANDBOX_INTEGRATION=1; never pulls images", +) + + +def test_container_has_mandatory_boundaries(tmp_path): + cmd = sandbox._command("docker", "sha256:" + "a" * 64, "test", tmp_path) + for flag in ( + "--pull=never", + "--network=none", + "--read-only", + "--user=65534:65534", + "--cap-drop=ALL", + "--security-opt=no-new-privileges", + "--pids-limit=32", + "--memory=128m", + "--memory-swap=128m", + "--cpus=1", + "--log-driver=none", + ): + assert flag in cmd + assert cmd.count("--mount") == 1 + assert f"type=bind,src={tmp_path},dst=/input,readonly" in cmd + assert "-i" in cmd + assert not any("docker.sock" in x for x in cmd) + + +def test_missing_docker_fails_closed(monkeypatch): + monkeypatch.setattr(sandbox.shutil, "which", lambda _: None) + with pytest.raises(sandbox.SandboxUnavailable, match="no host fallback"): + sandbox.ensure_available() + + +def test_missing_image_does_not_pull(monkeypatch): + calls = [] + + def unavailable(cmd, **kwargs): + calls.append(cmd) + raise sandbox.subprocess.CalledProcessError(1, cmd) + + monkeypatch.setattr(sandbox.subprocess, "run", unavailable) + with pytest.raises(sandbox.SandboxUnavailable, match="no automatic pull"): + sandbox._local_image("docker") + assert len(calls) == 1 + assert calls[0][1:3] == ["image", "inspect"] + + +@pytest.fixture +def example(monkeypatch): + monkeypatch.syspath_prepend(str(EXAMPLES)) + monkeypatch.setitem(__import__("sys").modules, "_sandbox", sandbox) + spec = importlib.util.spec_from_file_location( + "code_example", EXAMPLES / "01_code_pytest.py" + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_preflight_prevents_model_request(example, monkeypatch, tmp_path): + call = Mock(side_effect=AssertionError("model must not be called")) + monkeypatch.setattr(example, "call_claude", call) + monkeypatch.setattr( + example, + "ensure_available", + Mock(side_effect=sandbox.SandboxUnavailable("offline")), + ) + with pytest.raises(sandbox.SandboxUnavailable): + example.one_iteration(None, tmp_path, "", "") + call.assert_not_called() + + +def test_cleanup_failure_stops_baseline(example, monkeypatch, tmp_path): + iteration = Mock(side_effect=sandbox.SandboxUnavailable("cleanup unconfirmed")) + monkeypatch.setattr(example, "one_iteration", iteration) + with pytest.raises(sandbox.SandboxUnavailable): + example.baseline_run(None, tmp_path) + iteration.assert_called_once() + + +@INTEGRATION +def test_benign_candidate_pass_fail(): + status, output = sandbox.execute_pytest( + "def add(a,b): return a+b", + "from solution import add\ndef test_pass(): assert add(1,2)==3\ndef test_fail(): assert add(1,2)==4", + ) + assert status == 1 + assert "1 failed, 1 passed" in output + + +@INTEGRATION +def test_effective_container_boundaries(monkeypatch): + monkeypatch.setenv("SYNTHETIC_HOST_SECRET", "must-not-reach-candidate") + worker = """import os,json,pathlib +try: + pathlib.Path('/input/request.json').write_text('changed') + writable=True +except OSError: + writable=False +pathlib.Path('/tmp/scratch').write_text('ok') +print(json.dumps(dict(uid=os.getuid(), env=dict(os.environ), writable=writable, + memory=pathlib.Path('/sys/fs/cgroup/memory.max').read_text().strip(), + pids=pathlib.Path('/sys/fs/cgroup/pids.max').read_text().strip(), + routes=pathlib.Path('/proc/net/route').read_text().splitlines(), + status=pathlib.Path('/proc/self/status').read_text())))""" + data = json.loads(sandbox._run(sandbox.ensure_available(), worker, {}, 10)) + assert data["uid"] == 65534 + assert "SYNTHETIC_HOST_SECRET" not in data["env"] + assert set(data["env"]) <= { + "PATH", + "HOME", + "LC_CTYPE", + "PYTEST_DISABLE_PLUGIN_AUTOLOAD", + } + assert not data["writable"] + assert data["memory"] == "134217728" + assert data["pids"] == "32" + assert len(data["routes"]) == 1 # no routes beyond the table header + assert "CapEff:\t0000000000000000" in data["status"] + assert "NoNewPrivs:\t1" in data["status"] + + +@INTEGRATION +@pytest.mark.parametrize( + "worker,timeout,match", + [ + ("import time; time.sleep(30)", 0.5, "timeout"), + ("print('x'*70000)", 10, "output limit"), + ], +) +def test_limits_remove_container(monkeypatch, worker, timeout, match): + from types import SimpleNamespace + + name = "loopgain-example-candidate-synthetic-limit-test" + monkeypatch.setattr( + sandbox.uuid, "uuid4", lambda: SimpleNamespace(hex="synthetic-limit-test") + ) + image = sandbox._local_image(sandbox._docker()) + with pytest.raises(sandbox.SandboxExecutionError, match=match): + sandbox._run(image, worker, {}, timeout) + result = sandbox.subprocess.run( + [sandbox._docker(), "container", "inspect", name], + capture_output=True, + timeout=10, + ) + assert result.returncode != 0 + assert b"No such" in result.stderr + + +def test_cleanup_failure_stops_run(monkeypatch): + import subprocess + + monkeypatch.setattr(sandbox, "_docker", lambda: "docker") + + def failed_create_and_cleanup(cmd, **kwargs): + if cmd[1] == "create": + raise subprocess.CalledProcessError(1, cmd) + return subprocess.CompletedProcess(cmd, 1, b"", b"daemon unavailable") + + monkeypatch.setattr(sandbox.subprocess, "run", failed_create_and_cleanup) + with pytest.raises(sandbox.SandboxUnavailable, match="cleanup"): + sandbox._run("sha256:" + "a" * 64, "print('unused')", {}, 1) + + +@INTEGRATION +def test_worker_exit_removes_background_child(monkeypatch): + from types import SimpleNamespace + + monkeypatch.setattr( + sandbox.uuid, "uuid4", lambda: SimpleNamespace(hex="synthetic-child-test") + ) + image = sandbox._local_image(sandbox._docker()) + result = sandbox._run( + image, + "import subprocess; subprocess.Popen(['sleep','30']); print('done')", + {}, + 10, + ) + assert result.strip() == "done" + check = sandbox.subprocess.run( + [ + sandbox._docker(), + "container", + "inspect", + "loopgain-example-candidate-synthetic-child-test", + ], + capture_output=True, + timeout=10, + check=False, + ) + assert check.returncode != 0 + assert b"No such" in check.stderr + + +@INTEGRATION +def test_actual_example_fixed_fixture_runs_in_container(example, tmp_path): + (tmp_path / "solution.py").write_text( + 'def format_duration(seconds):\n if seconds < 0: raise ValueError()\n return "now"' + ) + (tmp_path / "test_solution.py").write_text(example.TESTS) + failures, output = example.run_pytest(tmp_path) + assert failures == 13 + assert "13 failed, 2 passed" in output + + +def test_main_preflights_before_client_creation(example, monkeypatch): + client = Mock(side_effect=AssertionError("client must not be created")) + monkeypatch.setattr(example, "get_client", client) + monkeypatch.setattr( + example, + "ensure_available", + Mock(side_effect=sandbox.SandboxUnavailable("offline")), + ) + with pytest.raises(sandbox.SandboxUnavailable): + example.main() + client.assert_not_called() + + +def test_cleanup_failure_stops_loopgain(example, monkeypatch, tmp_path): + from types import SimpleNamespace + + iteration = Mock(side_effect=sandbox.SandboxUnavailable("cleanup unconfirmed")) + observe = Mock(side_effect=AssertionError("fatal failure must not be scored")) + monkeypatch.setattr(example, "one_iteration", iteration) + monkeypatch.setattr( + example, + "LoopGain", + lambda **kwargs: SimpleNamespace(should_continue=lambda: True, observe=observe), + ) + with pytest.raises(sandbox.SandboxUnavailable): + example.loopgain_run(None, tmp_path) + iteration.assert_called_once() + observe.assert_not_called()