sltcnb/cumulonimbus

## 演示

# Cumulonimbus [](https://github.com/sltcnb/cumulonimbus/actions/workflows/ci.yml) **云取证与应急响应工具包。** 从 AWS、Azure、GCP 和 Kubernetes 收集云原生日志及 artifacts，并将其解析、标准化为 **Elastic Common Schema (ECS) v8**，运行跨云分析，并导出 随时可用于 Elasticsearch、Splunk、Sigma、STIX pipelines 或 Citadel 的时间线。 以风暴云命名——因为这通常是 IR 项目的真实 写照。 ## 目录 - [为什么选择它](#why) - [核心概念](#concepts) - [安装说明](#install) - [快速开始](#quick-start) - [提供商与数据集](#providers--datasets) - [命令参考](#command-reference) - [`collect` (分提供商)](#collect-per-provider) - [`parse`](#parse) - [`analyze`](#analyze) - [`export`](#export) - [`push`](#push) - [辅助命令](#helpers) - [ECS 数据模型](#ecs-data-model) - [分析阶段](#analysis-passes) - [身份验证](#authentication) - [IAM / 最小权限](#iam--least-privilege) - [部署](#deployment) - [检测](#detections-sigma) - [扩展 Cumulonimbus](#extending-cumulonimbus) - [测试](#testing) - [仓库结构](#repository-layout) - [安全说明](#security-notes) - [路线图](#roadmap) - [开源许可](#license) ## 为什么选择它 云 IR 人工成本高、依赖特定提供商，且缺乏标准化的 工具。各控制台相互孤立；商业平台价格昂贵。Cumulonimbus 是开源的，支持四大主流云平台，并输出单一的标准化 schema，因此 一条时间线即可跨越所有提供商——并且可以直接对接你 正在使用的工具。 **设计原则** - **保留原生数据。** `collect` 会将提供商的原生证据原封不动地写入 `raw/`。标准化是一个独立且可重复的步骤——原始数据 绝不会遭到修改。 - **离线解析。** `parse`、`analyze` 和 `export` 不需要云 凭证，也不需要网络。只有 `collect`/`push` 会与提供商进行通信。 - **统一 schema。** 所有内容都会转化为 ECS v8，因此 AWS + Azure + GCP + K8s 共享一条 时间线、相同的字段名以及相同的检测规则。 - **柔性故障。** 单个格式错误的记录或损坏的 enricher 永远不会中止 整个运行过程。 ## 核心概念 该工具是一个三阶段 pipeline，并在其基础上附加了分析和导出功能： ``` collect --▶ /raw/*.jsonl provider-native records, untouched parse --▶ /ecs/*.ecs.jsonl ECS v8, enriched analyze --▶ reports (tables / JSON) timeline, users, network, IOCs, correlation export --▶ timeline.csv | .jsonl | .stix | bulk.ndjson | citadel bundle push --▶ Citadel / Splunk HEC / Elasticsearch ``` | 术语 | 含义 | |------|---------| | **Collector** | 从提供商 API/日志源获取原始记录；写入 `raw/.jsonl`。 | | **Parser** | 将一条原生记录映射为 -> `ForensicEvent` (ECS)。按 dataset 名称注册。 | | **Dataset** | 一个 `provider.service` 字符串，例如 `aws.cloudtrail`。它也是原始文件名的词干——这是 `parse` 选择 parser 的方式。 | | **Normalizer** | Enrichment 阶段：始终会判定网络方向；可选开启 GeoIP/ASN、reverse-DNS 和 IOC 匹配。 | | **Exporter** | 将 `ForensicEvent` 序列化为 jsonl/csv/STIX/ES-bulk/Citadel。 | ## 安装说明 需要 Python 3.9+。 ``` pip install -e ".[aws]" # + boto3 pip install -e ".[azure]" # + azure-identity / azure-mgmt-monitor / requests pip install -e ".[gcp]" # + google-cloud-logging / google-cloud-securitycenter pip install -e ".[k8s]" # + kubernetes pip install -e ".[all]" # every provider SDK pip install -e ".[dev]" # pytest / pytest-cov / moto (development) ``` 基础安装（无附加依赖）已经可以运行 `parse`、`analyze` 和基于文件的 `export`——你只需安装对应云平台的附加依赖即可从中执行 `collect`。 ## 快速开始 ``` # 1. 收集（需要 AWS credentials） cumulonimbus aws collect --service all \ --profile prod --region us-east-1 \ --start-time 2024-01-01T00:00Z \ --output ./case/ # 2. 解析 + 标准化为 ECS（离线） cumulonimbus parse ./case/raw -o ./case # 3. 分析 cumulonimbus analyze ./case/ecs # 4. 导出时间线 cumulonimbus export ./case/ecs -o timeline.csv --format csv ``` case 目录最终看起来像这样： ``` case/ |-- raw/ # provider-native, verbatim | |-- aws.cloudtrail.jsonl | |-- aws.guardduty.jsonl | `-- aws.vpcflow.jsonl `-- ecs/ # normalized ECS v8 |-- aws.cloudtrail.ecs.jsonl |-- aws.guardduty.ecs.jsonl `-- aws.vpcflow.ecs.jsonl ``` ## 提供商与数据集 | 提供商 | 数据集 | 收集来源 | |----------|----------|----------------| | **AWS** | `aws.cloudtrail`, `aws.guardduty`, `aws.vpcflow`, `aws.s3access`, `aws.ec2`, `aws.iam`, `aws.lambda`, `aws.rds` | boto3 (LookupEvents, GuardDuty, Describe\*), S3 日志对象 | | **Azure** | `azure.activity`, `azure.signin`, `azure.audit`, `azure.nsgflow` | azure-mgmt-monitor, Microsoft Graph | | **GCP** | `gcp.audit`, `gcp.vpcflow`, `gcp.scc` | Cloud Logging, Security Command Center | | **Kubernetes** | `k8s.audit`, `k8s.event`, `k8s.container`, `k8s.etcd` | 审计日志文件, CoreV1 事件, pod/container 清单, 解码后的 etcd 快照 | `cumulonimbus parsers` 会打印所有已注册的 parser。 VPC Flow (AWS) 和 NSG Flow (Azure) 也可直接接收**原生日志行**，因此 投递到 S3/Storage 的日志无需结构化的 collector 即可进行解析。 ## 命令参考 ### `collect` (分提供商) 每个提供商都是一个命令组。所有命令都会将数据写入 `

/raw/`。 ``` # AWS — 基于 API 的服务 cumulonimbus aws collect \ --service {cloudtrail|guardduty|ec2|iam|lambda|rds|all} \ [--profile NAME] [--region REGION] \ [--start-time ISO8601] [--end-time ISO8601] \ -o ./case/ # AWS — 传送到 S3 bucket 的日志（access logs 或 VPC flow logs） cumulonimbus aws collect-s3 \ --bucket my-log-bucket [--prefix path/] \ --dataset {aws.s3access|aws.vpcflow} \ [--profile NAME] [--region REGION] \ -o ./case/ # Azure cumulonimbus azure collect \ --service {activity|signin|audit|all} \ [--subscription SUB_ID] [--tenant TENANT_ID] \ [--start-time ISO8601] [--end-time ISO8601] \ -o ./case/ # GCP cumulonimbus gcp collect \ --service {audit|vpcflow|scc|all} \ --project PROJECT_ID \ [--start-time ISO8601] [--end-time ISO8601] \ -o ./case/ # Kubernetes cumulonimbus k8s collect --service event [--kubeconfig PATH] [--in-cluster] -o ./case/ cumulonimbus k8s collect --service container [--kubeconfig PATH] [--in-cluster] -o ./case/ cumulonimbus k8s collect --service audit --audit-log /var/log/kube-audit.log -o ./case/ cumulonimbus k8s collect --service etcd --etcd-export ./etcd-decoded.jsonl -o ./case/ ``` 时间为 ISO-8601 格式；末尾带有 `Z` 或无时区将被视为 UTC。 **Container 与 etcd 取证。** `k8s.container` 会盘点所有运行中的 container（镜像、特权标志、host-path 挂载、重启次数），并映射到 ECS `container.*`。`k8s.etcd` 会接收**已解码的** etcd 快照——etcd 是一个 二进制 bolt DB，存储了所有处于静态的 API 对象（包括 Secrets），因此请先使用 `auger extract` 等工具解码 `etcdctl snapshot save` 文件，然后传入 JSON `{key, value}` 对象。静态的 Secret 物料会被标记 （`k8s.contains_secret_material`, `k8s.data_keys`）。 ### `parse` ``` cumulonimbus parse RAW_DIR -o OUTPUT_DIR [--dataset NAME] ``` - 读取 `RAW_DIR` 中的所有 `*.jsonl` 文件，根据每个文件的词干推断 parser （`aws.cloudtrail.jsonl` -> `aws.cloudtrail`），进行标准化，并写入 `OUTPUT_DIR/ecs/.ecs.jsonl`。 - 将 `RAW_DIR` 指向 `collect` 创建的 `raw/` 文件夹。 - `--dataset` 会强制为所有文件使用同一个 parser（适用于命名奇怪的输入）。 **Enrichment**（`parse` 上的可选标志）： ``` cumulonimbus parse ./case/raw -o ./case \ --geoip-city GeoLite2-City.mmdb --geoip-asn GeoLite2-ASN.mmdb \ --rdns \ --ioc iocs.txt ``` | 标志 | 效果 | |------|--------| | `--geoip-city` / `--geoip-asn` | 从 MaxMind 数据库填充 `source/destination.geo.*` 和 `.as_number`（需要 `[geoip]` 附加依赖） | | `--rdns` | 将公网 IP 反向 DNS 解析为 `.domain`（带缓存；会执行网络查询） | | `--ioc FILE` | 标记源/目的 IP 匹配到 indicator 的事件；设置 `threat.matched` 并将 `event.kind` 提升为 `alert`。FILE 为每行一个 IP 或 STIX 2.x bundle | 网络方向标记始终会运行。Enricher 采用尽力而为原则——查询 失败永远不会中止运行。 ### `analyze` ``` cumulonimbus analyze ECS_INPUT [--report REPORT] [--json] [--limit N] ``` `ECS_INPUT` 是一个包含 `*.ecs.jsonl` 的目录或单个文件。报告： | `--report` | 输出 | |------------|--------| | `all` (默认) | 用户 + 网络 + 提权 + 数据外传 + 关联分析 | | `timeline` | 按 `@timestamp` 排序的所有事件 | | `users` | 每个 principal 的操作计数、源 IP、失败情况、首次/最后出现时间 | | `network` | 按字节数排名的 Top talkers（`--limit` 行） | | `privesc` | 匹配已知 IAM 权限提升技术的 CloudTrail 事件 | | `exfil` | 大量出站流量 + 批量 S3 读取活动 | | `correlate` | 在 ≥2 个云提供商中发现的相同身份或源 IP | `--json` 输出机器可读内容，而不是 Rich 表格。 ### `export` ``` cumulonimbus export ECS_INPUT -o OUTPUT --format FMT [--gzip] [--es-index NAME] [--case-id ID] ``` | `--format` | 说明 | |------------|-------| | `jsonl` (默认) | 每行一个 ECS doc；支持 `--gzip` | | `csv` | 扁平化的分析师列；支持 `--gzip` | | `stix` | STIX 2.1 bundle（IP observables + observed-data，确定性 ID） | | `es-bulk` | Elasticsearch `_bulk` NDJSON；通过 `--es-index` 指定索引 | | `citadel` | Citadel 摄取 bundle (`citadel.bundle.v1`)；使用 `--case-id` 打标签 | ### `push` 通过 HTTP 将标准化后的事件直接发送到平台。 ``` cumulonimbus push ECS_INPUT --target citadel \ --url https://citadel.local/api/ingest --token $TOKEN --case-id C-42 cumulonimbus push ECS_INPUT --target splunk-hec \ --url https://splunk:8088/services/collector --token $HEC_TOKEN cumulonimbus push ECS_INPUT --target es-bulk \ --url https://es:9200 --token $API_KEY --es-index cases ``` `--no-verify` 会禁用 TLS 验证（仅供实验室使用）。 ### 辅助命令 ``` cumulonimbus parsers # list every registered parser cumulonimbus iam-policy # print the minimum AWS IAM policy (JSON) cumulonimbus --version ``` ## ECS 数据模型 每个 parser 都会输出一个 `ForensicEvent`（见 `cumulonimbus/ecs/schema.py`），这是一个基于 ECS v8 子集的 pydantic 模型。特定于提供商的字段会保留在 `aws` / `azure` / `gcp` namespaces 下（以及额外的 `k8s` namespace 中）； 空值会在导出时被裁剪。 **CloudTrail -> ECS** ``` { "@timestamp": "2024-01-15T10:30:00Z", "event": { "action": "ConsoleLogin", "category": ["authentication"], "type": ["access"], "outcome": "success", "provider": "aws", "kind": "event", "dataset": "aws.cloudtrail" }, "user": { "name": "admin", "id": "AIDACKCEVSQ6C2EXAMPLE" }, "source": { "ip": "203.0.113.42" }, "cloud": { "provider": "aws", "account_id": "123456789012", "region": "us-east-1", "service_name": "signin.amazonaws.com" }, "aws": { "cloudtrail": { "event_source": "signin.amazonaws.com", "event_name": "ConsoleLogin", "read_only": false } } } ``` **VPC Flow -> ECS** ``` { "@timestamp": "2024-01-15T10:30:00Z", "event": { "action": "ACCEPT", "category": ["network"], "type": ["connection"], "outcome": "success", "provider": "aws" }, "source": { "ip": "10.0.1.42", "port": 54321 }, "destination": { "ip": "198.51.100.10", "port": 443 }, "network": { "transport": "tcp", "protocol": "6", "bytes": 1024, "packets": 10, "direction": "outbound" }, "aws": { "vpc": { "interface_id": "eni-12345678" } } } ``` 结果、类别和 event.type 在所有提供商中均保持一致地进行了标准化 （例如，无论是 AWS、Azure 还是 GCP，失败的登录都会标记为 `outcome: "failure"`）。 **Kubernetes** 会映射到 ECS `orchestrator.*` 和 `container.*`，而不是映射到 provider namespace： ``` { "event": { "action": "get:secrets", "provider": "kubernetes", "dataset": "k8s.audit", "outcome": "success" }, "user": { "name": "system:anonymous" }, "orchestrator": { "type": "kubernetes", "namespace": "kube-system", "resource": { "type": "secrets", "name": "admin-token" } }, "k8s": { "verb": "get", "sensitive": true, "response_code": 200 } } ``` 一个很小的额外 `k8s` namespace 承载了 ECS 中没有对应字段的取证细节 （动词、响应代码、敏感/特权标志、etcd 元数据）。 ## 分析阶段 所有阶段均作为处理 ECS dict 的纯函数位于 `cumulonimbus/core/analysis.py` 中—— 可独立导入和测试。 - **`timeline`** — 将所有事件合并为一个按时间顺序排列的序列。 - **`user_activity`** — 每个 principal 的操作直方图、源 IP、失败 次数、首次/最后出现时间。 - **`top_talkers`** — 按 (源 IP, 目的 IP, 目的端口) 聚合的网络 flows，并按 字节数排名。 - **`privesc_indicators`** — 标记已知 权限提升集合（`PutUserPolicy`, `AttachUserPolicy`, `CreateAccessKey`, `UpdateAssumeRolePolicy`, …）中的 IAM API 调用。 - **`exfil_indicators`** — 大量出站传输（默认 ≥ 100 MB）以及 批量 S3 读取 API。 - **`correlate_identities`** — 跨云视角的透视：出现在多个提供商中的相同 principal （已剥离 domain）或源 IP。这就是将四份独立的日志集 转化为一个安全事件的关键。 ## 身份验证 | 提供商 | 方式 | |----------|---------| | **AWS** | 指定的配置文件 (`--profile`)、环境变量、实例/任务角色、跨账户 assume-role、MFA —— 凡是 `boto3.Session` 能解析的均可。 | | **Azure** | `DefaultAzureCredential`：环境变量 (service principal)、托管标识、Azure CLI、VS Code。登录/审计日志使用 Graph token。 | | **GCP** | 应用默认凭证或通过 `GOOGLE_APPLICATION_CREDENTIALS` 提供的 service-account JSON key。 | | **Kubernetes** | kubeconfig (`--kubeconfig`) 或集群内的 service account (`--in-cluster`)。审计日志从文件路径读取。 | 凭证永远不会写入磁盘或记录在日志中。 ## IAM / 最小权限 `cumulonimbus iam-policy` 会打印 collectors 所需的最低只读 AWS 策略 （同样见于 `cumulonimbus/providers/aws/iam_permissions.py`）—— 包括 CloudTrail `LookupEvents`、GuardDuty list/get、EC2/IAM/Lambda/RDS `Describe`/ `List`，以及用于日志存储桶的 S3 读取权限。请将其附加到专用的 IR 角色上。 ## 部署 ### Docker ``` docker run -v $(pwd)/output:/data -e AWS_PROFILE=prod \ sltcnb/cumulonimbus:latest aws collect --service all --output /data ``` 镜像以非 root 用户身份运行。 ### Kubernetes Job ``` kubectl apply -f k8s/job.yaml ``` 从 `aws-creds` Secret 读取凭证，并写入到 `cumulonimbus-output` PVC。 hardened `securityContext`（非 root、禁止 privilege escalation、只读 root FS）。 ### Helm ``` helm install ir ./k8s/helm \ --set provider=aws --set service=all \ --set schedule="0 */6 * * *" \ --set credentialsSecret=cloud-creds ``` 设置了 `schedule` 时会渲染为 `CronJob`，否则渲染为一次性 `Job`。通过更改 `--set provider=` 即可适用于任何提供商。 ## 检测 `sigma/` 中的预置规则： | 规则 | 触发条件 | |------|----------| | `aws_root_console_login.yml` | root 账户控制台身份验证 | | `k8s_secret_access_anonymous.yml` | 匿名/未经身份验证的 principal 读取 Secrets | | `multicloud_impossible_travel.yml` | 同一个身份/IP 在多个云中活跃 | 规则基于 ECS 字段名进行匹配，因此它们可以对导出的 jsonl 运行，或者在数据摄取后在你的 SIEM 中运行。 ## 扩展 Cumulonimbus 只需两步即可添加一个 dataset。 **1. Parser** — 继承 `Parser`，按 dataset 名称注册它： ``` from cumulonimbus.core.parser import Parser, register from cumulonimbus.ecs.schema import Event, ForensicEvent @register("aws.route53") class Route53Parser(Parser): def parse_record(self, record): if "query_name" not in record: return None # returning None skips the record return ForensicEvent( **{"@timestamp": record["query_timestamp"]}, event=Event(action="dns_query", category=["network"], provider="aws", dataset="aws.route53"), # ... ) ``` 从其提供商的 `parsers/__init__.py` 导入该模块，以便运行装饰器。 现在 `parse` 可以自动处理 `aws.route53.jsonl` 了。 **2. Collector**（可选，用于实时拉取数据） — 继承 `Collector`，设置 `dataset`，实现 `collect()` 以 yield 原始 dicts，并将其添加到 提供商的 `COLLECTORS` 映射中。`collect_to()` 会处理流式传输到磁盘的过程。 ## 测试 ``` pip install -e ".[dev]" pytest -q ``` 58 个单元测试覆盖了每个 parser（带有 golden fixtures）、normalizer、所有 分析阶段、每个 export 编码器、enrichers 以及所有 collectors。AWS collectors 会在 **moto** 模拟的 API（EC2、IAM、S3、Lambda、 RDS）上运行端到端测试；Azure/GCP/Kubernetes collectors 会针对注入的伪 SDK 模块运行测试； 基于文件的 K8s collectors（audit、etcd）会直接进行测试。此外，还会在 CI 中的 Python 3.9–3.12 环境下运行，并辅以 ruff lint 门禁。 ## 仓库结构 ``` cumulonimbus/ |-- pyproject.toml |-- Dockerfile |-- k8s/ | |-- job.yaml | `-- helm/ # Chart.yaml, values.yaml, templates/cronjob.yaml |-- sigma/ # detection rules |-- cumulonimbus/ | |-- cli.py # Click CLI: collect / parse / analyze / export / push | |-- core/ | | |-- collector.py # Collector base | | |-- parser.py # Parser base + registry | | |-- normalizer.py # enrichment | | |-- analysis.py # timeline / users / privesc / exfil / correlate | | |-- exporter.py # streaming jsonl / csv | | `-- exports.py # STIX / ES-bulk / Citadel encoders | |-- ecs/ | | `-- schema.py # ForensicEvent (ECS v8) | `-- providers/ | |-- aws/ { collectors/, parsers/, iam_permissions.py } | |-- azure/ { collectors/, parsers/ } | |-- gcp/ { collectors/, parsers/ } | `-- k8s/ { collectors/, parsers/ } `-- tests/ |-- unit/ # parsers, analysis, exports, multicloud `-- fixtures/ ``` ## 安全说明 - **只读。** Collectors 仅读取云状态。不进行任何修改，不采取任何响应 操作。 - **凭证。** 绝不会被记录，也绝不会写入到输出中。 - **保留原始数据。** `collect` 原封不动地存储证据；解析是 下游且可重复的步骤，从而保持了干净的监管链。 - **TLS。** `push` 默认验证 TLS；`--no-verify` 仅限实验室使用。 ## 路线图 - SaaS 审计日志（Okta、GitHub、Microsoft 365） - 增量收集（按 dataset 记录上次运行的书签） - 取证快照自动化（EBS / Azure Disk / GCE Disk） - Kubernetes 的 Container/etcd 取证 - 针对用户行为和网络 flows 的异常检测 ## 开源许可 Apache-2.0

标签：库, 应急响应, 数字取证, 自动化脚本, 逆向工具