diff --git a/tests/test_shell_bwrap.py b/tests/test_shell_bwrap.py new file mode 100644 index 0000000..07747fe --- /dev/null +++ b/tests/test_shell_bwrap.py @@ -0,0 +1,124 @@ +"""Bubblewrap boundary tests: fail-closed launch and host filesystem separation.""" +import os +import shutil +import sys + +import pytest + +from tiny_harness import Agent, Config, Prompt +from tiny_harness._sandbox import SandboxUnavailable, build_bubblewrap_argv + + +@pytest.fixture +def workspace(tmp_path): + d = tmp_path / "workspace" + d.mkdir() + return d + + +def test_sandbox_command_is_structured_and_never_interpolated(workspace, monkeypatch): + monkeypatch.setattr("tiny_harness._sandbox.shutil.which", + lambda name, path=None: "/usr/bin/bwrap") + marker = "echo hello; printf 'unsafe?'; echo --ro-bind" + argv = build_bubblewrap_argv(marker, workspace=str(workspace), cwd=None) + assert argv[0] == "/usr/bin/bwrap" + assert "--unshare-all" in argv + assert "--clearenv" in argv + assert "--die-with-parent" in argv + assert "--new-session" in argv + assert "--bind" in argv and str(workspace) in argv + assert "--ro-bind" in argv + assert "--proc" in argv and "--dev" in argv + assert "--tmpfs" in argv + assert "--share-net" not in argv + assert "--ro-bind-try" not in argv + assert argv[-3:] == ["/bin/sh", "-c", marker] + assert argv[argv.index("--chdir") + 1] == "/workspace" + # The container root is isolated; host / and /etc are not bulk mounted. + assert not any( + argv[i:i+3] in (["--ro-bind", "/", "/"], ["--ro-bind", "/etc", "/etc"]) + for i in range(len(argv) - 2) + ) + + +def test_sandbox_maps_nested_working_directory(workspace, monkeypatch): + (workspace / "nested").mkdir() + monkeypatch.setattr("tiny_harness._sandbox.shutil.which", + lambda name, path=None: "/usr/bin/bwrap") + argv = build_bubblewrap_argv( + "pwd", workspace=str(workspace), cwd=str(workspace / "nested") + ) + assert argv[argv.index("--chdir") + 1] == "/workspace/nested" + + +def test_sandbox_blocks_workspace_escape_and_missing_cwd(workspace, tmp_path, monkeypatch): + monkeypatch.setattr("tiny_harness._sandbox.shutil.which", + lambda name, path=None: "/usr/bin/bwrap") + for cwd in (str(tmp_path), str(workspace / "missing")): + with pytest.raises(SandboxUnavailable): + build_bubblewrap_argv("true", workspace=str(workspace), cwd=cwd) + + +@pytest.mark.asyncio +async def test_missing_bwrap_fails_closed_for_registered_tool(workspace, monkeypatch): + monkeypatch.setattr("tiny_harness._sandbox.shutil.which", + lambda name, path=None: None) + agent = Agent(Prompt("fixture"), Config( + model="fixture", api_key="unused", workspace=str(workspace), + no_hitl=True, allow_unsafe_shell=True, shell_sandbox="bubblewrap", + )) + agent.load_skill("shell") + result = await agent._tool_executor.execute( + "run_command", {"command": "echo expected"}, "c1" + ) + assert not result.success + assert "bubblewrap executable is unavailable" in result.content + + +def test_invalid_sandbox_config_is_rejected(workspace): + with pytest.raises(ValueError, match="shell_sandbox"): + Agent(Prompt("fixture"), Config( + model="fixture", api_key="unused", workspace=str(workspace), + shell_sandbox="unknown" + )) + + +def _bwrap_works(workspace): + if sys.platform != "linux" or shutil.which("bwrap", path="/usr/bin:/bin") is None: + return False + import subprocess + try: + argv = build_bubblewrap_argv("true", workspace=str(workspace), cwd=None) + return subprocess.run( + argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + timeout=5, + ).returncode == 0 + except (OSError, subprocess.TimeoutExpired, SandboxUnavailable): + return False + + +@pytest.mark.asyncio +async def test_real_namespace_hides_host_files_and_writes_only_workspace(workspace, tmp_path): + # CI may disable unprivileged user namespaces. Report a skip instead of + # claiming an isolation smoke test passed when the host cannot run bwrap. + if not _bwrap_works(workspace): + pytest.skip("bubblewrap or unprivileged namespaces unavailable on runner") + + outside = tmp_path / "host-secret-marker.txt" + outside.write_text("not-visible") + agent = Agent(Prompt("fixture"), Config( + model="fixture", api_key="unused", workspace=str(workspace), + no_hitl=True, allow_unsafe_shell=True, shell_sandbox="bubblewrap", + )) + agent.load_skill("shell") + command = ( + 'pwd; printf "sandbox-write" > generated.txt; ' + 'test ! -e /etc/passwd && test ! -e ' + str(outside) + ) + result = await agent._tool_executor.execute( + "run_command", {"command": command}, "c2" + ) + assert result.success, result.content + assert "/workspace" in result.content + assert (workspace / "generated.txt").read_text() == "sandbox-write" + assert outside.read_text() == "not-visible" diff --git a/tiny_harness/_config.py b/tiny_harness/_config.py index 855e9a5..011f890 100644 --- a/tiny_harness/_config.py +++ b/tiny_harness/_config.py @@ -20,6 +20,7 @@ class AgentConfig: approval_timeout_ms: int = 120_000 no_hitl: bool = False allow_unsafe_shell: bool = False + shell_sandbox: str = "off" # 'off' or 'bubblewrap' (Linux only) class Prompt: diff --git a/tiny_harness/_core.py b/tiny_harness/_core.py index ea3b75e..881c21c 100644 --- a/tiny_harness/_core.py +++ b/tiny_harness/_core.py @@ -22,6 +22,8 @@ class Agent: def __init__(self, prompt: Prompt, config: AgentConfig, store: SessionStore | None = None): self._config = config self._prompt = prompt + if config.shell_sandbox not in ("off", "bubblewrap"): + raise ValueError(f"Invalid shell_sandbox: {config.shell_sandbox!r}") if config.context_window <= config.max_tokens: raise ValueError("context_window must exceed max_tokens (reserved output budget)") self._messages = MessageManager( diff --git a/tiny_harness/_sandbox.py b/tiny_harness/_sandbox.py new file mode 100644 index 0000000..22018fb --- /dev/null +++ b/tiny_harness/_sandbox.py @@ -0,0 +1,84 @@ +"""Optional Linux bubblewrap backend for the *registered* async Shell tool. + +This constrains visible mounts and removes the host network namespace. It +does not turn Python path guards into OS security, constrain arbitrary other +plugins, or replace cgroup limits / kernel hardening. +""" +from __future__ import annotations + +import os +import shutil +import sys + + +class SandboxUnavailable(RuntimeError): + """Sandbox requested, but cannot be created safely.""" + + +# Runtime directories containing shared libraries and executables. Never +# recursively bind /, /home, /etc, /root, /run or arbitrary host paths. +_RUNTIME_DIRS = ("/usr", "/bin", "/sbin", "/lib", "/lib64") +_RUNTIME_FILES = ("/etc/ld.so.cache", "/etc/nsswitch.conf") + + +def build_bubblewrap_argv(command: str, *, workspace: str, cwd: str | None) -> list[str]: + """Build an argv array; no untrusted command bytes become bwrap options. + + User commands run through the shell *inside* the namespace. An unavailable + bwrap executable or unsupported host causes an error, never an unsafe + fallback to an ordinary subprocess. + """ + if sys.platform != "linux": + raise SandboxUnavailable("bubblewrap sandbox is available only on Linux") + + # Resolve bwrap from trusted system binary directories, not cwd/PATH. + binary = shutil.which("bwrap", path="/usr/bin:/bin") + if binary is None: + raise SandboxUnavailable( + "bubblewrap executable is unavailable; install bwrap or choose " + "shell_sandbox='off' explicitly" + ) + if not os.path.isdir(workspace): + raise SandboxUnavailable("Sandbox workspace must be an existing directory") + + root = os.path.realpath(workspace) + working_dir = os.path.realpath(cwd or root) + if os.path.commonpath((root, working_dir)) != root: + raise SandboxUnavailable("Sandbox cwd is outside workspace") + if not os.path.isdir(working_dir): + raise SandboxUnavailable("Sandbox cwd must be an existing directory") + + relative = os.path.relpath(working_dir, root) + sandbox_cwd = "/workspace" if relative == "." else "/workspace/" + relative.replace(os.sep, "/") + + argv = [ + binary, + "--unshare-all", # Includes the network and PID namespaces. + "--die-with-parent", + "--new-session", + "--clearenv", + "--setenv", "PATH", "/usr/bin:/bin", + "--setenv", "HOME", "/tmp", + "--setenv", "TMPDIR", "/tmp", + "--setenv", "LANG", "C.UTF-8", + "--tmpfs", "/", # An empty filesystem root. + ] + + for path in _RUNTIME_DIRS: + if os.path.isdir(path): + argv.extend(("--ro-bind", path, path)) + for path in _RUNTIME_FILES: + if os.path.isfile(path): + argv.extend(("--ro-bind", path, path)) + + argv.extend(( + "--proc", "/proc", + "--dev", "/dev", + "--tmpfs", "/tmp", + "--dir", "/workspace", + "--bind", root, "/workspace", # Only writable host mount. + "--chdir", sandbox_cwd, + "--", + "/bin/sh", "-c", command, + )) + return argv diff --git a/tiny_harness/cli.py b/tiny_harness/cli.py index aeadfba..aa16598 100644 --- a/tiny_harness/cli.py +++ b/tiny_harness/cli.py @@ -17,7 +17,8 @@ def parse_args() -> Namespace: parser.add_argument("--max-iterations", type=int, default=25, help="Max loop iterations") parser.add_argument("--skills", default="", help="Comma-separated skill names (default: files)") parser.add_argument("--no-hitl", action="store_true", help="Disable human-in-the-loop approval prompts") - parser.add_argument("--allow-unsafe-shell", action="store_true", help="Explicitly permit unsandboxed shell with --no-hitl (dangerous)") + parser.add_argument("--allow-unsafe-shell", action="store_true", help="Explicitly permit Shell without human approval (dangerous)") + parser.add_argument("--shell-sandbox", choices=("off", "bubblewrap"), default="off", help="Optional Linux bwrap Shell sandbox; fails closed if unavailable") parser.add_argument("--tui", action="store_true", help="Use rich TUI mode (requires 'rich' package)") parser.add_argument("--api-key-env", default=None, help="Env var for API key (defaults to provider-specific variable)") return parser.parse_args() @@ -162,7 +163,7 @@ def main(): from tiny_harness import Agent, Prompt, Config prompt = Prompt("You are a helpful AI assistant.") - config = Config(model=args.model, api_key=api_key, workspace=args.workspace, provider=args.provider, api_base_url=args.api_base_url, max_iterations=args.max_iterations, no_hitl=args.no_hitl, allow_unsafe_shell=args.allow_unsafe_shell) + config = Config(model=args.model, api_key=api_key, workspace=args.workspace, provider=args.provider, api_base_url=args.api_base_url, max_iterations=args.max_iterations, no_hitl=args.no_hitl, allow_unsafe_shell=args.allow_unsafe_shell, shell_sandbox=args.shell_sandbox) agent = Agent(prompt=prompt, config=config) if not args.no_hitl: diff --git a/tiny_harness/skills/shell.py b/tiny_harness/skills/shell.py index a562a73..8463161 100644 --- a/tiny_harness/skills/shell.py +++ b/tiny_harness/skills/shell.py @@ -92,6 +92,6 @@ def register(agent) -> None: risk_level="dangerous", path_args=("cwd",), ), - partial(run_command_async, structured=True), + partial(run_command_async, structured=True, sandbox_mode=agent._config.shell_sandbox, workspace=agent._guard.guard(".")), ) agent._prompt.append(SHELL_PROMPT) diff --git a/tiny_harness/tools/shell.py b/tiny_harness/tools/shell.py index 0387b63..3c2cb4b 100644 --- a/tiny_harness/tools/shell.py +++ b/tiny_harness/tools/shell.py @@ -76,7 +76,7 @@ def run_command(args: dict) -> str: # The agent registers the async entrypoint below. Unlike the legacy sync helper # it can stop its whole process group when the executor cancels the coroutine. -async def run_command_async(args: dict, *, structured: bool = False) -> str | ToolOutcome: +async def run_command_async(args: dict, *, structured: bool = False, sandbox_mode: str = "off", workspace: str | None = None) -> str | ToolOutcome: import asyncio import os import signal @@ -94,7 +94,20 @@ async def run_command_async(args: dict, *, structured: bool = False) -> str | To if key in safe_keys or key.startswith("LC_") } - cmd, use_shell = _build_args(command) + if sandbox_mode not in ("off", "bubblewrap"): + raise ValueError(f"Unknown shell sandbox mode: {sandbox_mode!r}") + if sandbox_mode == "bubblewrap": + from tiny_harness._sandbox import build_bubblewrap_argv + if workspace is None: + raise ValueError("A trusted workspace is required for bubblewrap") + # Fail closed: do not use an ordinary shell if bwrap is absent. + cmd, use_shell = build_bubblewrap_argv( + command, workspace=workspace, cwd=cwd, + ), False + cwd = workspace + else: + cmd, use_shell = _build_args(command) + options = { "cwd": cwd, "stdin": asyncio.subprocess.DEVNULL,