test_mcp_sandbox.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_mcp/test_mcp_sandbox.py
 
"""
Purpose
-------
Regression tests for :mod:`oct.mcp.sandbox` — ``SandboxExecutor`` and
environment sanitisation.
 
Responsibilities
----------------
- Verify environment sanitisation strips sensitive variables.
- Verify timeout enforcement and output-size limits.
- Verify T-04 (command chaining) and shell=False enforcement.
- Verify exit-code forwarding from subprocess.
 
Diagnostics
-----------
Domain: MCP-TESTS
Levels:
    L2 — test lifecycle
    L3 — assertion details
    L4 — deep tracing
 
Contracts
---------
- Tests use mock subprocess; no real commands are executed.
"""
 
from __future__ import annotations
 
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
 
import pytest
 
from oct.mcp.sandbox import (
    SandboxExecutor,
    SandboxResult,
    _sanitise_env,
    _SENSITIVE_ENV_SUBSTRINGS,
)
 
 
# -------------------------------------------------------------------
# _sanitise_env
# -------------------------------------------------------------------
 
class TestSanitiseEnv:
    def test_pythonpath_stripped(self, tmp_path: Path, monkeypatch):
        monkeypatch.setenv("PYTHONPATH", "/some/path")
        env = _sanitise_env(tmp_path)
        assert "PYTHONPATH" not in env
 
    def test_path_preserved(self, tmp_path: Path):
        env = _sanitise_env(tmp_path)
        assert "PATH" in env
 
    def test_home_preserved(self, tmp_path: Path):
        env = _sanitise_env(tmp_path)
        # HOME or USERPROFILE (Windows)
        assert "HOME" in env or "USERPROFILE" in env
 
    def test_secret_key_stripped(self, tmp_path: Path, monkeypatch):
        monkeypatch.setenv("MY_API_KEY", "topsecret")
        env = _sanitise_env(tmp_path)
        assert "MY_API_KEY" not in env
 
    def test_password_key_stripped(self, tmp_path: Path, monkeypatch):
        monkeypatch.setenv("DB_PASSWORD", "hunter2")
        env = _sanitise_env(tmp_path)
        assert "DB_PASSWORD" not in env
 
    def test_token_key_stripped(self, tmp_path: Path, monkeypatch):
        monkeypatch.setenv("GITHUB_TOKEN", "ghp_abc")
        env = _sanitise_env(tmp_path)
        assert "GITHUB_TOKEN" not in env
 
    def test_cwd_injected(self, tmp_path: Path):
        env = _sanitise_env(tmp_path)
        assert env.get("OCT_MCP_CWD") == str(tmp_path)
 
    def test_virtual_env_preserved(self, tmp_path: Path, monkeypatch):
        monkeypatch.setenv("VIRTUAL_ENV", "/venv")
        env = _sanitise_env(tmp_path)
        assert "VIRTUAL_ENV" in env
 
    def test_returns_new_dict(self, tmp_path: Path):
        env = _sanitise_env(tmp_path)
        assert env is not os.environ
 
 
# -------------------------------------------------------------------
# SandboxResult
# -------------------------------------------------------------------
 
class TestSandboxResult:
    def test_defaults(self):
        r = SandboxResult(exit_code=0, stdout="out", stderr="err")
        assert r.timed_out is False
        assert r.output_truncated is False
        assert r.error_message == ""
 
 
# -------------------------------------------------------------------
# SandboxExecutor.run — normal execution
# -------------------------------------------------------------------
 
class TestSandboxExecutorRun:
    def test_runs_simple_command(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            [sys.executable, "-c", "print('hello')"],
            cwd=tmp_path,
            timeout=10,
        )
        assert result.exit_code == 0
        assert "hello" in result.stdout
 
    def test_captures_exit_code_nonzero(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            [sys.executable, "-c", "import sys; sys.exit(42)"],
            cwd=tmp_path,
            timeout=10,
        )
        assert result.exit_code == 42
 
    def test_captures_stderr(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            [sys.executable, "-c", "import sys; print('err', file=sys.stderr)"],
            cwd=tmp_path,
            timeout=10,
        )
        assert "err" in result.stderr
 
    def test_shell_false_enforced(self, tmp_path: Path):
        """T-04: shell=True must never be used — semicolons are not executed."""
        executor = SandboxExecutor()
        # Passing shell injection via argv list items should not execute the shell.
        # The subprocess receives the literal string as an argument, not a shell command.
        result = executor.run(
            [sys.executable, "-c", "print('safe')"],
            cwd=tmp_path,
            timeout=10,
        )
        assert result.exit_code == 0
        assert result.timed_out is False
 
    def test_command_not_found_returns_error(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            ["this_command_does_not_exist_xyz123"],
            cwd=tmp_path,
            timeout=10,
        )
        assert result.exit_code == 1
        assert result.error_message != ""
 
    def test_never_raises(self, tmp_path: Path):
        executor = SandboxExecutor()
        # Should not raise even with a bad command.
        result = executor.run(
            ["nonexistent_cmd_abc"],
            cwd=tmp_path,
            timeout=5,
        )
        assert isinstance(result, SandboxResult)
 
    def test_working_directory_is_set(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            [sys.executable, "-c", "import os; print(os.getcwd())"],
            cwd=tmp_path,
            timeout=10,
        )
        assert result.exit_code == 0
        # The cwd should match tmp_path (resolving any symlinks for comparison)
        assert Path(result.stdout.strip()).resolve() == tmp_path.resolve()
 
 
# -------------------------------------------------------------------
# SandboxExecutor.run — timeout
# -------------------------------------------------------------------
 
class TestSandboxExecutorTimeout:
    def test_timeout_returns_timed_out_true(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            [sys.executable, "-c", "import time; time.sleep(30)"],
            cwd=tmp_path,
            timeout=1,
        )
        assert result.timed_out is True
        assert result.exit_code == 1
 
    def test_timeout_not_flagged_for_fast_command(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            [sys.executable, "-c", "print('done')"],
            cwd=tmp_path,
            timeout=10,
        )
        assert result.timed_out is False
 
 
# -------------------------------------------------------------------
# SandboxExecutor.run — output size limit
# -------------------------------------------------------------------
 
class TestSandboxExecutorOutputLimit:
    def test_output_within_limit_not_truncated(self, tmp_path: Path):
        executor = SandboxExecutor()
        result = executor.run(
            [sys.executable, "-c", "print('short')"],
            cwd=tmp_path,
            timeout=10,
            max_output_bytes=1024,
        )
        assert result.output_truncated is False
 
    def test_output_over_limit_truncated(self, tmp_path: Path):
        executor = SandboxExecutor()
        # Generate a large output.
        result = executor.run(
            [sys.executable, "-c", "print('x' * 2000)"],
            cwd=tmp_path,
            timeout=10,
            max_output_bytes=100,
        )
        assert result.output_truncated is True
        # Combined output should not exceed limit significantly.
        combined = len((result.stdout + result.stderr).encode("utf-8"))
        assert combined <= 200  # some slack for truncation boundary