FastGPT 沙箱隔离方案分析

一、总体架构

FastGPT 的代码沙箱采用进程池 + 语言级安全加固的混合隔离方案，不依赖 Docker/VM 等重量级虚拟化技术，而是通过 Node.js worker 和 Python worker 的进程隔离 + 代码层安全强化实现。

HTTP Request → Hono Server → Process Pool → Worker (long-lived) → Result
                                ↓
                        ┌──────────────┐
                        │  JS Workers  │  node worker.js (×N)
                        │  Py Workers  │  python3 worker.py (×N)
                        └──────────────┘
                        stdin: JSON task → stdout: JSON result

关键特性：

  • 预热长驻 worker 进程（默认 20 个 JS + 20 个 Python）
  • 通过 stdin/stdout JSON 行协议通信
  • 进程崩溃后自动 respawn
  • 每种语言独立的进程池

二、沙箱技术架构

2.1 隔离技术选型

技术方案 FastGPT 采用
容器级隔离 (Docker/gVisor/Firecracker) 否
VM 级隔离 否
进程级隔离 + 语言安全加固 是

实现路径：

  • worker.ts (JS): Node.js 子进程，通过 Function 构造器冻结、原型链遮蔽、模块白名单实现
  • worker.py (Python): Python3 子进程，通过 __import__ 拦截、AST 静态检查、execglobals 隔离实现

2.2 进程池管理

核心文件：src/pool/base-process-pool.ts

BaseProcessPool
  ├── ProcessPool (JS worker: tsx/node worker.ts)
  └── PythonProcessPool (Python worker: python3 worker.py)

生命周期管理：

  • 启动时预热 N 个 worker（默认 20）
  • Worker 通过 sh -c "exec tsx worker.ts" 启动
  • ping/pong 健康检查（30s 间隔，5s 超时）
  • Worker 崩溃自动 kill 并 respawn
  • 请求超时后 kill 并 respawn
  • 池满自动排队（acquire/release 模式）

资源初始化：

// worker 启动时接收 init 消息
{ type: "init", allowedModules: [...], requestLimits: {...} }
// 回复
{ type: "ready" }
// 之后每条任务消息
{ code: "...", variables: {}, timeoutMs: 10000 }
// 返回结果
{ success: true, data: { codeReturn, log } }

三、网络隔离方案

3.1 统一收口

所有网络请求必须通过 SystemHelper.httpRequest() 收口，禁止直接使用 fetch/XMLHttpRequest/WebSocket。

JS 端 (worker.ts):

lockGlobal('fetch', undefined);
lockGlobal('XMLHttpRequest', undefined);
lockGlobal('WebSocket', undefined);

const SystemHelper = {
  async httpRequest(url, opts) {
    // 1. 先检查 URL 是否指向内部地址
    if (await isInternalAddress(url)) throw new Error('Request to private network not allowed');

    // 2. DNS 解析获取 IP
    const ips = await dnsResolve(parsed.hostname);

    // 3. DNS rebinding TOCTOU 防护：对 IP 再次校验
    if (ips.some(ip => isInternalResolvedIP(ip))) throw new Error('...');

    // 4. 协议白名单
    if (!['http:', 'https:'].includes(parsed.protocol)) throw new Error('Protocol not allowed');

    // 5. 请求限制
    if (++requestCount > maxRequests) throw new Error('Request limit exceeded');

    // 6. DNS-pinned HTTP 连接（强制连接到预解析的 IP）
    http.request({ hostname: resolvedIP, servername: hostname })
  }
};

Python 端 (worker.py):

class _PinnedHTTPConnection(_http_client.HTTPConnection):
    """强制连接到预解析的 IP，防止 DNS rebinding"""
    def connect(self):
        self.sock = _socket.create_connection(
            (self._pinned_ip or self.host, ...), self.timeout
        )

3.2 内网 IP 黑名单

// worker.ts
const _BLOCKED_CIDRS = [
  '10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16',
  '169.254.0.0/16', '127.0.0.0/8', '0.0.0.0/8',
  '::1/128', '::', 'fc00::/7', 'fe80::/10'
];
// 云元数据 IP
'169.254.169.254' (AWS), '100.100.100.200' (阿里云), 'fd00:ec2::254' (AWS IPv6)

3.3 请求限制

const REQUEST_LIMITS = {
  maxRequests: 30,           // 单次执行最大请求数
  timeoutMs: 60000,         // 单次请求超时
  maxResponseSize: 10MB,    // 最大响应体
  maxRequestBodySize: 5MB,  // 最大请求体
  allowedProtocols: ['http:', 'https:']
};

四、资源限制

4.1 内存限制

跨平台 RSS 轮询监控（base-process-pool.ts）:

const RSS_POLL_INTERVAL = 500; // 500ms 轮询
const limitMB = env.SANDBOX_MAX_MEMORY_MB + RUNTIME_MEMORY_OVERHEAD_MB;

rssTimer = setInterval(async () => {
  const rss = await this.getWorkerRSSMB(worker.proc.pid!);
  if (rss > limitMB) {
    this.killAndRespawn(worker);
    settle({ success: false, message: `Memory limit exceeded (RSS: ${rss}MB)` });
  }
}, RSS_POLL_INTERVAL);

Linux: 读取 /proc/{pid}/status 中 VmRSS（单位 kB）
macOS/其他: ps -o rss= -p {pid}（单位 kB）

默认限制：256MB 用户代码 + 50MB 运行时开销 = 306MB 实际进程限制

4.2 CPU / 超时限制

JS 端：

// worker.ts
const timeoutPromise = new Promise((_, reject) => {
  timer = setTimeout(() => reject(new Error(`Script execution timed out after ${timeoutMs}ms`)), timeoutMs || 10000);
});
const result = await Promise.race([resultPromise, timeoutPromise]);

Python 端：

# worker.py - 双重 SIGALRM 机制
def _timeout_handler(signum, frame):
    global _timeout_stage
    _timeout_stage += 1
    if _timeout_stage >= 2:
        raise SystemExit("Script execution timed out (forced)")
    signal.alarm(1)  # 1s 后的兜底 alarm
    raise TimeoutError("Script execution timed out")

signal.signal(signal.SIGALRM, _timeout_handler)
signal.alarm(timeout_s)

默认超时：60s（可配置 1s - 600s）

4.3 日志输出限制

const MAX_LOG_SIZE = 1024 * 1024; // 1MB

五、文件系统隔离

5.1 JS 端

通过 AST 分析拦截动态 import：

// worker.ts
function assertNoDynamicImport(code: string): void {
  const ast = parse(code, { ecmaVersion: 'latest', sourceType: 'script' });
  walk(ast, {
    ImportExpression() {
      throw new Error('Dynamic import() is not allowed in sandbox.');
    }
  });
}

require 白名单：只有 lodash, dayjs, moment, uuid, crypto-js, qs, url, querystring 可用

5.2 Python 端

open() 拦截：

# worker.py
def _restricted_open(*args, **kwargs):
    global _open_guard
    if _open_guard:
        return _original_open(*args, **kwargs)
    _open_guard = True
    try:
        stack = _tb.extract_stack()
    finally:
        _open_guard = False
    if len(stack) >= 2:
        caller_fn = stack[-2].filename or ''
        # 用户代码（<string>）不允许直接 open
        if caller_fn in ('<string>', '<test>', '<module>'):
            raise PermissionError("File system access is not allowed in sandbox")
    return _original_open(*args, **kwargs)

_safe_builtins['open'] = _restricted_open

预检黑名单模块（即使在 stdlib 中也禁止）：

_DANGEROUS_STDLIB = frozenset({
    'os', 'subprocess', 'shutil', 'pathlib', 'glob', 'tempfile',
    'multiprocessing', 'threading', 'concurrent', 'ctypes', 'importlib',
    'runpy', 'code', 'codeop', 'compileall', 'socket', 'http', 'urllib',
    'ftplib', 'smtplib', 'poplib', 'imaplib', 'xmlrpc', 'socketserver',
    'ssl', 'asyncio', 'selectors', 'select', 'signal', 'resource', 'pty',
    'termios', 'tty', 'fcntl', 'mmap', 'dbm', 'sqlite3', 'shelve',
    'webbrowser', 'turtle', 'tkinter', 'idlelib', 'venv', 'ensurepip',
    'pip', 'site', 'gc', 'sys', 'builtins', 'marshal', 'pickle'
})

六、通信机制

6.1 IPC 协议

主进程                     Worker 进程
   |--- stdin: JSON init --->|
   |<-- stdout: {type:ready} -|
   |                           |
   |--- stdin: JSON task ----->|
   |<-- stdout: JSON result --|
   |                           |
   |--- stdin: {type:ping} --->|
   |<-- stdout: {type:pong} --|

任务消息格式：

{
  "code": "async function main(variables) { return { sum: variables.a + variables.b }; }",
  "variables": { "a": 1, "b": 2 },
  "timeoutMs": 10000
}

结果消息格式：

{
  "success": true,
  "data": {
    "codeReturn": { "sum": 3 },
    "log": "console output here"
  }
}

6.2 安全通信设计

  • Worker stdin/stdout 通过 pipe 连接，不暴露给外部
  • JSON 协议，无序列化漏洞
  • 每次执行后清理 require cache，防止模块污染

七、安全性设计（防逃逸）

7.1 JS 端防护链

攻击向量 防护措施
constructor.constructor 逃逸 Function.prototype.constructor 被替换为 SafeFunction
eval / new Function 替换为抛出异常的 stub
原型链污染 Object.getPrototypeOf 拦截，返回空对象
__proto__ / setPrototypeOf 返回 false 或返回空对象
Reflect.construct 拦截 Function 构造器
AsyncFunction/GeneratorFunction 构造器 替换原型.constructor 为 SafeFunction
process.binding('fs') 启动时即删除该方法
process.kill/exit 启动后删除（等白名单模块加载完再删）
process.env 敏感变量 清理含 secret/password/token 等关键字的变量
动态 import AST 分析拦截 ImportExpression
require 白名单外模块 safeRequire 抛出异常

核心防御代码（worker.ts）:

// 危险全局对象通过函数参数遮蔽（用户代码通过参数获取，而非 globalThis）
const userFn = new Function(
  'require', 'console', 'SystemHelper', 'countToken', ..., 'globalThis', 'process', 'Bun',
  '"use strict";\n' + code + '\nreturn main;'
);

// process 对象被替换为冻结的最小化对象
const _sandboxProcess = Object.freeze({
  env: Object.freeze({}),
  cwd: () => '/sandbox',
  version: process.version,
  platform: process.platform
});

// 模块导出只读包装
const readonlyView = new Proxy(obj, {
  set() { throw new Error('Sandbox module exports are read-only'); },
  deleteProperty() { throw new Error('Sandbox module exports are read-only'); }
});

7.2 Python 端防护链

攻击向量 防护措施
__import__ 逃逸 替换 builtins.import，基于调用栈帧检测用户代码
exec/eval 内 import 通过调用栈检测拦截用户代码
__subclasses__ 逃逸 AST 分析 + 替换 object 为 SafeObject
builtins 篡改 _BuiltinsProxy 代理对象，静默忽略覆盖
模块状态污染 每次执行前快照、执行后恢复
open() 文件读写 基于调用栈帧检测用户代码拦截

核心防御代码（worker.py）:

# __import__ 拦截 - 基于调用栈帧检测
def _safe_import(name, *args, **kwargs):
    global _import_guard
    if _import_guard:
        return _original_import(name, *args, **kwargs)

    # 白名单检查
    top_level = name.split('.')[0]
    if top_level in _STDLIB_MODULES and top_level not in _DANGEROUS_STDLIB:
        return _original_import(name, *args, **kwargs)
    if top_level in _allowed_modules:
        return _original_import(name, *args, **kwargs)

    # 检查调用栈
    _import_guard = True
    try:
        stack = _tb.extract_stack()
    finally:
        _import_guard = False

    if len(stack) >= 2:
        caller_fn = stack[-2].filename or ''
        if caller_fn in ('<string>', '<test>', '<module>'):
            raise ImportError(f"Module '{name}' is not in the allowlist.")

    return _original_import(name, *args, **kwargs)

# object.__subclasses__ 屏蔽
class _SafeObject(object):
    __subclasses__ = None
_safe_builtins['object'] = _SafeObject

7.3 内网访问防护

双重检查机制（DNS rebinding TOCTOU 防护）:

// 1. URL 预检 - 阻止内网主机名
if (await isInternalAddress(url)) throw new Error('...');

// 2. DNS 解析后 IP 检查 - 防止 DNS rebinding
const ips = await dnsResolve(parsed.hostname);
if (ips.some(ip => isInternalResolvedIP(ip))) throw new Error('...');

// 3. DNS-pinned 连接 - 强制连接到预解析的 IP
lib.request({ hostname: resolvedIP, servername: hostname })

被阻止的主机名：

  • localhost
  • metadata.google.internal (GCP)
  • kubernetes.default.svc / kubernetes (K8s)
  • 云元数据域名等

支持的 IP 绕过变体检测：

  • 十进制: 2130706433 (=127.0.0.1)
  • 十六进制: 0x7f000001
  • 八进制: 0177.0.0.01
  • IPv4-mapped IPv6: ::ffff:127.0.0.1

八、部署配置

8.1 Docker 镜像（projects/code-sandbox/Dockerfile）

FROM node:24-alpine AS builder
RUN npm install -g pnpm@10.33.2
COPY .../pnpm-lock.yaml .../packages/ .../projects/code-sandbox/ ./
RUN pnpm install --frozen-lockfile --ignore-scripts
RUN pnpm --filter @fastgpt-sdk/otel --filter @fastgpt-sdk/storage build
RUN cd /app/projects/code-sandbox && pnpm build

FROM node:24-alpine AS runner
# 安装 Python 和依赖
RUN apk add python3 py3-pip libffi util-linux gcc g++ musl-dev python3-dev libffi-dev
RUN pip3 install -r requirements.txt  # numpy, pandas, matplotlib

# 非 root 用户运行
RUN addgroup -S sandbox && adduser -S sandbox -G sandbox && chown -R sandbox:sandbox /app
USER sandbox

EXPOSE 3000
CMD ["node", "/app/code-sandbox/index.js"]

8.2 关键环境变量

SANDBOX_PORT=3000                # 服务端口
SANDBOX_TOKEN=                   # Bearer token 认证（生产必填）
SANDBOX_POOL_SIZE=20             # 每种语言 worker 数

SANDBOX_MAX_TIMEOUT=60000        # 超时上限(ms)
SANDBOX_MAX_MEMORY_MB=256        # 内存上限(MB)

CHECK_INTERNAL_IP=false          # 是否检查私网 IP
SANDBOX_REQUEST_MAX_COUNT=30     # 单次执行最大请求数
SANDBOX_REQUEST_TIMEOUT=60000    # 单次 HTTP 请求超时(ms)
SANDBOX_REQUEST_MAX_RESPONSE_MB=10
SANDBOX_REQUEST_MAX_BODY_MB=5

SANDBOX_JS_ALLOWED_MODULES=lodash,dayjs,moment,uuid,crypto-js,qs,url,querystring
SANDBOX_PYTHON_ALLOWED_MODULES=math,random,datetime,json,hashlib,numpy,pandas,matplotlib,...

九、关键组件一览

组件 文件 职责
服务入口 src/index.ts Hono HTTP Server，路由 /sandbox/js 和 /sandbox/python
进程池基类 src/pool/base-process-pool.ts Worker 生命周期管理、健康检查、超时/内存监控
JS 进程池 src/pool/process-pool.ts JS worker 池配置（tsx/node + worker.ts）
JS Worker src/pool/worker.ts JS 代码执行器，含完整安全加固
Python 进程池 src/pool/python-process-pool.ts Python worker 池配置（python3 + worker.py）
Python Worker src/pool/worker.py Python 代码执行器，含完整安全加固
环境变量 src/env.ts 配置加载和校验
网络检查 src/utils/ipCheck.util.ts 内网 IP/URL 检测
信号量 src/utils/semaphore.ts 并发控制

十、隔离机制原理总结

FastGPT 的沙箱采用多层防御策略：

┌─────────────────────────────────────────────────────────────┐
│                    第一层：进程隔离                          │
│  每个 worker 是独立的 Node.js/Python 子进程                 │
│  崩溃互不影响，超时/kill 后自动 respawn                      │
├─────────────────────────────────────────────────────────────┤
│                    第二层：语言级安全                        │
│  JS: Function 构造器冻结 + 原型链遮蔽 + require 白名单       │
│  Py: __import__ 拦截 + open() 拦截 + 模块状态快照/恢复        │
├─────────────────────────────────────────────────────────────┤
│                    第三层：网络隔离                          │
│  所有请求必须走 SystemHelper.httpRequest()                   │
│  DNS-pinned 连接 + 内网 IP 黑名单 + SSRF 防护                │
├─────────────────────────────────────────────────────────────┤
│                    第四层：资源限制                          │
│  RSS 内存轮询监控 + Promise.race/SIGALRM 超时               │
│  请求数/请求体/响应体大小限制                                │
├─────────────────────────────────────────────────────────────┤
│                    第五层：通信协议                          │
│  stdin/stdout JSON 行协议，无序列化漏洞                       │
│  每次执行后清理 require cache                               │
└─────────────────────────────────────────────────────────────┘

不依赖容器/VM 的原因：

  • 轻量化，启动快（进程池预热）
  • 满足代码执行隔离需求
  • 部署简单，不依赖 Kubernetes 等基础设施
# 开发语言
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

当 AI 修图师装上“万能插头“：MCP 如何改变修图工作流

avatar AI编程社区
cover

2026年8月开发者实战指南：ChatGPT Plus / Pro + Codex 如何协同完成需求分析、编码、测试、调试与代码审查

avatar AI编程社区
cover

投资一台B300，部署Claude Opus 5多久才能回本？

avatar AI编程社区