---

name: automating-ioc-enrichment description: > 使用 SOAR 平台、Python 流水线或 TIP Playbook 自动化对原始失陷指标（IOC）进行多源威胁情报上下文富化，以减少分析师分诊时间并标准化富化输出。适用于构建与 SIEM 告警、邮件提交流水线或威胁情报批量 IOC 处理集成的自动化富化工作流时使用。 domain: cybersecurity subdomain: threat-intelligence tags: [SOAR, enrichment, IOC, Cortex-XSOAR, Splunk-SOAR, VirusTotal, automation, CTI, NIST-CSF] version: 1.0.0 author: team-cybersecurity license: Apache-2.0

自动化 IOC 富化

适用场景

在以下情况下使用本技能：

• 构建 SOAR Playbook，在将 SIEM 告警路由给分析师之前自动为其添加威胁情报上下文
• 创建用于对钓鱼邮件提交进行批量 IOC 富化的 Python 流水线
• 通过预填充来自 VirusTotal、Shodan 和 MISP 的告警上下文，降低分析师平均分诊时间（MTTT）

不适用于在没有人工审查的情况下进行全自动封锁决策——富化自动化应该辅助决策，而非对高影响行动自主执行封锁。

前置条件

• SOAR 平台（Cortex XSOAR、Splunk SOAR、Tines 或 n8n）或 Python 3.9+ 环境
• API 密钥：VirusTotal、AbuseIPDB、Shodan，以及至少一个 TIP（MISP 或 OpenCTI）
• 用于告警消费的 SIEM 集成端点
• 记录在案的每个 API 速率限制预算（VirusTotal：免费版 4/分钟，企业版 500/分钟）

工作流程

步骤 1：设计富化流水线架构

为每种 IOC 类型定义富化流程：

SIEM 告警 → 提取 IOC → 分类类型 → 路由到富化函数
  IP 地址 → AbuseIPDB + Shodan + VirusTotal IP + MISP
  域名 → VirusTotal Domain + PassiveTotal + Shodan + MISP
  URL → URLScan.io + VirusTotal URL + Google Safe Browse
  文件哈希 → VirusTotal Files + MalwareBazaar + MISP
→ 聚合结果 → 计算置信度分数 → 更新告警 → 通知分析师

步骤 2：实现 Python 富化函数

import requests
import time
from dataclasses import dataclass, field
from typing import Optional

RATE_LIMIT_DELAY = 0.25  # VirusTotal 免费版速率限制：4 请求/秒

@dataclass
class EnrichmentResult:
    ioc_value: str
    ioc_type: str
    vt_malicious: int = 0
    vt_total: int = 0
    abuse_confidence: int = 0
    shodan_ports: list = field(default_factory=list)
    misp_events: list = field(default_factory=list)
    confidence_score: int = 0

def enrich_ip(ip: str, vt_key: str, abuse_key: str, shodan_key: str) -> EnrichmentResult:
    result = EnrichmentResult(ip, "ip")

    # VirusTotal IP 查询
    vt_resp = requests.get(
        f"https://www.virustotal.com/api/v3/ip_addresses/{ip}",
        headers={"x-apikey": vt_key}
    )
    if vt_resp.status_code == 200:
        stats = vt_resp.json()["data"]["attributes"]["last_analysis_stats"]
        result.vt_malicious = stats.get("malicious", 0)
        result.vt_total = sum(stats.values())

    time.sleep(RATE_LIMIT_DELAY)

    # AbuseIPDB
    abuse_resp = requests.get(
        "https://api.abuseipdb.com/api/v2/check",
        headers={"Key": abuse_key, "Accept": "application/json"},
        params={"ipAddress": ip, "maxAgeInDays": 90}
    )
    if abuse_resp.status_code == 200:
        result.abuse_confidence = abuse_resp.json()["data"]["abuseConfidenceScore"]

    # 计算复合置信度分数
    result.confidence_score = min(
        (result.vt_malicious / max(result.vt_total, 1)) * 60 +
        (result.abuse_confidence / 100) * 40, 100
    )

    return result

def enrich_hash(sha256: str, vt_key: str) -> EnrichmentResult:
    result = EnrichmentResult(sha256, "sha256")
    vt_resp = requests.get(
        f"https://www.virustotal.com/api/v3/files/{sha256}",
        headers={"x-apikey": vt_key}
    )
    if vt_resp.status_code == 200:
        stats = vt_resp.json()["data"]["attributes"]["last_analysis_stats"]
        result.vt_malicious = stats.get("malicious", 0)
        result.vt_total = sum(stats.values())
        result.confidence_score = int((result.vt_malicious / max(result.vt_total, 1)) * 100)
    return result

步骤 3：构建 SOAR Playbook（Cortex XSOAR）

在 Cortex XSOAR 中创建富化 Playbook：

1. 触发器：SIEM 中创建告警（通过 Webhook 或轮询）
2. 提取 IOC：使用"提取指标"任务，配合 IP、域名、URL、哈希的正则模式
3. 并行富化：扇出到多个富化任务同步执行
4. VirusTotal 富化：调用 !vt-file-scan 或 !vt-ip-scan 命令
5. AbuseIPDB 检查：调用 !abuseipdb-check-ip 命令
6. MISP 查询：调用 !misp-search 进行交叉核对
7. 分数聚合：计算复合分数的 Python 转换任务
8. 条件路由：分数 ≥70 → 高优先级队列；40-69 → 中等；<40 → 自动关闭并记录
9. 告警富化：将富化结果写入告警上下文供分析师查看

步骤 4：处理速率限制和失败

import time
from functools import wraps

def rate_limited(max_per_second):
    min_interval = 1.0 / max_per_second
    def decorator(func):
        last_called = [0.0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait = min_interval - elapsed
            if wait > 0:
                time.sleep(wait)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator

def retry_on_429(max_retries=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                response = func(*args, **kwargs)
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 60))
                    time.sleep(retry_after)
                else:
                    return response
        return wrapper
    return decorator

步骤 5：指标和调优

每周追踪流水线性能：

• 富化延迟：目标 <30 秒（从告警触发到富化输出）
• API 成功率：目标 >99%（识别速率限制或中断事件）
• 真正例率：追踪分析师对自动化置信度分数的覆盖情况
• 成本：追踪 API 调用量与预算对比（VirusTotal Enterprise：每 100 万次查询 $X）

核心概念

术语	定义
SOAR	安全编排、自动化和响应——用于自动化安全工作流程并集成不同工具的平台
富化 Playbook	向原始安全事件添加上下文情报的自动化工作流序列
速率限制	API 提供商对请求频率的限制（如 VirusTotal 免费版：4 请求/分钟）；流水线必须遵守这些限制
复合置信度分数	使用加权公式从多个富化来源聚合信号的单一分数
扇出模式	并行执行多个富化查询以最小化总富化延迟

工具与系统

• Cortex XSOAR（Palo Alto）：企业级 SOAR，拥有 700+ 市场集成，包括 VirusTotal、MISP、Shodan 和 AbuseIPDB
• Splunk SOAR（Phantom）：基于 Python Playbook 的 SOAR 平台；原生 Splunk SIEM 集成
• Tines：无代码 SOAR 平台，基于 Webhook 的自动化；对小团队具有成本效益
• TheHive + Cortex：开源 IR/富化平台，通过 Cortex 分析器进行可观察对象富化

常见陷阱

• 富化延迟导致阻塞：如果富化超过 5 分钟，分析师会开始处理未富化的告警，使富化失去意义。设置超时限制并提供部分结果。
• 无缓存：对同一 IOC 查询 50 次会产生不必要的 API 费用。默认将富化结果缓存 24 小时。
• 静默忽略 API 失败：失败的富化调用应被记录并触发回退逻辑，而非静默产生看似干净的空结果。
• 仅凭富化分数自动封锁：复合分数包含误报；对共享基础设施的封锁决策需要人工确认。