从零实现 AI Agent 安全沙箱:FastAPI、Docker 与策略引擎实战
项目地址:https://github.com/ycx666994/ai-agent-security-sandbox
一、项目背景
随着智能体的快速发展,现在的AI Agent 能调用搜索、HTTP 请求、Shell 命令等工具,已经初步展示出了强大且惊人的实力,但“能调用”不代表“应该无条件调用”。(也就是agent存在一些风险,所以我大胆预测,十年内,AI安全有关的方向必定是个大热门,因为需求真的会越来越大,越来越高)
例如,一个不受约束的 Agent 可能:
1.访问不该访问的内部地址;
2.执行危险命令;
3.通过超长响应或海量输出拖垮服务;
4.伪造 agent_id 冒充其他 Agent;
5.在超时后留下未清理的执行容器。
等等......
那么,这些潜在风险,会给我们带来不必要的麻烦。介于这些诸多的风险,因此,我实现了一个 AI Agent Security Sandbox(AI Agent 安全执行沙箱)。它不是让 Agent 直接执行工具,而是在 Agent 与真实执行环境之间增加一个安全网关:(我简单介绍一下它的工作流程)
1. Agent 先向网关发起工具调用请求;
2. 网关先验证调用方身份;
3. 策略引擎判断本次调用是否允许;
4. 允许的 Shell 命令交由隔离 runner 执行;
5. 所有关键行为写入审计日志;
6. 网关返回统一格式的执行结果。
最终目标是:默认拒绝,按策略最小授权,执行环境隔离,行为可审计。
二、最终成果展示
目前为止,这个AI Agent 安全沙箱已经做到一定的程度了。
我展示一下当初我手写的架构图:(字有点丑)

下面的这个是Docker Desktop 容器运行状态的截图:

项目运行效果:
启动后,网关只监听本机地址:
http://127.0.0.1:8001/docs
健康检查结果:
{
"status": "ok"
}
匿名调用会被拒绝:
HTTP 401 Unauthorized
即使携带了合法 API Key,访问不在允许列表中的站点也会被拒绝:
{
"request_id": "96bd1fd5-31ca-4c3b-aebe-cfd4cf2873c9",
"status": "denied",
"result": null,
"reason": "host is not in the network allow-list"
}
而允许的 Shell 命令能够在隔离容器中成功执行:
{
"request_id": "7d997d46-f3dd-4556-adaf-c7c981fbf308",
"status": "allowed",
"result": {
"exit_code": 0,
"stdout": "sandbox-ok\n",
"stderr": "",
"output_truncated": false
},
"reason": null
}
测试结果:
9 passed
以下截图为使用 Docker Compose 构建并启动安全沙箱。gateway 和 runner 两个服务均成功启动。


然后以下截图为 Agent 携带正确 API Key 调用允许列表中的 echo 命令。请求经过网关认证、策略校验和内部 runner 隔离执行后,成功返回> sandbox-ok。

然后我GitHub 仓库首页截图如下:

三、相关的技术栈
就大致说一下我做这个项目的流程就行了:
我用的是VS Code编写了Python 项目代码,使用Python 3.12 和 FastAPI 搭建 API 网关。
然后,我使用 Docker Desktop 将网关、runner 和实际执行命令的容器隔离开来。
接着,我使用 PowerShell 调用接口,验证匿名请求、越权请求和合法请求的返回结果。
最后,我使用 Git 和 GitHub 管理版本,并将项目推送到仓库:
https://github.com/ycx666994/ai-agent-security-sandbox
我也把我用到的一些核心技术整理出来了:
技术 用途
FastAPI 提供工具调用 API
Pydantic 校验请求数据结构
httpx 执行安全的 HTTP 请求、调用内部 runner
Docker Desktop 提供命令隔离环境
Docker Engine API 创建、启动、等待、删除任务容器
Docker Compose 编排 gateway 和 runner
pytest 测试认证、策略和 API 行为
GitHub 源码托管和项目展示
四、第一步:用 FastAPI 建立工具调用网关
我首先用 FastAPI 定义了统一的工具调用接口:
@app.post("/v1/tool-calls", response_model=ToolResult)
async def invoke_tool(call: ToolCall, x_api_key: str | None = Header(default=None)) -> ToolResult:
request_id = call.request_id or str(uuid.uuid4())
try:
authenticate(x_api_key, call.agent_id)
except HTTPException:
audit.write(
request_id=request_id,
agent_id=call.agent_id,
tool=call.tool,
decision="denied",
reason="authentication failed",
)
raise
allowed, reason = policy.evaluate(call)
if not allowed:
audit.write(
request_id=request_id,
agent_id=call.agent_id,
tool=call.tool,
decision="denied",
reason=reason,
)
return ToolResult(
request_id=request_id,
status="denied",
reason=reason,
)
这段代码体现了一个重要原则:执行前必须认证,认证后必须鉴权,鉴权通过后才能执行。
处理顺序如下:
1. 生成 request_id,方便追踪一次调用;
2. 验证 API Key;
3. 检查该 Key 是否有权代表请求中的 agent_id;
4. 调用策略引擎;
5. 不允许时写审计日志并直接返回;
6. 只有策略允许时才进入执行阶段。
五、第二步:实现 API Key 和 Agent 身份绑定
仅仅让客户端传递 agent_id 是不安全的,因为任何人都可以伪造:
{
"agent_id": "admin-agent"
}
所以我使用环境变量 SANDBOX_API_TOKENS 保存“API Key 到 Agent 身份”的映射。
例如:
$env:SANDBOX_API_TOKENS='{"local-dev-token":"research-agent"}'
核心认证代码如下:
def authenticate(api_key: str | None, agent_id: str) -> None:
if not api_key:
raise HTTPException(status_code=401, detail="missing API key")
configured = json.loads(os.environ["SANDBOX_API_TOKENS"])
matched_agent = None
for token, identity in configured.items():
if secrets.compare_digest(api_key, token):
matched_agent = identity
if matched_agent is None:
raise HTTPException(status_code=401, detail="invalid API key")
if not secrets.compare_digest(agent_id, matched_agent):
raise HTTPException(
status_code=403,
detail="API key is not authorized for this agent",
)
这里使用了 secrets.compare_digest(),而不是普通的 ==。
原因是普通字符串比较可能存在时间差异,而compare_digest() 更适合比较密钥、令牌等敏感信息,可以降低时序攻击风险。
最终效果:
场景 返回结果
没有 API Key 401
API Key 不正确 401
API Key 正确但 agent_id 不匹配 403
API Key 和 agent_id 都正确 进入策略校验
六、第三步:实现默认拒绝的策略引擎
项目中的 policy.json 是安全规则的核心:
{
"version": "1.0",
"default_action": "deny",
"tools": {
"http_get": {
"enabled": true,
"allowed_hosts": ["api.github.com"],
"timeout_seconds": 5,
"max_response_bytes": 65536,
"max_url_chars": 2048
},
"shell": {
"enabled": true,
"allowed_commands": ["echo", "date", "whoami"],
"timeout_seconds": 5,
"max_command_chars": 4096
}
}
}
我跟大家讲一下这个代码的具体策略
默认拒绝所有工具;
HTTP 工具只允许访问 api.github.com;
Shell 工具只允许执行 echo、date、whoami;
HTTP 响应最大读取 64 KiB;
命令和 URL 都有长度限制;
单次执行最长 5 秒。
URL 校验核心代码:
parsed = urlparse(value)
if parsed.scheme != "https" or not parsed.hostname:
return False, "only absolute https URLs are allowed"
try:
ipaddress.ip_address(parsed.hostname)
return False, "IP address targets are not allowed"
except ValueError:
pass
if parsed.port not in (None, 443):
return False, "only HTTPS port 443 is allowed"
if parsed.hostname.lower() not in allowed_hosts:
return False, "host is not in the network allow-list"
这段代码做了四层限制:
1. 只接受完整 HTTPS URL;
2. 拒绝直接填写 IP 地址;
3. 只允许 HTTPS 默认端口 443;
4. 主机名必须在允许列表中。
例如下面这些请求会被拒绝:
http://example.com
https://127.0.0.1
https://192.168.1.1
https://api.github.com:8443
https://example.com
这能减少 SSRF、内部网络探测和端口扫描风险。
七、第四步:限制 Shell 命令,而不是直接执行任意字符串
Shell 命令策略校验如下:
tokens = shlex.split(value)
if not tokens or tokens[0] not in rule.get("allowed_commands", []):
return False, "command is not in the allow-list"
if any(token in value for token in (";", "&&", "||", "|", ">", "<", "$(", "`")):
return False, "shell operators are not allowed"
它的作用是:
1.先用 shlex.split() 安全解析命令;
2.检查第一个命令是否在允许列表;
3.拒绝管道、重定向、命令替换等危险 Shell 操作符。
因此:echo sandbox-ok可以通过。
但:echo ok; whoami会被拒绝,因为包含 ;。
这体现了“允许明确的能力,而不是阻止无限多的危险行为”的安全思想。
八、第五步:为什么还要把 gateway 和 runner 分开
一开始,最简单的方案是让 FastAPI 服务直接挂载 Docker socket 并执行 Docker 命令。
但 Docker socket 的权限非常高,接近宿主机管理员权限。如果对外网关直接拥有它,风险太大。
因此,我使用 Docker Compose 将服务拆分成两个角色:
1.gateway:对外提供 API,不挂载 Docker socket
2.runner:仅在 Docker 内网中运行,拥有 Docker socket
核心 Compose 配置如下:
gateway:
ports:
- "127.0.0.1:8001:8000"
environment:
RUNNER_URL: http://runner:8000
read_only: true
cap_drop:
- ALL
runner:
volumes:
- /var/run/docker.sock:/var/run/docker.sock
read_only: true
cap_drop:
- ALL
这样做的效果是:
1.网关只能通过 http://runner:8000 请求 runner;
2.runner 没有暴露宿主机端口;
3.外部用户不能直接访问 runner;
4.Docker socket 不再暴露给对外 API 网关;
5.网关只绑定 127.0.0.1,默认不对局域网开放。
九、第六步:用 Docker Engine API 创建受限容器
runner 不再依赖 docker run 命令行,而是通过 Docker Engine Unix Socket API 创建容器。
核心配置如下:
payload = {
"Image": image,
"Cmd": shlex.split(command),
"User": "65532:65532",
"HostConfig": {
"NetworkMode": "none",
"ReadonlyRootfs": True,
"CapDrop": ["ALL"],
"SecurityOpt": ["no-new-privileges"],
"PidsLimit": 64,
"Memory": 128 * 1024 * 1024,
"NanoCpus": 500_000_000,
"Tmpfs": {
"/tmp": "rw,noexec,nosuid,size=16m"
}
}
}
这里每一项都对应一个安全控制:

这使得即使允许执行 echo sandbox-ok,它也只能在一个受限容器中执行,而不是直接在宿主机执行。
十、第七步:解决“截断不等于限制”的问题
一个常见错误是这样写:
body = response.content[:65536]
表面上看,只返回 64 KiB。但实际上,response.content 已经先将完整响应读到内存中。如果远程服务返回几百 MB,网关仍然可能内存耗尽。
所以我改成流式读取:
body = bytearray()
async for chunk in response.aiter_bytes():
remaining = max_response_bytes - len(body)
if remaining <= 0:
truncated = True
break
body.extend(chunk[:remaining])
if len(chunk) > remaining:
truncated = True
break
这样程序只会把最大允许大小的数据保存在内存中。
Shell 输出也采用同样思想:读取时限制输出,而不是“命令执行完后再截断”。
十一、第八步:限制请求体大小
仅限制 URL 和命令长度还不够。攻击者可以在 JSON 中塞入一个巨大但未使用的字段,
例如:
{
"agent_id": "research-agent",
"tool": "shell",
"arguments": {
"command": "echo ok",
"unused": "这里可以是几百 MB 的无意义数据"
}
}
因此我加入了 ASGI 中间件,在 JSON 解析之前限制请求体:
class RequestBodyLimitMiddleware:
def __init__(self, app, max_body_bytes: int = 16384):
self.app = app
self.max_body_bytes = max_body_bytes
当请求超过 16 KiB 时,直接返回:
HTTP 413 Payload Too Large
这样就可以避免大请求在进入业务逻辑前消耗过多内存。
十二、第九步:审计日志
每次调用都会记录为 JSONL 格式:
{
"timestamp": "2026-07-29T05:10:07.816267+00:00",
"request_id": "xxxx",
"agent_id": "research-agent",
"tool": "shell",
"decision": "allowed",
"result_summary": {
"exit_code": 0
}
}
而我们学会写审计日志有什么好处吗呢,就是我们能够弄清楚:
1.哪个 Agent 调用了什么工具?
2.请求是否被拒绝?
3.被拒绝的原因是什么?
4.命令是否执行成功?
5.某次安全事件对应哪个 request_id?
需要注意的是,本项目的 JSONL 日志适合本地开发和演示。生产环境中应将日志发送到远程、追加式、权限隔离的日志系统。
十三、测试结果
我使用 pytest 对关键安全边界进行了测试:
9 passed
测试覆盖包括:
1.允许列表中的 HTTPS 主机可以访问;
2.不在允许列表中的主机被拒绝;
3.字面 IP 地址被拒绝;
4.非 443 HTTPS 端口被拒绝;
5.Shell 操作符注入被拒绝;
6.超长命令被拒绝;
7.API Key 与 Agent 身份绑定;
8.未配置认证时默认拒绝;
9.请求体超过限制时返回 413。
运行测试:
pytest -q
十四、如何运行项目
先配置本地开发环境变量:
$env:SANDBOX_API_TOKENS='{"local-dev-token":"research-agent"}'
$env:RUNNER_SHARED_TOKEN='replace-with-a-long-random-secret'
$env:DOCKER_GID='0'
启动服务:
docker-compose up --build -d
测试健康检查:
Invoke-RestMethod http://127.0.0.1:8001/health
测试允许的 Shell 命令:
Invoke-RestMethod `
-Uri 'http://127.0.0.1:8001/v1/tool-calls' `
-Method Post `
-ContentType 'application/json' `
-Headers @{ 'X-API-Key' = 'local-dev-token' } `
-Body '{"agent_id":"research-agent","tool":"shell","arguments":{"command":"echo sandbox-ok"}}'
十五、项目总结
这个项目让我完整实践了从“功能能运行”到“功能有安全边界”的过程。
我先用 FastAPI 搭建工具调用接口;然后使用 API Key 解决调用方身份问题;接着用 JSON 策略实现默认拒绝与允许列表;之后使用Docker 将 Shell 命令隔离;最后通过流式输出限制、请求体限制、超时清理、审计日志和测试补齐安全细节。
目前项目已经实现了:
身份认证;
Agent 身份绑定;
默认拒绝策略;
HTTP 域名白名单;
Shell 命令白名单;
容器隔离;
非 root 执行;
CPU、内存、PID 限制;
无网络与只读根文件系统;
超时清理;
请求与输出大小限制;
结构化审计;
自动化测试。
当然,Docker socket 对 runner 仍然是高权限能力。但是,在真正的生产环境中,其实那些企业都是用的是Kubernetes Job、受限容器平台或专门的任务执行服务,并搭配 NetworkPolicy、准入控制和远程审计系统。(但是要把这些东西用好有极高的门槛,初学者不太建议)
谢谢大家!
更多推荐


所有评论(0)