automating-ioc-enrichment

自动化 IOC 富化

何时使用

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

• 构建 SOAR playbook，在将 SIEM 警报路由给分析师之前，自动用威胁情报上下文富化警报
• 创建 Python 管道，用于对网络钓鱼邮件提交中的 IOC 进行批量富化
• 通过使用 VT、Shodan 和 MISP 数据预先填充警报上下文，缩短分析师的分类平均时间（MTTT）

不要将本技能用于无需人工审查的全自动阻断决策——对于高影响操作，富化自动化应当为决策提供信息，而不是自主执行阻断。

前置条件

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

工作流

步骤 1：设计富化管道架构

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

SIEM Alert → Extract IOCs → Classify Type → Route to enrichment functions
  IP Address → AbuseIPDB + Shodan + VirusTotal IP + MISP
  Domain → VirusTotal Domain + PassiveTotal + Shodan + MISP
  URL → URLScan.io + VirusTotal URL + Google Safe Browse
  File Hash → VirusTotal Files + MalwareBazaar + MISP
→ Aggregate results → Calculate confidence score → Update alert → Notify analyst

步骤 2：实现 Python 富化函数

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

RATE_LIMIT_DELAY = 0.25  # 4 requests/second for VT free tier

@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 lookup
    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"]

    # Calculate composite confidence score
    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：使用 “Extract Indicators” 任务，并使用针对 IP、域名、URL、哈希的正则表达式模式
3. 并行富化：同时扇出到多个富化任务
4. VT 富化：调用 !vt-file-scan 或 !vt-ip-scan 命令
5. AbuseIPDB 检查：调用 !abuseipdb-check-ip 命令
6. MISP 查询：调用 !misp-search 进行交叉引用
7. 分数聚合：Python transform 任务计算综合分数
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 调用量（VT Enterprise：每 100 万次查询 $X）

关键概念

术语	定义
SOAR	Security Orchestration, Automation, and Response——用于自动化安全工作流并集成各种工具的平台
富化 Playbook	为原始安全事件添加上下文情报的自动化工作流序列
速率限制	API 提供商对请求频率的限制（例如，VT 免费版：4 次请求/分钟）；管道必须遵守这些限制
综合置信度分数	使用加权公式聚合多个富化来源信号的单一分数
扇出模式	同时并行执行多个富化查询，以最小化总富化延迟

工具与系统

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

常见陷阱

• 因富化延迟而阻塞：如果富化耗时超过 5 分钟，分析人员就会开始处理未富化的告警，从而违背了初衷。应设置超时限制并提供部分结果。
• 没有缓存：对同一个 IOC 查询 50 次会产生不必要的 API 成本。默认将富化结果缓存 24 小时。
• 静默忽略 API 失败：失败的富化调用应被记录并触发回退逻辑，而不是静默产生空结果，让这些结果看起来像是干净的 IOC。
• 仅凭富化评分自动执行封禁：综合评分包含误报；针对共享基础设施的封禁决策需要人工确认。