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 分钟
处置结果：  真阳性 — 已升级至事件响应
```