DaPhilll/SOAR-Playbook-Engineering-Incident-Response-Automation

[](https://github.com/DaPhilll) # SOAR Playbook 工程与事件响应自动化 ## 仓库结构 ``` /scripts virustotal_enrichment.py wazuh_endpoint_isolate.py anyrun_sandbox_submit.py deploy-shuffle.sh requirements.txt LICENSE README.md ``` ## 1. 执行摘要与目标 * **问题陈述：** 在活跃事件期间，手动进行威胁情报查询和控制台切换会增加平均响应时间（MTTR），并在时间压力下引发错误。 * **解决方案概述：** 本项目在开源的 Shuffle 平台上构建了安全编排、自动化与响应（SOAR）工作流。它通过 webhook 接收 SIEM 警报，通过 VirusTotal API 丰富文件哈希指标，并通过 Wazuh API 隔离受感染的 endpoint，从而实现原本需要手动执行的分类步骤自动化。 * **核心能力：** * 通过经过认证的 API 调用实现 endpoint 遏制。 * 自动化丰富文件哈希失陷指标（IOC）。 * 在路由分发给分析师之前，通过条件逻辑过滤良性结果。 * 向 SOC 通信通道输出标准化通知。 ## 2. 架构与环境拓扑 此工作流运行在共享实验室环境（VMware Workstation Pro，`10.10.0.0/24`）中，并与 `SRV-SOC01` 上的 Wazuh 部署直接集成。Endpoint 隔离操作针对 `WKSTN-01`。 * **编排主机：** Ubuntu Server — `SRV-SOC01`，已容器化。 * **编排核心：** Shuffle SOAR 引擎（前端、后端、Orborus 执行引擎、OpenSearch 后端），通过 Docker Compose 部署。 * **集成平面：** HTTP webhook，用于从上游 SIEM 警报系统接收 JSON payload。 * **集成的 API：** VirusTotal v3 REST API、AnyRun Sandbox API 以及 Wazuh manager API（`55000/tcp`）。 ## 3. 工程思考过程与方法论 * **设计考量：** 开源、厂商中立的 SOAR 平台避免了供应商锁定，并允许将集成构建为普通的 Python 脚本，而非专有插件。 * **技术挑战与解决方案：** * **挑战：** 原始的 SIEM JSON 警报结构不可预测，可能会破坏解析逻辑。在脚本节点内硬编码 API token 会带来安全风险。 * **解决方案：** 在所有接收节点中进行标准化的 JSON schema 处理与异常捕获。所有凭据均被移出脚本主体，转存入 Shuffle 的加密身份验证保管库中。 ## 4. 网络杀戮链与威胁生命周期映射 * **安装与利用：** 在执行后立即从进程遥测数据中提取可疑文件。 * **命令与控制：** 通过主机隔离切断持久的 beacon 连接。 * **目标操作：** 通过自动化的网络隔离来限制横向移动和数据暂存。 ## 5. MITRE ATT&CK 矩阵对齐 | 战术 | 技术 ID | 技术名称 | 缓解机制 | | :--- | :--- | :--- | :--- | | **执行** | T1204.002 | 恶意文件 | Endpoint 警报接收，并通过 VirusTotal 立即进行哈希信誉检查。 | | **横向移动** | T1570 | 横向工具传输 | 自动化的网络边界限制，以阻断资产到资产的文件复制。 | | **缓解** | M1040 | Endpoint 隔离 | 向 Wazuh manager 发起经过认证的 API 调用，以应用立即的主机阻断。 | ## 6. 集成的威胁情报工具 * **VirusTotal v3 API：** 针对文件指标提供历史信誉评分、多引擎检测计数以及行为元数据。 * **AnyRun 交互式沙箱 API：** 在隔离沙箱中自动化执行 payload，以捕获行为遥测数据。 ## 7. 实现与代码 ### 基础设施初始化 `scripts/deploy-shuffle.sh` ``` # 克隆 Shuffle 仓库 git clone https://github.com/Shuffle/Shuffle cd Shuffle # 初始化 frontend、backend、database 和 orchestration worker 容器 sudo docker-compose up -d ``` ### 用例 1：VirusTotal 信誉查询 `scripts/virustotal_enrichment.py` ``` import requests import json def check_vt_reputation(file_hash, api_key): url = f"https://www.virustotal.com/api/v3/files/{file_hash}" headers = {"x-apikey": api_key} response = requests.get(url, headers=headers) if response.status_code == 200: data = response.json() malicious_votes = data['data']['attributes']['last_analysis_stats']['malicious'] undetected_votes = data['data']['attributes']['last_analysis_stats']['undetected'] return json.dumps({"hash": file_hash, "malicious_score": malicious_votes, "undetected": undetected_votes}) else: return json.dumps({"error": f"API request failed with status code {response.status_code}"}) ``` ### 用例 2：通过 Wazuh API 隔离主机 `scripts/wazuh_endpoint_isolate.py` ``` import requests import json import urllib3 # 抑制本地实验室证书的不安全 HTTPS 警告 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) def isolate_endpoint(wazuh_ip, jwt_token, agent_id): base_url = f"https://{wazuh_ip}:55000/active-response/send" headers = { "Authorization": f"Bearer {jwt_token}", "Content-Type": "application/json" } payload = { "command": "host-isolate", "arguments": ["-", "user"], "custom": False, "agent_list": [agent_id] } response = requests.put(base_url, headers=headers, data=json.dumps(payload), verify=False) if response.status_code == 200: return {"status": "Success", "message": f"Isolation command deployed to Agent {agent_id}."} else: return {"status": "Failed", "error": response.text} ``` ### 用例 3：通过 ANY.RUN SDK 进行沙箱引爆 `scripts/anyrun_sandbox_submit.py` — 一旦 VirusTotal 步骤超过恶意评分阈值，便使用官方的 `anyrun-sdk` 包提交标记的 URL 以进行自动引爆。 ``` import os from anyrun.connectors import SandboxConnector def detonate_url(target_url, api_key): with SandboxConnector.windows(api_key) as connector: analysis_id = connector.run_url_analysis(target_url) for status in connector.get_task_status(analysis_id): print(status) verdict = connector.get_analysis_verdict(analysis_id) if verdict in ("Suspicious", "Malicious"): connector.get_analysis_report(analysis_id, report_format="html", filepath="./reports") return {"analysis_id": analysis_id, "verdict": verdict} ``` ## 8. 工作流逻辑与输出示例 ### 恶意 Payload 分类流水线 ``` [ SIEM Webhook Ingest ] │ ▼ [ VirusTotal Hash Lookup ] │ ▼ [ Logic Gate: Is Malicious Score >= 5? ] ├──► (No) ──► [ Append Case Notes ] ──► [ Terminate Workflow ] │ └──► (Yes) ──► [ ANY.RUN Sandbox Detonation ] ──► [ Trigger Wazuh Host Isolation API ] ──► [ Send SOC Alert ] ``` 下方的 JSON 展示了流经每个节点的数据。该文件哈希是已知的空字符串 MD5 值，使用它是为了确保其不会被误认为是真实的指标。 **丰富节点输出（VirusTotal 结果）：** ``` { "hash": "d41d8cd98f00b204e9800998ecf8427e", "malicious_score": 47, "undetected": 12 } ``` **遏制节点输出（Wazuh API 返回值）：** ``` { "status": "Success", "message": "Isolation command deployed to Agent 01." } ``` **SOC 通知 Payload：** ``` { "alert_id": "SIEM-LAB-001", "file_hash": "d41d8cd98f00b204e9800998ecf8427e", "callback_ip": "198.51.100.7", "agent_id": "01", "vt_malicious_score": 47, "action_taken": "host-isolate", "status": "Contained" } ``` ## 9. 强化与未来增强 * **当前状态：** Shuffle 容器运行在隔离的内部桥接网络上。应用之间的凭据共享范围仅限于加密的身份验证保管库，以防止在脚本调试期间发生明文暴露。 * **未来路线图：** * [ ] 在对生产级基础设施执行遏制操作之前，通过工单 API（GLPI）添加手动确认步骤。 * [ ] 为出站连接目标添加 IP 信誉查询，以加快网络层级的阻断速度。 ## 依赖项 使用以下命令安装脚本依赖项： ``` pip install -r requirements.txt ``` 有关确切的版本，请参见 `requirements.txt`。 ## 许可证 MIT — 详见 [许可证](./LICENSE)。


[](https://github.com/DaPhilll)

标签：Ask搜索, Shuffle, SOAR, VirusTotal, Wazuh, 安全运营, 库, 应急响应, 扫描框架, 版权保护, 网络信息收集, 自动化剧本, 请求拦截, 逆向工具