ChengenHsieh0225/kafka-sigma-engine
GitHub: ChengenHsieh0225/kafka-sigma-engine
基于 Kafka、Elasticsearch 和 Kubernetes 构建的实时安全日志分析引擎，使用 Sigma 规则进行威胁检测。
Stars: 0 | Forks: 0
# Kafka Sigma Engine
一个可水平扩展的日志接入与威胁检测 pipeline，用于模拟云原生 XDR (Extended Detection and Response) 平台。该系统接入合成安全事件，使用水平扩展的 worker 池根据 [Sigma](https://github.com/SigmaHQ/sigma) 检测规则对其进行实时评估，并将匹配的 Alerts 持久化到 Elasticsearch —— 所有这些都可通过实时的 Grafana dashboard 进行观察。在 4-CPU 的 minikube 实例上，持续吞吐量达到 ≥ 5,000 EPS 且零延迟增长；该架构在专用生产节点上的目标为 10,000+ EPS。
## 目录
- [项目概述](#project-overview)
- [架构与数据流](#architecture--data-flow)
- [项目结构](#project-structure)
- [快速入门 — Kubernetes (minikube)](#getting-started--kubernetes-minikube)
- [配置](#configuration)
- [测试指南](#testing-guide)
- [可观测性](#observability)
- [关键工程决策](#key-engineering-decisions)
## 项目概述
| 能力 | 实现 |
|---|---|
| 高吞吐量接入 | Kafka `raw-logs` topic，8 个分区，以源主机为 key |
| 并行威胁检测 | 最多 8 个 asyncio Rule Engine workers（每个分区一个） |
| 内存规则评估 | Level 2 Sigma 条件解析器 —— 字段相等、字符串修饰符、布尔逻辑 |
| 时间窗口聚合 | 每个 worker 的滑动窗口无需外部状态即可检测突发模式（例如暴力破解） |
| 容错处理 | 手动至少一次 Kafka offset 提交；重启时不会丢失日志 |
| 实时规则管理 | 带有类型化 JSON 封装的 `rule-updates` topic —— 无需重启即可添加、更新和删除规则 |
| Alert 去重 | Elasticsearch 的 `_id` 设为 `alert_id`；幂等 upsert 消除了 Kafka 至少一次产生的重复 |
| 逼真的负载生成 | 每台主机的状态机产生相关的攻击序列；HTTP admin endpoint 用于运行时 EPS 控制 |
| 高效的 Alert 存储 | 微批量刷新至 Elasticsearch `_bulk` API（数量 ≥ 500 或耗时 ≥ 5 秒） |
| 实时可观测性 | Prometheus metrics + 预构建的 Grafana dashboard |
| Kubernetes 部署 | `k8s/` 中的完整 stack 针对 minikube；Rule Engine 可通过 `kubectl scale` 扩展至 8 个副本 |
**实测性能（4-CPU / 8 GB minikube，8 个 Rule Engine 副本）：**
| 指标 | 实测结果 | 备注 |
|---|---|---|
| 持续吞吐量 | **≥ 5,000 EPS** | 在 minikube 上测试至 5,000 EPS，延迟保持接近于零 |
| 单事件规则评估 | 亚毫秒级 | 8 条规则，纯 Python，无 I/O |
| worker 重启时的日志丢失 | 零 | 至少一次提交；重复项由 `alert_id` 去重 |
吞吐量的实现归功于即发即忘的 alert 发布（`producer.send()`）和批量消费者 offset 提交（每 100 条消息或 5 秒一次）。10,000+ EPS 的目标需要专用的 Kubernetes 节点，以消除服务间的 CPU 竞争。
## 架构与数据流
```
┌───────────────────────────────────────────────────┐
│ Log Generator (Python asyncio) │
│ │
│ HostStateMachine — per-host states: │
│ idle → brute_forcing → compromised │
│ → lateral_moving │
│ │
│ HTTP admin POST /rate GET /rate :8080 │
└───────────────────────┬───────────────────────────┘
│ raw-logs (8 partitions, keyed by host)
▼
┌───────────────────────────────────────────────────────────────────┐
│ Apache Kafka │
│ raw-logs (8 partitions) │ alerts (1 partition) │ rule-updates │
└───┬───────────────────────────────┬──────────────────┬───────────┘
│ │ │
│ Consumer group: rule-engine │ │ Fan-out: unique
▼ │ │ group per worker
┌──────────────────────────────┐ │ ┌────────┴────────┐
│ Rule Engine Worker ×N │ │ │ rule-updates │
│ (N ≤ 8, Python asyncio) │ │ │ consumer per │
│ │ │ │ worker (fan-out)│
│ RuleEngineService │ │ └─────────────────┘
│ ├─ load_rules() on startup │ │
│ ├─ apply_rule_update() │ │ Envelope format (ADR-0011):
│ │ add | update | delete │ │ {"op":"add"|"update"|"delete",
│ ├─ evaluate() — single-log │────┘ "rule_id":"...", "rule":{...}}
│ ├─ SlidingWindow — bursts │
│ └─ /metrics :8001 │
└──────────────────────────────┘
│ alerts (1 partition)
▼
┌──────────────────────────────┐
│ Alert Storage Service │ Micro-batch buffer → Elasticsearch _bulk
│ (Python asyncio) │ _id = alert_id → idempotent upsert (ADR-0014)
│ Flush: size ≥ 500 OR 5 s │
└──────────────┬───────────────┘
▼
┌──────────────────────┐
│ Elasticsearch │ Index: alerts (alert_id, rule_id, host → keyword)
│ (single-node) │
└──────────────────────┘
```
### Kafka Topics
| Topic | 分区 | Key | 用途 |
|---|---|---|---|
| `raw-logs` | 8 | `host` 字段 (UTF-8) | Raw Log 流；来自同一主机的所有事件始终路由到同一个分区 |
| `alerts` | 1 | — | 由 Alert Storage 消费的匹配 Alert 流 |
| `rule-updates` | 1 | — | 用于添加/更新/删除规则生命周期操作的类型化 JSON 封装；分发给所有 workers |
### Rule Engine 并发
每个 worker 进程都由 Kafka 的 consumer-group 重均衡器分配一个或多个 `raw-logs` 分区。分区数为 8；1 到 8 之间的任何副本数都是有效的 —— Kafka 会自动分配分区。由于 `raw-logs` 以主机为 key，来自同一台机器的所有事件总是会到达同一个 worker，这使得每个 worker 在内存中的滑动窗口无需任何跨 worker 协调即可保持正确 (ADR-0010)。
对于 `rule-updates`，每个 worker 使用**唯一的 consumer group ID**（派生自其 `WORKER_ID`），以便每个 worker 都能收到每一个规则封装 —— 这是分发（fan-out），而不是竞争消费。
### Sigma 规则支持
Rule Engine 实现了自己的 YAML 到条件解析器，支持：
- **字段相等：** `event_id: '4625'`
- **字符串修饰符：** `field|contains`, `field|startswith`, `field|endswith`
- **列表值 (OR)：** `event_id: ['4624', '4625', '4648']`
- **布尔逻辑：** `and`, `or`, `not`，括号表达式
- **聚合 (时间窗口)：** `selection | count() by host > N` 带有 `timeframe: s` 键 —— 通过每个 worker 的 `SlidingWindow` 进行评估
### Log Generator 状态机
Log Generator 运行每台主机的状态机 (ADR-0016)，产生相关的攻击序列，而不是纯随机事件：
| 状态 | 触发的事件 |
|---|---|
| `idle` | 随机混合 —— 基准噪声 |
| `brute_forcing` | 快速 `event_id 4625`（登录失败）突发 |
| `compromised` | `event_id 4624`（登录成功） → `event_id 4672`（特权使用） |
| `lateral_moving` | 带有可疑 `process_name` 的 `event_id 4688`（进程创建） |
转移概率是加权随机的，因此主机会自然地循环经历各个攻击阶段。可以通过 HTTP admin endpoint 在不重启容器的情况下更改运行时 EPS 速率。
## 项目结构
```
kafka-sigma-engine/
├── CONTEXT.md # Single-context domain doc: data flow, glossary, constraints
├── pyproject.toml # Project metadata, pytest config, mypy config
├── requirements.txt # Pinned runtime dependencies
│
├── docs/
│ ├── adr/ # Architecture Decision Records (ADR-0001 … ADR-0016)
│ └── agents/ # Agent skill docs: issue-tracker, triage-labels, domain
│
├── k8s/ # Kubernetes manifests (minikube)
│ ├── namespace.yaml
│ ├── kafka/
│ │ ├── kafka.yaml # Apache Kafka 3.8.1 StatefulSet, Services, topic-provisioning Job
│ │ └── values.yaml # Legacy Bitnami Helm chart values (reference only)
│ ├── elasticsearch/ # Deployment + ClusterIP Service
│ ├── prometheus/ # Deployment, Service, ConfigMap, RBAC for kubernetes_sd_configs
│ ├── grafana/ # Deployment, Service, ConfigMaps (datasource, dashboard provider, dashboard JSON)
│ ├── rule-engine/ # Deployment + headless Service
│ ├── alert-storage/ # Deployment
│ └── log-generator/ # Deployment + LoadBalancer Service (port 8080)
│
├── services/ # One Dockerfile per microservice
│ ├── alert_storage/
│ ├── log_generator/
│ └── rule_engine/
│
├── sigma_rules/ # Bundled Sigma Rule YAML files (loaded at startup)
│ ├── windows_failed_login.yml # event_id 4625 → medium
│ ├── windows_successful_login.yml # event_id 4624 → low
│ ├── windows_explicit_credentials_logon.yml # event_id 4648 → medium
│ ├── windows_privilege_use.yml # event_id 4672 → high
│ ├── windows_suspicious_process.yml # event_id 4688 + process_name|contains → medium
│ ├── windows_brute_force.yml # aggregation: >5 × 4625 within 60 s → high
│ ├── cloudtrail_bucket_delete.yml # action startswith DeleteBucket → high
│ └── cloudtrail_iam_user_create.yml # action = CreateUser → high
│
├── src/
│ ├── exceptions.py # Domain exception types
│ ├── models.py # SigmaRule, Alert dataclasses
│ ├── alert_storage/
│ │ ├── main.py # Service entry point; _ESIndexer sets _id=alert_id
│ │ └── service.py # AlertStorageService: micro-batch buffer + flush logic
│ ├── log_generator/
│ │ ├── __main__.py # Entry point; wires HostStateMachine + admin server
│ │ ├── generator.py # generate_raw_log() + HostStateMachine (state machine)
│ │ ├── admin.py # LogAdminHandler: GET/POST /rate (ADR-0015)
│ │ └── service.py # LogGeneratorService: async publish loop + EPS control
│ └── rule_engine/
│ ├── main.py # Entry point: raw-logs consumer + rule-updates fan-out
│ ├── evaluator.py # evaluate(raw_log, rules) → list[Alert] (single-log)
│ ├── loader.py # load_rules(path) → list[SigmaRule]
│ ├── service.py # RuleEngineService: rule set management + evaluate_log()
│ └── window.py # SlidingWindow: per-host time-window event counter
│
└── tests/
├── conftest.py
├── integration/
│ └── test_pipeline.py # 11 end-to-end tests against the live minikube stack
└── unit/
├── test_aggregation_rule.py # SlidingWindow + brute-force rule evaluation
├── test_alert_storage.py # AlertStorageService: micro-batch + flush logic
├── test_k8s_manifests.py # Smoke-checks on k8s YAML structure
├── test_log_generator.py # LogGeneratorService: publish loop + EPS control
├── test_log_generator_admin.py # LogAdminHandler: GET/POST /rate
├── test_log_generator_state_machine.py # HostStateMachine state transitions
├── test_models.py # SigmaRule, Alert dataclasses
├── test_rule_engine.py # evaluate(): field equality, modifiers, boolean logic
├── test_rule_engine_service.py # RuleEngineService: rule lifecycle + evaluate_log()
├── test_sigma_rules.py # All bundled .yml rules parse and match correctly
└── test_window.py # SlidingWindow: eviction, threshold, edge cases
```
## 快速入门 — Kubernetes (minikube)
`k8s/` 目录在 minikube 上部署完整的 stack，其中 Rule Engine 以 8 个副本运行。
### 前置条件
- 带有 Docker 驱动程序的 [minikube](https://minikube.sigs.k8s.io/docs/start/)
- [kubectl](https://kubernetes.io/docs/tasks/tools/)
### 1. 克隆仓库
```
git clone git@github.com:ChengenHsieh0225/kafka-sigma-engine.git
cd kafka-sigma-engine
```
### 2. 启动 minikube
```
minikube start --driver=docker --cpus=4 --memory=8g
```
### 3. 将 Docker 指向 minikube 的守护进程并构建镜像
```
eval $(minikube docker-env)
docker build -t kafka-sigma-engine/rule-engine:latest -f services/rule_engine/Dockerfile .
docker build -t kafka-sigma-engine/alert-storage:latest -f services/alert_storage/Dockerfile .
docker build -t kafka-sigma-engine/log-generator:latest -f services/log_generator/Dockerfile .
```
### 4. 部署 Kafka
```
kubectl apply -f k8s/namespace.yaml
kubectl apply -f k8s/kafka/kafka.yaml
```
等待 Kafka 准备就绪并且 topic 配置 Job 完成：
```
kubectl wait --for=condition=ready pod/kafka-controller-0 -n kafka-sigma-engine --timeout=120s
kubectl wait --for=condition=complete job/kafka-topic-provisioning -n kafka-sigma-engine --timeout=120s
```
### 5. 部署剩余 stack
```
kubectl apply -f k8s/elasticsearch/
kubectl apply -f k8s/prometheus/
kubectl apply -f k8s/grafana/
kubectl apply -f k8s/rule-engine/
kubectl apply -f k8s/alert-storage/
kubectl apply -f k8s/log-generator/
```
等待所有 pods 达到 Running 状态：
```
kubectl get pods -n kafka-sigma-engine --watch
```
### 6. 访问 dashboard
```
minikube tunnel # Keep running in a separate terminal
```
获取每个服务的 URL 并在浏览器中打开：
```
minikube service grafana -n kafka-sigma-engine --url # Grafana
minikube service prometheus -n kafka-sigma-engine --url # Prometheus
```
- **minikube dashboard：** `minikube dashboard` —— 然后将命名空间下拉菜单（左上角）从 **default** 切换到 **kafka-sigma-engine** 以查看 pods
### 7. 扩展 Rule Engine
```
# 扩展至 8 个 replicas（最大值 — 匹配 8 个 partitions）
kubectl scale deployment rule-engine -n kafka-sigma-engine --replicas=8
# 缩减 — Kafka 自动重新分配 partitions
kubectl scale deployment rule-engine -n kafka-sigma-engine --replicas=4
```
### 8. 调整 Log Generator 速率
```
LOG_GEN_URL=$(minikube service log-generator -n kafka-sigma-engine --url)
curl -X POST "$LOG_GEN_URL/rate" -H "Content-Type: application/json" -d '{"eps": 5000}'
curl "$LOG_GEN_URL/rate"
```
### 9. 在运行时管理规则 (热重载)
可以在 pipeline 运行时添加、更新或删除规则。向 `rule-updates` topic 发布类型化的 JSON 封装 —— 所有 workers 无需重启即可应用更改。
**添加新规则：**
```
kubectl exec -n kafka-sigma-engine kafka-controller-0 -- \
/opt/kafka/bin/kafka-console-producer.sh \
--bootstrap-server localhost:9092 \
--topic rule-updates <<'EOF'
{"op":"add","rule_id":"win-rdp-001","rule":{"id":"win-rdp-001","title":"RDP Logon","level":"medium","detection":{"sel":{"log_type":"windows_event","event_id":"4624"},"condition":"sel"}}}
EOF
```
**删除规则：**
```
kubectl exec -n kafka-sigma-engine kafka-controller-0 -- \
/opt/kafka/bin/kafka-console-producer.sh \
--bootstrap-server localhost:9092 \
--topic rule-updates <<'EOF'
{"op":"delete","rule_id":"win-rdp-001"}
EOF
```
**更新规则** —— 发布一个带有相同 `rule_id` 和更新后规则体的 `"op":"update"` 封装。
### 10. 运行集成测试
Kafka 在其元数据响应中广播其内部 pod 主机名。将其添加到 `/etc/hosts`，以便它可以通过端口转发隧道解析（每台机器只需设置一次）：
```
echo "127.0.0.1 kafka-controller-0.kafka-controller-headless.kafka-sigma-engine.svc.cluster.local" \
| sudo tee -a /etc/hosts
```
在运行测试前将 Log Generator 扩展为 0。在 1000 EPS 时，它会淹没 `raw-logs` 分区 —— 测试消息会被掩埋在积压中并超时：
```
kubectl scale deployment log-generator -n kafka-sigma-engine --replicas=0
```
将 Kafka 和 Elasticsearch 转发到 localhost。如果已在运行则跳过（使用 `lsof -ti tcp:9092` 检查）：
```
kubectl port-forward -n kafka-sigma-engine svc/kafka 9092:9092 &
kubectl port-forward -n kafka-sigma-engine svc/elasticsearch 9200:9200 &
```
运行测试套件：
```
pytest -m integration tests/integration/
```
完成后恢复 Log Generator：
```
kubectl scale deployment log-generator -n kafka-sigma-engine --replicas=1
```
## 配置
所有服务均通过环境变量进行配置。
### Log Generator
| 变量 | 默认值 | 描述 |
|---|---|---|
| `KAFKA_BOOTSTRAP_SERVERS` | `kafka:9092` | Kafka broker 地址 |
| `LOG_GENERATOR_TOPIC` | `raw-logs` | 目标 Kafka topic |
| `LOG_GENERATOR_EPS` | `1000` | 初始每秒事件数（可在运行时通过 `POST /rate` 调整） |
| `LOG_GENERATOR_ADMIN_PORT` | `8080` | `GET /rate` 和 `POST /rate` HTTP admin endpoint 的端口 |
Log Generator 始终使用基于主机的状态机。初始 EPS 由 `LOG_GENERATOR_EPS` 设置；使用 `POST /rate` 可在不重启的情况下更改它。
### Rule Engine (每个 worker)
| 变量 | 默认值 | 描述 |
|---|---|---|
| `KAFKA_BOOTSTRAP_SERVERS` | `kafka:9092` | Kafka broker 地址 |
| `SIGMA_RULES_DIR` | `sigma_rules` | 启动时扫描 `*.yml` Sigma 规则的目录 |
| `METRICS_PORT` | `8001` | Prometheus `/metrics` HTTP endpoint 的端口 |
| `WORKER_ID` | *(随机 UUID)* | 唯一标识符；用于派生唯一的 `rule-updates` consumer group ID |
| `COMMIT_EVERY_N` | `100` | 处理指定数量的消息后提交消费者 offset（取决于哪个阈值先达到） |
| `COMMIT_EVERY_S` | `5.0` | 距离上次提交指定秒数后提交消费者 offset |
### Alert Storage Service
| 变量 | 默认值 | 描述 |
|---|---|---|
| `KAFKA_BOOTSTRAP_SERVERS` | `kafka:9092` | Kafka broker 地址 |
| `ELASTICSEARCH_URL` | `http://elasticsearch:9200` | Elasticsearch endpoint |
| `BATCH_SIZE` | `500` | 当缓冲了指定数量的 Alerts 时刷新微批量缓冲区 |
| `FLUSH_INTERVAL` | `5` | 当距离上次刷新过去指定秒数时刷新缓冲区 |
## 测试指南
### 安装开发依赖
```
pip install -e ".[dev]"
```
### 单元测试
单元测试没有外部依赖，并且在没有 Kubernetes 的情况下运行。它们涵盖 `evaluate()`、`RuleEngineService`（规则生命周期 + 窗口评估）、`SlidingWindow`、`AlertStorageService`（微批量 + 注入时钟）、`HostStateMachine` 和 `LogAdminHandler`。
```
pytest tests/unit/
```
运行并显示详细输出：
```
pytest tests/unit/ -v
```
运行静态类型检查：
```
mypy src/
```
### 集成测试
集成测试针对实时 minikube stack 运行，并执行完整的 Kafka → Rule Engine → Elasticsearch 路径。它们使用真实的 Kafka 和 Elasticsearch 连接 —— 没有 mock 任何外部客户端。
Kafka 在其元数据响应中会广播其内部集群主机名。在端口转发能够端到端工作之前，请将该主机名添加到您的 `/etc/hosts` 中，以便它能通过隧道解析回本地：
```
echo "127.0.0.1 kafka-controller-0.kafka-controller-headless.kafka-sigma-engine.svc.cluster.local" \
| sudo tee -a /etc/hosts
```
在运行测试套件之前，将 Log Generator 扩展为 0 个副本 —— 其 1000 EPS 的洪流会将测试消息掩埋在分区积压中并导致超时：
```
kubectl scale deployment log-generator -n kafka-sigma-engine --replicas=0
```
然后转发端口（如果已经在运行则跳过 —— 使用 `lsof -ti tcp:9092` 检查）：
```
kubectl port-forward -n kafka-sigma-engine svc/kafka 9092:9092 &
kubectl port-forward -n kafka-sigma-engine svc/elasticsearch 9200:9200 &
```
运行测试套件：
```
pytest -m integration tests/integration/
```
完成后恢复 Log Generator：
```
kubectl scale deployment log-generator -n kafka-sigma-engine --replicas=1
```
如果端口转发地址不同，`KAFKA_BOOTSTRAP` 和 `ES_URL` 都可以被覆盖：
```
KAFKA_BOOTSTRAP=localhost:9092 ES_URL=http://localhost:9200 pytest -m integration tests/integration/
```
集成测试套件验证：
| 测试 | 检查内容 |
|---|---|
| `test_windows_failed_login_creates_alert_in_elasticsearch` | 基础的端到端路径：raw log → Rule Engine → ES |
| `test_cloudtrail_bucket_delete_creates_alert_in_elasticsearch` | `startswith` 字符串修饰符 |
| `test_non_matching_log_produces_no_alert` | 无误报 |
| `test_duplicate_alert_message_is_deduplicated_in_elasticsearch` | `_id=alert_id` 幂等 upsert |
| `test_rule_update_add_envelope_creates_alert` | 类型化 JSON 装热重载 (添加) |
| `test_windows_successful_login_creates_alert` | `event_id 4624` 规则 |
| `test_windows_explicit_credentials_creates_alert` | `event_id 4648` 规则 |
| `test_windows_privilege_use_creates_alert` | `event_id 4672` 规则 |
| `test_windows_suspicious_process_creates_alert` | `event_id 4688` + `process_name\|contains` |
| `test_cloudtrail_iam_user_create_creates_alert` | CloudTrail CreateUser 规则 |
| `test_brute_force_aggregation_rule_creates_alert` | 滑动窗口聚合（>5 个事件 / 60 秒） |
排除集成测试：
```
pytest -m "not integration" tests/
```
## 可观测性
服务通过 `minikube service`（Grafana、Prometheus）或 `kubectl port-forward`（Elasticsearch）进行访问。请先在单独的终端中运行 `minikube tunnel`，以便分配 LoadBalancer IP。
### Grafana
```
minikube service grafana -n kafka-sigma-engine --url
```
将打印出的 URL 复制到浏览器中。dashboard 会在启动时自动配置（无需登录）。导航至 **Dashboards → Kafka Sigma Engine**。
| 面板 | 指标 | 显示内容 |
|---|---|---|
| **EPS 吞吐量** | `rate(logs_processed_total[1m])` 跨所有 workers 的总和 | Rule Engine 每秒处理的实时事件数 |
| **p99 匹配延迟** | `histogram_quantile(0.99, rule_evaluation_duration_seconds_bucket)` | 针对所有加载的规则评估一条 Raw Log 的 99 百分位耗时 |
| **消费者延迟** | 每个 pod 的 `kafka_consumer_lag` | 未处理消息的积压；激增表明 Rule Engine 落后于 Log Generator |
Prometheus 使用 `kubernetes_sd_configs` 自动发现所有的 Rule Engine pods —— 所有副本都会作为 scrape targets 出现，无需任何静态配置。
### Prometheus
```
minikube service prometheus -n kafka-sigma-engine --url
```
将打印出的 URL 复制到浏览器中。验证所有的 Rule Engine scrape targets 是否在 **Status → Targets** 处于活动状态（在您输入 PromQL 查询之前，主页上的表达式浏览器会显示 "No data queried yet" —— 这是正常的）。
### Elasticsearch
Elasticsearch 具有 ClusterIP 服务，不对外暴露。通过端口转发访问它：
```
kubectl port-forward -n kafka-sigma-engine svc/elasticsearch 9200:9200 &
```
然后在 `http://localhost:9200` 进行查询：
| 查询 | 描述 |
|---|---|
| `GET /alerts/_count` | 已索引的 Alert 文档总数 |
| `GET /alerts/_search?q=severity:high` | 高严重性 alerts |
| `GET /alerts/_search?q=rule_id:win-brute-force-001` | 暴力破解聚合 alerts |
| `GET /alerts/_search?q=alert_id:` | 每个 `alert_id` 精确对应一个文档（去重） |
`alert_id`、`severity`、`rule_id` 和 `host` 字段被索引为 `keyword`，支持精确匹配过滤和桶聚合。
### 内置 Sigma 规则
仓库附带八条规则，并在启动时由每个 Rule Engine worker 加载：
| 文件 | 规则 ID | 触发条件 | 严重性 |
|---|---|---|---|
| `windows_failed_login.yml` | `win-failed-login-001` | `event_id = 4625` (登录失败) | `medium` |
| `windows_successful_login.yml` | `win-successful-login-001` | `event_id = 4624` (登录成功) | `low` |
| `windows_explicit_credentials_logon.yml` | `win-explicit-creds-001` | `event_id = 4648` (显式凭据使用) | `medium` |
| `windows_privilege_use.yml` | `win-privilege-use-001` | `event_id = 4672` (分配特殊权限) | `high` |
| `windows_suspicious_process.yml` | `win-suspicious-process-001` | `event_id = 4688` 且 `process_name` 包含 `powershell` 或 `cmd` | `medium` |
| `windows_brute_force.yml` | `win-brute-force-001` | 同一主机在 60 秒内 `event_id = 4625` 超过 5 次 | `high` |
| `cloudtrail_bucket_delete.yml` | `aws-s3-delete-001` | `action` 以 `DeleteBucket` 开头 | `high` |
| `cloudtrail_iam_user_create.yml` | `aws-iam-create-user-001` | `action = CreateUser` | `high` |
通过将 `.yml` 文件放在 `sigma_rules/` 中并重启 workers 来添加新规则，或者通过在 `rule-updates` topic 上发送 `{"op":"add", ...}` 封装进行实时传递。
## 关键工程决策
完整的理由记录在 `docs/adr/` 中：
| ADR | 决策 |
|---|---|
| `0001` | 使用 Kafka `rule-updates` topic 交付 Sigma 规则（对比 REST API 或 文件监视器） |
| `0002` | Rule Engine 使用 Python asyncio（对比 Go） |
| `0003` | 使用多个独立的 asyncio 进程作为 worker 池（对比 `ProcessPoolExecutor`） |
| `0004` | 使用 Elasticsearch 进行 Alert 存储（对比 PostgreSQL） |
| `0005` | `raw-logs`：最初为 4 个分区，以主机为 key —— 被 ADR-0012 取代 |
| `0006` | 手动至少一次 offset 提交（对比自动提交或仅一次） |
| `0007` | 双重刷新触发器：大小 ≥ 500 或 耗时 ≥ 5 秒 |
| `0008` | Alert 文档嵌入完整的 Raw Log（对比通过 offset 引用） |
| `0009` | 使用 Prometheus + Grafana 进行可观测性监控 |
| `0010` | 每个 worker 使用内存滑动窗口处理聚合规则（对比基于 Redis 的共享计数器） |
| `0011` | 用于规则生命周期操作（添加、更新、删除）的类型化 JSON 封装（取代仅支持添加的纯 YAML） |
| `0012` | `raw-logs` topic 增加到 8 个分区以实现 Kubernetes 水平扩展（取代 ADR-0005） |
| `0013` | 使用 minikube 在 `k8s/` 中部署完整的 Kubernetes stack（对比混合 Docker Compose + K8s） |
| `0014` | 通过 Elasticsearch `_id = alert_id` 进行 Alert 去重（对比查询时折叠或内存集合） |
| `0015` | Log Generator HTTP admin endpoint 用于运行时 EPS 调整（对比 Kafka 控制 topic 或信号） |
| `0016` | Log Generator 中每台主机的状态机用于生成相关的攻击序列（对比 YAML 场景手册） |
标签：AMSI绕过, Elasticsearch, Grafana, Kafka, SonarQube插件, 威胁检测, 子域名突变, 安全日志分析, 自定义请求头, 计算机取证, 请求拦截, 软件成分分析, 逆向工具