Cyrus-fr/AC-2035
GitHub: Cyrus-fr/AC-2035
一套面向 GCP/GKE 的开源 honeytoken 取证归因系统，通过 eBPF 检测、遥测回溯和图数据库重建攻击链路并触发自动化阻断。
Stars: 0 | Forks: 0
# AC-2035
基于 GCP/GKE 的开源 honeytoken 取证归因系统。
## 项目概述
AC-2035 在云基础设施中部署并监控 honeytoken（诱饵凭证、API 密钥和文件）。当 honeytoken 被触碰时，系统会捕获触发信号，对历史遥测数据（VPC Flow Logs、Cloud Logging、Pub/Sub 事件）执行取证回溯以重建访问链，在 Neo4j 中构建归因图，并可触发紧急阻断以撤销违规主体的访问权限。
## 架构摘要
| 组件 | 作用 |
|--------------|--------------------------------------------------------------------|
| `detector/` | eBPF 探针 + 遥测代理 — 底层触发/事件捕获 |
| `deployer/` | 生成、注入和轮换 honeytoken (GCP Secret Manager) |
| `collector/` | 订阅 Pub/Sub，读取历史日志以进行 T-30 分钟回溯 |
| `graph/` | Neo4j schema + 写入器 — 归因图构建 |
| `backtrace/` | 基于收集的遥测数据进行取证重建逻辑 |
| `killswitch/`| 在确认归因后撤销受损主体/访问权限 |
| `api/` | 暴露系统状态和控制端点的 FastAPI 服务 |
| `dashboard/` | 用于查看触发器、图表和归因结果的前端 |
| `infra/` | 用于 GKE、VPC、Pub/Sub 和 IAM 的 Terraform |
Dashboard/API 的身份认证由 **Zitadel** 提供；本地开发环境通过 Docker Compose 运行 Neo4j、Zitadel（+ Postgres）和 API。生产环境运行在由 `infra/` 中的 Terraform 配置的 GKE 上（VPC 原生、Workload Identity）。
## 前置条件
- Python 3.11+
- Docker + Docker Compose v2 (`docker compose ...`)
- Terraform ~> 1.5, `hashicorp/google` provider ~> 5.0
- 启用结算功能的 GCP 项目（用于 `infra/` 部署）
- 已通过认证的 `gcloud` CLI (`gcloud auth application-default login`)
## 本地开发设置
```
cp .env.example .env # fill in real values — never commit .env
docker compose up -d
```
这将启动：
| 服务 | URL | 说明 |
|-----------|-------------------------------|---------------------------------|
| Neo4j | http://localhost:7474 | 浏览器 UI；Bolt 端口为 `:7687` |
| Zitadel | http://localhost:8081 | 身份提供者（+ Postgres） |
| API | http://localhost:8000/health | 返回 `{"status": "ok"}` |
用于容器外本地脚本/工具的 Python 依赖项：
```
python -m venv .venv
.venv/Scripts/activate # Windows
pip install -r requirements.txt
```
## GCP 部署
```
cd infra
terraform init
terraform plan -var="project_id="
terraform apply -var="project_id="
```
这将配置：一个自定义 VPC + 子网（启用 Flow Logs）、一个具有 Workload Identity 的 VPC 原生 GKE 集群、`honeytoken-triggers` 和 `telemetry-raw` Pub/Sub 主题/订阅，以及用于 `collector`、`deployer` 和 `killswitch` 的最小权限服务账号（每个账号通过 Workload Identity 绑定到匹配的 Kubernetes 服务账号）。
## Honeytoken 部署器
`deployer/` 生成虚假凭证，将其注入 GKE 工作负载和 GCP Secret Manager，在本地 SQLite 注册表中进行跟踪，并按计划自动轮换。
```
python deployer/main.py deploy --pod --namespace --types gcp_key,gcp_api_key,db_connection,api_token
python deployer/main.py status
python deployer/main.py rotate --all
python deployer/main.py serve # run the auto-rotation scheduler in the foreground
```
- Token 类型：`gcp_key`（虚假服务账号 JSON）、`gcp_api_key` (`AIza...`)、`db_connection` (`postgresql://`/`mysql://`)、`api_token` (`Bearer` + 256 位十六进制)。均不是活动的、有效的凭证。
- GKE 注入会将目标 pod 解析为其所属的 Deployment，并修补 pod 模板（触发滚动更新），而不是直接修改实时 pod —— 因为在真实集群中，Pod 规范在创建后是不可变的。
- 无需 GCP 凭证或实时集群即可工作：如果未设置 `GCP_PROJECT_ID` 或未找到 kubeconfig/集群内配置，每个注入器将记录警告并跳过 —— token 仍会被生成并注册。
- 注册表位于 `deployer/registry.db`（已 gitignore）；轮换间隔由 `.env` 中的 `ROTATION_INTERVAL_HOURS` 控制（默认 24 小时）。
- Token 的值绝不会被记录 —— 仅记录 `token_id` 和目标位置。
## 遥测收集器
`collector/` 对 honeytoken 触发器做出反应，从每个可用数据源中提取触发前 30 分钟的遥测数据，并将其合并为一个按时间排序的 JSON 时间线。
- `pubsub_listener.py` — 订阅 `honeytoken-triggers` Pub/Sub 主题，将每条消息解析为 `TriggerEvent`，并通过 pipeline 运行它。
- `gcp_logs.py` — 提取触发前 30 分钟窗口内的 GCP Cloud Logging 条目（`k8s_container` 资源，匹配命名空间）。
- `vpc_flow.py` — 提取同一时间窗口内的 VPC Flow Log 条目，并提取源/目标 IP 和端口。
- `cloudflare_logs.py` — 通过 Logpull API 提取同一时间窗口内的 Cloudflare 访问日志，使用 CF-Ray ID 标记每个事件以供后续关联。
- `normalizer.py` — 将所有数据源合并为按时间戳排序的 `NormalizedEvent` 列表，并将其保存至 `collector/timelines/{token_id}_{timestamp}.json`。
运行本地模拟（无需 GCP/Cloudflare）：
```
python collector/simulate_trigger.py
```
这会伪造一个 `TriggerEvent` 以及 20-30 个真实的混合事件（`k8s_log` / `vpc_flow` / `cloudflare_access`），将它们合并并排序，保存时间线 JSON，并将其打印到控制台。
`pubsub_listener.py` 遵循与 deployer 相同的本地优先模式：如果 `GCP_PROJECT_ID` 为空，它将进入模拟模式，从 stdin 读取 `TriggerEvent` JSON 行，而不是订阅 Pub/Sub。
预期会出现部分时间线，这完全没问题 —— 如果缺少 GCP 或 Cloudflare 凭证（或某个数据源的请求失败），该获取器将记录警告并返回空列表，而不是中止运行，因此时间线仅由成功获取的数据源构建。
## Neo4j 图导入器
`graph/` 将 Phase 2 的时间线作为可查询的归因图加载到 Neo4j 中 —— 为每个参与者/资源建立节点，为每次交互建立边。
- `schema.py` — 应用唯一性约束（`ExternalIP`、`Pod`、`Service`、`Identity`、`Honeytoken`、`Technique`）和索引，并管理所有其他模块连接所用的单例 `get_driver()`。每条语句都是 `IF NOT EXISTS` 的 —— 多次运行是安全的。
- `ingestor.py` — 以每个事务最多 100 个事件的形式批量写入时间线。节点始终通过 `MERGE` 创建（重新导入时间线绝不会创建重复项）；边始终通过 `CREATE` 创建（每个事件都是时间上的一个独立事件）。
- `queries.py` — Phase 4（回溯引擎）直接调用的 Cypher 查询库：`find_paths_to_token`、`find_cf_ray_chain`、`find_vpc_flow_chain`、`get_full_graph`（针对 Phase 7 的 Cytoscape.js 格式）和 `clear_graph`。
运行端到端演示（加载最新的 `collector/timelines/` 文件，应用 schema，导入数据，然后运行路径查询）：
```
python collector/simulate_trigger.py # produces a timeline first, if you don't have one
python graph/demo_ingest.py
```
在 http://localhost:7474 打开 Neo4j 浏览器查看生成的图表。由于节点写入是通过每个节点的唯一键进行 `MERGE` 的，因此针对同一条时间线重新运行演示是安全的 —— 它绝不会复制 `ExternalIP`/`Pod`/`Honeytoken` 节点，只会添加新边。
## 取证回溯引擎
**这是 AC-2035 的创新贡献。** `backtrace/` 接收 honeytoken 触发器，通过 **将 CF-Ray 标头和 VPC Flow 链作为确定性密钥** 关联 Neo4j 图表，重建攻击者的完整路径 —— 从外部入口点经过每一跳直到被盗取的 token —— 为每一跳进行置信度评分，并标记 MITRE ATT&CK 技术。它会输出一个结构化的 `AttackObject`。
- `engine.py` — 编排器；`run_backtrace(trigger_event)` 运行完整的加载 → 导入 → 关联 → 查找路径 → 评分 → MITRE 标记流程，并返回 `AttackObject`。
- `correlator.py` — 确定性关联核心：`find_cf_ray`（哪个 CF-Ray 到达了触发 pod）、`trace_entry`（CF-Ray 背后的外部 IP）、`find_vpc_chain`（VPC Flow 横向移动链）。
- `path_finder.py` — 封装 `graph/queries.py`；查找从入口 IP 到 token 的候选路径，并将其提取为带评分的 `PathHop`。
- `scorer.py` — 为每一跳评级为 HIGH（CF-Ray + VPC 匹配）/ MEDIUM（仅一项匹配）/ LOW（仅基于时间）；一条路径的强度仅取决于其最弱的一跳。
- `mitre_tagger.py` — 将各跳映射到 ATT&CK 技术（T1190, T1021, T1552, T1083），当其 STIX bundle 存在时使用 mitreattack-python 验证 ID，否则回退到硬编码的元数据。
**CF-Ray 是主要的关联密钥**；当无法将任何 CF-Ray 绑定到触发 pod 时，引擎会 **回退到时间临近性** 评分（并且绝不会崩溃 —— 当无法重建路径时，它会返回一个空的、低置信度的 `AttackObject`）。
运行端到端演示（无需实时 GCP）：
```
python backtrace/demo_backtrace.py
```
它运行 Phase 2 的收集器模拟，构建一个连贯的攻击场景（外部 IP → pod → honeytoken），对其进行回溯，并打印完整的 `AttackObject` JSON 以及带有逐跳置信度和 MITRE 技术的人类可读攻击摘要。
## 紧急阻断编排器
`killswitch/` 接收一个 Phase 4 的 `AttackObject`，并 **并行** 触发三个遏制操作 —— 撤销受损身份的 GCP IAM 角色、在 Cloudflare 封锁攻击者 IP，以及终止其 Zitadel 会话 —— 为每次尝试写入完整的 JSON 审计追踪。
- `orchestrator.py` — 协调器；`execute(attack, mode)` 和 `approve(pending_id)` 通过 `ThreadPoolExecutor` 触发三个处理程序并写入审计日志。
- `gcp_iam.py` — `revoke()` 从项目的 IAM 策略中剥离绑定到该身份的每个角色（通过 google-auth 使用 Cloud Resource Manager REST）。
- `cloudflare.py` — `block_ip()` 针对入口 IP 创建一个 Firewall Rules 封锁。
- `zitadel.py` — `kill_sessions()` 通过 Management API 查找用户并删除每个活动会话。
**自动与手动模式：** `execute(attack, mode="auto")` 会立即触发所有三个操作；`mode="manual"` 会存储攻击并返回一个状态为 `pending` 的 `pending_id`，之后分析师可以调用 `approve(pending_id)` 来触发它们（记录为 `triggered_by: analyst`）。
**部分成功是按操作分别跟踪的，而不是全有或全无：** 状态包括 `executed`（三个操作全部成功）、`partial`（部分成功）或 `failed`（全部失败）—— 一个操作失败绝不会影响其他操作。当缺少凭证时（这是本地开发中的预期结果），每个处理程序都会优雅降级，返回一个 `ActionResult` 而不是崩溃。
这三个操作 **并行触发**（ThreadPoolExecutor）。审计日志保存至 `killswitch/audit/{token_id}_{timestamp}.json`。
运行端到端演示（无需实时 GCP/Cloudflare/Zitadel）：
```
python killswitch/demo_killswitch.py
```
它获取一个 `AttackObject`（复用 Phase 4 引擎），以自动模式触发紧急阻断，打印 `KillSwitchResult` JSON + 审计摘要，验证审计文件是否已写入，并演示手动 → 批准的流程。
## 自定义 eBPF 检测器
`detector/` 是一个替代 Falco 的内核空间 honeytoken 监视器。一个自定义的 eBPF C 程序在 GKE 节点级别挂钩 Linux Security Module 层；当任何进程触碰已部署的 honeytoken 文件时，它会向用户空间发出一个完全归因的事件（PID/UID/comm/inode/pod identity/timestamp），Python 代理随后将其作为 Phase 2 的 `TriggerEvent` 发送到 Pub/Sub。
- `ebpf/honeytoken_watch.c` — eBPF C 程（4 个钩子，ring-buffer 输出）。
- `ebpf/Makefile` — clang + libbpf CO-RE 构建（`make vmlinux`, `make all`）。
- `ebpf/loader.py` — 通过 libbpf (ctypes) 加载 `.o` 文件，附加钩子，并在运行时填充/更新 `watched_inodes` 映射。
- `telemetry_agent/agent.py` — 读取 ring-buffer 事件，根据注册表将 `token_id_hash` 解析为 `token_id`，构建 `TriggerEvent` 并发布到 Pub/Sub。
- `telemetry_agent/struct_defs.py` — C 结构体的逐字节精确 ctypes 镜像（自检布局断言）+ FNV-1a 哈希。
**四个内核钩子：**
| 钩子 | 捕获内容 |
|------|---------|
| `lsm/file_open` | 进程 `open()` 一个受监视的 honeytoken inode |
| `lsm/file_permission` | 绕过全新 `file_open` 的 `read()` / `mmap()` 访问 |
| `kprobe/sys_execve` | honeytoken pod 内的进程执行谱系 |
| `tp/sched/sched_process_exit` | 进程退出（进程在触碰 token 后立即退出是一个强烈的信号） |
**LSM 钩子仅作观察 —— 它们始终返回 0（允许）并且绝不阻断访问。** 检测器负责归因访问；它不负责阻止它。
**内核要求：** Linux，root 权限，且内核为 **5.7+** 并启用了 BPF LSM (`lsm=...,bpf`)。该 C 程序需要 clang + libbpf + bpftool 才能构建。在非 Linux 环境下（或缺少这些工具时），一切都会回退到模拟模式。
模拟模式（可在 Windows 上运行，无需 root/内核）：
```
python detector/demo_detector.py --simulate
```
这会伪造一个 honeytoken 事件，并将其通过真实的 decode → `TriggerEvent` → Pub/Sub pipeline 运行，打印结果。
真实模式（Linux + root + 内核 5.7+）：
```
sudo python detector/demo_detector.py --real
```
这将编译 eBPF 对象，加载它，监视一个测试文件的 inode，触碰该文件以触发检测，打印捕获的事件，然后卸载。
**Deployer 集成：** `deployer/injector.register_inodes_with_detector()` 会获取每个基于文件的 honeytoken 的 inode，并将其添加到 eBPF 的 `watched_inodes` 映射中 —— 从而形成闭环：*部署 → 注入 → 注册 inode → eBPF 监视 → 触发 → Pub/Sub → 收集器 → 回溯 → 紧急阻断*。
## FastAPI 后端
`api/` 是将每个阶段串联在一起的控制平面 —— Phase 8 dashboard 仅与此处通信。交互式 API 文档自动生成于 **http://localhost:8000/docs**。
**路由组：**
- `/api/graph` — 作为 Cytoscape.js JSON 格式的 Neo4j 图：`GET /full`、`GET /attack/{token_id}`（高亮显示攻击路径，如无则返回 404）、`GET /stats`（节点/边/类型计数）、`POST /clear`。
- `/api/alerts` — `GET /`（紧急阻断审计历史记录，`?limit=N`）、`GET /{token_id}`，以及 `POST /trigger` — **完整的端到端 pipeline**。
- `/api/tokens` — `GET /`（注册表，`?status=`）、`GET /{token_id}`、`POST /rotate`。`token_value` 永远不会被暴露。
- `/api/killswitch` — `GET /pending`、`POST /approve/{pending_id}`、`POST /execute`（针对提供的 AttackObject 自动触发）。
- `/ws` — WebSocket 告警流（见下文）。
- `GET /health` — 存活探针，`{"status": "ok"}`。
**触发 pipeline** — `POST /api/alerts/trigger` 接收一个 `TriggerEvent`，运行 `backtrace.engine.run_backtrace()` 重建攻击，通过 `orchestrator.execute(mode=KILLSWITCH_MODE)` 触发（或暂存）紧急阻断，广播实时告警，并返回 `AttackObject` + `KillSwitchResult`。这是一次调用实现整个系统功能的过程：检测 → 回溯 → 遏制。
**WebSocket 告警流** — 连接到 `/ws`；连接时你会收到 `{"type": "connected", ...}`，然后每 30 秒收到一次保活 ping 以及触发时的实时告警。告警 schema：
```
{
"type": "honeytoken_trigger | killswitch_fired | token_rotated | system_info",
"token_id": "…or null",
"timestamp": "ISO-8601",
"data": { }
}
```
**Neo4j → Cytoscape.js** — `api/serializers.py` 将 `graph.queries.get_full_graph()` 重塑为类型化的 Cytoscape `{nodes, edges}` 元素；`attack_object_to_cytoscape()` 渲染一条重建路径，并将其中每个元素标记为 `attack_path: true` 以便高亮显示。
**`KILLSWITCH_MODE`**（`.env`）控制 `POST /api/alerts/trigger` 如何触发紧急阻断：`manual`（默认值，更安全 —— 通过 `/api/killswitch/approve/{id}` 暂存以供分析师批准）或 `auto`（立即触发）。`CORS_ORIGINS`（默认 `*`）设置允许的 dashboard 源。
API 会随 `docker compose up -d` 启动；如果启动时无法连接到 Neo4j，它会记录警告并继续运行（在恢复之前路由将返回 500）。
## React Dashboard
`dashboard/` 是运维人员 UI —— 一个单页的 React + TypeScript + Vite 应用，具有独特的 **Hyprland 遇上网络欺骗行动** 外观（深空海军蓝基调，青色/紫色/绿色/红色/琥珀色点缀，CRT 扫描线叠加效果，微妙的 40px 网格背景，每个卡片上的棱角科技感边角括号，以及发光的边框 + 脉动状态点）。没有使用任何组件库 —— 每个组件都是定制的；图表渲染使用带有 dagre 布局的 Cytoscape.js。
**五个视图**（持久侧边栏，可折叠为图标）：
1. **攻击图** — 在 Cytoscape.js 中展示的完整 Neo4j 图（dagre 布局，类型化的节点形状/颜色）；通过 token_id 加载攻击路径以高亮显示；点击节点会出现滑入式详情面板。
2. **告警流** — 基于 WebSocket 的终端风格实时流；新告警从顶部滑入；按类型过滤；展开任意告警以查看原始 JSON。
3. **Token 面板** — 每个 honeytoken 对应一个卡片，带有状态发光边框（active=绿色脉冲，triggered=红色，rotated=琥珀色）；提供全部轮换 + 过滤功能。
4. **攻击时间线** — 某个 token 的重建交互的水平时间线，带有悬停工具提示和事件表。
5. **紧急阻断面板** — 左侧显示待批准项（批准/驳回），右侧显示带有逐操作 ✓/✗ 结果的紧急阻断审计日志。
运行它：
```
cd dashboard
npm install
npm run dev # http://localhost:5173
```
**环境变量**（`dashboard/.env`，从 `.env.example` 复制）：
- `VITE_API_URL`（默认 `http://localhost:8000`）— HTTP API 基础地址。
- `VITE_WS_URL`（默认 `ws://localhost:8000/ws`）— 告警流 URL。
它 **必须** 显式使用 `ws://`，并且 **绝不能通过 Vite 进行代理**（仅 `/api` 会被代理）—— 这可以避免 ws:// 与 http:// 的协议混淆。
WebSocket **会以指数退避算法自动重连**（1s → 2s → … 上限为 30s），并在恢复时向流中推送一条“WebSocket reconnected”系统告警。**`token_value` 绝不会在任何地方渲染** —— UI 始终显示 `••••••••`。
## 构建阶段
- [x] **Phase 0** — 环境设置（仓库结构、Docker Compose、Terraform 基础）
- [x] **Phase 1** — Honeytoken 部署器（生成、注入、轮换）
- [x] **Phase 2** — 遥测收集器（Pub/Sub 监听器、GCP 日志、VPC Flow、Cloudflare、标准化器）
- [x] **Phase 3** — Neo4j 图表导入器（时间线 → 图表节点和边）
- [x] **Phase 4** — 取证回溯引擎（CF-Ray + VPC Flow 关联、Cypher 路径重建、置信度评分、MITRE 标记）
- [x] **Phase 5** — 紧急阻断编排器（GCP IAM 撤销、Cloudflare IP 封锁、Zitadel 会话终止）
- [x] **Phase 6** — 自定义 eBPF 检测器（替代 Falco，内核空间 honeytoken 监视器）
- [x] **Phase 7** — FastAPI 后端（REST + WebSocket + Neo4j→Cytoscape 序列化器）
- [x] **Phase 8** — React Dashboard（Cytoscape.js 图表、实时告警、时间线回放、紧急阻断控制）
## 运行完整系统
完整的 AC-2035 技术栈，端到端：
```
# 1. Backend stack — Neo4j、Zitadel（+ Postgres）以及 FastAPI API
docker compose up -d
# API 位于 http://localhost:8000（文档位于 /docs），Neo4j 位于 http://localhost:7474
# 2. Dashboard
cd dashboard && npm install && npm run dev
# UI 位于 http://localhost:5173
# 3. 生成 telemetry timeline（无需 GCP）
python collector/simulate_trigger.py
# 4. 为该 token 触发 full pipeline — backtrace + kill-switch —
# 这也会向 dashboard 广播 live alert：
curl -X POST http://localhost:8000/api/alerts/trigger \
-H "Content-Type: application/json" \
-d '{"token_id":"","token_type":"api_token",
"trigger_time":"2026-07-12T00:00:00+00:00","pod_name":"checkout-api",
"pod_namespace":"prod","process_name":"python3","pid":4242,"source":"ebpf"}'
# 5. 打开 dashboard 并看着它到达
open http://localhost:5173
```
所有内容均在本地运行并具有优雅降级特性 —— 本地演示流程不需要实时的 GCP、Cloudflare 或 Zitadel 凭证。
标签：AV绕过, ECS, FastAPI, GCP, StruQ, Terraform, 安全取证与溯源, 蜜罐与欺骗防御, 请求拦截, 逆向工具