killvxk 使用 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 资产配置：
{
"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 资产：
{
"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 资产：
{
"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 剧本格式）创建自动化钓鱼响应剧本：
"""
钓鱼邮件分诊自动化剧本
触发条件：通过 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 告警的富化：
"""
通用告警富化剧本
对每个新事件运行，在分析师审查前添加上下文
"""
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：为高影响操作实施审批门控
为关键操作添加人工参与环节：
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 中设置事件触发器：
{
"playbook_name": "phishing_triage_automation",
"trigger": {
"type": "event_created",
"conditions": {
"label": ["phishing", "notable"],
"severity": ["high", "medium"]
}
},
"active": true,
"run_as": "automation_user"
}
步骤 6：监控剧本性能
使用 SOAR 指标追踪自动化效果：
# 查询 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 分钟
处置结果： 真阳性 — 已升级至事件响应