# Automating Ioc Enrichment

> 使用 SOAR 平台、Python 流水线或 TIP Playbook 自动化对原始失陷指标（IOC）进行多源威胁情报上下文富化，以减少分析师分诊时间并标准化富化输出。适用于构建与 SIEM 告警、邮件提交流水线或威胁情报批量 IOC 处理集成的自动化富化工作流时使用。

- Skill: `killvxk/automating-ioc-enrichment` (Agent Skill, multi-file: 4 files)
- Install (CLI): `npx skillmds@latest add killvxk/automating-ioc-enrichment`
- Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/automating-ioc-enrichment/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: Apache-2.0
- Author: killvxk (https://skillmd.com/u/killvxk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/killvxk/automating-ioc-enrichment

---

# 自动化 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 富化函数

```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：处理速率限制和失败

```python
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 失败**：失败的富化调用应被记录并触发回退逻辑，而非静默产生看似干净的空结果。
- **仅凭富化分数自动封锁**：复合分数包含误报；对共享基础设施的封锁决策需要人工确认。