代码智能体不应直接获得文件系统、Shell 和网络的完整能力，而应通过一个受控执行层调用工具。本文用 Node.js 实现可运行的最小版本，并说明应用级策略能够防住什么，以及何时必须升级到容器或系统级沙箱。

先定义威胁模型

当模型只能生成文本时，错误通常停留在回答层；一旦接入工具，提示注入、模型误判或恶意输入就可能变成真实副作用。例如读取工作目录外的密钥、执行带有破坏性的命令，或者把源码发送到外部地址。

这里不尝试判断模型“是否可信”，而是假设每次工具参数都可能有问题，并在模型与操作系统之间增加执行层：

  1. 默认拒绝，没有明确策略的工具不能运行。
  2. 参数必须结构化，命令执行不经过 Shell 拼接。
  3. 权限按工具分别授予，而不是给智能体一个万能终端。
  4. 高风险动作执行前确认，执行前后都留下审计记录。
  5. 输出、运行时间和响应体大小都要有限制。

本文示例只提供只读文件工具。文件写入、删除和重命名应设计成独立工具，分别配置目录、扩展名、覆盖规则和确认流程，不要复用一个任意文件操作接口。

三类工具的策略

执行层应先把自然语言意图收敛为少量、稳定的工具协议，再逐项校验参数。

工具默认策略主要限制
read_file可自动执行仅工作目录内，只读指定扩展名，限制文件大小
run_command人工确认命令和参数组合白名单，禁用 Shell，限制时间与输出
http_get人工确认仅 HTTPS 和指定主机，禁止自动重定向，限制响应体

命令白名单不能只检查前缀。例如允许 git 并不等于所有 git 子命令都安全；更稳妥的方式是把“查看状态”“读取差异”定义成不同动作，并为每个动作固定参数形状。网络白名单也要检查协议、主机、端口和重定向，否则允许的站点可能把请求转向未授权地址。

实现最小执行层

下面示例只使用 Node.js 内置 API，要求 Node.js 20 或更高版本。保存为 index.mjs 即可运行。

import { appendFile, mkdir, readFile, realpath, stat } from 'node:fs/promises';
import { createHash, randomUUID } from 'node:crypto';
import { extname, isAbsolute, relative, resolve } from 'node:path';
import { spawn } from 'node:child_process';
import { createInterface } from 'node:readline/promises';

const ROOT = resolve(process.env.AGENT_ROOT ?? './workspace');
const AUDIT_FILE = resolve(process.env.AUDIT_FILE ?? './audit.ndjson');
const MAX_BYTES = 1024 * 1024;
const TIMEOUT_MS = 5000;
const ALLOWED_EXTENSIONS = new Set(['.txt', '.md', '.json']);
const ALLOWED_HOSTS = new Set(['example.com']);

await mkdir(ROOT, { recursive: true });
const REAL_ROOT = await realpath(ROOT);

function assertInsideRoot(target) {
  const rel = relative(REAL_ROOT, target);
  if (rel === '..' || rel.startsWith(`..${process.platform === 'win32' ? '\\' : '/'}`) || isAbsolute(rel)) {
    throw new Error('路径超出工作目录');
  }
}

async function audit(event) {
  await appendFile(AUDIT_FILE, `${JSON.stringify({
    time: new Date().toISOString(),
    ...event
  })}\n`, { encoding: 'utf8', mode: 0o600 });
}

function requestView(tool, input) {
  if (tool === 'http_get') {
    const url = new URL(input.url);
    return { url: `${url.origin}${url.pathname}` };
  }
  return input;
}

function resultView(result) {
  const data = JSON.stringify(result);
  return {
    bytes: Buffer.byteLength(data),
    sha256: createHash('sha256').update(data).digest('hex')
  };
}

async function readFileTool(input) {
  if (typeof input.path !== 'string') throw new Error('path 必须是字符串');
  const candidate = resolve(REAL_ROOT, input.path);
  assertInsideRoot(candidate);

  const target = await realpath(candidate);
  assertInsideRoot(target);
  if (!ALLOWED_EXTENSIONS.has(extname(target).toLowerCase())) {
    throw new Error('文件扩展名不在白名单');
  }

  const info = await stat(target);
  if (!info.isFile() || info.size > MAX_BYTES) throw new Error('文件类型或大小不允许');
  return { path: relative(REAL_ROOT, target), content: await readFile(target, 'utf8') };
}

async function runCommandTool(input) {
  const allowed = input.command === 'node-version'
    && Array.isArray(input.args)
    && input.args.length === 0;
  if (!allowed) throw new Error('命令或参数不在白名单');

  return await new Promise((resolvePromise, reject) => {
    const child = spawn(process.execPath, ['--version'], {
      cwd: REAL_ROOT,
      shell: false,
      stdio: ['ignore', 'pipe', 'pipe'],
      env: { PATH: process.env.PATH ?? '' }
    });

    const stdout = [];
    const stderr = [];
    let size = 0;
    let reason = '';
    const collect = (list, chunk) => {
      size += chunk.length;
      if (size > MAX_BYTES) {
        reason = '命令输出超过限制';
        child.kill('SIGKILL');
      } else {
        list.push(chunk);
      }
    };

    child.stdout.on('data', chunk => collect(stdout, chunk));
    child.stderr.on('data', chunk => collect(stderr, chunk));
    child.on('error', reject);

    const timer = setTimeout(() => {
      reason = '命令执行超时';
      child.kill('SIGKILL');
    }, TIMEOUT_MS);

    child.on('close', code => {
      clearTimeout(timer);
      if (reason) return reject(new Error(reason));
      resolvePromise({
        exitCode: code,
        stdout: Buffer.concat(stdout).toString('utf8'),
        stderr: Buffer.concat(stderr).toString('utf8')
      });
    });
  });
}

async function httpGetTool(input) {
  if (typeof input.url !== 'string') throw new Error('url 必须是字符串');
  const url = new URL(input.url);
  if (url.protocol !== 'https:' || url.username || url.password) throw new Error('只允许无凭据 HTTPS URL');
  if (!ALLOWED_HOSTS.has(url.hostname) || (url.port && url.port !== '443')) {
    throw new Error('目标主机或端口不在白名单');
  }

  const response = await fetch(url, {
    method: 'GET',
    redirect: 'manual',
    signal: AbortSignal.timeout(TIMEOUT_MS),
    headers: { 'user-agent': 'auditable-agent-layer/1.0' }
  });
  if (response.status >= 300 && response.status < 400) throw new Error('不允许自动重定向');
  if (!response.body) return { status: response.status, body: '' };

  const reader = response.body.getReader();
  const chunks = [];
  let size = 0;
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    size += value.byteLength;
    if (size > MAX_BYTES) {
      await reader.cancel();
      throw new Error('网络响应超过限制');
    }
    chunks.push(Buffer.from(value));
  }
  return {
    status: response.status,
    contentType: response.headers.get('content-type'),
    body: Buffer.concat(chunks).toString('utf8')
  };
}

async function confirm(tool, input) {
  if (!process.stdin.isTTY) return false;
  const rl = createInterface({ input: process.stdin, output: process.stdout });
  const answer = await rl.question(`允许执行 ${tool} ${JSON.stringify(requestView(tool, input))}？[y/N] `);
  rl.close();
  return answer.trim().toLowerCase() === 'y';
}

async function execute(request) {
  const id = randomUUID();
  const started = Date.now();
  const { tool, input = {} } = request;
  await audit({ id, phase: 'requested', tool, input: requestView(tool, input) });

  try {
    if (!['read_file', 'run_command', 'http_get'].includes(tool)) throw new Error('未知工具');
    if (tool !== 'read_file' && !await confirm(tool, input)) throw new Error('人工确认未通过');

    const handlers = { read_file: readFileTool, run_command: runCommandTool, http_get: httpGetTool };
    const result = await handlers[tool](input);
    await audit({ id, phase: 'completed', tool, durationMs: Date.now() - started, result: resultView(result) });
    return result;
  } catch (error) {
    await audit({ id, phase: 'failed', tool, durationMs: Date.now() - started, error: String(error.message) });
    throw error;
  }
}

if (!process.argv[2]) {
  console.error('用法：node index.mjs <JSON 请求>');
  process.exit(1);
}

try {
  console.log(JSON.stringify(await execute(JSON.parse(process.argv[2])), null, 2));
} catch (error) {
  console.error(error.message);
  process.exitCode = 1;
}

运行、确认与审计

先创建测试文件，再发起只读请求：

node -e "require('fs').mkdirSync('workspace',{recursive:true});require('fs').writeFileSync('workspace/hello.txt','hello\n')"
node index.mjs '{"tool":"read_file","input":{"path":"hello.txt"}}'

命令和网络请求会进入人工确认流程：

node index.mjs '{"tool":"run_command","input":{"command":"node-version","args":[]}}'
node index.mjs '{"tool":"http_get","input":{"url":"https://example.com/"}}'

每次请求至少写入一条 requested 记录，完成或失败后再写一条结果记录。结果正文不直接进入日志，只记录字节数和 SHA-256，避免把文件内容或网络响应复制到审计系统。URL 查询参数也被移除，因为其中经常包含令牌和用户数据。

生产环境还应记录调用者、会话、模型、策略版本和审批人，并把日志发送到智能体无权修改的外部存储。示例中的本地 NDJSON 便于演示，但进程崩溃、磁盘故障以及日志文件被宿主用户修改等问题，不能靠 appendFile 彻底解决。

应用级防护的边界

这套执行层适合权限较小、工具集合明确的智能体，但它不是安全沙箱。工作目录校验能拦截常见的 ../ 和符号链接逃逸，却难以彻底消除路径检查与文件打开之间的竞态条件。命令超时调用 kill，也不保证所有平台上由目标程序派生的子进程都会一起退出。

网络层还存在 DNS 重绑定、代理配置、IPv6、服务端内容变化等问题。若要访问内网，应在网络出口实施独立策略，而不是只依赖 URL 字符串检查。

出现以下情况时，应使用容器、虚拟机或操作系统级沙箱：

  • 允许模型生成并运行任意代码或任意命令；
  • 多租户共享宿主机，租户之间必须强隔离；
  • 工作负载会处理密钥、生产数据或不可信压缩包；
  • 需要可靠限制 CPU、内存、进程数、磁盘和网络；
  • 需要阻止系统调用、挂载、设备访问或子进程逃逸。

即使采用容器，应用级策略仍然有价值：容器负责限制最坏结果，工具层负责表达业务许可、人工审批和审计语义，两者不是替代关系。

总结

  • 用少量结构化工具替代万能 Shell，并坚持默认拒绝。
  • 文件访问同时检查规范路径、真实路径、类型、扩展名和大小。
  • 命令采用动作白名单，关闭 Shell，并限制时间与输出。
  • 网络请求限制 HTTPS、主机、端口、重定向和响应体。
  • 高风险操作执行前人工确认，执行前后写入脱敏审计日志。
  • 应用执行层不能提供强隔离；面对任意代码、多租户或敏感数据时，应叠加容器或系统级沙箱。