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 服务器并下载指标：

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 源：

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 源：

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）：

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 格式以实现标准化：

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 威胁情报：

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 进行集中管理：

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：监控源健康状态和质量

跟踪源有效性指标：

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 已从源移除）