3.2 审计日志与回放

3.2 审计日志与回放

本节摘要：沙箱与策略负责「尽量不出事」，审计负责「出了事（或差点出事）之后，真相可查」。本节先定审计事件 schema——从请求、决策、审批到执行结果，每个字段的取证价值逐一说清；再给最小审计器 auditor.py：JSONL 逐行追加、写后即同步、哈希链防篡改（每一行携带前一行的摘要，改任何一行都会让链条断掉）；最后是三层回放思路：时间线重建（发生了什么）、重放对照（怎么发生的）、指标统计（会不会再发生），并说明审计如何为第 4 章供应链取证与第 6 章红队演练提供判据。至此，第 0.1 节的三层防线全部落地且互相咬合。

学习目标

• 设计审计事件 schema，解释决策、审批、结果三类事件的取证价值。
• 部署 auditor.py 并验证哈希链的防篡改能力。
• 用三层回放思路完成一次事件的完整复盘。
• 说清审计与红队演练、供应链取证的关系。

一、审计事件 schema：一行一个事实

审计日志的原则：机器写、人不改、事后可验证。格式选 JSONL（每行一个独立 JSON 对象）：追加即落盘、逐行可流式处理、坏了也只坏一行。事件 schema：

💡 command 坚持存参数数组而不存拼接字符串：注入攻击常利用引号、分号、空格歧义伪装，数组形态让「当时真正要执行的东西」无可辩驳。stdout_digest 存哈希而非全文：既留了「输出是否与回放一致」的比对锚点，又不让日志库变成泄露面。

二、最小审计器：auditor.py

# auditor.py —— 最小审计器：JSONL 追加 + 写后即同步 + 哈希链防篡改（示意实现）
import hashlib, json, os, time, uuid

LOG_PATH = "audit/audit.jsonl"

def _last_digest() -> str:
    try:
        with open(LOG_PATH, "rb") as f:
            lines = f.read().strip().splitlines()
        return hashlib.sha256(lines[-1]).hexdigest() if lines else "GENESIS"
    except FileNotFoundError:
        return "GENESIS"

def emit(event: dict) -> dict:
    event = {"event_id": uuid.uuid4().hex,
             "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), **event}
    event["prev_hash"] = _last_digest()
    event["digest"] = hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
    os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
    with open(LOG_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(event, ensure_ascii=False) + "\n")
        f.flush()
        os.fsync(f.fileno())      # 写后即同步：崩溃也不丢最后一行
    return event

def verify() -> bool:
    prev = "GENESIS"
    with open(LOG_PATH, encoding="utf-8") as f:
        for line in f:
            e = json.loads(line)
            body = {k: v for k, v in e.items() if k != "digest"}
            if e["prev_hash"] != prev or hashlib.sha256(
                    json.dumps(body, sort_keys=True, ensure_ascii=False).encode()).hexdigest() != e["digest"]:
                return False
            prev = e["digest"]
    return True

if __name__ == "__main__":
    emit({"phase": "decision", "run_id": "demo", "decision": "deny",
          "command": ["curl", "http://evil.example"], "reason": "命中拒绝规则: *curl*",
          "sandbox": "docker-hardened"})
    print("链条完整:", verify())
​

接入方式：把 gateway.py 与 sandbox_run.py 里原来各自写的 JSONL 追加，统一替换为 from auditor import emit——所有事件进同一条链，一个 verify() 覆盖全部。篡改实验：手工改掉日志中间一行的任意字符再跑 verify()，期望返回 False。

⚠️ 边界诚实说明：哈希链防「事后悄悄改历史」，不防「有 root 权限的人整段重写」。生产级还要加远程转发（审计旁路到另一台机器）与定期外部校验——思路一致，搬运工不同。

三、回放：三层思路

第一层 · 时间线重建（发生了什么）。 以 run_id 为主键过滤，按 ts 排序重放：

# replay_query.py —— 按 run_id 重建命令时间线（最小示例）
import json, sys
for line in open("audit/audit.jsonl", encoding="utf-8"):
    e = json.loads(line)
    if e.get("run_id") == sys.argv[1]:
        print(e["ts"], e["phase"], e["decision"], " ".join(e["command"]), e.get("reason", ""))
​

典型输出：一条 request 对一条 decision；若出现「同一 run 连续多次 deny 后紧跟一次 allow」，就是注入在试探策略边界的经典指纹。

第二层 · 重放对照（怎么发生的）。 把 decision=allow 的命令序列在干净沙箱里重新执行一遍，比对 exit_code 与 stdout_digest：一致，说明行为可复现（脚本化复盘）；不一致，说明当时结果依赖外部状态（网络、时间、被投毒的依赖）——差异点就是调查起点。

第三层 · 指标统计（会不会再发生）。 三个日常指标就够起步：deny 率与 top deny 命令（策略与工作流的摩擦点）、审批通过率（通过率长期趋近百分之百，说明审批门形同虚设，回看第 3.1 节「预览」教训）、超时率（沙箱内死循环或资源限额过紧）。

四、审计的下游：取证与演练的判据

审计日志不是合规摆设，它直接喂给两个下游场景：其一，第 4 章供应链治理——当恶意依赖或恶意 Skill 事发，command 数组与 reason 是「它到底让 Agent 执行了什么」的第一现场；其二，第 6 章红队自测——演练剧本的判据不是「感觉挡住了」，而是「攻击命令在 decision 事件的 deny 记录里」。在没有日志的地方谈安全，等于在没有病历的地方谈疗效。

本节要点回顾

• schema 关键字段：phase 四段（请求 / 决策 / 审批 / 结果）、command 永远存数组、审批记 reviewer、输出存摘要。
• auditor.py：JSONL 追加 + fsync + 哈希链；verify() 一行校验全链；防篡改不等于防重写，生产加远程转发。
• 回放三层：时间线重建、重放对照、指标统计；「连续 deny 后 allow」是试探指纹，审批通过率趋近全过是门失效信号。
• 审计是供应链取证与红队演练的共同判据——三层防线至此全部落地。

规则、沙箱、审批、审计已经咬合成环，但它守的只是「执行侧」。还有一整条战线本书尚未触及：Agent 吃进去的依赖、工具与 Skill 本身就可能是有毒的——恶意 Skill 已有官方 CVE 记录。第 4 章《供应链治理》从你刚落地的审计日志出发，去管住「进料」这一关。