test_mcp_sandbox_backends.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_mcp/test_mcp_sandbox_backends.py
 
"""
Purpose
-------
Phase 5C tests for pluggable sandbox backends (``NativeBackend``,
``BubblewrapBackend``, ``SandboxExecBackend``) and the
``make_sandbox_backend`` factory.
 
Responsibilities
----------------
- Verify ``NativeBackend.run()`` returns correctly and passes sanitised
  env.
- Verify ``BubblewrapBackend`` and ``SandboxExecBackend`` command
  structure.
- Verify ``make_sandbox_backend`` factory for all backend names and
  auto-detection per platform.
- Verify ``SandboxExecutor`` delegation, memory-limit conversion, output
  capping, and error handling (timeout, OSError, FileNotFoundError).
- Verify ``_sanitise_env`` strips sensitive variables and adds
  ``OCT_MCP_CWD``.
 
Diagnostics
-----------
Domain: MCP-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- Tests use mock subprocess and platform stubs; no real sandbox commands
  are executed.
"""
 
from __future__ import annotations
 
import os
import subprocess
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
 
import pytest
 
from oct.mcp.sandbox import (
    BubblewrapBackend,
    NativeBackend,
    SandboxBackend,
    SandboxExecBackend,
    SandboxExecutor,
    _sanitise_env,
    make_sandbox_backend,
)
 
 
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
 
def _fake_config(
    sandbox_backend: str = "auto",
    memory_limit_mb: int = 0,
) -> object:
    """Minimal stand-in for McpConfig."""
    cfg = MagicMock()
    cfg.sandbox_backend = sandbox_backend
    cfg.memory_limit_mb = memory_limit_mb
    return cfg
 
 
# ---------------------------------------------------------------------------
# SandboxBackend Protocol
# ---------------------------------------------------------------------------
 
class TestSandboxBackendProtocol:
    def test_native_backend_is_sandbox_backend(self):
        assert isinstance(NativeBackend(), SandboxBackend)
 
    def test_bubblewrap_backend_is_sandbox_backend(self):
        assert isinstance(BubblewrapBackend(), SandboxBackend)
 
    def test_sandbox_exec_backend_is_sandbox_backend(self):
        assert isinstance(SandboxExecBackend(), SandboxBackend)
 
 
# ---------------------------------------------------------------------------
# NativeBackend
# ---------------------------------------------------------------------------
 
class TestNativeBackend:
    def test_run_returns_tuple_of_three(self, tmp_path: Path):
        backend = NativeBackend()
        result = backend.run(
            cmd=[sys.executable, "-c", "print('hello')"],
            env={"PATH": os.environ.get("PATH", "")},
            cwd=tmp_path,
            timeout=10,
            memory_limit_bytes=None,
        )
        assert isinstance(result, tuple)
        assert len(result) == 3
        exit_code, stdout, stderr = result
        assert isinstance(exit_code, int)
        assert isinstance(stdout, str)
        assert isinstance(stderr, str)
 
    def test_run_exit_code_zero_on_success(self, tmp_path: Path):
        backend = NativeBackend()
        exit_code, _, _ = backend.run(
            cmd=[sys.executable, "-c", "import sys; sys.exit(0)"],
            env={"PATH": os.environ.get("PATH", "")},
            cwd=tmp_path,
            timeout=10,
            memory_limit_bytes=None,
        )
        assert exit_code == 0
 
    def test_run_nonzero_exit_code(self, tmp_path: Path):
        backend = NativeBackend()
        exit_code, _, _ = backend.run(
            cmd=[sys.executable, "-c", "import sys; sys.exit(42)"],
            env={"PATH": os.environ.get("PATH", "")},
            cwd=tmp_path,
            timeout=10,
            memory_limit_bytes=None,
        )
        assert exit_code == 42
 
    def test_run_captures_stdout(self, tmp_path: Path):
        backend = NativeBackend()
        _, stdout, _ = backend.run(
            cmd=[sys.executable, "-c", "print('captured_output')"],
            env={"PATH": os.environ.get("PATH", "")},
            cwd=tmp_path,
            timeout=10,
            memory_limit_bytes=None,
        )
        assert "captured_output" in stdout
 
    def test_run_captures_stderr(self, tmp_path: Path):
        backend = NativeBackend()
        _, _, stderr = backend.run(
            cmd=[sys.executable, "-c", "import sys; print('err', file=sys.stderr)"],
            env={"PATH": os.environ.get("PATH", "")},
            cwd=tmp_path,
            timeout=10,
            memory_limit_bytes=None,
        )
        assert "err" in stderr
 
    def test_preexec_fn_is_none_on_windows(self, tmp_path: Path):
        """On Windows, memory limit is ignored (preexec_fn must be None)."""
        backend = NativeBackend()
        captured = {}
 
        original_run = subprocess.run
 
        def mock_run(*args, **kwargs):
            captured["preexec_fn"] = kwargs.get("preexec_fn")
            return original_run(*args, **kwargs)
 
        with patch("oct.mcp.sandbox.subprocess.run", side_effect=mock_run):
            with patch("oct.mcp.sandbox.os.name", "nt"):
                backend.run(
                    cmd=[sys.executable, "-c", "pass"],
                    env={"PATH": os.environ.get("PATH", "")},
                    cwd=tmp_path,
                    timeout=10,
                    memory_limit_bytes=536_870_912,
                )
        assert captured["preexec_fn"] is None
 
    @pytest.mark.skipif(
        os.name == "nt",
        reason=(
            "POSIX-only: NativeBackend uses preexec_fn (a fork-only "
            "subprocess feature) to apply setrlimit before exec; the "
            "Windows subprocess module rejects preexec_fn entirely. "
            "Behaviour is exercised by Linux/macOS CI."
        ),
    )
    def test_preexec_fn_set_on_posix_with_limit(self, tmp_path: Path):
        """On POSIX with memory_limit_bytes set, preexec_fn should be non-None."""
        backend = NativeBackend()
        captured = {}
 
        original_run = subprocess.run
 
        def mock_run(*args, **kwargs):
            captured["preexec_fn"] = kwargs.get("preexec_fn")
            # Don't actually run to avoid setrlimit side effects.
            m = MagicMock()
            m.returncode = 0
            m.stdout = ""
            m.stderr = ""
            return m
 
        with patch("oct.mcp.sandbox.subprocess.run", side_effect=mock_run):
            backend.run(
                cmd=[sys.executable, "-c", "pass"],
                env={"PATH": os.environ.get("PATH", "")},
                cwd=tmp_path,
                timeout=10,
                memory_limit_bytes=536_870_912,
            )
        assert captured["preexec_fn"] is not None
        assert callable(captured["preexec_fn"])
 
    def test_preexec_fn_none_when_no_memory_limit(self, tmp_path: Path):
        """No memory limit → preexec_fn must be None regardless of platform."""
        backend = NativeBackend()
        captured = {}
 
        original_run = subprocess.run
 
        def mock_run(*args, **kwargs):
            captured["preexec_fn"] = kwargs.get("preexec_fn")
            m = MagicMock()
            m.returncode = 0
            m.stdout = ""
            m.stderr = ""
            return m
 
        with patch("oct.mcp.sandbox.subprocess.run", side_effect=mock_run):
            backend.run(
                cmd=[sys.executable, "-c", "pass"],
                env={"PATH": os.environ.get("PATH", "")},
                cwd=tmp_path,
                timeout=10,
                memory_limit_bytes=None,
            )
        assert captured["preexec_fn"] is None
 
    def test_shell_is_always_false(self, tmp_path: Path):
        """shell=True must never be passed to subprocess.run."""
        backend = NativeBackend()
        captured = {}
 
        original_run = subprocess.run
 
        def mock_run(*args, **kwargs):
            captured["shell"] = kwargs.get("shell", False)
            m = MagicMock()
            m.returncode = 0
            m.stdout = ""
            m.stderr = ""
            return m
 
        with patch("oct.mcp.sandbox.subprocess.run", side_effect=mock_run):
            backend.run(
                cmd=[sys.executable, "-c", "pass"],
                env={},
                cwd=tmp_path,
                timeout=10,
                memory_limit_bytes=None,
            )
        assert captured["shell"] is False
 
 
# ---------------------------------------------------------------------------
# BubblewrapBackend
# ---------------------------------------------------------------------------
 
class TestBubblewrapBackend:
    def test_bwrap_cmd_starts_with_bwrap(self, tmp_path: Path):
        backend = BubblewrapBackend()
        result = backend._bwrap_cmd(["echo", "hello"], tmp_path)
        assert result[0] == "bwrap"
 
    def test_bwrap_cmd_contains_unshare_net(self, tmp_path: Path):
        backend = BubblewrapBackend()
        result = backend._bwrap_cmd(["echo", "hello"], tmp_path)
        assert "--unshare-net" in result
 
    def test_bwrap_cmd_contains_die_with_parent(self, tmp_path: Path):
        backend = BubblewrapBackend()
        result = backend._bwrap_cmd(["echo", "hello"], tmp_path)
        assert "--die-with-parent" in result
 
    def test_bwrap_cmd_binds_cwd(self, tmp_path: Path):
        backend = BubblewrapBackend()
        result = backend._bwrap_cmd(["echo", "hello"], tmp_path)
        assert "--bind" in result
        bind_idx = result.index("--bind")
        # bind source and target should both be the cwd
        assert result[bind_idx + 1] == str(tmp_path)
        assert result[bind_idx + 2] == str(tmp_path)
 
    def test_bwrap_cmd_contains_chdir(self, tmp_path: Path):
        backend = BubblewrapBackend()
        result = backend._bwrap_cmd(["echo", "hello"], tmp_path)
        assert "--chdir" in result
        chdir_idx = result.index("--chdir")
        assert result[chdir_idx + 1] == str(tmp_path)
 
    def test_bwrap_cmd_ends_with_separator_then_cmd(self, tmp_path: Path):
        backend = BubblewrapBackend()
        cmd = ["echo", "hello"]
        result = backend._bwrap_cmd(cmd, tmp_path)
        assert "--" in result
        sep_idx = result.index("--")
        assert result[sep_idx + 1:] == cmd
 
    def test_bwrap_cmd_proc_dev_tmpfs(self, tmp_path: Path):
        backend = BubblewrapBackend()
        result = backend._bwrap_cmd(["echo"], tmp_path)
        assert "--proc" in result
        assert "--dev" in result
        assert "--tmpfs" in result
 
    def test_shell_is_always_false(self, tmp_path: Path):
        backend = BubblewrapBackend()
        captured = {}
 
        original_run = subprocess.run
 
        def mock_run(*args, **kwargs):
            captured["shell"] = kwargs.get("shell", False)
            m = MagicMock()
            m.returncode = 0
            m.stdout = ""
            m.stderr = ""
            return m
 
        with patch("oct.mcp.sandbox.subprocess.run", side_effect=mock_run):
            backend.run(
                cmd=["echo", "hi"],
                env={},
                cwd=tmp_path,
                timeout=10,
                memory_limit_bytes=None,
            )
        assert captured["shell"] is False
 
 
# ---------------------------------------------------------------------------
# SandboxExecBackend
# ---------------------------------------------------------------------------
 
class TestSandboxExecBackend:
    def test_sandboxed_cmd_starts_with_sandbox_exec(self):
        backend = SandboxExecBackend()
        result = backend._sandboxed_cmd(["echo", "hi"])
        assert result[0] == "sandbox-exec"
 
    def test_sandboxed_cmd_has_p_flag(self):
        backend = SandboxExecBackend()
        result = backend._sandboxed_cmd(["echo", "hi"])
        assert "-p" in result
 
    def test_sandboxed_cmd_contains_deny_network(self):
        backend = SandboxExecBackend()
        result = backend._sandboxed_cmd(["echo", "hi"])
        profile_idx = result.index("-p") + 1
        assert "deny network" in result[profile_idx]
 
    def test_sandboxed_cmd_ends_with_original_cmd(self):
        backend = SandboxExecBackend()
        cmd = ["python", "-c", "pass"]
        result = backend._sandboxed_cmd(cmd)
        assert result[-len(cmd):] == cmd
 
    def test_profile_denies_network(self):
        backend = SandboxExecBackend()
        assert "deny network" in backend._PROFILE
 
    def test_shell_is_always_false(self, tmp_path: Path):
        backend = SandboxExecBackend()
        captured = {}
 
        original_run = subprocess.run
 
        def mock_run(*args, **kwargs):
            captured["shell"] = kwargs.get("shell", False)
            m = MagicMock()
            m.returncode = 0
            m.stdout = ""
            m.stderr = ""
            return m
 
        with patch("oct.mcp.sandbox.subprocess.run", side_effect=mock_run):
            backend.run(
                cmd=["echo", "hi"],
                env={},
                cwd=tmp_path,
                timeout=10,
                memory_limit_bytes=None,
            )
        assert captured["shell"] is False
 
 
# ---------------------------------------------------------------------------
# make_sandbox_backend factory
# ---------------------------------------------------------------------------
 
class TestMakeSandboxBackend:
    def test_native_always_returns_native_backend(self):
        cfg = _fake_config(sandbox_backend="native")
        result = make_sandbox_backend(cfg)
        assert isinstance(result, NativeBackend)
 
    def test_auto_returns_native_on_windows(self):
        cfg = _fake_config(sandbox_backend="auto")
        with patch("oct.mcp.sandbox.sys.platform", "win32"):
            result = make_sandbox_backend(cfg)
        assert isinstance(result, NativeBackend)
 
    def test_auto_returns_bubblewrap_on_linux_when_bwrap_found(self):
        cfg = _fake_config(sandbox_backend="auto")
        with patch("oct.mcp.sandbox.sys.platform", "linux"):
            with patch("oct.mcp.sandbox.shutil.which", return_value="/usr/bin/bwrap"):
                result = make_sandbox_backend(cfg)
        assert isinstance(result, BubblewrapBackend)
 
    def test_auto_returns_native_on_linux_when_bwrap_missing(self):
        cfg = _fake_config(sandbox_backend="auto")
        with patch("oct.mcp.sandbox.sys.platform", "linux"):
            with patch("oct.mcp.sandbox.shutil.which", return_value=None):
                result = make_sandbox_backend(cfg)
        assert isinstance(result, NativeBackend)
 
    def test_auto_returns_sandbox_exec_on_macos_when_found(self):
        cfg = _fake_config(sandbox_backend="auto")
        with patch("oct.mcp.sandbox.sys.platform", "darwin"):
            with patch("oct.mcp.sandbox.shutil.which", return_value="/usr/bin/sandbox-exec"):
                result = make_sandbox_backend(cfg)
        assert isinstance(result, SandboxExecBackend)
 
    def test_auto_returns_native_on_macos_when_sandbox_exec_missing(self):
        cfg = _fake_config(sandbox_backend="auto")
        with patch("oct.mcp.sandbox.sys.platform", "darwin"):
            with patch("oct.mcp.sandbox.shutil.which", return_value=None):
                result = make_sandbox_backend(cfg)
        assert isinstance(result, NativeBackend)
 
    def test_forced_bubblewrap_raises_when_bwrap_absent(self):
        cfg = _fake_config(sandbox_backend="bubblewrap")
        with patch("oct.mcp.sandbox.shutil.which", return_value=None):
            with pytest.raises(RuntimeError, match="bwrap"):
                make_sandbox_backend(cfg)
 
    def test_forced_bubblewrap_returns_backend_when_bwrap_present(self):
        cfg = _fake_config(sandbox_backend="bubblewrap")
        with patch("oct.mcp.sandbox.shutil.which", return_value="/usr/bin/bwrap"):
            result = make_sandbox_backend(cfg)
        assert isinstance(result, BubblewrapBackend)
 
    def test_forced_sandbox_exec_raises_when_absent(self):
        cfg = _fake_config(sandbox_backend="sandbox_exec")
        with patch("oct.mcp.sandbox.shutil.which", return_value=None):
            with pytest.raises(RuntimeError, match="sandbox-exec"):
                make_sandbox_backend(cfg)
 
    def test_forced_sandbox_exec_returns_backend_when_present(self):
        cfg = _fake_config(sandbox_backend="sandbox_exec")
        with patch("oct.mcp.sandbox.shutil.which", return_value="/usr/bin/sandbox-exec"):
            result = make_sandbox_backend(cfg)
        assert isinstance(result, SandboxExecBackend)
 
    def test_unknown_name_falls_through_to_native(self):
        """Unknown sandbox_backend value (e.g. config validation missed it) → NativeBackend."""
        cfg = _fake_config(sandbox_backend="unknown_backend")
        result = make_sandbox_backend(cfg)
        assert isinstance(result, NativeBackend)
 
 
# ---------------------------------------------------------------------------
# SandboxExecutor — backend delegation
# ---------------------------------------------------------------------------
 
class TestSandboxExecutorDelegation:
    def test_executor_calls_backend_run(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "ok", "")
        executor = SandboxExecutor(backend=mock_backend)
        executor.run(cmd=["echo"], cwd=tmp_path, timeout=10)
        mock_backend.run.assert_called_once()
 
    def test_executor_passes_cwd_to_backend(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "", "")
        executor = SandboxExecutor(backend=mock_backend)
        executor.run(cmd=["echo"], cwd=tmp_path, timeout=10)
        call_kwargs = mock_backend.run.call_args[1]
        assert call_kwargs["cwd"] == tmp_path
 
    def test_executor_passes_timeout_to_backend(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "", "")
        executor = SandboxExecutor(backend=mock_backend)
        executor.run(cmd=["echo"], cwd=tmp_path, timeout=42)
        call_kwargs = mock_backend.run.call_args[1]
        assert call_kwargs["timeout"] == 42
 
    def test_executor_memory_limit_zero_passes_none(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "", "")
        executor = SandboxExecutor(backend=mock_backend, memory_limit_mb=0)
        executor.run(cmd=["echo"], cwd=tmp_path, timeout=10)
        call_kwargs = mock_backend.run.call_args[1]
        assert call_kwargs["memory_limit_bytes"] is None
 
    def test_executor_memory_limit_512mb_passes_bytes(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "", "")
        executor = SandboxExecutor(backend=mock_backend, memory_limit_mb=512)
        executor.run(cmd=["echo"], cwd=tmp_path, timeout=10)
        call_kwargs = mock_backend.run.call_args[1]
        assert call_kwargs["memory_limit_bytes"] == 512 * 1024 * 1024
 
    def test_executor_defaults_to_native_backend(self):
        executor = SandboxExecutor()
        assert isinstance(executor._backend, NativeBackend)
 
    def test_executor_output_capped_at_max_output_bytes(self, tmp_path: Path):
        large_output = "x" * 2000
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, large_output, "")
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["echo"], cwd=tmp_path, timeout=10, max_output_bytes=100)
        assert len(result.stdout) <= 100
        assert result.output_truncated is True
 
    def test_executor_returns_sandbox_result_on_success(self, tmp_path: Path):
        from oct.mcp.sandbox import SandboxResult
        mock_backend = MagicMock()
        mock_backend.run.return_value = (0, "output", "")
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["echo"], cwd=tmp_path, timeout=10)
        assert isinstance(result, SandboxResult)
        assert result.exit_code == 0
        assert result.stdout == "output"
 
    def test_executor_handles_timeout_expired(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.side_effect = subprocess.TimeoutExpired(cmd="echo", timeout=5)
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["echo"], cwd=tmp_path, timeout=5)
        assert result.timed_out is True
        assert result.exit_code == 1
 
    def test_executor_handles_file_not_found(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.side_effect = FileNotFoundError("no such file")
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["nonexistent_command"], cwd=tmp_path, timeout=10)
        assert result.exit_code == 1
        assert result.error_message != ""
 
    def test_executor_handles_os_error(self, tmp_path: Path):
        mock_backend = MagicMock()
        mock_backend.run.side_effect = OSError("permission denied")
        executor = SandboxExecutor(backend=mock_backend)
        result = executor.run(cmd=["echo"], cwd=tmp_path, timeout=10)
        assert result.exit_code == 1
        assert result.error_message != ""
 
 
# ---------------------------------------------------------------------------
# _sanitise_env
# ---------------------------------------------------------------------------
 
class TestSanitiseEnv:
    def test_strips_pythonpath(self, tmp_path: Path):
        with patch.dict(os.environ, {"PYTHONPATH": "/some/path"}, clear=False):
            env = _sanitise_env(tmp_path)
        assert "PYTHONPATH" not in env
 
    def test_strips_api_key(self, tmp_path: Path):
        with patch.dict(os.environ, {"MY_API_KEY": "super_secret"}, clear=False):
            env = _sanitise_env(tmp_path)
        assert "MY_API_KEY" not in env
 
    def test_strips_password_key(self, tmp_path: Path):
        with patch.dict(os.environ, {"DB_PASSWORD": "pass123"}, clear=False):
            env = _sanitise_env(tmp_path)
        assert "DB_PASSWORD" not in env
 
    def test_strips_secret_key(self, tmp_path: Path):
        with patch.dict(os.environ, {"AWS_SECRET_ACCESS_KEY": "xxx"}, clear=False):
            env = _sanitise_env(tmp_path)
        assert "AWS_SECRET_ACCESS_KEY" not in env
 
    def test_preserves_path(self, tmp_path: Path):
        env = _sanitise_env(tmp_path)
        # PATH should be present (or at least not stripped) if it existed
        if "PATH" in os.environ:
            assert "PATH" in env
 
    def test_preserves_home(self, tmp_path: Path):
        if "HOME" in os.environ:
            env = _sanitise_env(tmp_path)
            assert "HOME" in env
 
    def test_adds_oct_mcp_cwd(self, tmp_path: Path):
        env = _sanitise_env(tmp_path)
        assert "OCT_MCP_CWD" in env
        assert env["OCT_MCP_CWD"] == str(tmp_path)
 
    def test_returns_new_dict_not_os_environ(self, tmp_path: Path):
        env = _sanitise_env(tmp_path)
        assert env is not os.environ