shahzain-khan450/cloudsec-ai

# CloudSec AI — 自主云安全运营中心 (SOC)     ## 目录 - [项目概述与问题陈述](#project-overview-and-problem-statement) - [架构图](#architecture-diagram) - [技术栈](#technology-stack) - [仓库结构](#repository-structure) - [前置条件](#prerequisites) - [安装与部署说明](#installation-and-deployment-instructions) - [CI/CD 工作流](#cicd-workflow) - [基础设施详情](#infrastructure-details) - [监控与日志设置](#monitoring-and-logging-setup) - [安全考量](#security-considerations) - [截图](#screenshots) - [挑战与经验教训](#challenges-and-lessons-learned) - [未来改进](#future-improvements) - [展示技能](#skills-demonstrated) - [许可证与作者](#license-and-author) ## 项目概述与问题陈述 云安全事件很少以巨大的警报开始——它们往往始于深埋在 CloudTrail 中的一个小事件、来自容器的一行 runtime 日志，或者没有人实时监控的网络特征。在大多数团队中，将这些原始信号转化为“发生了什么、为什么重要、如何修复”仍然是一个手动过程：工程师必须注意到警报，理解日志格式，将其与 MITRE ATT&CK 等框架进行交叉比对，然后决定是否采取行动。 **CloudSec AI** 实现了该 pipeline 的端到端自动化。它从 AWS CloudTrail、Falco（容器/K8s runtime）、Wazuh（主机/SIEM）和 Suricata（网络 IDS）摄取安全信号；使用确定性规则引擎和用于细微判别的 LLM 对它们进行分类；将它们映射到 MITRE 风格的战术/技术；自动修复一类可以安全快速自动修复的配置错误（公开的 S3 存储桶）；并通过 Prometheus/Grafana 和 Slack/电子邮件呈现所有内容。 **示例流程：** 开发人员将 S3 存储桶设为公开 → CloudTrail 记录该事件 → EventBridge 触发 Lambda → Lambda 调用 AI 引擎 → AI 引擎用通俗易懂的语言解释风险并将其映射到 MITRE 技术 → 修复 Lambda 将存储桶恢复为私有 → Slack 收到通知 → Grafana dashboard 更新——在演示场景中，通常需要数小时手动调查的配置错误，在不到一分钟内即可得到解释和修复。 ## 架构图 ``` CloudTrail / Container logs / Falco / Wazuh / Suricata / Trivy │ ▼ EventBridge / Lambda │ ▼ AI Engine (FastAPI + LLM + RAG) ┌───────────┼───────────────┐ ▼ ▼ ▼ Auto-Remediate Grafana Slack/Email (Terraform/ Dashboard Notification Lambda) ``` 有关完整的数据流分解以及混合规则 + LLM 分类器设计背后的原因，请参阅 [`docs/architecture.md`](docs/architecture.md)。 ## 技术栈 | 层级 | 技术 | |---|---| | API / AI 引擎 | Python 3.12, FastAPI, Pydantic, Uvicorn | | AI / ML | 混合规则引擎 + LLM（Ollama/Llama 3，带有模板化回退机制），基于初始 runbook 集的 TF-IDF RAG (scikit-learn) | | 云 | AWS (CloudTrail, EventBridge, Lambda, SNS, IAM, S3), boto3 | | IaC | Terraform (AWS provider) | | 容器 / 编排 | Docker, Docker Compose, Kubernetes (Deployment, DaemonSet, Service) | | runtime 与网络安全 | Falco (eBPF/kernel-module runtime 检测), Wazuh (主机/SIEM), Suricata (网络 IDS) | | 可观测性 | Prometheus (自定义 `/metrics`), Grafana (配置化的 dashboard) | | CI/CD | GitHub Actions, pytest, Trivy (镜像扫描), Gitleaks (机密扫描), OWASP Dependency-Check, `terraform validate` | | 测试 | pytest | ## 仓库结构 ``` cloudsec-ai/ ├── ai-engine/ FastAPI service: threat classifier, IAM/cost analyzers, │ Falco/Wazuh/Suricata webhooks, incident reports, │ RAG log analyzer, Prometheus metrics ├── terraform/ CloudTrail, EventBridge, Lambda (remediation + daily scan), │ SNS, IAM ├── lambda/ remediate_s3.py (auto-fix), daily_scan.py (IAM+cost report) ├── kubernetes/ Namespace, ai-engine Deployment/Service, real Falco DaemonSet ├── grafana/ Provisioned datasource + dashboard (events by severity/source, │ remediations, latency) ├── prometheus/ Scrape config (targets the real /metrics endpoint) ├── scripts/ Sample events for every source (CloudTrail, Falco, Wazuh, Suricata) ├── docs/ architecture.md, threat-model.md, setup.md, │ k8s-validation-checklist.md ├── .github/workflows/ci.yml CI pipeline └── docker-compose.yml Local demo stack (ai-engine + Prometheus + Grafana) ``` ## 前置条件 **运行本地演示（推荐的起点）：** - Docker 和 Docker Compose **部署 AWS 端：** - AWS 账户（Free Tier 即可） - 已配置 AWS 凭证（`aws configure` 或环境变量） - Terraform >= 1.5 **部署到 Kubernetes：** - 一个真正的基于 VM 的 k3s（或完整的 Kubernetes）节点——Falco DaemonSet 需要 `privileged: true` 和 host 内核访问权限，因此基于容器的“Docker 中的 Kubernetes”设置（如 kind 或 minikube-on-Docker）无法用于该部分 - `kubectl` **在本地不使用 Docker 运行测试或开发：** - Python 3.12 - `pip` ## 安装与部署说明 ### 1. 本地演示（无需 AWS） ``` git clone https://github.com//cloudsec-ai.git cd cloudsec-ai docker compose up --build ``` - **AI 引擎 / Swagger 文档：** http://localhost:8000/docs - **Grafana：** http://localhost:3000 (`admin` / `admin`) - **Prometheus：** http://localhost:9090 发送示例事件： ``` curl -X POST http://localhost:8000/analyze/event \ -H "Content-Type: application/json" \ -d @scripts/sample_events/s3_public_bucket.json ``` 打开 Grafana——**“CloudSec AI — 安全概览”** dashboard 是自动配置的，在您发送几个事件后应该已经显示数据。 可选——运行本地 LLM 而不是模板化回退： ``` docker compose --profile llm up --build docker exec -it cloudsec-ai-ollama-1 ollama pull llama3 ``` **其他可尝试的 endpoint**——完整列表和示例 payload 见 [`docs/setup.md`](docs/setup.md)： ``` curl -X POST http://localhost:8000/analyze/remediate -H "Content-Type: application/json" \ -d '{"event_type": "s3_public_bucket", "resource_id": "customer-invoices-prod", "dry_run": true}' curl -X POST http://localhost:8000/webhooks/falco -H "Content-Type: application/json" \ -d @scripts/sample_events/falco_shell_spawn.json curl -X POST http://localhost:8000/logs/analyze -H "Content-Type: application/json" \ -d '{"log_text": "sshd: Failed password for root from 185.220.101.44, 250 attempts in 3 minutes"}' ``` ### 2. 运行测试 ``` cd ai-engine pip install -r requirements.txt pytest tests/ -v ``` ### 3. 部署 AWS 端 ``` cd terraform terraform init terraform plan \ -var="notification_email=you@example.com" \ -var="ai_engine_url=https://your-deployed-ai-engine/analyze/event" terraform apply ``` 要进行端到端测试：在控制台中将一个存储桶设为公开，等待 1-2 分钟让 CloudTrail 传递事件，然后检查 Lambda 的 CloudWatch Logs——它应该已经恢复了公开访问并发送了 SNS 通知。 ### 4. 部署到 Kubernetes ``` kubectl apply -f kubernetes/namespace.yaml kubectl apply -f kubernetes/ai-engine-deployment.yaml kubectl apply -f kubernetes/falco-daemonset.yaml kubectl get pods -n cloudsec-ai ``` ## CI/CD 工作流 `.github/workflows/ci.yml` 会在每次向 `main` 发起 push/PR 时运行，包含五个作业： | 作业 | 功能 | |---|---| | `test` | 安装依赖并针对 AI 引擎运行 pytest 套件 | | `build-and-scan` | 构建 `ai-engine` Docker 镜像并使用 **Trivy** 进行扫描（CRITICAL/HIGH 严重级别） | | `secret-scan` | 对完整的 git 历史记录进行 **Gitleaks** 机密扫描 | | `dependency-check` | 针对 `ai-engine` 的依赖项进行 **OWASP Dependency-Check** 检查，CVSS ≥ 9 时失败，并上传 HTML 报告 artifact | | `terraform-validate` | `terraform fmt -check` + `terraform init -backend=false` + `terraform validate` | ## 基础设施详情 ### Terraform (`terraform/`) 提供 pipeline 的实际 AWS 端配置： - 多区域 **CloudTrail** trail，记录到专用的、访问权限锁定的 S3 存储桶 - **EventBridge** 规则，模式匹配 `PutBucketAcl` / `PutBucketPolicy` 调用 - **Lambda** (`remediate_s3.py`)，重新应用 S3 Block Public Access 并发布到 SNS - 每日调度的 **Lambda** (`daily_scan.py`)，运行 IAM 和成本分析器并通过电子邮件发送摘要 - 带有可选电子邮件订阅的 **SNS** 主题 所有这些在 AWS Free Tier 内足以满足演示/作品集使用；S3 数据事件日志记录是唯一值得禁用的设置（`main.tf` 中的 `data_resource` 块），如果指向具有大量真实流量的存储桶的话。 ### Kubernetes (`kubernetes/`) - `namespace.yaml` — 专用的 `cloudsec-ai` namespace - `ai-engine-deployment.yaml` — 2 副本 Deployment，具有资源 requests/limits（100m/256Mi requests，500m/512Mi limits），针对 `/health` 的 readiness 和 liveness probes，以及一个 `ClusterIP` Service - `falco-daemonset.yaml` — 一个真实的 Falco DaemonSet，使用 Falco 内置的 `http_output` 将每个 alert 直接 POST 到 ai-engine 的 `/webhooks/falco` endpoint；以 `hostNetwork`/`hostPID`/`privileged: true` 运行，并挂载 `/proc`、`/dev`、`/lib/modules`，这就是为什么它需要真正的基于 VM 的节点，而不是 kind/minikube-on-Docker 的原因 这些清单已经过正确性审查，但尚未在此环境中的实时 cluster 中进行过实际测试——有关填补这一差距的确切步骤以及该测试的实际状态，请参阅 [`docs/k8s-validation-checklist.md`](docs/k8s-validation-checklist.md)。 ## 监控与日志设置 - **Prometheus** (`prometheus/prometheus.yml`) 抓取 AI 引擎真实的 `/metrics` endpoint——不仅仅是健康 ping，还包括按来源/严重程度统计的事件数、修复结果以及分析延迟直方图。 - **Grafana** 在启动时自动配置其 Prometheus 数据源和 dashboard（`grafana/provisioning/`，`grafana/dashboards/cloudsec-overview.json`），包含四个面板： - 按严重程度分析的事件 - 按来源统计的事件 - 修复（成功 vs 失败） - 分析延迟 (p95) - 来自 Falco 的结构化 JSON 告警（`json_output=true`）直接输入到 webhook pipeline 中，因此 runtime alert 会与 CloudTrail 驱动的事件显示在同一个 dashboard 中。 ## 安全考量 完整细节见 [`docs/threat-model.md`](docs/threat-model.md)；以下是重点： - **自动修复被刻意限制在极小范围内。** 只有 `s3_public_bucket` 会被自动修复，因为那里的误报成本低且可逆。IAM 和成本发现被设计为仅报告——分离策略或停止实例可能会破坏人类尚未审查的内容，因此这些内容会进入每日 SNS 摘要供人工采取行动。 - **在范围内、已实施的检测：** 公开的 S3 存储桶 (CloudTrail)、暴力登录 (Wazuh)、容器 shell 生成 (Falco)、IAM 配置错误（管理员策略、未使用的用户、非活跃密钥、缺少 MFA——通过 boto3）、空闲/未挂载的成本发现，以及端口扫描/C2 信标 (Suricata)。 - **已记录但尚未实施：** root-account-usage 检测、Falco `container_host_mount`（需要实时 cluster 才能触发）、挖矿行为（需要 Falco + Prometheus CPU 指标关联）以及真正的“异常 API 调用量”基线检测器。 - **CI 安全门控：** 每次构建都使用 Trivy（镜像 CVE）、Gitleaks（机密）和 OWASP Dependency-Check（依赖项 CVE，CVSS 超过 9 则失败）进行扫描。 - **明确的非目标：** 这是一个作品集/演示系统，而不是经过认证的生产级 SOC 工具——它不作为符合 PCI/SOC2 标准的工具展示，并且将自动修复扩展到这一低影响范围之外的操作首先需要经过人工批准步骤。 ## 截图 ### API 文档 (Swagger UI)   ### Grafana Dashboard  ## 挑战与经验教训 - **决定何时信任 LLM 与规则表。** 早期，人们很想通过 LLM 路由每个事件。最终的设计对任何明确的情况（公开的 S3 存储桶始终值得标记）使用确定性规则传递，并将 LLM 保留给真正基于判断的情况，例如将 300 次失败的登录与真正的暴力破解尝试区分开来。这使得系统保持快速、可审计且可用于 CI，而无需 GPU 或 API key——模板化回退意味着如果无法访问 Ollama，它仍然可以给出真实的答案。 - **真实的 alert-schema 解析与简单的测试格式。** Falco、Wazuh 和 Suricata 都有自己原生的 alert JSON 格式。构建解析*实际* schema 的适配器 (`source_adapters.py`)——而不是简化的替代品——需要更多的前期研究，但这意味着 webhook 处理程序是真正符合生产级别的，而不仅仅是仅用于演示的 stub。 - **知道不该自动化什么。** IAM 和成本分析器刻意停留在报告阶段，而不是自动修复，因为错误自动修复（分离策略、杀死实例）的影响范围比恢复存储桶的公开访问要大得多。明确划定这条线——并写下*原因*——与代码本身一样重要。 - **诚实面对未经测试的基础设施。** Kubernetes 清单，特别是 Falco DaemonSet，编写正确，但尚未在此环境中的实时 k3s 节点上运行。我们没有声称它已通过验证，而是这一差距在 `docs/k8s-validation-checklist.md` 中得到了明确跟踪，并带了填补该差距的确切步骤。 ## 未来改进 - 针对实时 k3s 节点验证 Falco DaemonSet（参见检查清单），并将 threat model 从“尚未经过 cluster 测试”更新为带有日期和版本的验证说明。 - 添加 threat model 中描述的 `root_account_usage`、`container_host_mount`、`crypto_miner_behavior` 以及真正的基于基线的 `unusual_api_call_volume` 检测器。 - 将 Falco 进程生成事件与 Prometheus CPU 指标相关联，以进行真正的挖矿检测，而不是单一信号检查。 - 添加人工审批工作流（例如 Slack 交互式消息），作为在将自动修复扩展到 S3 公开访问修复之外的先决条件。 - 录制简短的演示视频，并为作品集展示添加真实的 dashboard/API 截图。 - 考虑使用 `falcosidekick` 进行多目标 alert 路由（Slack + PagerDuty + webhook），而不是 Falco 的单一 `http_output`。 ## 展示技能 - **云安全工程：** AWS CloudTrail、EventBridge、Lambda、SNS、IAM 审计、S3 自动修复 - **基础设施即代码：** Terraform（多资源 AWS stack，变量，输出） - **容器编排：** Kubernetes 清单（Deployment、DaemonSet、Service、namespace、资源限制、health probes） - **Runtime 与网络安全工具：** Falco（基于 eBPF 的 runtime 检测）、Wazuh（主机/SIEM）、Suricata（网络 IDS）、真实的原生 alert-schema 解析 - **后端/API 开发：** Python、FastAPI、Pydantic schema 设计、REST API 设计 - **应用 AI/ML：** 混合规则 + LLM 分类设计、MITRE ATT&CK 风格映射、基于 runbook 语料库的 TF-IDF RAG 检索 - **可观测性：** 自定义 Prometheus 指标，Grafana dashboard 配置 - **DevSecOps / CI-CD：** GitHub Actions pipeline，Trivy 镜像扫描，Gitleaks 机密扫描，OWASP Dependency-Check，Terraform 验证门控 - **安全架构与 threat modeling：** 记录范围、检测覆盖率和明确的非目标 ## 许可证与作者 **许可证：** MIT — 见 [LICENSE](LICENSE) **作者：** Shahzain Iftikhar — DevOps 工程师，CNCF KubeStronaut

标签：AI安全分析, AI风险缓解, ECS, Grafana, Metaprompt, Terraform, 子域名突变, 安全规则引擎, 安全运营中心, 网络映射, 自动化响应, 自定义请求头, 请求拦截, 逆向工具