Ayyadi266/soar-playbook
GitHub: Ayyadi266/soar-playbook
一个可审计的 SOAR 告警富化与分诊框架，通过严格分层架构和 fail-safe 设计确保安全决策的离线可重现性与 MTTR 的量化度量。
Stars: 0 | Forks: 0
# soar-playbook
[](../../actions/workflows/ci.yml)
[](../../actions/workflows/release-gate.yml)
一个可审计的 SOAR 富化和分诊 playbook：接收包含指标（indicators）的告警，结合威胁情报 API 对其进行富化、决策、幂等执行，并报告量化的 MTTR。
Python 3.11+。无数据库，无 Web UI，无 async，无 ML。
## 结果
结果待实时捕获 —— 参见 `soar preflight`。
## 本项目旨在解决的核心问题
富化自动化中真正致命的故障模式并非漏报，而是**自动关闭了一个根本没人看过的告警**。
这种故障有其特定的机制：在 pipeline 的某个环节，“我们没有数据”变成了“我们没发现任何坏东西”，进而变成了“这是安全的”。当 provider 返回的报告数为零，并以此得出 `0.0` 的分数时，它发生了。当查询报错，而空结果集满足了 `all(results_are_benign)` 时，它发生了。当某种 IOC 类型没有可用的 provider，而同样的空真值关闭了案例时，它也发生了。
这里的每一个结构性决策都是为了防止此类情况发生。
**Provider 只报告观察结果，从不作结论。** `EnrichmentState` 的状态为 `HAS_DATA / NO_DATA / ERROR`。“0.19 算不算坏？”这是一个组织层面的决策问题，应该由带有版本控制的策略文件来回答，而不是 API 响应的固有属性。
**`score` 的类型是 `float | None`，绝不默认为 `0.0`。** 默认值为零恰恰是将“缺乏证据”等同于“证据不存在”的具体机制。
**`Decision` 在不安全的情况下无法构建为 AUTO_CLOSE。** 只要存在任何性能降级、基于空的证据，或是基于任何非 BENIGN（良性）的判定，在构建时都会直接抛出异常。
### 曾经依然发布过一次 fail-open
当一次 fail-open 被发布时，上述所有机制其实都完好无损，而且问题就出在代码库中最不起眼的那个函数里。VirusTotal 解析器将缺失的检测计数默认为零，因此，如果 API 字段被重命名，代码就会面对一个无法理解的 payload，并得出一个确信的“安全”分数——这发生在所有防护机制的上游，因为没有任何防护机制会检查那里。大约 400 个通过的测试漏掉了它；而为另一个无关原因编写的变异测试（mutation pass）在几分钟内就发现了它。
**[阅读详细分析 →](docs/incidents/2026-07-virustotal-fail-open.md)**
## 架构
```
flowchart TB
subgraph ingest[" "]
A[alert.json] --> CLI[soar cli]
end
CLI --> P[pipeline]
P --> E[enrichers
abuseipdb · virustotal] E --> H[http layer
quota · retry · redaction] H -.->|"the only socket"| NET((threat-intel APIs)) E -->|EnrichmentResult
HAS_DATA / NO_DATA / ERROR| T[triage
pure function] POL[(policy yaml
versioned cut points)] --> T CAP[capability snapshot
passed as data] --> T T -->|Decision| ACT[actions] ACT --> L[(O_EXCL ledger
one action per key)] T --> AUD[audit] ACT --> AUD AUD --> CASE[(cases/*.json
raw payloads + decision)] AUD --> EV[(events.jsonl
monotonic durations)] EV --> M[report-mttr] CASE -->|offline, no network| R[replay] RED{{redaction}} -.->|every sink| AUD RED -.-> H classDef pure fill:#1f6f43,stroke:#0d3b23,color:#fff classDef store fill:#334155,stroke:#1e293b,color:#fff classDef net fill:#7c2d12,stroke:#431407,color:#fff class T pure class CASE,EV,L,POL store class NET net ``` 绿色代表纯函数 —— 没有 I/O、时钟或网络操作。红色是唯一会打开 socket 的路径。分层通过 `import-linter` 强制执行，而非口头约定： - **分层架构** —— 高层导入低层，绝不反向导入。 - **分诊不执行任何 I/O 且不读取时钟** —— 这一点是跨*全传递依赖图*检查的。策略类型存放在 `soar.models.policy` 中，正是为了让分诊在使用它们时无需触及 `soar.policy`（后者会导入 `pathlib` 和 `yaml`）。 - **决策路径绝不导入 enricher** —— `triage`、`policy`、`actions`、`audit`、`mttr` 和 `cli` 无法引用任何 provider。 - **Enricher 绝不导入决策路径** —— provider 代码无法针对自身的判定做特例处理。 以上每一条都经过了反向测试：将 `import time` 注入 `triage.py` 会破坏其中四分之三的规则。 ### 添加 provider 一个新的 provider 只需要在 `src/soar/enrichers/` 中添加一个模块，在 `registry.py` 中加一行代码，以及一个 fixture 目录 —— **无需修改** triage、actions、policy schema 或 CLI。能力（Capability）以*数据*的形式到达 triage，并被持久化到案例文件中，因此即使在添加了第三个 provider 之后回放旧案例，系统依然会推导出原始决策，而不是得出一个新决策。 ## 快速开始 ``` python -m pip install -e ".[dev]" cp .env.example .env # then fill in; .env is gitignored soar validate-policy policies/triage-v1.yaml soar run examples/alert-sample.json ``` ### 凭据 这两个 provider 都有免费层级。密钥应放在 `.env` 中，不存放到其他任何地方 —— 刻意没有提供 `--api-key` 参数，因为参数会将密钥暴露在 shell 历史记录和进程列表中。 | Provider | 免费层级 | 获取密钥地址 | | --- | --- | --- | | AbuseIPDB | 1,000 次检查/天 | |
| VirusTotal | 4 次请求/分钟，500 次/天 | |
```
ABUSEIPDB_API_KEY=your_key_here
VIRUSTOTAL_API_KEY=your_key_here
```
仅凭其中任意一个密钥就足以运行。如果两者都没有，每个告警都会进入 `NO_CAPABLE_PROVIDER` 并路由到人工复核 —— 这是正确的行为，也是 `scripts/seed-run.sh` 在这种状态下拒绝记录遥测数据的原因。
在未配置凭据的情况下，最后一条命令*理应*进入人工复核并返回退出码 3 —— 没有可用的 provider，因此什么也没有被检查：
```
decision: manual_review
rationale: degraded:no_capable_provider
degradations:
- no_capable_provider: no registered enricher supports ipv4
- no_capable_provider: no registered enricher supports sha256
```
### 命令
| 命令 | 功能说明 |
| --- | --- |
| `soar run ALERT.json` | 富化、分诊、执行、持久化。`--dry-run` 只做决策不执行；`--source NAME` 限制 provider。 |
| `soar replay CASE.json` | 从存储的 payload 中离线重新推导。`--policy` 会使用不同的策略对历史记录进行评估；`--allow-version-drift` 会在更改过的解析器代码下重新推导。 |
| `soar report-mttr` | 根据记录的遥测数据输出百分位数。 |
| `soar validate-policy P.yaml` | 拒绝结果不可达的策略。 |
| `soar preflight` | 发布门禁。如果任何声明缺乏证据，则返回非零值。 |
| `soar record-fixture PROVIDER IOC` | **唯一进行真实网络调用的命令。** |
### 退出码
退出码具有明确意义，因为本项目旨在通过调度器运行。
| 代码 | 含义 |
| --- | --- |
| 0 | 自动关闭，或命令成功上报 |
| 1 | 已升级处理 |
| 2 | 使用错误 |
| 3 | 需要人工复核 |
| 4 | 回放结果出现偏差 |
| 5 | 配置或策略错误 |
| 6 | `record-fixture` 中止：脱敏并非无操作 |
| 7 | 回放发生偏移（判定相同，但解析器代码不同） |
| 8 | preflight 失败：发布门禁已关闭 |
特意将 `7` 和 `0` 分开，将 `4` 和 `7` 分开：“今天的解析器一致同意”与“原始结果可重现”是截然不同的两个声明，下游的任何环节都不应将它们混为一谈。
## 离线重现决策
这里的声明是：在网络不可达的情况下，仅凭案例文件就足以推导出其原始判定。`soar replay` 就是这一声明的可执行证明。
```
$ soar replay cases/90/909c53ad...json
original: manual_review
re-derived: manual_review
result: MATCH
```
它会使用存储的 `FetchMeta`，在每个 provider 存储的原始 payload 上重新运行其纯函数 `parse()`，然后使用*存储的* `evaluated_at` 重新运行分诊 —— 因此，一个在 2020 年做出决策的案例在回放时会得出其原始判定，而不是显示为“现在一切都过期了”。
### 当解析器发生变动时
如果 provider 的 `version` 发生了变化，回放默认会**拒绝执行**，因为你不能用决策当时根本没运行过的代码来重现一个决策：
```
$ soar replay cases/90/909c53ad...json
soar: cannot replay 909c53ad4200: virustotal is at version 2.0.0 but the case
was decided under 1.0.0; the score mapping may differ. Pass
allow_version_drift to re-derive with current code, which produces a DRIFTED
result, not a clean reproduction.
```
`--allow-version-drift` 回答了另一个确实有价值的问题 —— *今天的解析器是否依然会得出相同的判定？* —— 且绝不将结果作为一次“重现”来呈现：
```
$ soar replay cases/90/909c53ad...json --allow-version-drift
original: auto_close
re-derived: auto_close
result: DRIFTED (same verdict, different parser code)
virustotal: decided under 1.0.0, replayed under 2.0.0
This is not a reproduction of the original decision. It shows that the current
parser reaches the same verdict from the stored payloads.
$ echo $?
7
```
这就是你针对真实历史记录测试解析器更改的方法：在新版本下回放语料库，看看哪些决策发生了变化。这种隔离通过三种方式强制执行：`DecisionRevision.is_drifted`、绝不打印 `MATCH` 字样，以及退出码 `7`。
## 不变性如何强制执行
| 不变性 | 机制 |
| --- | --- |
| Fail-safe，绝不 fail-open | `Decision` 拒绝构建不安全的 AUTO_CLOSE；`AutoCloseAction` 依然会进行二次检查并抛出异常 |
| Fail-safe | `Policy.effective_min_sources()` 将覆盖率限制在可用 provider 的数量内；`validate_against()` 会在加载时拒绝无法满足的策略 |
| 决策可重现 | `Enricher.parse()` 是纯函数；`soar replay` 可离线重新推导；`enricher_version` 让映射的变更变得可见 |
| 密钥绝不落盘或出现在日志中 | `SecretStr`，对所有输出端使用精确值 `Redactor`，在写入案例前执行 `assert_clean`，没有凭据参数，也没有基于文件的凭据读取路径 |
| 从第一次提交起就植入埋点 | `telemetry.stage()` 使用 `monotonic_ns` 进行测量；MTTR 的计算不读取任何其他数据 |
| 测试不触碰网络 | 自动生效的 socket 守卫，该守卫本身由 `tests/test_network_guard.py` 验证 |
| Fixtures 严格为 provider 的真实响应 | 除非脱敏操作是无操作，否则 `record-fixture` 会中止 |
```
pytest -q
ruff check .
mypy
lint-imports
soar preflight # currently, and correctly, fails
```
CI 在运行前四项测试时，环境中**没有任何密钥** —— 需要凭据的测试套件即是触碰网络的套件。
## MTTR 方法论
- **自动化时间是基于每个阶段的 `StageEvent` 记录测量的。** 没有任何代码路径会手动写入延迟数据。
- **使用百分位数而非均值。** 一次 30 秒的超时就能毁掉一个均值。采用最近排名法，因为插值法会凭空捏造一个请求从未经历过的延迟。
- **始终打印 `n`。** 当 n < 20 时，p95 会被*隐藏*，而不是被标注为“低置信度” —— 因为一个打印出来的数字总是会被引用，无论你附加什么警告。
- **配额等待时间**会从网络时间中剥离出来。
- **人工基准是一个估算值（ESTIMATE）**，需要引用来源，并且在每次渲染时都会进行标注。策略初始将其设为 `null`。
完整的操作流程 —— 告警格式、混合数据集的种子注入，以及人工计时方法 —— 详见 **[docs/runbook-mttr.md](docs/runbook-mttr.md)**。
## 局限性
这些都是刻意为之的记录。每一项都是决策，而非疏忽。
### 幂等性基于文件，而非事务
Actions 的去重基于 `f"{alert_uid}:{action_type}"`，并使用 `os.open(O_CREAT | O_EXCL)` 进行声明锁定 —— 在 POSIX 和 Windows 上均具备原子性，无需锁库，也没有关于 action 状态的第二个事实来源。
1. **崩溃窗口。** 一个已创建但未最终确定的声明会卡住告警。它会以 `UNRESOLVED_ACTION_CLAIM` 的状态呈现并路由到人工复核，因此崩溃需要人工介入 —— 但绝不会导致错误的关闭。
2. **无跨主机协调。** 两台拥有独立文件系统的主机会各自声明成功。这对于单主机 CLI 是正确的；但在两台机器上则是错误的。
3. **NFS。** `O_EXCL` 在 NFS 上不保证可靠的原子性。在网络共享的案例目录上，这一保证会退化为“通常”可靠。
这三点都是 fail-safe 的：它们都倾向于退化为人工复核，绝不会退化为静默关闭。这就是为什么它们是可以接受的，且不值得为了引入 SQLite 而增加第二套存储。
### 单主机架构设计
上述所有内容都源于一个选择：这是一个运行在单台机器上的 CLI，而不是一个服务。账本、速率限制器和案例存储都假定使用单一的文件系统，且同一时间只有一个进程。并发运行两个实例不是受支持的配置，这需要真正的分布式协调原语。
### 速率限制仅限于进程本地
两个并发运行的任务都认为自己拥有全部配额，因此会联合超出限制。此时的故障模式是 HTTP 429，重试机制会通过 `Retry-After` 进行处理，因此最终结果只是吞吐量下降，而不会导致错误的决策。
### 无富化缓存
每次运行都会重新查询每一个指标。在 4 次请求/分钟的免费层级下，这是主要成本所在；同一个 IOC 出现在十个告警中就会产生十次查询。基于 `(provider, ioc)` 键、存储原始 payload 的缓存已经设计好，但被刻意搁置：它会与数据过期、来源追踪以及 `_redaction: noop` 保证产生交互，如果在未充分测试的情况下将其与其他代码一起发布，将是这里最站不住脚的做法。
### 串行富化带来真实的延迟代价
刻意不使用 async。遥测系统会单独记录 `rate_limit_wait_ns`，因此报告中会将计算时间和配额等待时间分开显示。两者都是实际测量得出的。
###支持 VirusTotal URL
VT 使用 base64 标识符而非 URL 本身来定位 URL，而半吊子的编码会导致对错误对象产生自信的查询结果。因此，URL 没有可用的 provider，并总是被路由到人工复核。
### 跨 provider 的分数具有损耗性
VirusTotal 的 0.19（72 个引擎中的 14 个）与 AbuseIPDB 的 0.19（置信度 19）含义不同。`score_basis` 在每个结果中保留了归一化之前的推导过程，并且存在按 provider 设定的截断点覆盖机制，防止单一的全局阈值断言一种虚假的等价性。
### 覆盖率上限存在隐性影响
`min_sources_per_ioc` 受限于可用 provider 的数量；如果没有这个上限，对于任何只有单一 provider 的 IOC 类型来说，AUTO_CLOSE 将永远无法触达。代价是：单 provider 部署会静默地获得比其策略要求更弱的限制。这会在每个决策中被记录为 `coverage_caps_applied`，并通过 `soar validate-policy` 报告。
### 内置的 fixtures 是合成的
手工根据已发布的 API 文档编写，**非实时捕获**。每个 fixture 都带有 `"_provenance": "synthetic"` 标记，只要它们还存在，`soar preflight` 就会失败；而一旦出现未分类的捕获文件，`tests/test_fixture_shape.py` 就会立刻报错。参见 [`fixtures/README.md`](fixtures/README.md)。
### 静态穷举检查依赖于 mypy 运行
`EnrichmentState` 上的 `assert_never` 会在出现新状态时中断类型检查 —— 但前提是 CI 中运行了 `mypy`。
## 后续步骤
按顺序排列，且均不涉及代码：
1. **在 `.env` 中添加至少一个 API 密钥**（参见[凭据](#credentials)）。
2. **捕获六个实时的 fixture 案例** —— 运行 `./scripts/capture.sh`，然后在 `EXPECTED_STATE` 中声明每个新文件。
3. **运行告警语料库** —— 运行 `./scripts/seed-run.sh`。[`examples/alerts/`](examples/alerts/) 中已准备好 25 个告警，这些告警基于可验证的公共指标（EICAR、已发布的 DNS 解析器、RFC 5737 文档地址段）构建，混合比例大致为 1/3 恶意、1/3 干净、1/3 模糊。
4. **手动对 5 个告警进行计时**，并根据[操作手册](docs/runbook-mttr.md)在策略中记录中位数、其计算方法以及 `n` 的值。
5. **运行 `soar preflight`** 直到其通过，然后**仅**根据 `soar report-mttr` 的输出填写结果（Results）部分。
在此之后，按大致的优先级顺序排列：实现 `(provider, ioc)` 缓存；添加第三个 provider，利用真实的对象而非合成的测试替身来验证其可插拔性声明；支持 VT URL 标识符；如果未来需要跨越单机运行，则加入跨进程协调机制。
## 仓库结构
| 路径 | 内容 |
| --- | --- |
| `src/soar/models/` | 冻结的、`extra="forbid"` 的领域类型。无 I/O，无时钟。 |
| `src/soar/http/` | 唯一会打开 socket 的地方。 |
| `src/soar/enrichers/` | Provider 实现及其注册表。 |
| `src/soar/triage.py` | 纯决策函数。 |
| `src/soar/actions/` | 副作用处理及 O_EXCL 账本。 |
| `src/soar/preflight.py` | 发布门禁。 |
| `policies/` | 带版本控制的截断点。这是阈值配置的唯一存放地。 |
| `fixtures/` | Provider 响应（目前为合成数据）。 |
| `scripts/capture.sh` | 唯一触及网络的脚本。 |
| `docs/runbook-mttr.md` | MTTR 数据的强制生产规范。 |
| `docs/incidents/` | 事后复盘记录。 |
| `examples/alerts/` | 基于可验证的公共指标构建的 25 条告警语料库。 |
| `DEVELOPMENT.md` | 设计约束、强制执行机制及开发规范。 |
abuseipdb · virustotal] E --> H[http layer
quota · retry · redaction] H -.->|"the only socket"| NET((threat-intel APIs)) E -->|EnrichmentResult
HAS_DATA / NO_DATA / ERROR| T[triage
pure function] POL[(policy yaml
versioned cut points)] --> T CAP[capability snapshot
passed as data] --> T T -->|Decision| ACT[actions] ACT --> L[(O_EXCL ledger
one action per key)] T --> AUD[audit] ACT --> AUD AUD --> CASE[(cases/*.json
raw payloads + decision)] AUD --> EV[(events.jsonl
monotonic durations)] EV --> M[report-mttr] CASE -->|offline, no network| R[replay] RED{{redaction}} -.->|every sink| AUD RED -.-> H classDef pure fill:#1f6f43,stroke:#0d3b23,color:#fff classDef store fill:#334155,stroke:#1e293b,color:#fff classDef net fill:#7c2d12,stroke:#431407,color:#fff class T pure class CASE,EV,L,POL store class NET net ``` 绿色代表纯函数 —— 没有 I/O、时钟或网络操作。红色是唯一会打开 socket 的路径。分层通过 `import-linter` 强制执行，而非口头约定： - **分层架构** —— 高层导入低层，绝不反向导入。 - **分诊不执行任何 I/O 且不读取时钟** —— 这一点是跨*全传递依赖图*检查的。策略类型存放在 `soar.models.policy` 中，正是为了让分诊在使用它们时无需触及 `soar.policy`（后者会导入 `pathlib` 和 `yaml`）。 - **决策路径绝不导入 enricher** —— `triage`、`policy`、`actions`、`audit`、`mttr` 和 `cli` 无法引用任何 provider。 - **Enricher 绝不导入决策路径** —— provider 代码无法针对自身的判定做特例处理。 以上每一条都经过了反向测试：将 `import time` 注入 `triage.py` 会破坏其中四分之三的规则。 ### 添加 provider 一个新的 provider 只需要在 `src/soar/enrichers/` 中添加一个模块，在 `registry.py` 中加一行代码，以及一个 fixture 目录 —— **无需修改** triage、actions、policy schema 或 CLI。能力（Capability）以*数据*的形式到达 triage，并被持久化到案例文件中，因此即使在添加了第三个 provider 之后回放旧案例，系统依然会推导出原始决策，而不是得出一个新决策。 ## 快速开始 ``` python -m pip install -e ".[dev]" cp .env.example .env # then fill in; .env is gitignored soar validate-policy policies/triage-v1.yaml soar run examples/alert-sample.json ``` ### 凭据 这两个 provider 都有免费层级。密钥应放在 `.env` 中，不存放到其他任何地方 —— 刻意没有提供 `--api-key` 参数，因为参数会将密钥暴露在 shell 历史记录和进程列表中。 | Provider | 免费层级 | 获取密钥地址 | | --- | --- | --- | | AbuseIPDB | 1,000 次检查/天 |
标签：Python, SOAR, 告警分诊, 威胁情报, 安全规则引擎, 安全运营, 开发者工具, 扫描框架, 无后门, 自动化响应