SlateGitOrg/sentinel-triage
GitHub: SlateGitOrg/sentinel-triage
一个利用 LLM 辅助 SOC 告警分诊的工具，通过关联日志、匹配 Sigma 规则并自动总结，帮助分析师聚焦真实威胁并显著缩短平均分诊时间。
Stars: 0 | Forks: 0
# sentinel-triage
**在不让您的 SOC 误报狼来了的前提下，缩短平均分诊时间。**
## 真正的问题
SIEM 每天会生成数千个低保真警报。一线分析师将大部分工作时间花在手动关联跨用户、主机和 IP 的日志行上，仅仅为了弄清楚十次登录失败到底是一次攻击，还是某个在长周末忘记了密码的人——而等到他们手动完成关联时，疲劳感早已袭来，真正的安全事件也开始被遗漏。团队需要的是能够切实缩短平均分诊时间的自动化工具，而不是另一个嘈杂、产生幻觉并误报，导致分析师更加不信任工具的额外负担。
`sentinel-triage` 是这个问题的一个可独立构建、精简的切入点：单一日志源（端点/认证事件），关联成候选事件，与映射到 MITRE ATT&CK 的 Sigma 规则进行匹配，由必须引用其所使用的确切日志行的 LLM 进行总结，最后放入一个轻量级分析师处理队列——并配备一个评估工具，用于报告针对已标注数据集的精确率、召回率和误报率，从而用具体数字取代“它看起来很智能”的主观感觉。
## 架构
```
flowchart TD
A[Endpoint / auth logs
labeled dataset] --> B[Streaming ingestion
Redis Streams] B --> C[(OpenSearch
event store)] C --> D[Correlation engine
time-window + entity clustering] D --> E[Sigma / MITRE ATT&CK
detection layer] E --> F[LLM triage summarizer
citation grounding + confidence score] F --> G[Analyst review queue
approve / reject / escalate] D -.feeds.-> H[[Eval harness
precision / recall / FPR
on labeled set]] E -.feeds.-> H F -.feeds.-> H ``` | 阶段 | 模块 | 功能说明 | |---|---|---| | 数据接入 | `sentinel_triage.ingestion` | 从 CSV 加载规范事件模式；通过类 Redis-Streams 接口进行缓冲；索引至 OpenSearch（或用于离线运行时的内存模拟对象） | | 关联分析 | `sentinel_triage.correlation.entity_clustering` | 按实体（用户）分组事件，通过滑动时间窗口聚类生成候选事件 | | 检测 | `sentinel_triage.detection.sigma_engine` | 针对每个事件评估 Sigma 规则的实用子集（字段选择 + 特定规则时间范围内的 `count`/`distinct_count` 聚合），并映射至 MITRE ATT&CK 战术/技术 | | 分诊 | `sentinel_triage.triage` | 通过函数调用调用 LLM 进行总结，引用特定的 `event_id`，并分配带有置信度评分的严重性；每一条引用都会根据事件的真实事件进行机械化校验（`validate_citation_grounding`）——如果出现幻觉 id，评估将被拒绝，而不会获得默认信任 | | 审核 | `sentinel_triage.review_queue` | 由 SQLite 支持的分析师处理队列：批准 / 拒绝 / 升级，按严重性及随后按置信度排序 | | 评估 | `sentinel_triage.eval.harness` | 在标注数据集上运行整个 pipeline，并报告精确率 / 召回率 / F1 分数 / 误报率，包含总体和按攻击类型分类的结果 | ## 快速开始 需要 Python 3.11+。 ``` python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt pip install -e . # 生成/刷新 bundled synthetic labeled dataset（确定性，seeded） python scripts/generate_dataset.py # 运行 unit test suite（离线，无需 Docker/网络/API key） python -m pytest -v # 针对 labeled dataset 运行 eval harness python scripts/run_eval.py # 运行完整的离线 demo pipeline 并打印 analyst 所见内容 python scripts/ingest_and_triage.py ``` ### 可选：真实的 OpenSearch + Redis + Grafana 技术栈 ``` docker compose up -d docker compose exec app python scripts/ingest_and_triage.py ``` 这一步完全是可选的——上述所有代码路径均可在内存模拟对象上运行，绝不涉及 Docker、网络套接字或付费 API。这是一个深思熟虑的设计选择（参见下文的“设计上可离线验证”），而非缺失的功能。 ## 工作原理 **1. 数据接入。** `ingestion.loader` 从 CSV 中流式读取规范事件模式（`event_id`, `timestamp`, `host`, `user`, `src_ip`, `event_type`, `raw`, `country`）。`ingestion.stream.InMemoryEventStream` 使用与 Redis Stream（`XADD`/`XREAD`）相同的添加/读取游标结构来缓冲事件；`RedisEventStream` 是用于生产环境的真实适配器。`ingestion.opensearch_client.FakeOpenSearchIndexer` 将事件存储在内存中，并提供与真实的 `OpenSearchIndexer` 相同的过滤/范围查询接口。 **2. 关联分析。** `correlation.entity_clustering.correlate_events` 按实体（用户名）划分事件，然后在每个实体的按时间顺序排列的时间线中，只要与上一个事件的间隔超过关联窗口（默认 15 分钟），就会开启一个新的候选事件。这是沿时间轴的单链接聚类：一个持续 10 个事件的突发情况（其中连续事件之间相隔几分钟）会合并为一个事件，即使第一个和最后一个事件相隔 30 分钟以上，这也符合分析师实际阅读会话的方式。不同用户的记录绝不会仅仅因为时间戳重叠而被合并。 | 规则 | MITRE | 逻辑 | |---|---|---| | 暴力破解认证尝试 | T1110 / T1110.001 | 10分钟内出现 ≥5 次 `login_failure` | | 成功登录间的不可能旅行 | T1078 | 30分钟内出现来自 ≥2 个不同国家的 `login_success` | | 认证连续失败后的权限滥用 | T1078 / T1068 | 15分钟内出现 ≥3 次 `login_failure` **且** ≥1 次 `privilege_use` | | 登录失败突发后的账户锁定 | T1110.001 | 10分钟内出现 ≥4 次 `login_failure` **且** ≥1 次 `account_lockout` | **4. 分诊。** `triage.llm_client` 定义了一个单一的 OpenAI 工具调用函数模式 `submit_triage_assessment(summary, severity, confidence, cited_event_ids)`。`FakeLLMClient` 是该契约的一个确定性、离线实现（默认使用，也用于测试和评估工具），其引用直接从检测引擎匹配到的事件中构建，因此从结构上讲，它不可能引用事件范围之外的记录。`OpenAIFunctionCallingClient` 是真实的后端；通过向 `run_pipeline` / `evaluate_dataset` 传递一个实例（而不是 `FakeLLMClient()`）即可将其替换进来。无论哪种方式，`validate_citation_grounding` 都会在调用返回后针对真实事件重新校验每一条引用，如果每个被引用的 id 都是捏造的，则会引发 `GroundingError`（丢弃该分诊结果而不是选择信任它）。 **5. 审核队列。** `review_queue.queue.ReviewQueue` 是一个小型的 SQLite 表。经过分诊的事件以 `pending`（待处理）状态进入，并按严重性及随后按置信度进行排序；分析师对每个事件仅执行一次 `approve`（批准） / `reject`（拒绝） / `escalate`（升级）操作（对已做出决定的事件重新进行审核会引发 `InvalidTransitionError`，而不是悄无声息地覆盖审计追踪）。 **6. 评估工具。** `eval.harness.evaluate_dataset` 在已标注的 CSV 上运行整个 pipeline，并在*事件*级别（即分析师审核的单位）对其进行评分：如果一个事件包含的任何记录在源数据中被标记为恶意，则该事件为真实阳性（ground-truth positive）；如果分诊分配的严重性高于 `low`（等价于：至少触发了一条 Sigma 规则），则判定为预测阳性。 ### 实测结果（内置合成数据集，388 个事件 / 296 个候选事件） ``` precision: 1.0000 recall: 0.8333 f1: 0.9091 false_positive_rate: 0.0000 per-attack-type recall: brute_force 1.00, privilege_escalation 1.00, impossible_travel 0.50 ``` 使用 `python scripts/run_eval.py` 进行复现。客观存在的差距：四个“不可能旅行”事件中有两个被漏报，这是因为在特定场景下，两次登录的发生时间可能会超过 15 分钟的关联窗口，因此关联引擎在检测层将其作为一个整体处理之前，就将其拆分成了两个单一国家的事件——这是基于窗口的关联分析的客观局限性，而非检测规则的 Bug（参见“已知局限性”）。 ### 替换为真实数据集 任何包含 `event_id, timestamp, host, user, src_ip, event_type, raw`（以及可选的 `country`；对于评估，还需要 `is_malicious`/`attack_type`）列的 CSV 都无需修改即可配合 `ingestion.load_events_from_csv` 使用。这个精简的原型项目附带了一个由 `scripts/generate_dataset.py` 生成的合成数据集——其模型结构参考了公开的认证日志语料库（例如 LANL 综合多源网络安全事件数据集）——而不是直接分发庞大的公共数据集，这样既能保持代码库体积小巧，又能确保测试的封闭性。将 `--dataset` 指向真实的数据集（在完成其列名映射之后）即可获得真实的评估数据。 ## 已知局限性 - **关联窗口 / 规则时间范围不匹配。** 如上所述：如果一次攻击的事件跨度超过了实体关联窗口，检测层将永远无法将它们视为一个整体事件。生产级版本应该在重叠窗口上进行滚动关联，而不是单次贪心遍历。 - **单一日志源。** 目前仅对端点/认证事件进行了建模。真实 SOC 环境中的噪音（网络、DNS、EDR 遥测数据）不在此切片版本的范围内。 - **Sigma 子集，而非完整的 Sigma。** 不支持通配符/正则表达式字段匹配，不支持嵌套的布尔值括号，也不支持 `1 of selection*` 语法。需要这些功能的规则必须进行转换。 - **FakeLLMClient 只是一个替身。** 它证明了落地校验契约并能提供确定性的、可测试的数值；真实的 LLM 会更好地措辞总结，但仍然必须通过相同的 `validate_citation_grounding` 校验。 ## 拓展目标（已记录在文档中，但尚未实现） - 支持多格式接入（syslog, CEF, JSON）及格式自动检测，超越此处构建的单一 CSV/认证日志格式。 - 完整集成 pySigma 后端，以支持完整的 Sigma 条件语法和社区规则包。 - 跨实体关联（例如，追踪共享同一源 IP 的多个用户间的横向移动），而不是仅仅局限于单实体时间窗口聚类。 - 提供真实的 Grafana 仪表盘定义（`docker-compose.yml` 启动了 Grafana，但尚未配置任何仪表盘 JSON）。 ## 项目结构 ``` src/sentinel_triage/ events.py canonical Event schema ingestion/ CSV loader, Redis-Streams-shaped buffer, OpenSearch (+fake) indexer correlation/ time-window + entity clustering detection/ Sigma/MITRE rule engine triage/ LLM function-calling client (fake + real) + citation grounding review_queue/ SQLite analyst queue eval/ precision/recall/FPR harness pipeline.py wires it all together sigma_rules/ 4 bundled detection rules (YAML) data/ bundled synthetic labeled dataset scripts/ generate_dataset.py, run_eval.py, ingest_and_triage.py tests/ pytest suite (unit + offline integration) ``` ## 测试 ``` python -m pytest -v # full offline suite python -m pytest -v --cov=sentinel_triage # with coverage python -m pytest -m integration # (none bundled runnable offline; documents the marker) ``` CI (`.github/workflows/ci.yml`) 会在每次 push/PR 时运行数据集生成确定性检查、完整的测试套件、评估工具以及 Docker 构建任务。 ## 设计上可离线验证 构建或测试此项目不需要任何密钥、付费 API 调用，也不需要运行 Docker 守护进程或集群。`FakeOpenSearchIndexer`、`InMemoryEventStream` 和 `FakeLLMClient` 实现了与真实组件（`OpenSearchIndexer`、`RedisEventStream`、`OpenAIFunctionCallingClient`）完全相同的接口，这些真实组件采用延迟导入机制，只有在您明确实例化它们时才会被调用。唯一一个标记了 `@pytest.mark.integration` 的测试明确了这一边界，并通过 `pyproject.toml` 的 `addopts` 配置将其排除在默认的 `pytest` 运行之外。 ## 许可证 MIT — 查看 [许可证](LICENSE)。
labeled dataset] --> B[Streaming ingestion
Redis Streams] B --> C[(OpenSearch
event store)] C --> D[Correlation engine
time-window + entity clustering] D --> E[Sigma / MITRE ATT&CK
detection layer] E --> F[LLM triage summarizer
citation grounding + confidence score] F --> G[Analyst review queue
approve / reject / escalate] D -.feeds.-> H[[Eval harness
precision / recall / FPR
on labeled set]] E -.feeds.-> H F -.feeds.-> H ``` | 阶段 | 模块 | 功能说明 | |---|---|---| | 数据接入 | `sentinel_triage.ingestion` | 从 CSV 加载规范事件模式；通过类 Redis-Streams 接口进行缓冲；索引至 OpenSearch（或用于离线运行时的内存模拟对象） | | 关联分析 | `sentinel_triage.correlation.entity_clustering` | 按实体（用户）分组事件，通过滑动时间窗口聚类生成候选事件 | | 检测 | `sentinel_triage.detection.sigma_engine` | 针对每个事件评估 Sigma 规则的实用子集（字段选择 + 特定规则时间范围内的 `count`/`distinct_count` 聚合），并映射至 MITRE ATT&CK 战术/技术 | | 分诊 | `sentinel_triage.triage` | 通过函数调用调用 LLM 进行总结，引用特定的 `event_id`，并分配带有置信度评分的严重性；每一条引用都会根据事件的真实事件进行机械化校验（`validate_citation_grounding`）——如果出现幻觉 id，评估将被拒绝，而不会获得默认信任 | | 审核 | `sentinel_triage.review_queue` | 由 SQLite 支持的分析师处理队列：批准 / 拒绝 / 升级，按严重性及随后按置信度排序 | | 评估 | `sentinel_triage.eval.harness` | 在标注数据集上运行整个 pipeline，并报告精确率 / 召回率 / F1 分数 / 误报率，包含总体和按攻击类型分类的结果 | ## 快速开始 需要 Python 3.11+。 ``` python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt pip install -e . # 生成/刷新 bundled synthetic labeled dataset（确定性，seeded） python scripts/generate_dataset.py # 运行 unit test suite（离线，无需 Docker/网络/API key） python -m pytest -v # 针对 labeled dataset 运行 eval harness python scripts/run_eval.py # 运行完整的离线 demo pipeline 并打印 analyst 所见内容 python scripts/ingest_and_triage.py ``` ### 可选：真实的 OpenSearch + Redis + Grafana 技术栈 ``` docker compose up -d docker compose exec app python scripts/ingest_and_triage.py ``` 这一步完全是可选的——上述所有代码路径均可在内存模拟对象上运行，绝不涉及 Docker、网络套接字或付费 API。这是一个深思熟虑的设计选择（参见下文的“设计上可离线验证”），而非缺失的功能。 ## 工作原理 **1. 数据接入。** `ingestion.loader` 从 CSV 中流式读取规范事件模式（`event_id`, `timestamp`, `host`, `user`, `src_ip`, `event_type`, `raw`, `country`）。`ingestion.stream.InMemoryEventStream` 使用与 Redis Stream（`XADD`/`XREAD`）相同的添加/读取游标结构来缓冲事件；`RedisEventStream` 是用于生产环境的真实适配器。`ingestion.opensearch_client.FakeOpenSearchIndexer` 将事件存储在内存中，并提供与真实的 `OpenSearchIndexer` 相同的过滤/范围查询接口。 **2. 关联分析。** `correlation.entity_clustering.correlate_events` 按实体（用户名）划分事件，然后在每个实体的按时间顺序排列的时间线中，只要与上一个事件的间隔超过关联窗口（默认 15 分钟），就会开启一个新的候选事件。这是沿时间轴的单链接聚类：一个持续 10 个事件的突发情况（其中连续事件之间相隔几分钟）会合并为一个事件，即使第一个和最后一个事件相隔 30 分钟以上，这也符合分析师实际阅读会话的方式。不同用户的记录绝不会仅仅因为时间戳重叠而被合并。 | 规则 | MITRE | 逻辑 | |---|---|---| | 暴力破解认证尝试 | T1110 / T1110.001 | 10分钟内出现 ≥5 次 `login_failure` | | 成功登录间的不可能旅行 | T1078 | 30分钟内出现来自 ≥2 个不同国家的 `login_success` | | 认证连续失败后的权限滥用 | T1078 / T1068 | 15分钟内出现 ≥3 次 `login_failure` **且** ≥1 次 `privilege_use` | | 登录失败突发后的账户锁定 | T1110.001 | 10分钟内出现 ≥4 次 `login_failure` **且** ≥1 次 `account_lockout` | **4. 分诊。** `triage.llm_client` 定义了一个单一的 OpenAI 工具调用函数模式 `submit_triage_assessment(summary, severity, confidence, cited_event_ids)`。`FakeLLMClient` 是该契约的一个确定性、离线实现（默认使用，也用于测试和评估工具），其引用直接从检测引擎匹配到的事件中构建，因此从结构上讲，它不可能引用事件范围之外的记录。`OpenAIFunctionCallingClient` 是真实的后端；通过向 `run_pipeline` / `evaluate_dataset` 传递一个实例（而不是 `FakeLLMClient()`）即可将其替换进来。无论哪种方式，`validate_citation_grounding` 都会在调用返回后针对真实事件重新校验每一条引用，如果每个被引用的 id 都是捏造的，则会引发 `GroundingError`（丢弃该分诊结果而不是选择信任它）。 **5. 审核队列。** `review_queue.queue.ReviewQueue` 是一个小型的 SQLite 表。经过分诊的事件以 `pending`（待处理）状态进入，并按严重性及随后按置信度进行排序；分析师对每个事件仅执行一次 `approve`（批准） / `reject`（拒绝） / `escalate`（升级）操作（对已做出决定的事件重新进行审核会引发 `InvalidTransitionError`，而不是悄无声息地覆盖审计追踪）。 **6. 评估工具。** `eval.harness.evaluate_dataset` 在已标注的 CSV 上运行整个 pipeline，并在*事件*级别（即分析师审核的单位）对其进行评分：如果一个事件包含的任何记录在源数据中被标记为恶意，则该事件为真实阳性（ground-truth positive）；如果分诊分配的严重性高于 `low`（等价于：至少触发了一条 Sigma 规则），则判定为预测阳性。 ### 实测结果（内置合成数据集，388 个事件 / 296 个候选事件） ``` precision: 1.0000 recall: 0.8333 f1: 0.9091 false_positive_rate: 0.0000 per-attack-type recall: brute_force 1.00, privilege_escalation 1.00, impossible_travel 0.50 ``` 使用 `python scripts/run_eval.py` 进行复现。客观存在的差距：四个“不可能旅行”事件中有两个被漏报，这是因为在特定场景下，两次登录的发生时间可能会超过 15 分钟的关联窗口，因此关联引擎在检测层将其作为一个整体处理之前，就将其拆分成了两个单一国家的事件——这是基于窗口的关联分析的客观局限性，而非检测规则的 Bug（参见“已知局限性”）。 ### 替换为真实数据集 任何包含 `event_id, timestamp, host, user, src_ip, event_type, raw`（以及可选的 `country`；对于评估，还需要 `is_malicious`/`attack_type`）列的 CSV 都无需修改即可配合 `ingestion.load_events_from_csv` 使用。这个精简的原型项目附带了一个由 `scripts/generate_dataset.py` 生成的合成数据集——其模型结构参考了公开的认证日志语料库（例如 LANL 综合多源网络安全事件数据集）——而不是直接分发庞大的公共数据集，这样既能保持代码库体积小巧，又能确保测试的封闭性。将 `--dataset` 指向真实的数据集（在完成其列名映射之后）即可获得真实的评估数据。 ## 已知局限性 - **关联窗口 / 规则时间范围不匹配。** 如上所述：如果一次攻击的事件跨度超过了实体关联窗口，检测层将永远无法将它们视为一个整体事件。生产级版本应该在重叠窗口上进行滚动关联，而不是单次贪心遍历。 - **单一日志源。** 目前仅对端点/认证事件进行了建模。真实 SOC 环境中的噪音（网络、DNS、EDR 遥测数据）不在此切片版本的范围内。 - **Sigma 子集，而非完整的 Sigma。** 不支持通配符/正则表达式字段匹配，不支持嵌套的布尔值括号，也不支持 `1 of selection*` 语法。需要这些功能的规则必须进行转换。 - **FakeLLMClient 只是一个替身。** 它证明了落地校验契约并能提供确定性的、可测试的数值；真实的 LLM 会更好地措辞总结，但仍然必须通过相同的 `validate_citation_grounding` 校验。 ## 拓展目标（已记录在文档中，但尚未实现） - 支持多格式接入（syslog, CEF, JSON）及格式自动检测，超越此处构建的单一 CSV/认证日志格式。 - 完整集成 pySigma 后端，以支持完整的 Sigma 条件语法和社区规则包。 - 跨实体关联（例如，追踪共享同一源 IP 的多个用户间的横向移动），而不是仅仅局限于单实体时间窗口聚类。 - 提供真实的 Grafana 仪表盘定义（`docker-compose.yml` 启动了 Grafana，但尚未配置任何仪表盘 JSON）。 ## 项目结构 ``` src/sentinel_triage/ events.py canonical Event schema ingestion/ CSV loader, Redis-Streams-shaped buffer, OpenSearch (+fake) indexer correlation/ time-window + entity clustering detection/ Sigma/MITRE rule engine triage/ LLM function-calling client (fake + real) + citation grounding review_queue/ SQLite analyst queue eval/ precision/recall/FPR harness pipeline.py wires it all together sigma_rules/ 4 bundled detection rules (YAML) data/ bundled synthetic labeled dataset scripts/ generate_dataset.py, run_eval.py, ingest_and_triage.py tests/ pytest suite (unit + offline integration) ``` ## 测试 ``` python -m pytest -v # full offline suite python -m pytest -v --cov=sentinel_triage # with coverage python -m pytest -m integration # (none bundled runnable offline; documents the marker) ``` CI (`.github/workflows/ci.yml`) 会在每次 push/PR 时运行数据集生成确定性检查、完整的测试套件、评估工具以及 Docker 构建任务。 ## 设计上可离线验证 构建或测试此项目不需要任何密钥、付费 API 调用，也不需要运行 Docker 守护进程或集群。`FakeOpenSearchIndexer`、`InMemoryEventStream` 和 `FakeLLMClient` 实现了与真实组件（`OpenSearchIndexer`、`RedisEventStream`、`OpenAIFunctionCallingClient`）完全相同的接口，这些真实组件采用延迟导入机制，只有在您明确实例化它们时才会被调用。唯一一个标记了 `@pytest.mark.integration` 的测试明确了这一边界，并通过 `pyproject.toml` 的 `addopts` 配置将其排除在默认的 `pytest` 运行之外。 ## 许可证 MIT — 查看 [许可证](LICENSE)。
标签：Petitpotam, Sigma规则, 告警分诊, 安全运营, 扫描框架, 搜索引擎查询, 目标导入, 请求拦截, 逆向工具