sandbox.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/sandbox.py
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Module docstring prose, not a secret"
"""
Purpose
-------
Execution Sandbox (Layer 4, outer wrapper) for the ``oct-mcp``
six-layer pipeline. Provides environment sanitisation, working-directory
pinning, timeout enforcement, and output-size limiting for subprocess
execution.
 
Phase 5C adds pluggable OS-level sandbox *backends*:
 
* :class:`NativeBackend` - existing behavior (subprocess + env
  sanitization). Always available; default on Windows.
* :class:`BubblewrapBackend` - Linux only. Wraps every command in
  ``bwrap --unshare-net --die-with-parent``. Requires ``bwrap`` in PATH.
* :class:`SandboxExecBackend` - macOS only. Prepends
  ``sandbox-exec -p "(deny network*)(allow default)"``. Requires
  ``sandbox-exec`` in PATH.
 
Use :func:`make_sandbox_backend` to select the appropriate backend based
on :class:`~oct.mcp.config.McpConfig`.
 
Responsibilities
----------------
- Strip sensitive environment variables before forking (``PYTHONPATH``,
  common secret-named vars).
- Pin the subprocess working directory to the validated project root.
- Enforce per-tool timeouts and kill the process on expiry.
- Cap combined stdout+stderr output to ``max_output_bytes``.
- Never use ``shell=True``.
- Return a :class:`SandboxResult` - never raise for subprocess errors.
 
Diagnostics
-----------
Domain: MCP
Levels:
    L1 - errors: process launch failures, unexpected errors
    L2 - lifecycle: process start, process end, backend selection
    L3 - command details (argv)
    L4 - deep trace: per-line output, env keys
 
Contracts
---------
- ``shell=False`` is enforced unconditionally in all backends - no
  caller path can override it.
- :meth:`SandboxExecutor.run` never raises. Any ``OSError`` or
  ``subprocess.SubprocessError`` is caught, logged at level 1, and
  returned as a ``SandboxResult`` with a non-zero exit code.
- The sanitised environment always preserves ``PATH``, ``HOME``,
  ``LANG``, ``TERM``, ``USER``, ``USERNAME``, ``LOGNAME``,
  ``VIRTUAL_ENV``, ``CONDA_DEFAULT_ENV``, and
  ``OCT_MCP_SESSION_ID`` / ``OCT_MCP_REQUEST_ID`` correlation IDs.
- :func:`make_sandbox_backend` never raises - it always returns a
  usable backend (falls back to :class:`NativeBackend`).
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- bwrap (not imported; the bubblewrap CLI that BubblewrapBackend wraps every command in on Linux)
"""
 
from __future__ import annotations
 
import os
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol, runtime_checkable
 
from oct.core.diagnostics import _dbg
 
 
# ---------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------
 
#: Environment variable keys (or key substrings, case-insensitive) that
#: must be stripped before passing the environment to a subprocess.
#: Full key names are exact-matched first; substrings are checked second.
_SENSITIVE_ENV_EXACT: frozenset[str] = frozenset({
    "PYTHONPATH",
})
 
#: Case-insensitive substrings. Any env key containing one of these
#: tokens is stripped.
_SENSITIVE_ENV_SUBSTRINGS: tuple[str, ...] = (
    "api_key", "apikey",
    "password", "passwd", "pwd",
    "secret",
    "token",
    "credentials",
    "private_key",
    "auth",
    "access_key",
)
 
#: Environment variable keys that are always preserved regardless of
#: name-based filtering.
_SAFE_ENV_KEYS: frozenset[str] = frozenset({
    "PATH",
    "HOME",
    "LANG",
    "LC_ALL",
    "LC_CTYPE",
    "TERM",
    "COLORTERM",
    "USER",
    "USERNAME",
    "LOGNAME",
    "USERPROFILE",         # Windows home directory
    "SYSTEMROOT",          # Windows system root
    "VIRTUAL_ENV",
    "CONDA_DEFAULT_ENV",
    "CONDA_PREFIX",
    "OCT_MCP_SESSION_ID",
    "OCT_MCP_REQUEST_ID",
    # Diagnostics env vars
    "OCT_DBG_LEVEL",
    "OCT_DBG_FILE",
})
 
 
# ---------------------------------------------------------------------
# Environment sanitisation (shared by all backends)
# ---------------------------------------------------------------------
 
 
def _sanitise_env(project_root: Path) -> dict[str, str]:
    """Build a sanitised environment for subprocess execution.
 
    Starts from ``os.environ``, preserves :data:`_SAFE_ENV_KEYS`,
    strips :data:`_SENSITIVE_ENV_EXACT` exact keys and anything whose
    key contains a :data:`_SENSITIVE_ENV_SUBSTRINGS` token (case-
    insensitive), and adds ``OCT_MCP_CWD`` so the subprocess knows
    its sandboxed working directory.
 
    Returns
    -------
    A new ``dict[str, str]`` - never modifies ``os.environ``.
    """
    func = "_sanitise_env"
    clean: dict[str, str] = {}
 
    for key, value in os.environ.items():
        key_upper = key.upper()
 
        # Always keep safe keys.
        if key_upper in {k.upper() for k in _SAFE_ENV_KEYS}:
            clean[key] = value
            continue
 
        # Strip exact sensitive keys.
        if key_upper in {k.upper() for k in _SENSITIVE_ENV_EXACT}:
            _dbg("MCP", func, f"stripped env key={key}", 4)
            continue
 
        # Strip substring-matched keys.
        if any(sub in key_upper.lower() for sub in _SENSITIVE_ENV_SUBSTRINGS):
            _dbg("MCP", func, f"stripped env key={key} (substring match)", 4)
            continue
 
        clean[key] = value
 
    # Pin working directory as an env var for informational purposes.
    clean["OCT_MCP_CWD"] = str(project_root)
 
    return clean
 
 
# ---------------------------------------------------------------------
# Result type
# ---------------------------------------------------------------------
 
 
@dataclass
class SandboxResult:
    """Result of a sandboxed subprocess execution.
 
    Attributes
    ----------
    exit_code
        The subprocess return code, or ``1`` on launch failure.
    stdout
        Captured standard output (possibly truncated).
    stderr
        Captured standard error (possibly truncated).
    timed_out
        ``True`` if the process was killed due to timeout.
    output_truncated
        ``True`` if combined stdout+stderr exceeded ``max_output_bytes``.
    error_message
        Non-empty if the process could not be launched (e.g. not found
        on PATH). Distinct from stderr - set on ``OSError``.
    """
 
    exit_code: int
    stdout: str
    stderr: str
    timed_out: bool = False
    output_truncated: bool = False
    error_message: str = ""
 
 
# ---------------------------------------------------------------------
# Backend protocol
# ---------------------------------------------------------------------
 
 
@runtime_checkable
class SandboxBackend(Protocol):
    """Protocol for OS-level sandbox backends.
 
    Each backend wraps the subprocess call with platform-specific
    isolation (network namespace, resource limits, etc.). The backend
    receives a sanitised environment and returns raw
    ``(exit_code, stdout, stderr)`` - output capping and error handling
    live in :class:`SandboxExecutor`.
    """
 
    def run(
        self,
        cmd: list[str],
        env: dict[str, str],
        cwd: Path,
        timeout: int,
        memory_limit_bytes: int | None,
    ) -> tuple[int, str, str]:
        """Run *cmd* and return ``(exit_code, stdout, stderr)``.
 
        Parameters
        ----------
        cmd
            Command + arguments list. Must not use ``shell=True``.
        env
            Sanitised environment dict.
        cwd
            Working directory (validated project root).
        timeout
            Maximum wall-clock seconds.
        memory_limit_bytes
            Virtual memory ceiling in bytes, or ``None`` to skip.
            Implementations on POSIX apply this via ``setrlimit``; on
            Windows it is silently ignored.
        """
        ...
 
 
# ---------------------------------------------------------------------
# Concrete backends
# ---------------------------------------------------------------------
 
 
class NativeBackend:
    """Subprocess-only backend - existing Phase 5A/5B behavior.
 
    On POSIX, when *memory_limit_bytes* is non-None and non-zero,
    applies ``resource.setrlimit(RLIMIT_AS, ...)`` via ``preexec_fn``
    to limit virtual address space. On Windows the limit is silently
    ignored (use Docker for Windows memory enforcement).
    """
 
    def run(
        self,
        cmd: list[str],
        env: dict[str, str],
        cwd: Path,
        timeout: int,
        memory_limit_bytes: int | None,
    ) -> tuple[int, str, str]:
        preexec = None
        if memory_limit_bytes and os.name != "nt":
            limit = memory_limit_bytes
 
            def _preexec() -> None:
                import resource
                resource.setrlimit(
                    resource.RLIMIT_AS, (limit, limit)
                )
 
            preexec = _preexec
 
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            cwd=str(cwd),
            env=env,
            timeout=timeout,
            shell=False,  # NEVER shell=True
            preexec_fn=preexec,
        )
        return result.returncode, result.stdout or "", result.stderr or ""
 
 
class BubblewrapBackend:
    """Linux bubblewrap backend - network-isolated, die-with-parent.
 
    Wraps every command in ``bwrap --unshare-net --die-with-parent``.
    Requires ``bwrap`` to be available in PATH.
 
    Network isolation is provided by ``--unshare-net``.  The project
    root is bind-mounted read-write; system paths are read-only.
    """
 
    def _bwrap_cmd(self, cmd: list[str], cwd: Path) -> list[str]:
        """Build the full ``bwrap ... -- cmd`` invocation."""
        bwrap = ["bwrap"]
 
        # Read-only system mounts (common paths; skip missing ones silently).
        for ro_path in ("/usr", "/lib", "/lib64", "/bin", "/sbin", "/etc",
                        "/run", "/opt"):
            if Path(ro_path).exists():
                bwrap += ["--ro-bind", ro_path, ro_path]
 
        bwrap += [
            "--proc", "/proc",
            "--dev", "/dev",
            "--tmpfs", "/tmp",
            # Project root bind-mounted read-write.
            "--bind", str(cwd), str(cwd),
            "--chdir", str(cwd),
            # Network isolation.
            "--unshare-net",
            # Kill sandbox when the parent exits.
            "--die-with-parent",
            "--",
        ]
        return bwrap + cmd
 
    def run(
        self,
        cmd: list[str],
        env: dict[str, str],
        cwd: Path,
        timeout: int,
        memory_limit_bytes: int | None,
    ) -> tuple[int, str, str]:
        wrapped = self._bwrap_cmd(cmd, cwd)
        result = subprocess.run(
            wrapped,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            cwd=str(cwd),
            env=env,
            timeout=timeout,
            shell=False,
        )
        return result.returncode, result.stdout or "", result.stderr or ""
 
 
class SandboxExecBackend:
    """macOS ``sandbox-exec`` backend - network-denied policy.
 
    Prepends ``sandbox-exec -p "<profile>"`` to every command.
    Requires ``sandbox-exec`` to be available in PATH.
 
    The policy denies all network access while allowing everything else
    (file I/O, process creation, etc.) needed to run ``oct`` commands.
    """
 
    _PROFILE: str = "(version 1)(deny network*)(allow default)"
 
    def _sandboxed_cmd(self, cmd: list[str]) -> list[str]:
        """Return ``["sandbox-exec", "-p", PROFILE] + cmd``."""
        return ["sandbox-exec", "-p", self._PROFILE] + cmd
 
    def run(
        self,
        cmd: list[str],
        env: dict[str, str],
        cwd: Path,
        timeout: int,
        memory_limit_bytes: int | None,
    ) -> tuple[int, str, str]:
        wrapped = self._sandboxed_cmd(cmd)
        result = subprocess.run(
            wrapped,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            cwd=str(cwd),
            env=env,
            timeout=timeout,
            shell=False,
        )
        return result.returncode, result.stdout or "", result.stderr or ""
 
 
# ---------------------------------------------------------------------
# Backend factory
# ---------------------------------------------------------------------
 
 
def make_sandbox_backend(config: object) -> SandboxBackend:
    """Select and return the best available :class:`SandboxBackend`.
 
    Parameters
    ----------
    config
        A :class:`~oct.mcp.config.McpConfig` instance. Reads the
        ``sandbox_backend`` field (``"auto"`` / ``"native"`` /
        ``"bubblewrap"`` / ``"sandbox_exec"``) and ``memory_limit_mb``.
 
    Returns
    -------
    A concrete backend instance. On ``"auto"``, picks:
 
    * :class:`BubblewrapBackend` on Linux when ``bwrap`` is in PATH.
    * :class:`SandboxExecBackend` on macOS when ``sandbox-exec`` is in PATH.
    * :class:`NativeBackend` otherwise (always safe).
 
    When a *specific* backend is forced (e.g. ``"bubblewrap"``) but its
    binary is not found, raises :exc:`RuntimeError`.
 
    Never raises on ``"auto"`` or ``"native"`` - always returns a
    usable backend.
    """
    func = "make_sandbox_backend"
    name: str = getattr(config, "sandbox_backend", "auto")
 
    if name == "bubblewrap":
        if not shutil.which("bwrap"):
            raise RuntimeError(
                "sandbox_backend='bubblewrap' requested but 'bwrap' was not "
                "found in PATH. Install bubblewrap or set sandbox_backend='auto'."
            )
        _dbg("MCP", func, "backend=BubblewrapBackend (forced)", 2)
        return BubblewrapBackend()
 
    if name == "sandbox_exec":
        if not shutil.which("sandbox-exec"):
            raise RuntimeError(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Error message naming the missing sandbox-exec binary, not a secret"
                "sandbox_backend='sandbox_exec' requested but 'sandbox-exec' "
                "was not found in PATH. This backend requires macOS."
            )
        _dbg("MCP", func, "backend=SandboxExecBackend (forced)", 2)
        return SandboxExecBackend()
 
    if name == "auto":
        if sys.platform == "linux" and shutil.which("bwrap"):
            _dbg("MCP", func, "backend=BubblewrapBackend (auto, Linux)", 2)
            return BubblewrapBackend()
        if sys.platform == "darwin" and shutil.which("sandbox-exec"):
            _dbg("MCP", func, "backend=SandboxExecBackend (auto, macOS)", 2)
            return SandboxExecBackend()
 
    # "native" or auto-fallback (e.g. Windows, or no bwrap/sandbox-exec found).
    _dbg("MCP", func, "backend=NativeBackend", 2)
    return NativeBackend()
 
 
# ---------------------------------------------------------------------
# Sandbox executor
# ---------------------------------------------------------------------
 
 
class SandboxExecutor:
    """Sandboxed subprocess runner for the MCP execution layer.
 
    Wraps a :class:`SandboxBackend` with environment sanitisation,
    output capping, and error handling. A single :class:`SandboxExecutor`
    instance may be reused across multiple tool calls within the same
    server session.
 
    Parameters
    ----------
    backend
        OS-level isolation backend. Defaults to :class:`NativeBackend`
        when not supplied (Phase 5A/5B backward compatibility).
    memory_limit_mb
        Per-subprocess virtual memory limit in megabytes. ``0`` means
        no limit. Forwarded to the backend as bytes. POSIX only -
        silently ignored on Windows.
    """
 
    def __init__(
        self,
        backend: SandboxBackend | None = None,
        memory_limit_mb: int = 0,
    ) -> None:
        self._backend: SandboxBackend = backend if backend is not None else NativeBackend()
        self._memory_limit_bytes: int | None = (
            memory_limit_mb * 1024 * 1024 if memory_limit_mb > 0 else None
        )
 
    def run(
        self,
        cmd: list[str],
        cwd: Path,
        timeout: int,
        max_output_bytes: int = 1_048_576,
    ) -> SandboxResult:
        """Execute *cmd* in a sanitised environment under *cwd*.
 
        Parameters
        ----------
        cmd
            The command and arguments to run. Must not contain shell
            metacharacters - this is enforced by passing ``shell=False``
            in every backend.
        cwd
            Working directory, pinned to the validated project root.
        timeout
            Maximum wall-clock seconds before the process is killed.
        max_output_bytes
            Combined stdout+stderr cap. When exceeded, output is
            truncated and :attr:`SandboxResult.output_truncated` is set.
 
        Returns
        -------
        :class:`SandboxResult` - never raises.
        """
        func = "SandboxExecutor.run"
        _dbg("MCP", func, f"cmd={cmd!r} cwd={cwd} timeout={timeout}s", 3)
 
        env = _sanitise_env(cwd)
 
        try:
            exit_code, stdout, stderr = self._backend.run(
                cmd=cmd,
                env=env,
                cwd=cwd,
                timeout=timeout,
                memory_limit_bytes=self._memory_limit_bytes,
            )
            truncated = False
 
            # Enforce combined output cap.
            combined_bytes = len(
                (stdout + stderr).encode("utf-8", errors="replace")
            )
            if combined_bytes > max_output_bytes:
                stdout_bytes = stdout.encode("utf-8", errors="replace")
                if len(stdout_bytes) > max_output_bytes:
                    stdout = stdout_bytes[:max_output_bytes].decode(
                        "utf-8", errors="replace"
                    )
                    stderr = ""
                else:
                    remaining = max_output_bytes - len(stdout_bytes)
                    stderr = (
                        stderr.encode("utf-8", errors="replace")[:remaining]
                        .decode("utf-8", errors="replace")
                    )
                truncated = True
                _dbg(
                    "MCP", func,
                    f"output truncated combined_bytes={combined_bytes} "
                    f"limit={max_output_bytes}",
                    2,
                )
 
            _dbg(
                "MCP", func,
                f"exit_code={exit_code} truncated={truncated}",
                2,
            )
            return SandboxResult(
                exit_code=exit_code,
                stdout=stdout,
                stderr=stderr,
                timed_out=False,
                output_truncated=truncated,
            )
 
        except subprocess.TimeoutExpired:
            _dbg("MCP", func, f"SYSTEM_ERROR: timeout after {timeout}s cmd={cmd!r}", 1)
            return SandboxResult(
                exit_code=1,
                stdout="",
                stderr=f"Process timed out after {timeout} seconds.",
                timed_out=True,
            )
        except FileNotFoundError:
            msg = f"Command not found: {cmd[0]!r}. Is 'oct' installed on PATH?"
            _dbg("MCP", func, f"SYSTEM_ERROR: {msg}", 1)
            return SandboxResult(
                exit_code=1,
                stdout="",
                stderr=msg,
                error_message=msg,
            )
        except OSError as exc:
            msg = f"Failed to launch subprocess: {exc}"
            _dbg("MCP", func, f"SYSTEM_ERROR: {msg}", 1)
            return SandboxResult(
                exit_code=1,
                stdout="",
                stderr=msg,
                error_message=msg,
            )