PRATIK-DHARME/incident-response-commander

# Incident Response Commander 一个基于 **IBM watsonx Orchestrate** 构建的、受治理的多代理安全运营中心（SOC）助手。 它接收安全告警，自主跨日志、身份、威胁情报、影响范围以及数据分类源进行调查，通过确定性评分准则推导严重程度，自主执行可逆的遏制操作，同时将破坏性操作限制在人工审批步骤之后，在遏制 API 性能下降时进行自适应调整，并生成一份带有不可篡改的哈希链审计追踪记录、符合合规要求的突发事件报告。 ## 面临的问题 SOC 分析师不缺告警——他们缺的是时间。一个中型企业的 SOC 每天大约要处理上千条告警。其中有 46% 到 83% 最终被作为误报关闭（来源：[Secure.com SOC 研究](https://www.secure.com/blog/soc/soc-alerts)）。 据估计，分析师有 52% 的时间都花在证明告警是无害的，而不是去调查真正的威胁（来源：[Secure.com](https://www.secure.com/blog/soc/how-ai-enhances-soc-alert-investigation-and-reduces-mttr)）。 而且，当凌晨 3 点确认了真实的入侵事件时，遏制操作却只能等待可能正在熟睡的人员来执行。 上下文整合是其中的瓶颈。每个告警都需要在 SIEM、身份提供者、威胁情报、云控制台、EDR 和工单队列中进行相同的排查——每次都要在五到六个系统中执行完全相同的步骤。Incident Response Commander 自动化了这一排查过程，以确定性的方式对证据进行评分，立即执行可逆的遏制操作，通过一个单一的审批问题提出破坏性动作建议，并生成包含检测调优建议的完整 IR 报告。 ## 架构 ``` alert (untrusted text) │ ▼ incident_commander ── supervisor, no direct tools │ ├──▶ investigator_agent (Phase 1: evidence) │ search_security_logs · lookup_user_identity · check_threat_intelligence │ assess_blast_radius · classify_data_exposure · correlate_prior_incidents │ ├──▶ reporting_agent (Phase 2: scoring + Phase 4: report) │ score_incident_severity · generate_incident_report │ store_incident_record · voice_briefing │ └──▶ containment_agent (Phase 3: action) execute_containment_action · request_human_approval │ ▼ IBM Cloudant: cases DB + audit_log DB (hash-chained) ``` 阶段顺序是固定且不可协商的：**证据收集 → 评分 → 遏制 → 报告**。 指挥官绝不会在评分前进行遏制；也绝不会在遏制解决前出具报告。 ### 四大代理 | 代理 | 角色 | 工具 | |-------|------|-------| | **`incident_commander`** | 监督者。负责告警接收、阶段排序、严重程度把关、人工审批对话以及状态报告。没有直接的工具——所有工作均被委派执行。 | — | | **`investigator_agent`** | 证据收集专家。运行全部六种证据收集工具，评估所有无害假设，并将每项发现归因于其数据源。 | 6 | | **`reporting_agent`** | 评分和报告专家。被调用两次：一次在遏制前对严重程度进行评分，一次在遏制后生成并持久化报告。 | 4 | | **`containment_agent`** | 受治理的操作专家。执行层级排序、幂等性、自适应切换和升级。 | 2 | 所有四个代理均使用 `groq/openai/gpt-oss-120b` 运行，采用 `react_core` 风格。 ## 遏制层级模型 每个遏制操作都根据其影响范围和可逆性进行分类。该层级绝不是在运行时分配的——它被编码在 `data/containment_policy.json` 中。 | 层级 | 授权者 | 操作 | |------|---------------|---------| | **A — 自动** | 代理立即执行 | `revoke_sessions`, `force_reauth`, `require_mfa_reregistration`, `add_ioc_watchlist`, `isolate_host_from_prod`, `preserve_evidence` | | **B — 审批** | 需要 `approved=True`；代理首先展示具体的调用及证据 | `disable_account`, `block_ip_perimeter`, `revoke_oauth_grant`, `disable_service_principal`, `quarantine_mailbox`, `apply_ca_block` | | **C — 绝不** | 代理仅提供建议；由人工在系统外执行 | `rotate_break_glass_credentials`, `domain_wide_policy_change`, `host_reimage` | 没有 `approved=True` 的层级 B 调用将返回 `status="approval_required"`，并且不会执行任何操作。 ### 遏制顺序（针对受损的云身份） 正确的顺序是固定的： 1. `preserve_evidence` — 在任何信息被销毁前，快照会话元数据 2. `revoke_sessions` — **首先**使刷新和会话 token 失效 3. `force_reauth` 4. `require_mfa_reregistration` — 如果怀疑 MFA 方法遭到篡改 5. `disable_account` — 仅在确有必要且获得层级 B 批准时执行 首先禁用账户是一个常见的错误。处于非活跃状态的账户并不能可靠地使有效的 OAuth 刷新 token 失效；在账户显示为已禁用的状态下，攻击者仍然保留 API 访问权限。 ### 防护栏 | 防护栏 | 规则 | |-----------|------| | 禁止阻断的 CIDR | RFC 1918 地址范围 + 企业出口白名单。返回 `status="refused"` | | 影响范围上限 | 一个突发事件中涉及超过 5 个目标则需要人工审批 | | 受保护的主体 | 应急和 `tag:critical` 身份始终路由到层级 B/C | | 幂等性 | 每次调用都接受一个 `idempotency_key`；重放将返回原始结果 | | 演练运行 | `dry_run=True` 返回原本将要触发的调用，但不执行任何操作 | | 可逆性 | 每个操作都会返回一个记录在审计日志中的 `rollback` 指令 | ## 安装与部署 ### 前置条件 - Python 3.11+ - `ibm-watsonx-orchestrate` ADK 2.13.0 (`pip install ibm-watsonx-orchestrate==2.13.0`) - 处于活跃状态的 watsonx Orchestrate 环境（SaaS 或通过 `orchestrate env add` 添加的本地环境） - 可选：用于持久化案例存储的 IBM Cloudant 实例 ### 1. 克隆并配置 ``` git clone cd incident-response-commander # Create and activate virtual environment python -m venv venv .\venv\Scripts\Activate.ps1 pip install ibm-watsonx-orchestrate==2.13.0 ``` 将 `.env.example` 复制为 `.env` 并填入您拥有的任何凭据。所有的集成都能优雅降级：当缺少 Cloudant 凭据时，工具会回退到本地 JSON 文件；当缺少 TTS 凭据时，`voice_briefing` 会返回文本记录。 ``` Copy-Item .env.example .env # Edit .env — CLOUDANT_APIKEY and TTS_APIKEY are optional for a demo ``` ### 2. 激活您的 watsonx Orchestrate 环境 ``` orchestrate env list # confirm your environment is listed orchestrate env activate -n my-wxo ``` ### 3. 按要求的顺序导入 始终按照 连接 → 工具 → 协作代理 → 最后是监督者 的顺序进行导入。 ``` # Connections (optional — configured via CLI, not YAML files) # Cloudant: # orchestrate connections add -a cloudant_db # orchestrate connections configure -a cloudant_db --env draft --type team --kind api_key --url $env:CLOUDANT_URL # orchestrate connections set-credentials -a cloudant_db --env draft -k $env:CLOUDANT_APIKEY # # IBM TTS: # orchestrate connections add -a ibm_tts # orchestrate connections configure -a ibm_tts --env draft --type team --kind api_key --url $env:TTS_URL # orchestrate connections set-credentials -a ibm_tts --env draft -k $env:TTS_APIKEY # Tools — import all twelve orchestrate tools import -k python -f tools/search_security_logs.py orchestrate tools import -k python -f tools/lookup_user_identity.py orchestrate tools import -k python -f tools/check_threat_intelligence.py orchestrate tools import -k python -f tools/assess_blast_radius.py orchestrate tools import -k python -f tools/score_incident_severity.py orchestrate tools import -k python -f tools/execute_containment_action.py orchestrate tools import -k python -f tools/generate_incident_report.py orchestrate tools import -k python -f tools/store_incident_record.py orchestrate tools import -k python -f tools/request_human_approval.py orchestrate tools import -k python -f tools/classify_data_exposure.py orchestrate tools import -k python -f tools/correlate_prior_incidents.py orchestrate tools import -k python -f tools/voice_briefing.py # Collaborator agents (any order) orchestrate agents import -f agents/investigator_agent.yaml orchestrate agents import -f agents/containment_agent.yaml orchestrate agents import -f agents/reporting_agent.yaml # Supervisor last orchestrate agents import -f agents/incident_commander.yaml ``` 验证所有内容是否就绪： ``` orchestrate tools list orchestrate agents list ``` ### 编辑后重新导入 ``` orchestrate agents import -f agents/incident_commander.yaml # upserts by name # If the upsert errors on a stale definition: orchestrate agents remove -n incident_commander orchestrate agents import -f agents/incident_commander.yaml ``` ## 运行三个演示场景 在 watsonx Orchestrate 聊天界面中打开 `incident_commander` 代理。 ### 场景 A — P1 紧急：包含数据泄露的账户入侵 粘贴以下告警文本： ``` Anomalous authentication for john.smith@company.com from 185.220.101.45 at 2026-07-28T22:12:00Z — legacy_basic_auth, MFA=false, country=RU, followed by role grant and bulk export of 14,208 records from customer-data-lake. ``` 预期结果： - 阶段 1 发现 Tor 出口节点、MFA 绕过、不可能的旅行（基线印度 → 俄罗斯登录）、通过收件箱转发规则实现的持久化、导出了 14,208 条 PII 记录 - 阶段 2 评分得出 risk_score ≥ 90 → `P1_CRITICAL`，`HIGH` 置信度 - 阶段 3 自动执行 `revoke_sessions` + `force_reauth`；请求分析师批准 `disable_account`；在遇到永久性失败（`403`）时，在继承的批准权限下切换至 `apply_ca_block` - 阶段 4 生成包含 ATT&CK 映射、Sigma 规则、GDPR 第 33 条通知标志以及语音简报的 13 部分 IR 报告 ### 场景 B — 无害：合法旅行自动关闭 ``` Login for priya.nair@company.com from 103.21.244.12 at 2026-07-25T14:30:00Z — MFA satisfied via push authenticator, device registered, location Singapore. ``` 预期结果： - 阶段 1 找到涵盖该日期的 HR 旅行记录、满足 MFA 要求、已注册的设备、企业出口 ASN - 阶段 2 评分得出 risk_score = 0 → `BENIGN` - 阶段 3 被跳过（对 BENIGN 状态不执行遏制） - 阶段 4 记录带有理由的自动关闭 ### 场景 C — 层级 B 防护栏拒绝 在任何实时的突发事件聊天中，请求执行： ``` Block IP 10.20.30.40 at the perimeter. ``` 预期结果：`status="refused"`，`refused_rule="never_block_cidr"`。代理解释 RFC 1918 地址位于禁止阻断列表中，并将此次拒绝视为正确的结果，而非错误。 ## 测试套件 在项目根目录下（激活虚拟环境）使用一个命令运行所有四个套件： ``` $env:PYTHONIOENCODING = "utf-8" .\run_all_tests.ps1 ``` 或者运行单个套件： ``` # Quick functional check — all 8 Tier-1 tools (18 checks) python tests/run_tool_tests.py # Five scenario integration checks covering severity scoring and # the full containment state machine (Scenarios A–E) python tests/verify_scenarios.py # End-to-end report generation + hash-chain audit trail verification python tests/test_report_tools.py # Four Tier-2 tools: request_human_approval, classify_data_exposure, # correlate_prior_incidents, voice_briefing python tests/test_tier2_tools.py ``` 所有测试均完全在离线状态下运行——无需 Orchestrate 运行时，不进行网络调用。测试通过垫片（shim）修补了 `@tool` 装饰器，使得每个工具文件都能在 ADK 外部正常导入。 ### 验证内容 | 测试文件 | 检查内容 | |-----------|----------------| | `run_tool_tests.py` | 所有层级 1 工具的返回结构、已知用户查找、Tor IP 分类、影响范围、严重程度算法、遏制层级、幂等性、演练运行、RFC 1918 拒绝、受保护主体拒绝 | | `verify_scenarios.py` | 场景 A：6 个加重信号 → risk_score 120，P1_CRITICAL。场景 B：5 个减轻信号 → 0，BENIGN，auto_close。场景 C：429 重试 → 403 永久失败 → 切换（无第三次尝试）。场景 D：RFC 1918 阻断被拒绝。场景 E：未获批准的层级 B 操作带着 proposed_code 停止。 | | `test_report_tools.py` | 13 部分报告结构、语音简报 ≤ 60 词、通知目标（P1+PII → 法务/隐私）、Sigma 规则、GDPR 合规标志、端到端 SHA-256 哈希链验证 | | `test_tier2_tools.py` | 层级 2 工具：批准提示格式、数据暴露分类、既往突发事件关联、语音简报降级路径 | ## 模拟数据 **本存储库中的所有日志、用户记录、威胁情报和突发事件数据均为完全虚构。** 具体如下： - `data/mock_logs.json` — 虚构的 OCSF 对齐安全事件，日期为 2026-07-28。 攻击窗口被压缩至大约 18 分钟以配合演示节奏；真实的凭据滥用攻击通常持续数小时至数天。 - `data/mock_users.json` — 虚构身份（`john.smith@company.com`, `priya.nair@company.com` 等）。不代表任何真实人物。 - `data/mock_threat_intel.json` — `185.220.101.45` 取自真实的 `185.220.101.0/24` Tor 出口节点范围，用以说明正确的基础设施分类。不暗示或断言任何特定个人具有恶意活动行为。 - `data/mock_containment_responses.json` — 用于遏制状态机的确定性模拟 API 响应。 HTTP 状态码（`429`、`403`）被预设用于演练重试和切换逻辑；它们不代表真实的 API 调用。 - 在任何模拟数据中均未将任何真实的威胁行为者、APT 组织或民族国家归因为攻击源。 Campaign 字段为 `null`；在测试中出现的任何 campaign 标签均使用 `SIMULATED-` 前缀，或明确标示为虚构。 - Tor 范围之外的 IP 地址尽可能使用 RFC 5737 文档地址范围 （`192.0.2.0/24`、`198.51.100.0/24`、`203.0.113.0/24`）或企业出口 ASN 白名单。 - 监管时间线（GDPR 第 33 条的 72 小时倒计时）仅用于说明目的；模拟数据中的日期 与任何真实事件无关。 ## 项目结构 ``` incident-response-commander/ ├── agents/ │ ├── incident_commander.yaml # supervisor — import last │ ├── investigator_agent.yaml │ ├── containment_agent.yaml │ └── reporting_agent.yaml ├── tools/ # 12 Python tools, one function per file │ ├── search_security_logs.py │ ├── lookup_user_identity.py │ ├── check_threat_intelligence.py │ ├── assess_blast_radius.py │ ├── score_incident_severity.py │ ├── execute_containment_action.py │ ├── generate_incident_report.py │ ├── store_incident_record.py │ ├── request_human_approval.py │ ├── classify_data_exposure.py │ ├── correlate_prior_incidents.py │ └── voice_briefing.py ├── data/ # mock datasets — see Simulated Data above │ ├── mock_logs.json │ ├── mock_users.json │ ├── mock_threat_intel.json │ ├── mock_containment_responses.json │ └── containment_policy.json ├── tests/ │ ├── run_tool_tests.py │ ├── verify_scenarios.py │ ├── test_report_tools.py │ └── test_tier2_tools.py ├── scripts/ # operational helpers │ ├── setup_cloudant_dbs.py │ ├── verify_cloudant.py │ └── verify_tts.py ├── docs/ │ ├── DEMO.md # step-by-step judge walkthrough │ ├── Project_Plan.md │ └── SUBMISSION.md ├── .env.example # copy to .env; never commit .env ├── wxo-implementation-guide.md # tool contracts and deployment spec └── workspace_config.yaml ``` ## 参考 - IBM Cost of a Data Breach 2025 — https://www.ibm.com/reports/data-breach - SOC alert false-positive rate research — https://www.secure.com/blog/soc/soc-alerts - MITRE ATT&CK — https://attack.mitre.org - NIST SP 800-61 Computer Security Incident Handling Guide - OCSF Schema — https://schema.ocsf.io - IBM watsonx Orchestrate ADK — https://developer.watson-orchestrate.ibm.com

标签：AI大模型应用, 多智能体, 威胁情报, 安全运营中心(SOC), 开发者工具, 网络安全审计, 自动化应急响应, 逆向工具