2.3 执行网关与工具代理

2.3 执行网关与工具代理

本节摘要：让 Agent 直接调用 docker 有三个硬伤：没有校验（谁来拦 curl？）、没有审计（事后无法回放）、暴露宿主凭据（能起容器就能干别的）。解法是三段式架构：Agent 只会「发请求」，执行网关负责白名单校验、路径作用域检查、结果截断与审计落盘，沙箱负责物理边界内的执行。本节给出标准库实现的 gateway.py 最小网关——拒收 shell 字符串、只接受参数数组，deny 规则先于 allow 规则，绝对路径必须落在工作区内，回传输出截断到四千字节——并演示如何把网关注册成 Agent 的 execute_command 工具。至此，第 0.1 节三层防线中的规则层有了物理载体，沙箱层与它正式合体。

学习目标

• 说清「Agent 直接碰 Docker」的三个问题与网关模式的对位解法。
• 部署 gateway.py，走通一次「请求 → 校验 → 沙箱执行 → 回传」全流程。
• 理解拒收 shell 字符串、deny 优先、路径作用域三个设计决定。
• 会把网关包装成 Agent 工具定义，并解释回传截断为何关乎提示词安全。

一、为什么不让 Agent 直接碰 Docker

三段式架构（本章最终形态）

 Agent（模型 / 编排器）         执行网关 gateway.py                沙箱 sandbox_run.py
┌────────────────┐          ┌────────────────────┐           ┌──────────────────┐
│ 我要执行：       │  HTTP    │ ① deny 规则先判      │  子进程    │ 加固容器 / runsc / │
│ {"command":     │ ───────▶ │ ② allow 白名单       │ ────────▶ │ Firecracker VM    │
│  ["pytest","-q"]}│         │ ③ 路径作用域检查      │           │ 四道边界内执行      │
│                 │ ◀─────── │ ④ 调用沙箱执行        │ ◀──────── │ stdout/stderr/码  │
│ 收到结构化结果    │  HTTP    │ ⑤ 输出截断 + 审计落盘 │           └──────────────────┘
└────────────────┘          └────────────────────┘
     不持有 Docker                  不执行业务命令                不做策略判断
​

分工铁律：网关只判断不执行，沙箱只执行不判断——两个角色都不越界，审计才完整。

二、最小网关：gateway.py

# gateway.py —— 执行网关最小实现：白名单校验 → 调用沙箱 → 截断回传 → 审计
import fnmatch, json, os, subprocess
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

ALLOW = [  # allow：fnmatch 模式，作用于整个命令行（写法示意，按项目裁剪）
    "python *", "python3 *", "pytest", "pytest *",
    "git status", "git diff", "git diff *", "git log", "git log *",
    "ls", "ls *", "cat *", "pip install -r requirements.txt",
]
DENY = [   # deny 优先于 allow
    "*curl*", "*wget*", "*nc *", "*sudo*", "*chmod*", "*chown*",
    "rm *", "git push*", "* > /dev/*", "*mkfs*", "*dd if=*",
]
MAX_OUT_BYTES = 4000   # 回传截断上限：防提示词淹没（示意值）

def check(argv: list) -> tuple:
    joined = " ".join(argv)
    for pat in DENY:
        if fnmatch.fnmatch(joined, pat):
            return False, f"命中拒绝规则: {pat}"
    if not any(fnmatch.fnmatch(joined, p) for p in ALLOW):
        return False, "不在白名单，需人工审批（见第 3.1 节审批门）"
    for a in argv[1:]:                       # 路径作用域：绝对路径必须在工作区内
        if a.startswith("~") or ".." in a.split("/"):
            return False, f"路径越界: {a}"
        if a.startswith("/") and not a.startswith("/workspace"):
            return False, f"路径越界: {a}"
    return True, "ok"

class Handler(BaseHTTPRequestHandler):
    def _reply(self, code, payload):
        body = json.dumps(payload, ensure_ascii=False).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        if self.path != "/exec":
            return self._reply(404, {"error": "not found"})
        req = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        argv = req.get("command")
        if not isinstance(argv, list) or not argv:      # 拒收 shell 字符串
            return self._reply(400, {"error": "command 必须是参数数组"})
        ok, reason = check(argv)
        if not ok:
            return self._reply(200, {"decision": "deny", "reason": reason})
        proc = subprocess.run(["python", "sandbox_run.py", *argv],
                              capture_output=True, text=True)
        try:
            result = json.loads(proc.stdout)
        except json.JSONDecodeError:
            return self._reply(500, {"error": "sandbox 内部错误"})
        return self._reply(200, {**result,
            "stdout_tail": result.get("stdout_tail", "")[:MAX_OUT_BYTES],
            "stderr_tail": result.get("stderr_tail", "")[:MAX_OUT_BYTES]})

if __name__ == "__main__":
    print("网关监听 http://127.0.0.1:8905/exec（仅本机，供 Agent 侧调用）")
    ThreadingHTTPServer(("127.0.0.1", 8905), Handler).serve_forever()
​

试运行（另开终端）：

# try_gateway.sh —— 网关冒烟测试：一次放行、一次拒绝（写法示意）
python gateway.py &
curl -s -X POST http://127.0.0.1:8905/exec -d '{"command":["python","-V"]}'
curl -s -X POST http://127.0.0.1:8905/exec -d '{"command":["curl","http://evil.example"]}'
# 期望：第一次返回 decision=finished（沙箱内执行结果），第二次 decision=deny
​

三个设计决定，每个都对应真实攻击：拒收 shell 字符串——一旦接受 "sh -c ..."，白名单就退化为装饰品，注入者可以用引号与变量组装出任意命令；deny 优先——allow 永远写不全，deny 名单兜住网络工具、提权与破坏性命令；路径作用域——白名单 cat * 若不限路径，cat /workspace/../etc/passwd 与 cat ~/.ssh/id_rsa 立刻打穿。

⚠️ 网关只监听 127.0.0.1：它本身没有任何认证，部署形态默认「与 Agent 同机」。跨机部署必须先加认证与 TLS，并把它纳入自己的威胁模型（网关失守 = 沙箱执行权失守）。

三、把网关接成 Agent 工具

网关对 Agent 来说就是一个工具（tool）。注册示例（结构示意，字段名以所用 Agent 框架为准）：

{
  "name": "execute_command",
  "description": "在沙箱内执行命令并返回结果。仅接受参数数组；可用命令见白名单；输出超长会被截断。",
  "input_schema": {
    "type": "object",
    "properties": {
      "command": {"type": "array", "items": {"type": "string"},
                  "description": "如 [\"pytest\", \"-q\"]，不要传 shell 字符串"}
    },
    "required": ["command"]
  }
}
​

工具侧拿到回执后的处理纪律：decision=deny 时把 reason 原样反馈给模型（让它换路子或求助），不要静默重试；decision=finished 时只把截断后的尾部输出放回上下文。

回传截断为什么是安全设计而不只是省 token：沙箱里一个命令可以在 stdout 里塞进几万字的注入文本（例如被投毒的测试输出），如果原样塞回上下文，等于把不可信输入直接递到模型嘴边——这正是第 0.1 节「命令来源不可信」的回环。截断 + 只回传尾部，把这条回流通道压到最窄。注入防御的系统论述见本站《AI 应用安全与对齐：提示注入防御与输出管控》文集。

💡 网关位置提醒：这里的白名单是「最小可用版」，真正的策略体系——命令模式匹配的粒度陷阱、审批门如何接人、审计如何防篡改——是第 3 章的全部内容。

本节要点回顾

• 三段式铁律：Agent 发请求不碰 Docker，网关判断不执行，沙箱执行不判断。
• gateway.py 四步：deny 先判、allow 白名单、路径作用域、截断回传；拒收 shell 字符串是底线。
• 输出截断防的是「被投毒的输出反向注入模型」，不只是省 token。
• 网关仅本机监听、无认证，跨机部署必须先补认证并纳入威胁模型。

三段式底座已完整：四道边界（2.1）、隔离层升级（2.2）、执行权收拢（2.3）。第 3 章在这块底座上织最后两张网：allowlist 与审批门的策略体系（3.1），以及让一切可回放的审计日志（3.2）。