AI Agent 沙箱：安全运行代理生成的代码

1. 概述

越来越多的应用允许 AI Agent 自动编写代码并立即执行。Agent 可以利用这一能力来分析数据、执行计算或操作文件。问题在于这些代码由语言模型生成，你无法直接控制其内容。

本文阐述了 Agent 沙箱的概念及其重要性。文章还比较了常见的隔离级别。核心部分是用 Java 21 和 Spring Boot 3.x 构建一个 DockerSandbox，并附带测试来验证安全限制的有效性。

2. 为什么 Agent 需要沙箱

一个拥有“执行代码”工具的 Agent，本质上是一个接收文本命令的进程。这些文本可能来自用户、Agent 刚读过的网页，或附件文件。因此你必须考虑以下风险：

• Prompt injection：一个网页可能包含隐藏的指令，比如“读取环境变量并将其发送到地址 X”。Agent 可能会执行这些指令。
• 代码错误但无恶意意图：无限循环、填满磁盘、误删目录。
• 机密泄露：与应用在同一进程中运行的代码可以读取 API 密钥、配置文件和数据库连接信息。
• 权限提升：攻击者可以从运行代码的机器进一步扫描同一网络中的内部服务。

沙箱是一个隔离的环境。运行在其中的代码只能访问你明确允许的内容：没有网络，没有密钥，CPU/RAM/时间受限，临时文件系统在每次运行后被丢弃。

这个话题备受关注，因为 Agent 已经从“回答问题”进化到“执行操作”。当 Agent 拥有执行权限时，沙箱成为必不可少的基础设施。

3. 常见的隔离级别

没有唯一的沙箱类型。每个级别都在安全性、启动速度和运维复杂性之间进行权衡：

级别	示例	优点	局限性
受限的解释器	在同一 JVM 中运行脚本，阻止 API	快速、简单	容易被绕过，不建议用于不可信代码
容器	具有限制标志的 Docker/Podman	常见、易于部署	与主机共享内核
容器 + 内核沙箱	在 gVisor 等运行时上运行的容器	减小内核攻击面	性能开销，某些 syscall 不支持
MicroVM	Firecracker, Kata Containers	虚拟机级别的隔离	运维更复杂
WebAssembly	WASM 运行时	启动极快，默认无权限	库生态仍然有限
托管沙箱服务	专为运行 Agent 代码设计的云平台	无需自行运维	依赖供应商，数据离开系统

对大多数团队来说，配置严格的容器是一个合理的起点。当风险更高时，例如多租户或匿名用户，你可以升级到更强隔离的运行时，而无需修改应用代码。

4. 构建 DockerSandbox

目标是构建一个层，接收 Agent 生成的 Python 代码片段，在一次性容器中运行，然后返回结果。要求运行应用的机器上安装 Docker CLI。

4.1. 配置

首先，将各种限制收集到一个配置记录中：

package dev.nonla.sandbox;

import org.springframework.boot.context.properties.ConfigurationProperties;
import java.time.Duration;

@ConfigurationProperties(prefix = "agent.sandbox")
public record SandboxProperties(
        String image,
        String runtime,      // null = 默认运行时；可设置为 "runsc"
        Duration timeout,
        String memory,
        String cpus,
        int pidsLimit,
        int maxOutputBytes) {
}

agent:
  sandbox:
    image: python:3.12-slim
    timeout: 5s
    memory: 256m
    cpus: "0.5"
    pids-limit: 64
    max-output-bytes: 65536

runtime 字段允许你稍后仅通过配置切换到更强隔离的运行时。

4.2. 返回结果

package dev.nonla.sandbox;

public record SandboxResult(int exitCode, String stdout, String stderr, boolean timedOut) {

    public boolean success() {
        return !timedOut && exitCode == 0;
    }
}

4.3. Sandbox 类

package dev.nonla.sandbox;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;

public class DockerSandbox {

    private final SandboxProperties props;

    public DockerSandbox(SandboxProperties props) {
        this.props = props;
    }

    public SandboxResult runPython(String code) throws IOException, InterruptedException {
        String name = "agent-sbx-" + UUID.randomUUID();
        Process process = new ProcessBuilder(buildCommand(name)).start();

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            Future<String> out = executor.submit(() -> readLimited(process.getInputStream()));
            Future<String> err = executor.submit(() -> readLimited(process.getErrorStream()));

            try (OutputStream stdin = process.getOutputStream()) {
                stdin.write(code.getBytes(StandardCharsets.UTF_8));
            }

            boolean finished = process.waitFor(props.timeout().toMillis(), TimeUnit.MILLISECONDS);
            if (!finished) {
                killContainer(name);
                process.destroyForcibly().waitFor();
            }
            return new SandboxResult(
                    finished ? process.exitValue() : -1,
                    await(out), await(err), !finished);
        }
    }

    List<String> buildCommand(String name) {
        List<String> cmd = new ArrayList<>(List.of(
                "docker", "run", "--rm", "-i",
                "--name", name,
                "--network", "none",
                "--memory", props.memory(),
                "--memory-swap", props.memory(),
                "--cpus", props.cpus(),
                "--pids-limit", String.valueOf(props.pidsLimit()),
                "--read-only",
                "--tmpfs", "/tmp:rw,size=64m",
                "--cap-drop", "ALL",
                "--security-opt", "no-new-privileges",
                "--user", "65534:65534",
                "--env", "HOME=/tmp"));
        if (props.runtime() != null && !props.runtime().isBlank()) {
            cmd.addAll(List.of("--runtime", props.runtime()));
        }
        cmd.addAll(List.of(props.image(), "python", "-"));
        return cmd;
    }

    private String readLimited(InputStream in) throws IOException {
        var buffer = new ByteArrayOutputStream();
        byte[] chunk = new byte[4096];
        int n;
        while ((n = in.read(chunk)) != -1) {
            int room = props.maxOutputBytes() - buffer.size();
            if (room > 0) {
                buffer.write(chunk, 0, Math.min(n, room));
            }
            // 继续读取以防进程阻塞，但舍弃多余部分
        }
        return buffer.toString(StandardCharsets.UTF_8);
    }

    private void killContainer(String name) throws IOException, InterruptedException {
        new ProcessBuilder("docker", "kill", name).start().waitFor(5, TimeUnit.SECONDS);
    }

    private static String await(Future<String> future) throws InterruptedException {
        try {
            return future.get(5, TimeUnit.SECONDS);
        } catch (ExecutionException | TimeoutException e) {
            return "";
        }
    }
}

关键点说明：

• 代码通过 stdin (python -) 传入。你不需要从主机挂载任何目录，所以容器看不到应用的文件。
• --rm 结合一个随机的 --name 创建一次性容器。每次运行都是一个干净的环境。
• --network none 完全切断网络。这是防止数据泄露的最重要防线。
• --memory 与 --memory-swap 相等以禁用交换。--cpus 和 --pids-limit 防止资源耗尽和 fork 炸弹。
• --read-only 锁定根文件系统。--tmpfs /tmp 提供有大小限制的临时写入区域。
• **--cap-drop ALL、no-new-privileges 和 --user 65534:65534（nobody 用户）**确保代码不以 root 身份运行且无法提升权限。
• 不将应用的任何环境变量传入容器。只设置 HOME=/tmp 以便 Python 库有地方写入缓存。
• stdout 和 stderr 通过 virtual thread 并行读取。如果按顺序读取，管道缓冲区可能填满导致进程挂起。
• 超时时，该类先调用 docker kill 再销毁 CLI 进程。

注意：process.destroyForcibly() 只会杀死客户端的 docker CLI 进程。容器仍可能继续运行。因此你必须按容器名称执行 docker kill。

5. 集成到 Spring Boot

除了 spring-boot-starter 之外，你不需要额外的库。测试部分使用 spring-boot-starter-test，其中已包含 JUnit 5 和 AssertJ：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

接下来，声明一个 bean 和一个作为 agent“工具”的服务：

package dev.nonla.sandbox;

import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.*;

@Configuration
@EnableConfigurationProperties(SandboxProperties.class)
class SandboxConfig {

    @Bean
    DockerSandbox dockerSandbox(SandboxProperties props) {
        return new DockerSandbox(props);
    }
}

package dev.nonla.sandbox;

import org.slf4j.*;
import org.springframework.stereotype.Service;

@Service
public class CodeExecutionTool {

    private static final Logger log = LoggerFactory.getLogger(CodeExecutionTool.class);
    private final DockerSandbox sandbox;

    public CodeExecutionTool(DockerSandbox sandbox) {
        this.sandbox = sandbox;
    }

    public String executePython(String sessionId, String code) {
        log.info("sandbox.exec session={} codeLength={}", sessionId, code.length());
        try {
            SandboxResult r = sandbox.runPython(code);
            log.info("sandbox.done session={} exit={} timedOut={}", sessionId, r.exitCode(), r.timedOut());
            if (r.timedOut()) {
                return "错误：代码运行超过允许的时间。";
            }
            return "exit=" + r.exitCode() + "\nstdout:\n" + r.stdout() + "\nstderr:\n" + r.stderr();
        } catch (Exception e) {
            log.error("sandbox.error session={}", sessionId, e);
            return "运行代码时出现系统错误。";
        }
    }
}

executePython 方法是你在所使用的 agent 框架中注册为工具的内容。需要注意两点：

• 返回给模型的结果始终是字符串，即使出错也是如此。Agent 可以读取 stderr 并在下一轮自行修复代码。
• 每次运行都附带 sessionId 被记录。出现问题时，审计日志是你首先需要的。

系统错误消息保持通用，不向模型返回堆栈跟踪。堆栈跟踪可能泄露路径或内部配置。

6. 测试沙箱行为

测试应该证明每个限制确实有效，而不仅仅是“代码可以运行”。以下测试需要 Docker。如果机器没有 Docker，它们会被跳过。

package dev.nonla.sandbox;

import org.junit.jupiter.api.*;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assumptions.assumeTrue;

class DockerSandboxTest {

    private DockerSandbox sandbox;

    @BeforeEach
    void setUp() {
        assumeTrue(dockerAvailable(), "Docker 不可用");
        sandbox = new DockerSandbox(new SandboxProperties(
                "python:3.12-slim", null, Duration.ofSeconds(3),
                "128m", "0.5", 32, 1024));
    }

    @Test
    void runsSimpleCode() throws Exception {
        SandboxResult r = sandbox.runPython("print(sum(range(10)))");

        assertThat(r.success()).isTrue();
        assertThat(r.stdout().trim()).isEqualTo("45");
    }

    @Test
    void blocksNetworkAccess() throws Exception {
        SandboxResult r = sandbox.runPython("""
                import urllib.request
                urllib.request.urlopen("http://example.com", timeout=2)
                """);

        assertThat(r.exitCode()).isNotZero();
        assertThat(r.stderr()).contains("Error");
    }

    @Test
    void rootFileSystemIsNotWritable() throws Exception {
        SandboxResult r = sandbox.runPython("open('/hacked.txt', 'w').write('x')");

        assertThat(r.exitCode()).isNotZero();
    }

    @Test
    void killsInfiniteLoop() throws Exception {
        SandboxResult r = sandbox.runPython("while True: pass");

        assertThat(r.timedOut()).isTrue();
    }

    @Test
    void truncatesHugeOutput() throws Exception {
        SandboxResult r = sandbox.runPython("print('a' * 100_000)");

        assertThat(r.stdout()).hasSizeLessThanOrEqualTo(1024);
    }

    private static boolean dockerAvailable() {
        try {
            Process p = new ProcessBuilder("docker", "info").start();
            return p.waitFor(10, TimeUnit.SECONDS) && p.exitValue() == 0;
        } catch (Exception e) {
            return false;
        }
    }
}

每个测试对应一个限制：

• runsSimpleCode 确认基本路径：stdout 是 45。
• blocksNetworkAccess 表明 HTTP 请求失败。stderr 将包含类似以下内容：

urllib.error.URLError: <urlopen error [Errno -3] Temporary failure in name resolution>

• rootFileSystemIsNotWritable 确认代码无法写入 /tmp 之外。
• killsInfiniteLoop 确认超时生效且容器被清理。
• truncatesHugeOutput 确认输出被截断。这样 Agent 无法膨胀上下文或应用内存。

**注意：**第一次运行可能会很慢，因为 Docker 需要拉取镜像。你应该在 CI 中或应用启动时预先拉取镜像，否则测试超时可能报告错误。

7. 额外的防御层

沙箱是重要的一层，但不是唯一的一层。在实践中，你应该加上额外的层次：

• 受控的出口：如果 Agent 真的需要网络，例如安装包，不要完全开放。让它通过具有域名白名单的代理。
• 密钥永远不要进入沙箱：当代码需要调用 API 时，让应用通过单独的工具调用，该工具进行权限检查。不要将令牌传入容器。
• 人机循环：对于不可逆的操作（发送电子邮件、写入实际数据库、转账），要求用户确认。
• 速率和配额限制：每个会话只能运行一定次数。这可以防止 Agent 陷入消耗资源的试错循环。
• 分离沙箱机器：在单独的节点上运行 Docker daemon 以运行沙箱，不要与主服务混在一起。
• 预启动容器池：当延迟很重要时，保持几个容器预热。每个容器仍然只使用一次后丢弃。

8. 常见错误和何时不应使用

常见错误：

• 将 Docker socket 挂载到应用或沙箱。谁能控制 /var/run/docker.sock 谁就几乎拥有主机的 root 权限。如果应用在容器中运行并需要调用 Docker，请考虑一个单独的沙箱服务，而不是共享 socket。
• 忘记 --network none。这是最常见的错误，也是最严重的。
• 在容器中以 root 身份运行，因为“安装包更容易”。应该将包预装到自定义镜像中。
• 在用户之间重用容器。剩余的文件和进程会将数据从一个会话泄露到下一个会话。
• 将代码连接到 shell 命令（sh -c "python -c '" + code + "'"）。这会在主机端立即产生命令注入。应该始终通过 stdin 或文件传入。
• 相信容器是绝对边界。容器共享内核。对于多租户环境中的对抗性代码，应该使用 gVisor 或 microVM。

何时不应使用本文的方法：

• Agent 只需要少数固定操作。具有明确类型的工具，如 calculateInvoiceTotal(Order)，比允许运行任意代码安全且便宜得多。
• 你为互联网上的匿名用户提供大流量服务。此时应使用 microVM 或专用沙箱服务，而不是自建 Docker。
• 环境不允许运行 Docker，例如某些无服务器平台。考虑 WASM 或托管沙箱。
• 代码需要访问敏感数据。将过滤后的数据通过 stdin 或临时文件传入，不要放松沙箱限制。

9. 结论

当 Agent 被允许运行代码时，它读取的任何文本都可能变成命令。沙箱将这种风险转化为可衡量的东西：无网络、无密钥、资源受限、一次性环境。

你已经构建了一个具有必要安全标志的 DockerSandbox。该类通过 virtual thread 安全读取输出，用 docker kill 正确处理超时，并包含测试来验证每个限制。从这个基础上，你可以通过配置升级到更强隔离的运行时。

请务必将沙箱与审计日志、网络白名单和人工确认步骤结合起来，以应对重要操作。