# Implementing Soar Automation With Phantom > 使用 Splunk SOAR（原 Phantom）实施安全编排、自动化和响应（SOAR）工作流， 自动化告警分诊、IOC 富化、遏制动作和事件响应剧本。 适用于 SOC 团队需要减少分析师手工工作、标准化响应流程， 或将多种安全工具集成到自动化工作流中时。 - Skill: `killvxk/implementing-soar-automation-with-phantom` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add killvxk/implementing-soar-automation-with-phantom` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/implementing-soar-automation-with-phantom/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/implementing-soar-automation-with-phantom --- # 使用 Phantom 实施 SOAR 自动化 ## 适用场景 以下情况使用本技能： - SOC 团队需要自动化高频告警的重复性分诊和富化任务 - 手动响应时间超出 SLA 要求，自动化可降低 MTTR - 多个安全工具（SIEM、EDR、防火墙、TIP）需要编排的响应动作 - 需要剧本标准化以确保跨班次分析师响应的一致性 **不适用于**无人类审批门控的全自动遏制——高影响操作（如禁用账户或主机隔离）必须包含分析师决策点。 ## 前置条件 - 已部署 Splunk SOAR（Phantom）6.x+ 并可访问 Web 界面 - 已配置应用连接器：VirusTotal、CrowdStrike、ServiceNow、Active Directory、Splunk ES - Splunk ES 集成用于将 notable 事件作为 SOAR 事件摄取 - 每个集成工具的 API 凭据已存储在 SOAR 资产配置中 - 了解 Python 用于自定义剧本动作 ## 工作流程 ### 步骤 1：配置资产连接 通过 SOAR 应用设置与安全工具的集成： **VirusTotal 资产配置：** ```json { "app": "VirusTotal v3", "asset_name": "virustotal_prod", "configuration": { "api_key": "YOUR_VT_API_KEY", "rate_limit": true, "max_requests_per_minute": 4 }, "product_vendor": "VirusTotal", "product_name": "VirusTotal" } ``` **CrowdStrike Falcon 资产：** ```json { "app": "CrowdStrike Falcon", "asset_name": "crowdstrike_prod", "configuration": { "client_id": "CS_CLIENT_ID", "client_secret": "CS_CLIENT_SECRET", "base_url": "https://api.crowdstrike.com" } } ``` **Active Directory 资产：** ```json { "app": "Active Directory", "asset_name": "ad_prod", "configuration": { "server": "dc01.company.com", "username": "soar_service@company.com", "password": "SERVICE_ACCOUNT_PASSWORD", "ssl": true } } ``` ### 步骤 2：构建网络钓鱼分诊剧本 用 Python（Phantom 剧本格式）创建自动化钓鱼响应剧本： ```python """ 钓鱼邮件分诊自动化剧本 触发条件：通过 Splunk ES notable 或邮件摄取报告的新钓鱼邮件 """ import phantom.rules as phantom import json def on_start(container): # 从容器中提取构件（URL、文件哈希、发件人） artifacts = phantom.get_artifacts(container_id=container["id"]) for artifact in artifacts: artifact_type = artifact.get("cef", {}).get("type", "") if artifact_type == "url": phantom.act("url reputation", targets=artifact, assets=["virustotal_prod"], callback=url_reputation_callback, name="url_reputation") elif artifact_type == "hash": phantom.act("file reputation", targets=artifact, assets=["virustotal_prod"], callback=hash_reputation_callback, name="file_reputation") elif artifact_type == "ip": phantom.act("ip reputation", targets=artifact, assets=["virustotal_prod"], callback=ip_reputation_callback, name="ip_reputation") def url_reputation_callback(action, success, container, results, handle): if not success: phantom.comment(container, "URL 信誉检查失败") return for result in results: data = result.get("data", [{}])[0] malicious_count = data.get("summary", {}).get("malicious", 0) total_engines = data.get("summary", {}).get("total_engines", 0) if malicious_count > 5: # 高置信度恶意 — 自动封锁并升级 phantom.act("block url", targets=result, assets=["palo_alto_prod"], name="block_malicious_url") phantom.set_severity(container, "high") phantom.set_status(container, "open") phantom.comment(container, f"URL 被 {malicious_count}/{total_engines} 个引擎标记为恶意。" f"已在防火墙封锁。正在升级到二级。") # 创建 ServiceNow 工单 phantom.act("create ticket", targets=container, assets=["servicenow_prod"], parameters=[{ "short_description": f"钓鱼攻击 - 检测到恶意 URL", "urgency": "2", "impact": "2" }], name="create_incident_ticket") elif malicious_count > 0: # 中等置信度 — 请求分析师审查 phantom.promote(container, template="钓鱼调查") phantom.comment(container, f"URL 被 {malicious_count}/{total_engines} 个引擎标记。" f"需要分析师审查。") else: # 干净 — 关闭并添加注释 phantom.set_status(container, "closed") phantom.comment(container, f"URL 无威胁：0/{total_engines} 个引擎标记。已自动关闭。") def hash_reputation_callback(action, success, container, results, handle): if not success: return for result in results: data = result.get("data", [{}])[0] positives = data.get("summary", {}).get("positives", 0) if positives > 10: # 已知恶意软件 — 隔离并封锁 phantom.act("quarantine device", targets=result, assets=["crowdstrike_prod"], name="isolate_endpoint") phantom.set_severity(container, "high") def ip_reputation_callback(action, success, container, results, handle): if not success: return for result in results: data = result.get("data", [{}])[0] malicious = data.get("summary", {}).get("malicious", 0) if malicious > 3: phantom.act("block ip", targets=result, assets=["palo_alto_prod"], name="block_malicious_ip") ``` ### 步骤 3：构建告警富化剧本 自动化所有传入 SIEM 告警的富化： ```python """ 通用告警富化剧本 对每个新事件运行，在分析师审查前添加上下文 """ import phantom.rules as phantom def on_start(container): # 获取所有构件 success, message, artifacts = phantom.get_artifacts( container_id=container["id"], full_data=True ) ip_artifacts = [a for a in artifacts if a.get("cef", {}).get("sourceAddress")] domain_artifacts = [a for a in artifacts if a.get("cef", {}).get("destinationDnsDomain")] # 并行富化 IP for artifact in ip_artifacts: ip = artifact["cef"]["sourceAddress"] # VirusTotal 查询 phantom.act("ip reputation", parameters=[{"ip": ip}], assets=["virustotal_prod"], callback=enrich_ip_callback, name=f"vt_ip_{ip}") # GeoIP 查询 phantom.act("geolocate ip", parameters=[{"ip": ip}], assets=["maxmind_prod"], callback=geoip_callback, name=f"geo_{ip}") # Whois 查询 phantom.act("whois ip", parameters=[{"ip": ip}], assets=["whois_prod"], name=f"whois_{ip}") # 富化域名 for artifact in domain_artifacts: domain = artifact["cef"]["destinationDnsDomain"] phantom.act("domain reputation", parameters=[{"domain": domain}], assets=["virustotal_prod"], name=f"vt_domain_{domain}") def enrich_ip_callback(action, success, container, results, handle): """使用富化数据更新容器""" if success: for result in results: summary = result.get("summary", {}) phantom.add_artifact(container, { "cef": { "vt_malicious": summary.get("malicious", 0), "vt_suspicious": summary.get("suspicious", 0), "enrichment_source": "VirusTotal" }, "label": "enrichment", "name": "VT IP 富化" }) ``` ### 步骤 4：为高影响操作实施审批门控 为关键操作添加人工参与环节： ```python def containment_decision(action, success, container, results, handle): """向分析师展示遏制选项""" phantom.prompt( container=container, user="soc_tier2", message=( "已确认恶意活动。\n" f"主机：{container['artifacts'][0]['cef'].get('sourceAddress')}\n" f"威胁：{results[0]['summary'].get('threat_name')}\n\n" "选择遏制动作：" ), respond_in_mins=15, options=["隔离主机", "禁用账户", "两者都执行", "仅监控"], callback=execute_containment ) def execute_containment(action, success, container, results, handle): response = results.get("response", "仅监控") if response in ["隔离主机", "两者都执行"]: phantom.act("quarantine device", parameters=[{"hostname": container["artifacts"][0]["cef"]["sourceHostName"]}], assets=["crowdstrike_prod"], name="isolate_host") if response in ["禁用账户", "两者都执行"]: phantom.act("disable user", parameters=[{"username": container["artifacts"][0]["cef"]["sourceUserName"]}], assets=["ad_prod"], name="disable_account") phantom.comment(container, f"分析师已批准：{response}") ``` ### 步骤 5：配置剧本调度和触发器 在 SOAR 中设置事件触发器： ```json { "playbook_name": "phishing_triage_automation", "trigger": { "type": "event_created", "conditions": { "label": ["phishing", "notable"], "severity": ["high", "medium"] } }, "active": true, "run_as": "automation_user" } ``` ### 步骤 6：监控剧本性能 使用 SOAR 指标追踪自动化效果： ```python # 查询 SOAR API 获取剧本执行统计 import requests headers = {"ph-auth-token": "YOUR_SOAR_TOKEN"} response = requests.get( "https://soar.company.com/rest/playbook_run", headers=headers, params={ "page_size": 100, "filter": '{"status":"success"}', "sort": "create_time", "order": "desc" } ) runs = response.json()["data"] # 计算自动化指标 total_runs = len(runs) avg_duration = sum(r["end_time"] - r["start_time"] for r in runs) / total_runs auto_closed = sum(1 for r in runs if r.get("auto_resolved")) print(f"总执行次数：{total_runs}") print(f"平均时长：{avg_duration:.1f}s") print(f"自动解决：{auto_closed}/{total_runs} ({auto_closed/total_runs*100:.0f}%)") ``` ## 核心概念 | 术语 | 定义 | |------|------| | **SOAR** | 安全编排、自动化和响应（Security Orchestration, Automation, and Response）——整合安全工具与自动化剧本的平台 | | **剧本（Playbook）** | 定义由安全事件触发的顺序和并行动作的自动化工作流 | | **资产（Asset）** | SOAR 中已连接安全工具的配置（API 端点、凭据、连接参数） | | **容器（Container）** | 包含来自已摄取告警或事件的构件（IOC）的 SOAR 事件对象 | | **构件（Artifact）** | 容器内的单个 IOC 或数据点（IP、哈希、URL、域名、电子邮件） | | **审批门控（Approval Gate）** | 在执行高影响自动化操作前需要分析师决策的人工参与步骤 | ## 工具与系统 - **Splunk SOAR（Phantom）**：具有 300+ 应用集成和可视化剧本编辑器的企业级 SOAR 平台 - **Splunk ES**：将 notable 事件作为容器输送给 SOAR 进行自动分诊的 SIEM 平台 - **CrowdStrike Falcon**：通过 SOAR 集成实现自动化主机隔离和威胁狩猎的 EDR 平台 - **ServiceNow**：通过集成实现自动化事件工单创建和跟踪的 ITSM 平台 - **Palo Alto NGFW**：通过 SOAR 剧本实现自动化 IP/URL 封锁的防火墙 ## 常见场景 - **钓鱼分诊**：自动提取 URL/附件，在沙箱引爆，封锁恶意内容，创建工单 - **恶意软件告警富化**：跨 VT/MalwareBazaar 自动富化文件哈希，确认恶意后隔离 - **暴力破解响应**：自动检查攻击是否成功，被攻陷则禁用账户，封锁源 IP - **威胁情报 IOC 处理**：自动摄取 TI 情报源 IOC，与内部日志比对，为匹配项创建封锁 - **漏洞告警响应**：自动查询资产数据库获取受影响系统，按优先级创建补丁工单 ## 输出格式 ``` SOAR 剧本执行报告 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 剧本： 钓鱼分诊自动化 v2.3 容器： SOAR-2024-08921 触发条件： 来自 Splunk ES 的 notable 事件（钓鱼） 已执行动作： [1] URL 信誉（VirusTotal） — 14/90 个引擎标记为恶意 [2.1s] [2] IP 信誉（AbuseIPDB） — 置信度：85% [1.3s] [3] 封锁 URL（Palo Alto） — 在 PA-5260 上封锁 [0.8s] [4] 封锁 IP（Palo Alto） — 在 PA-5260 上封锁 [0.7s] [5] 创建工单（ServiceNow） — INC0012345 已创建 [1.5s] [6] 提示分析师（二级） — 响应："隔离主机" [4m 12s] [7] 隔离设备（CrowdStrike） — WORKSTATION-042 已隔离 [3.2s] 总时长： 4m 22s（对比手动分诊平均 35 分钟） 节省时间： 约 31 分钟 处置结果： 真阳性 — 已升级至事件响应 ```