feat(shell): optional fail-closed Linux bubblewrap filesystem/network sandbox

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"

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:

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(

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

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:

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)

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,