# Building Threat Intelligence Feed Integration

> 构建自动化威胁情报（Threat Intelligence）源集成管道，将 STIX/TAXII 源、 开源威胁情报和商业 TI 平台接入 SIEM 和安全工具，实现实时 IOC 匹配和告警。 适用于 SOC 团队需要通过自动化源接入、标准化、评分和分发到检测系统来 将威胁情报付诸实践的场景。

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

---

# 构建威胁情报源集成

## 适用场景

以下情况使用本技能：
- SOC 团队需要将威胁情报源自动接入 SIEM 平台
- 多个 TI 源需要规范化为通用格式（STIX 2.1）
- 检测系统需要针对网络和端点遥测进行实时 IOC 匹配
- 需要建立 TI 源质量评估和去重流程

**不适用于**手动 IOC 查询——对于临时查询，请使用专用富化工具（VirusTotal、AbuseIPDB）。

## 前置条件

- MISP 实例或威胁情报平台（TIP）用于源聚合
- STIX/TAXII 客户端库（`taxii2-client`、`stix2` Python 包）
- 配置了 TI 框架的 SIEM 平台（Splunk ES、Elastic Security 或 Sentinel）
- 商业和开源源的 API 密钥（AlienVault OTX、Abuse.ch、CISA AIS）
- Python 3.8+ 用于源处理自动化

## 工作流程

### 步骤 1：识别并整理情报源

按类型、格式和更新频率映射可用源：

| 情报源 | 格式 | IOC 类型 | 更新频率 | 费用 |
|-------------|--------|-----------|-------------|------|
| AlienVault OTX | STIX/JSON | IP、域名、哈希、URL | 实时 | 免费 |
| Abuse.ch URLhaus | CSV/JSON | URL、域名 | 每 5 分钟 | 免费 |
| Abuse.ch MalwareBazaar | JSON API | 文件哈希 | 实时 | 免费 |
| CISA AIS | STIX/TAXII 2.1 | 全类型 | 每日 | 免费（美国政府） |
| CrowdStrike Intel | STIX/JSON | 全类型 + 威胁行为者 TTP | 实时 | 商业 |
| Mandiant Advantage | STIX 2.1 | 全类型 + 报告 | 实时 | 商业 |

### 步骤 2：接入 STIX/TAXII 源

连接到 TAXII 2.1 服务器并下载指标：

```python
from taxii2client.v21 import Server, Collection
from stix2 import parse

# 连接到 TAXII 服务器（示例：CISA AIS）
server = Server(
    "https://taxii.cisa.gov/taxii2/",
    user="your_username",
    password="your_password"
)

# 列出可用集合
for api_root in server.api_roots:
    print(f"API Root: {api_root.title}")
    for collection in api_root.collections:
        print(f"  Collection: {collection.title} (ID: {collection.id})")

# 从集合获取指标
collection = Collection(
    "https://taxii.cisa.gov/taxii2/collections/COLLECTION_ID/",
    user="your_username",
    password="your_password"
)

# 获取过去 24 小时添加的指标
from datetime import datetime, timedelta
added_after = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S.000Z")

response = collection.get_objects(added_after=added_after, type=["indicator"])
for obj in response.get("objects", []):
    indicator = parse(obj)
    print(f"Type: {indicator.type}")
    print(f"Pattern: {indicator.pattern}")
    print(f"Valid Until: {indicator.valid_until}")
    print(f"Confidence: {indicator.confidence}")
    print("---")
```

### 步骤 3：接入开源源

**Abuse.ch URLhaus 源：**

```python
import requests
import csv
from io import StringIO

# 下载 URLhaus 近期 URL
response = requests.get("https://urlhaus.abuse.ch/downloads/csv_recent/")
reader = csv.reader(StringIO(response.text), delimiter=',')

indicators = []
for row in reader:
    if row[0].startswith("#"):
        continue
    indicators.append({
        "id": row[0],
        "dateadded": row[1],
        "url": row[2],
        "url_status": row[3],
        "threat": row[5],
        "tags": row[6]
    })

print(f"从 URLhaus 接入了 {len(indicators)} 条 URL")

# 仅过滤活跃威胁
active = [i for i in indicators if i["url_status"] == "online"]
print(f"活跃威胁：{len(active)} 条")
```

**AlienVault OTX Pulse 源：**

```python
from OTXv2 import OTXv2, IndicatorTypes

otx = OTXv2("YOUR_OTX_API_KEY")

# 获取订阅的 pulse（过去 24 小时）
pulses = otx.getall(modified_since="2024-03-14T00:00:00")

for pulse in pulses:
    print(f"Pulse: {pulse['name']}")
    print(f"Tags: {pulse['tags']}")
    for indicator in pulse["indicators"]:
        print(f"  IOC: {indicator['indicator']} ({indicator['type']})")
```

**Abuse.ch Feodo Tracker（C2 IP）：**

```python
response = requests.get("https://feodotracker.abuse.ch/downloads/ipblocklist_recommended.json")
c2_data = response.json()

for entry in c2_data:
    print(f"IP: {entry['ip_address']}:{entry['port']}")
    print(f"Malware: {entry['malware']}")
    print(f"First Seen: {entry['first_seen']}")
    print(f"Last Online: {entry['last_online']}")
```

### 步骤 4：规范化与去重

将所有源转换为 STIX 2.1 格式以实现标准化：

```python
from stix2 import Indicator, Bundle
import hashlib

def create_stix_indicator(ioc_value, ioc_type, source, confidence=50):
    """将原始 IOC 转换为 STIX 2.1 指标"""
    pattern_map = {
        "ipv4": f"[ipv4-addr:value = '{ioc_value}']",
        "domain": f"[domain-name:value = '{ioc_value}']",
        "url": f"[url:value = '{ioc_value}']",
        "sha256": f"[file:hashes.'SHA-256' = '{ioc_value}']",
        "md5": f"[file:hashes.MD5 = '{ioc_value}']",
    }

    return Indicator(
        name=f"{ioc_type}: {ioc_value}",
        pattern=pattern_map[ioc_type],
        pattern_type="stix",
        valid_from="2024-03-15T00:00:00Z",
        confidence=confidence,
        labels=[source],
        custom_properties={"x_source_feed": source}
    )

# 跨源去重
seen_iocs = set()
unique_indicators = []

for ioc in all_collected_iocs:
    ioc_hash = hashlib.sha256(f"{ioc['type']}:{ioc['value']}".encode()).hexdigest()
    if ioc_hash not in seen_iocs:
        seen_iocs.add(ioc_hash)
        unique_indicators.append(
            create_stix_indicator(ioc["value"], ioc["type"], ioc["source"])
        )

bundle = Bundle(objects=unique_indicators)
print(f"唯一指标数：{len(unique_indicators)}")
```

### 步骤 5：推送至 SIEM 威胁情报框架

**推送至 Splunk ES 威胁情报：**

```python
import requests

splunk_url = "https://splunk.company.com:8089"
headers = {"Authorization": f"Bearer {splunk_token}"}

for indicator in unique_indicators:
    # 从 STIX 模式中提取 IOC 值
    ioc_value = indicator.pattern.split("'")[1]

    # 上传至 Splunk ES 威胁情报集合
    data = {
        "ip": ioc_value,
        "description": indicator.name,
        "weight": indicator.confidence // 10,
        "threat_key": indicator.id,
        "source_feed": indicator.get("x_source_feed", "unknown")
    }

    requests.post(
        f"{splunk_url}/services/data/threat_intel/item/ip_intel",
        headers=headers, data=data, verify=False
    )
```

**推送至 MISP 进行集中管理：**

```python
from pymisp import PyMISP, MISPEvent, MISPAttribute

misp = PyMISP("https://misp.company.com", "YOUR_MISP_API_KEY")

# 为源批次创建事件
event = MISPEvent()
event.info = f"TI Feed Import - {datetime.now().strftime('%Y-%m-%d')}"
event.threat_level_id = 2  # 中等
event.analysis = 2  # 已完成

# 将指标作为属性添加
for ioc in unique_indicators:
    attr = MISPAttribute()
    attr.type = "ip-dst" if "ipv4" in ioc.pattern else "domain"
    attr.value = ioc.pattern.split("'")[1]
    attr.to_ids = True
    attr.comment = f"Source: {ioc.get('x_source_feed', 'mixed')}"
    event.add_attribute(**attr)

result = misp.add_event(event)
print(f"MISP 事件已创建：{result['Event']['id']}")
```

### 步骤 6：监控源健康状态和质量

跟踪源有效性指标：

```spl
index=threat_intel sourcetype="threat_intel_manager"
| stats count AS total_iocs,
        dc(threat_key) AS unique_iocs,
        dc(source_feed) AS feed_count
  by source_feed
| join source_feed [
    search index=notable source="Threat Intelligence"
    | stats count AS matches by source_feed
  ]
| eval match_rate = round(matches / unique_iocs * 100, 2)
| sort - match_rate
| table source_feed, unique_iocs, matches, match_rate
```

## 核心概念

| 术语 | 定义 |
|------|-----------|
| **STIX 2.1** | 结构化威胁信息表达——用于共享威胁情报对象的标准化 JSON 格式 |
| **TAXII** | 指标信息可信自动化交换——通过 REST API 共享 STIX 数据的传输协议 |
| **TIP** | 威胁情报平台——用于聚合、评分和分发威胁情报的集中系统 |
| **IOC 评分（IOC Scoring）** | 根据源可靠性和相互印证为指标分配可信度值的过程 |
| **源去重（Feed Deduplication）** | 跨多个源删除重复 IOC，同时保留多源归因信息 |
| **IOC 过期（IOC Expiration）** | 删除过时指标的生存时间策略（IP：30 天，域名：90 天，哈希：1 年） |

## 工具与系统

- **MISP**：开源威胁情报平台，用于源聚合、关联和共享
- **AlienVault OTX**：免费威胁情报共享平台，提供社区 pulse 源
- **Abuse.ch**：免费威胁源套件（URLhaus、MalwareBazaar、Feodo Tracker、ThreatFox）
- **OpenCTI**：支持原生 STIX 2.1 存储的开源网络威胁情报平台
- **TAXII2 Client**：用于连接 STIX/TAXII 2.1 服务器以自动检索指标的 Python 库

## 常见场景

- **新源接入**：评估源质量，将字段映射到 STIX，配置自动化接入管道
- **多 SIEM 分发**：从 MISP 同时将规范化 IOC 推送至 Splunk、Elastic 和 Sentinel
- **误报减少**：按源数量和年龄对 IOC 评分，自动过期过时指标
- **源质量审计**：比较各源的检测匹配率，识别最高价值源
- **事件 IOC 共享**：将调查 IOC 打包为 STIX bundle，通过 TAXII 与 ISAC 共享

## 输出格式

```
威胁情报源状态 — 每日报告
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
日期：         2024-03-15
IOC 总数：     45,892 个活跃指标

源健康状态：
  源名称                IOC 数    匹配数  匹配率     状态
  Abuse.ch URLhaus      12,340    47      0.38%      健康
  AlienVault OTX        18,567    23      0.12%      健康
  Abuse.ch Feodo        1,203     12      1.00%      健康
  CISA AIS              8,945     8       0.09%      健康
  CrowdStrike Intel     4,837     31      0.64%      健康

今日操作：
  新增 IOC：        1,247 条
  过期 IOC：        892 条
  删除重复：        156 条
  SIEM 匹配：       121 个显著事件生成
  误报：            3 个（CDN IP 已从源移除）
```