本节摘要:策略看单条命令,行为看模式。第 3.2 节已经给出过一个人工观察:「同一 run 连续多次 deny 后紧跟一次 allow,是注入在试探策略边界的经典指纹」——本节把这个观察自动化。先立一张四类行为信号表:外联尝试、路径越界、资源尖峰、高频写,每类信号给观测点、典型指纹与阈值示意,观测点尽量复用已有构件(网关 deny 记录、审计 reason 字段、沙箱 result 事件),不新造监控体系;再给滑动窗口熔断器 breaker.py 的最小实现——窗口内按信号类别计数,达阈值触发动作分级(记录、降速、挂起、终止);最后讲接线方式与误报治理:先用「只记录不动作」模式跑一周,用真实流量定阈值,再逐步开启动作。
前三章的判断都发生在命令执行之前:策略拦单条命令,沙箱拦越界副作用。但有一类模式只有把多次执行放在一起看才显形——单次读 ~/.ssh/id_rsa 被拒是正常摩擦,六十秒内读家目录、试点点穿越、碰 /etc 连续被拒,就是有人在画你的边界。行为信号的价值正在于此:它不依赖对载荷的任何先验知识,只统计「尝试了什么、频率如何」。这一点对 Agent 场景尤其重要:供应链审查(第 4 章)已经证明静态侧拦不住精心混淆的载荷,而载荷一旦运行,行为是藏不住的——外联要碰网络栈,越界要过路径检查,耗资源要顶到限额。
四类信号与第 0.1 节四类攻击面一一对应(内核漏洞除外——它靠 2.2 节的隔离层兜底),观测点全部来自已有构件:
| 信号类别 | 观测点 | 典型指纹 | 阈值示意 |
|---|---|---|---|
| 外联尝试 | 网关 deny 记录;断网沙箱内的解析失败记录 | 反复尝试解析或连接外部地址 | 60 秒窗口内不少于 3 次 |
| 路径越界 | 审计 decision 事件的 reason 字段(含「越界」「封禁」) | 连续触碰工作区外、家目录、点点穿越 | 60 秒窗口内不少于 2 次 |
| 资源尖峰 | docker stats 采样;沙箱 result 事件的 duration 与 exit_code | 内存或 CPU 逼近限额、OOM、超时率上升 | 内存占用九成以上持续 3 个采样 |
| 高频写 | 工作区文件事件(inotify 思路)或定时清点文件数 | 每秒大量新建、改写文件 | 窗口内新增超过 200 个文件 |
阈值全部为示意值——真实数值必须用你自己的工作流校准,方法见第四节。
# breaker.py —— 滑动窗口熔断器:窗口内按信号类别计数,达阈值触发分级动作(示意实现) import time from collections import deque LIMITS = { # 60 秒窗口内的熔断阈值(示意值,先只记录校准后再启用) "net_attempt": 3, "path_escape": 2, "io_spike": 1, "res_spike": 2, } ACTIONS = { # 信号类别到动作级别的映射(示意,按风险偏好调整) "net_attempt": "throttle", "path_escape": "suspend", "io_spike": "suspend", "res_spike": "kill", } class WindowBreaker: def __init__(self, window_s: float = 60.0): self.window_s = window_s self.events = deque() self.tripped = None def record(self, kind: str): now = time.time() self.events.append((now, kind)) while self.events and now - self.events[0][0] > self.window_s: self.events.popleft() # 滑出窗口的旧事件不再计数 n = sum(1 for _, k in self.events if k == kind) if self.tripped is None and n >= LIMITS.get(kind, 99): self.tripped = kind return ACTIONS.get(kind, "suspend") # 返回应执行的动作级别 return None def reset(self): self.tripped = None self.events.clear() if __name__ == "__main__": b = WindowBreaker() for i in range(3): action = b.record("net_attempt") print("第三次外联尝试后动作:", action) # 期望输出: throttle
为什么是滑动窗口而不是累计计数:累计计数有个致命缺陷——攻击者只要把尝试间隔拉长(每五分钟碰一次边界),累计值照样涨,正常长会话却可能先被误伤。滑动窗口只统计最近六十秒,慢速试探不计入、正常流量不积累,两种错误同时压低。tripped 字段实现「熔断后保持」:触发过就停在熔断态,直到人工或脚本调用 reset()——这对应一个运维决定:熔断的恢复必须过人,不能自愈。
动作分级的语义(从轻到重):
| 级别 | 动作 | 网关侧落点 |
|---|---|---|
| 记录 | 只落审计,不干预 | 审计事件加 signal 字段 |
| 降速 | 给该 run 的每次执行加人为延迟 | 网关收到请求后先延时再校验 |
| 挂起 | 后续命令一律改道 needs_approval | 网关把灰区出口改为强制审批 |
| 终止 | 结束本次会话,杀掉沙箱实例 | 调用容器终止,审计记 result |
信号源就是审计事件流——第 3.2 节的 auditor 已经把每次决策写成了结构化事件,熔断器只是它的一个旁路消费者。最小接线(示意):在 gateway.py 处理请求的路径上,把 check 函数返回的拒绝原因做一次归类映射,喂给全局唯一的 WindowBreaker 实例;若返回了动作级别,按上表落点执行,并把熔断事件本身也 emit 进审计链:
# breaker_wire.py —— 网关侧接线示意:拒绝原因归类 → 喂熔断器 → 执行动作并落审计 from breaker import WindowBreaker from auditor import emit BREAKER = WindowBreaker() def classify(reason: str): if "命中拒绝规则" in reason and ("curl" in reason or "wget" in reason): return "net_attempt" if "路径越界" in reason or "封禁" in reason: return "path_escape" return None def on_deny(run_id: str, argv: list, reason: str): kind = classify(reason) if kind is None: return action = BREAKER.record(kind) emit({"phase": "decision", "run_id": run_id, "decision": "deny", "command": argv, "reason": reason, "signal": kind, "breaker_action": action or "record_only", "sandbox": "docker-hardened"}) if action == "suspend": pass # 此处把该 run 的后续命令改道 needs_approval(接第 3.1 节审批门) elif action == "kill": pass # 此处终止该 run 的沙箱实例(教学示意,落点按部署形态实现)
接线纪律两条:熔断器实例必须全局唯一(每个网关进程一个,按 run 维度聚合时改用字典索引);熔断动作必须回写审计——「为什么被终止」与「为什么被拒绝」同样重要,否则 5.2 节的事件响应没有起点。资源尖峰与高频写两类信号在审计流之外,取数方式分别是定时采样 docker stats 与定时清点工作区文件数,判定的思路一致:窗口内计数、达阈值调用同一个 record()——观测器可以多个,熔断器只有一个。
阈值是本节最大的坑:调紧了,一次跑全套测试的合法资源尖峰就会触发熔断;调松了,又回到没设防。落地次序(三步):
ACTIONS 全部置为 record,只观察一到两周,统计每类信号在正常工作流里的自然分布;⚠️ 两个常见误报源提前记下:其一,测试套件并行执行天然产生资源尖峰与高频写,给 CI 型 run 打标并在熔断器里豁免(或单独放宽);其二,第 3.2 节统计过的「deny 率 top 命令」往往指向策略与工作流的摩擦点——先修策略再调阈值,别用熔断掩盖策略缺陷。
熔断解决「发现了怎么办」,但还有一个更硬的问题没回答:如果攻击者不是试探边界,而是直接击穿了边界——逃逸发生了,损失如何计算、如何压缩?下一节把最坏情况纳入工程:压爆炸半径、隔离凭据、按五步走完事件响应。