FAAS19/autonomous-dfir-triage-agent

# 自主 DFIR 分诊 Agent ## 目录 - [概述](#overview) - [问题背景](#the-problem) - [解决方案架构](#solution-architecture) - [核心技术特性](#key-technical-features) - [仓库结构](#repository-structure) - [演示的课程概念](#course-concepts-demonstrated) - [快速开始](#getting-started) - [运行测试](#running-tests) - [安全设计](#security-design) - [商业影响](#business-impact) ## 概述 **自主 DFIR 分诊 Agent** 是一个完全基于 **Google Agent Development Kit (ADK)** 和 **FastMCP** 构建的智能流水线。它通过将确定性的 NLP 过滤与受治理的多 Agent 推理层相结合，自动化了事件响应中最耗时的阶段——解析和分诊 Windows Event ID 4688 (进程创建) 日志。 ## 问题背景 当网络安全事件超越最初的 SOC 分诊被升级时，它会交由专门的 **事件响应 (IR) 分析师** 或 **DFIR 从业者** 处理——而不是 SOC L1/L2 分析师。IR 分析师在数字取证、内存分析和证据保管链方面拥有专业技能——但他们不仅成本高昂、极其稀缺，而且时间紧迫。 核心瓶颈不在于检测。而在于**对完整数字证据时间线的取证分诊**。 通过像 [Plaso](https://github.com/log2timeline/plaso) 这样的取证工具处理的单个端点镜像会生成一个**超级时间线**——即系统上所有取证痕迹按时间顺序的合并。仅本项目的证据文件就包含了**横跨 9 种取证来源类型的 153,066 个事件**： | 来源 | 类型 | 数量 | 示例 | |---|---|---|---| | `REG` | 注册表项 | 88,588 | Run 键、服务、UserAssist、USB 条目、BagMRU | | `FILE` | 文件系统 (MFT) | 34,906 | 文件状态、创建/修改时间戳 | | `PE` | 可移植可执行文件 | 25,703 | 编译时间戳、导入表时间戳 | | `WEBHIST` | 浏览器历史记录 | 3,035 | MSIE 缓存、URL 记录 | | `LNK` | Windows 快捷方式 | 376 | Shell 项痕迹、最近文件访问 | | `EVT` | Windows 事件日志 | 188 | WinEVT 条目 | | `LOG` | 系统日志 | 235 | 系统痕迹 | | `OLECF` | OLE 复合文档 | 31 | Office 文档元数据 | | `RECBIN` | 回收站 | 4 | 已删除文件痕迹 | IR 分析师必须同时关联所有这些痕迹流，以重建攻击者的活动——识别持久化机制、横向移动、数据准备和防御规避。手动完成这项工作每个端点需要 **1 到 2 个分析师工作日**。 | 挑战 | 影响 | |---|---| | **数据量** | 每个端点时间线包含超过 15 万个痕迹 | | **噪音** | 注册表项变动和 PE 时间戳占据了主要信号 | | **人工关联** | 每次事件需要 1-2 个 IR 分析师工作日 | | **专家稀缺** | IR 分析师成本高昂；职业倦怠率极高 | | **MTTR 压力** | 在分诊期间，活跃的威胁行为者仍滞留在环境中 | ## 解决方案架构 该 Agent **在任何 LLM 看到数据之前，就在数学层面上消除了噪音**，采用了一个包含 4 个阶段的混合流水线： ``` ┌──────────────────────────────────────────────────────────────────────┐ │ EVIDENCE LAYER (Immutable) │ │ CSV Evidence → SHA-256 Hash → Chain of Custody │ └────────────────────────────────┬─────────────────────────────────────┘ │ ┌────────────────────────────────▼─────────────────────────────────────┐ │ MCP SERVER: triage_engine (5 Tools) │ │ │ │ ┌──────────────────┐ ┌───────────────────┐ ┌──────────────────┐ │ │ │ Phase 1 │ │ Phase 2 │ │ Phase 3 │ │ │ │ Frequency │→ │ NLP Embeddings │→ │ SIGMA Rule │ │ │ │ Stacking │ │ + Isolation │ │ Enrichment │ │ │ │ Filter O(N) │ │ Forest Scoring │ │ 3,100+ rules │ │ │ └──────────────────┘ └───────────────────┘ └──────────────────┘ │ │ Drops >98% noise text-embedding-004 MITRE ATT&CK │ │ in linear time TF-IDF fallback (offline) │ └────────────────────────────────┬─────────────────────────────────────┘ │ Top anomalies + SIGMA matches ┌────────────────────────────────▼─────────────────────────────────────┐ │ MULTI-AGENT ENGINE (Google ADK) │ │ │ │ ┌────────────────┐ ┌────────────────┐ ┌───────────────────────┐ │ │ │ Orchestrator │ → │ Log Analyst │ → │ Zero-Trust Validator │ │ │ │ (Coordinator) │ │ (SOC Expert) │ │ Gate │ │ │ └────────────────┘ └────────────────┘ └───────────────────────┘ │ │ ↓ 3x retry loop │ │ ↓ provenance check │ │ ↓ contradiction detect │ └────────────────────────────────┬─────────────────────────────────────┘ │ ┌────────────────────────────────▼─────────────────────────────────────┐ │ OUTPUT: Zero-Hallucination Report │ │ MITRE ATT&CK mappings │ Risk scores │ Confidence │ Citations │ │ Hash-chained provenance.jsonl (offline verifiable) │ └──────────────────────────────────────────────────────────────────────┘ ``` ### ADK 工作流图 多 Agent 分诊系统构建为一个结构化的 Google ADK 工作流图，协调专门的推理节点：

### 流水线阶段 | 阶段 | 方法 | 效果 | |---|---|---| | **1 — 频率堆叠** | O(N) 归一化频率计数；丢弃执行次数 >100 次的命令 | 以线性时间消除超过 98% 的日志量 | | **2 — NLP 异常检测** | `text-embedding-004` 向量 + 孤立森林评分 (TF-IDF 后备) | 呈现最罕见、最异常的命令模式 | | **3 — SIGMA 富化** | 离线匹配 3,100+ 捆绑的 SIGMA 规则 | 确定性地将异常预映射到 MITRE ATT&CK 战术 | | **4 — 多 Agent 分析** | ADK 编排器 → 日志分析师 → 零信任验证器 | LLM 仅对预过滤、预富化的异常进行推理 | ### 设计决策 | 决策 | 选择 | 理由 | |---|---|---| | 为什么不用 DBSCAN？ | 孤立森林 | DBSCAN 的内存复杂度为 O(N²) —— 在处理 50 万+ 事件时不可用 | | Embedding 模型 | `text-embedding-004` + TF-IDF 后备 | Google AI 对齐 + 无需 Gemini 密钥的离线弹性 | | 威胁情报来源 | 捆绑的 SIGMA YAML (离线) | 无网络依赖；确定性的 MITRE ATT&CK 映射 | | 防幻觉 | 来源网关 + 矛盾检测 + 3 次重试 | 每项发现都必须引用有效的 `tool_execution_id` | | 会话存储 | 内存中 (无数据库) | 本地部署无需任何基础设施 | ## 核心技术特性 ### 零幻觉网关 日志分析师生成的每一项发现都必须引用来自 MCP 服务器来源日志的有效 `tool_execution_id`。验证器 Agent 会丢弃任何缺乏引用或与证据相矛盾的发现。在编写报告之前，3 次重试循环会提升结果的置信度。 ### 哈希链来源追踪 每次工具调用都会将带有时间戳的记录写入 `outputs/provenance.jsonl`。该链条是只追加的，并且可以离线验证——提供了从原始 CSV 到最终结论的符合取证规范的审计追踪。 ### 只读受治理 MCP 服务器 FastMCP 的 `triage_engine` 向 Agent 暴露了 5 个工具。这里没有 `execute_shell` 功能。LLM 仅通过对内存中副本进行操作的类型化工具包装器与证据进行交互。原始 CSV 绝不会被修改。 ### 离线弹性 - 当没有 Gemini 密钥时，阶段 2 会自动从 `text-embedding-004` 回退到 `TfidfVectorizer` (scikit-learn) - 阶段 3 的 SIGMA 规则在本地捆绑——不需要 SIEM 或网络连接 - SIGMA 规则缓存 (`sigma_rules_cache.pkl`) 在重复运行时可提供 10 倍的加速 ### 交互式仪表板 自定义的 FastAPI 应用程序通过以下功能扩展了 ADK Web 基础： - 实时 Agent 流式视图 - 映射到 MITRE ATT&CK 的发现面板 - 哈希链来源审计查看器 - 用于编程访问的 `/api/findings` 端点 ## 仓库结构 ``` autonomous-dfir-triage-agent/ │ ├── .agents/ # Agent workspace conventions │ ├── AGENTS.md # Global coding rules and security constraints │ └── skills/dfir_triage/ # DFIR triage skill (Windows log guidelines + SIGMA) │ ├── agents/ # Specialist agent implementations │ ├── log_analyst.py # SOC Expert LLM agent (ADK) │ └── provenance_logger.py # Hash-chained audit trail writer │ ├── app/ # Application layer │ ├── agent.py # ADK Workflow: Orchestrator + pipeline nodes │ ├── fast_api_app.py # Custom FastAPI server (dashboard + ADK + A2A) │ ├── mcp_client.py # MCP stdio session manager │ ├── app_utils/ # Session, artifact, and A2A services │ └── templates/dashboard.html # Interactive DFIR dashboard UI │ ├── cases/ │ └── timeline.csv # Pre-committed forensic evidence (168,964 events, 69.5 MB) │ ├── mcp_server/ │ ├── server.py # FastMCP triage_engine (5 governed tools) │ ├── sigma_rules_cache.pkl # Pre-compiled SIGMA cache (10× speedup) │ └── sigma_rules/ # 3,100+ SIGMA rules (Windows, AWS, Azure, Linux, GCP) │ ├── outputs/ # Generated artefacts (committed as examples) │ ├── dfir_triage_report.md # Sample zero-hallucination incident report │ └── provenance.jsonl # Sample hash-chained audit trail │ ├── specs/ # Authoritative specification documents │ ├── 01_architecture.md │ ├── 02_data_schemas.yaml │ ├── 03_bdd_scenarios.md │ └── 05_agent_prompts.md │ ├── tests/ # Test suite │ ├── integration/ # End-to-end pipeline and server tests │ ├── unit/ # Unit tests │ └── eval/ # ADK evaluation datasets and runner │ ├── .env.example # API key template — copy to .env ├── pyproject.toml # Dependencies (managed by uv) ├── uv.lock # Locked dependency tree └── Dockerfile # Optional containerised deployment ``` ## 演示的课程概念 本项目演示了 *5 天 AI Agent 强化* 课程中的所有要求概念： | 概念 | 实现 | 文件 | |---|---|---| | **多 Agent 系统** | 编排器协调日志分析师 + 零信任验证器 | `app/agent.py` | | **MCP 服务器** | `triage_engine` — 5 个只读受治理的工具 | `mcp_server/server.py` | | **Agent 技能** | 包含 Windows 日志指南的 DFIR 分诊技能 | `.agents/skills/dfir_triage/SKILL.md` | | **安全与来源** | 哈希链审计追踪、零幻觉网关 | `agents/provenance_logger.py` | | **评估** | 带有 LLM-as-judge 评分的 ADK 评估数据集 | `tests/eval/` | | **规范驱动开发** | 架构、BDD 场景、数据 Schema、Prompt | `specs/` | | **可部署性** | FastAPI 仪表板 + CLI 运行器 + Dockerfile | `app/fast_api_app.py` | ## 快速开始 ### 前置条件 | 要求 | 详细信息 | |---|---| | **Python** | 3.11 或更高版本 | | **`uv`** | 快速 Python 包管理器 — [安装指南](https://docs.astral.sh/uv/getting-started/installation/) | | **OpenRouter API 密钥** | 免费，无需信用卡 — [在此获取](https://openrouter.ai/keys) | ### 安装说明 **1. 克隆仓库** ``` git clone https://github.com/FAAS19/autonomous-dfir-triage-agent.git cd autonomous-dfir-triage-agent ``` **2. 安装依赖** ``` uv sync ``` **3. 配置环境** ``` # Linux / macOS cp .env.example .env # Windows (Command Prompt) copy .env.example .env ``` 打开 `.env` 并设置你的 OpenRouter API 密钥： ``` OPENAI_API_KEY=your-openrouter-key-here OPENAI_API_BASE=https://openrouter.ai/api/v1 LOG_ANALYST_MODEL=openai/deepseek/deepseek-v4-flash ``` **4. 启动仪表板** ``` uv run python app/fast_api_app.py ``` 在浏览器中打开 **[http://localhost:8000/dashboard](http://localhost:8000/dashboard)**。 该仪表板提供： - 实时 Agent 流式传输及逐步的流水线进度 - 映射到 MITRE ATT&CK 的发现及其风险评分和置信度 - 哈希链来源审计追踪查看器 - 每项发现的原始痕迹证据面板

**5. (可选) 通过 CLI 运行无头分诊** ``` uv run python -m app.agent ``` 输出 `outputs/dfir_triage_report.md` 和 `outputs/provenance.jsonl`。 ## 运行测试 ``` uv run pytest ``` 所有 6 个测试 (集成 + 单元) 都应该通过。要进行 ADK 评估： ``` uv run python tests/eval/run_local_eval.py ``` ## 安全设计 | 特性 | 实现 | |---|---| | **只读证据** | MCP 工具在内存中的副本上操作——原始 CSV 绝不会被写入 | | **无 Shell 执行** | 所有工具定义中都不包含 `execute_shell` | | **零幻觉网关** | 验证器会丢弃任何没有有效 `tool_execution_id` 的发现 | | **哈希链来源追踪** | 带有 UTC 时间戳的只追加 `provenance.jsonl`——可离线验证 | | **图级别 RBAC** | 工具作用于特定的 Agent——没有过度授权的访问 | | **临时会话** | 内存会话服务——运行之间无持久状态 | ## 商业影响 | 指标 | 人工流程 | 使用此 Agent | |---|---|---| | 每次事件的分诊时间 | 1-2 个分析师工作日 | < 2 分钟 | | LLM 审查的事件数 | 168,964 | ~15-30 (仅限最高级别的异常) | | 日志量削减 | — | >98% (阶段 1 过滤) | | 幻觉率 | 不适用 | 0% (受来源网关限制) | | 威胁情报依赖 | 云 SIEM / 网络 | 完全离线 (捆绑 SIGMA) | ## 进阶：重新生成取证时间线 原始证据文件 (`cases/4DellLatitudeCPi.plaso`) 使用 Plaso 存储格式 `20190309`。需要固定版本的 Docker 镜像： ``` docker pull log2timeline/plaso:20210213 docker run -v "${PWD}/cases:/data" log2timeline/plaso:20210213 \ psort.py -o l2tcsv -w /data/timeline.csv /data/4DellLatitudeCPi.plaso ```

标签：多智能体, 安全运营, 库, 应急响应, 异常检测, 扫描框架, 逆向工具