IOC 自动化富集

自动化 IOC 富集

使用时机

此技能适用于：

• 构建 SOAR 剧本，在告警分发给分析师之前自动关联威胁情报上下文。
• 创建用于批量处理钓鱼邮件提交的 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 告警 → 提取 IOC → 分类类型 → 路由至富集函数
  IP 地址 → AbuseIPDB + Shodan + VirusTotal IP + MISP
  域名 → VirusTotal 域名 + PassiveTotal + Shodan + MISP
  URL → URLScan.io + VirusTotal URL + 谷歌安全浏览
  文件哈希 → VirusTotal 文件 + MalwareBazaar + MISP
→ 汇总结果 → 计算置信度分数 → 更新告警 → 通知分析师

步骤 2：实现 Python 富集函数

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

RATE_LIMIT_DELAY = 0.25  # VT 免费版 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 剧本（Cortex XSOAR）

在 Cortex XSOAR 中创建富集剧本：

1. 触发器：SIEM 创建告警（通过 webhook 或轮询）
2. 提取 IOC：使用“提取指标”任务，包含 IP、域名、URL、哈希的正则模式
3. 并行富集：同时向多个富集任务扇出
4. VT 富集：调用 !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 调用量是否在预算内（VT 企业版：每 100 万次查询 X 美元）

关键概念

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

工具与系统

• Cortex XSOAR（Palo Alto）：企业级 SOAR，拥有 700+ 市场集成，包括 VT、MISP、Shodan 和 AbuseIPDB
• Splunk SOAR（Phantom）：基于 Python 剧本的 SOAR 平台；原生集成 Splunk SIEM
• Tines：无代码 SOAR 平台，采用 webhook 驱动自动化；适合小型团队，成本效益高
• TheHive + Cortex：开源的事件响应/富集平台，通过 Cortex 分析器进行可观测对象富集

常见陷阱

• 因富集延迟而阻塞：如果富集耗时超过 5 分钟，分析师将开始处理未富集的告警，从而违背目的。应设置超时限制并提供部分结果。
• 无缓存：对同一个 IOC 查询 50 次会产生不必要的 API 成本。默认将富集结果缓存 24 小时。
• 静默忽略 API 故障：失败的富集调用应被记录并触发降级逻辑，而不是静默产生空白结果，否则会误判为干净的 IOC。
• 仅依赖富集分数进行自动封禁：复合分数存在误报；封锁共享基础设施的决策需要人工确认。