#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/sandbox.py
"""
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`).
"""
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
try:
from oc_diagnostics import _dbg as _real_dbg
def _dbg(*args, **kwargs) -> None:
_real_dbg(*args, **kwargs)
except ImportError: # pragma: no cover
def _dbg(*args, **kwargs) -> None:
return None
# ---------------------------------------------------------------------
# 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
# ---------------------------------------------------------------------
[docs]
@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
# ---------------------------------------------------------------------
[docs]
@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`.
"""
[docs]
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
# ---------------------------------------------------------------------
[docs]
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).
"""
[docs]
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 ""
[docs]
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
[docs]
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 ""
[docs]
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
[docs]
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
# ---------------------------------------------------------------------
[docs]
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(
"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
# ---------------------------------------------------------------------
[docs]
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
)
[docs]
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,
)
