MCP 代码执行工具与沙箱：解释器类工具的设计、隔离与审计

# 模型心算 vs 真实执行的典型差异
# 1) 大数乘法: 模型常给出"看起来对"的结果，实际错
# 2) 日期推算: 闰年、时区、工作日边界极易出错
# 3) 数据统计: 均值/中位数/分组聚合靠"感觉"
# 4) 正则匹配: 复杂正则的匹配结果不可预测
# 交给解释器: 确定性、可复现、可验证

{
  "name": "execute_python",
  "description": "在隔离沙箱中执行 Python 代码并返回 stdout/stderr 与产物。适合数据计算、文本处理与图表生成。",
  "inputSchema": {
    "type": "object",
    "properties": {
      "code": { "type": "string", "description": "要执行的 Python 源码" },
      "timeout_ms": { "type": "integer", "default": 10000, "maximum": 60000 },
      "session_id": { "type": "string", "description": "复用状态时可传上次返回的会话 ID" },
      "files": {
        "type": "array",
        "items": { "type": "string" },
        "description": "需要挂载进沙箱的只读输入文件路径"
      }
    },
    "required": ["code"]
  }
}

// 有状态会话：一次调用创建，后续调用复用
interface ExecResult {
  session_id: string;
  stdout: string;
  stderr: string;
  exit_code: number;
  truncated: boolean;
  artifacts: Array<{ name: string; uri: string; mime: string }>;
}

server.tool(
  "execute_python",
  "在隔离沙箱中执行 Python 代码",
  { code: z.string(), session_id: z.string().optional(), timeout_ms: z.number().optional() },
  async ({ code, session_id, timeout_ms }) => {
    const session = session_id
      ? await sessions.resume(session_id)
      : await sessions.create({ timeoutMs: timeout_ms ?? 10_000 });
    const result = await session.exec(code);
    return { content: [{ type: "text", text: render(result) }] };
  }
);

# 决策路径
# 1) 代码是否可信? 否 → 至少 gVisor 级别
# 2) 是否多租户共享节点? 是 → 拒绝普通容器，选 microVM/gVisor
# 3) 是否需要真实系统调用/原生依赖? 否 → WASM 是性价比最高
# 4) 是否需要 GPU/大内存? 是 → 容器 + 独占节点更现实
# 默认建议: 容器起步，多租户场景升级到 gVisor，敏感场景用 microVM

# docker run 关键安全参数
# 用户: 非 root，无 sudo
--user 65534:65534
# 只读根文件系统 + 独立可写 tmpfs
--read-only --tmpfs /tmp:size=64m,mode=1777
# 丢弃全部 capability，只按需加回
--cap-drop ALL --security-opt no-new-privileges
# 禁用特权与设备
--security-opt seccomp=/etc/mcp/seccomp.json
--pids-limit 64
# 资源上限
--memory 512m --memory-swap 512m --cpus 1.0
# 网络: 默认无
--network none

import resource, signal, subprocess

def run_with_limits(code_path: str, timeout_s: int = 10) -> dict:
    def preexec():
        # 地址空间上限 512MB
        resource.setrlimit(resource.RLIMIT_AS, (512 << 20, 512 << 20))
        # CPU 时间上限（秒）
        resource.setrlimit(resource.RLIMIT_CPU, (timeout_s, timeout_s))
        # 进程数上限
        resource.setrlimit(resource.RLIMIT_NPROC, (32, 32))
        # 单文件大小上限 64MB
        resource.setrlimit(resource.RLIMIT_FSIZE, (64 << 20, 64 << 20))

    proc = subprocess.Popen(
        ["python", "-I", "-S", code_path],  # -I 隔离模式: 忽略环境变量与用户 site
        preexec_fn=preexec,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        start_new_session=True,  # 独立进程组，便于整组 kill
    )
    try:
        out, err = proc.communicate(timeout=timeout_s)
        return {"stdout": out, "stderr": err, "exit_code": proc.returncode}
    except subprocess.TimeoutExpired:
        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)  # 杀整个进程组
        return {"stdout": b"", "stderr": b"timeout", "exit_code": -9}

# 从最严到最松
# L0 无网络: --network none，loopback 都不通（推荐默认）
# L1 仅 loopback: 允许本地进程间通信，不出口
# L2 白名单出口: 只允许访问指定域名/IP（包管理器、内部 API）
# L3 全出口: 几乎等于放弃网络隔离（仅可信场景）
# 绝大多数代码执行任务落在 L0/L1；确需装包时才用 L2

# 用 iptables 只放行指定目标，其余 DROP
iptables -P OUTPUT DROP
iptables -A OUTPUT -o lo -j ACCEPT
iptables -A OUTPUT -d 10.0.0.0/8 -p tcp --dport 443 -j ACCEPT   # 内部 API
iptables -A OUTPUT -d 1.2.3.4 -p tcp --dport 443 -j ACCEPT      # 镜像源
# DNS 也要限制，否则可被用作数据外泄通道
iptables -A OUTPUT -d 10.0.0.53 -p udp --dport 53 -j ACCEPT

# 两个常被忽略的出口
# 1) DNS 隧道: 把数据编码进子域名查询外泄 → 限制解析器
# 2) 云元数据端点 169.254.169.254: 可窃取实例凭证
#    → 强制 IMDSv2、iptables DROP 该地址、或用无角色实例
# 网络隔离的漏洞往往不在"能不能连百度"，而在这些侧信道

import os, shutil

ALLOWED_INPUT_ROOTS = ["/data/tasks"]

def stage_inputs(task_id: str, files: list[str]) -> str:
    workdir = f"/workspace/{task_id}"
    os.makedirs(workdir, exist_ok=True)
    for f in files:
        real = os.path.realpath(f)
        # 防路径穿越: 解析后的真实路径必须在允许根目录内
        if not any(real.startswith(r + os.sep) for r in ALLOWED_INPUT_ROOTS):
            raise PermissionError(f"input path not allowed: {f}")
        shutil.copy2(real, workdir)
    return workdir

def collect_outputs(workdir: str) -> list[dict]:
    arts = []
    for name in os.listdir(workdir):
        path = os.path.join(workdir, name)
        # 只回收约定产物，不回收中间文件
        if name.endswith((".png", ".csv", ".json")) and os.path.isfile(path):
            arts.append({"name": name, "size": os.path.getsize(path)})
    return arts

# 常见逃逸手法
# 1) 符号链接指向宿主敏感文件 → realpath 校验 + 禁 symlink 跟随
# 2) /proc/self/root 绕过路径检查 → 限制 /proc 挂载
# 3) 挂载传播把宿主目录带进来 → 挂载用 private 传播模式
# 4) 硬链接 + 可写目录组合 → 只挂载必要目录
# 文件系统隔离的正确姿势是"白名单挂载"，不是"黑名单屏蔽"

MAX_STDOUT = 32 * 1024   # 32 KB
MAX_STDERR = 8 * 1024

def capture(proc_result: dict) -> dict:
    out = proc_result["stdout"].decode("utf-8", "replace")
    err = proc_result["stderr"].decode("utf-8", "replace")
    return {
        "stdout": truncate(out, MAX_STDOUT),
        "stderr": truncate(err, MAX_STDERR),
        "exit_code": proc_result["exit_code"],
        # 明确告知是否被截断，让模型知道结果不完整
        "truncated": len(out) > MAX_STDOUT or len(err) > MAX_STDERR,
    }

def truncate(text: str, limit: int) -> str:
    if len(text) <= limit:
        return text
    head = text[: limit * 2 // 3]
    tail = text[-limit // 3 :]   # 保留头部与尾部，中间省略
    return f"{head}\n... [省略 {len(text) - limit} 字符] ...\n{tail}"

// 会话回收策略
const sessions = new Map<string, Session>();
const IDLE_MS = 5 * 60_000;

setInterval(() => {
  const now = Date.now();
  for (const [id, s] of sessions) {
    if (now - s.lastUsedAt > IDLE_MS) {
      s.kernel.kill();          // 杀解释器
      s.container.dispose();    // 销毁沙箱
      sessions.delete(id);      // 释放索引
    }
  }
}, 30_000);

# 有状态会话的三类污染
# 1) 变量污染: 上一轮定义的变量影响下一轮 → 会话按任务隔离
# 2) 文件污染: 临时文件残留 → 每会话独立 /workspace
# 3) 依赖污染: pip install 装进共享镜像 → 每会话 overlay 文件系统
# 会话 ID 必须绑定到"调用者身份"，不能跨用户复用

# 七层防线
# 1) 认证: 谁在调用（OAuth / mTLS）
# 2) 授权: 该用户能执行代码吗、能用多大配额
# 3) 沙箱: 内核级隔离（gVisor/microVM）
# 4) 资源: cgroup + rlimit
# 5) 网络: 默认无出口
# 6) 文件: 白名单挂载
# 7) 审计: 全量记录
# 任何一层都不是万能的，组合才有意义

{
  "event": "code_execution",
  "ts": "2026-10-04T10:12:33.481+08:00",
  "principal": "user:alice@example.com",
  "session_id": "sess_7f3a91",
  "code_hash": "sha256:9c1f...",
  "code_size": 2048,
  "sandbox": { "runtime": "gvisor", "image": "py-sandbox:3.12" },
  "limits": { "cpu": 1.0, "mem_mb": 512, "timeout_ms": 10000 },
  "network": "none",
  "exit_code": 0,
  "duration_ms": 1832,
  "stdout_bytes": 4096,
  "artifacts": ["plot.png"],
  "decision": "allow"
}

# 记录什么
# 1) 代码全文或哈希（合规要求留存时存全文，否则存哈希+样本）
# 2) 谁、什么时候、从哪个会话
# 3) 沙箱配置（镜像版本、限制参数）——复现问题靠它
# 4) 是否命中可疑模式（外联、大文件写、敏感路径访问）
# 5) 产物清单与去向
# 审计的价值在于"事后能还原"，而不是"事后有日志"