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

# Connect to TAXII server (example: CISA AIS)
server = Server(
    "https://taxii.cisa.gov/taxii2/",
    user="your_username",
    password="your_password"
)

# List available collections
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})")

# Fetch indicators from a collection
collection = Collection(
    "https://taxii.cisa.gov/taxii2/collections/COLLECTION_ID/",
    user="your_username",
    password="your_password"
)

# Get indicators added in last 24 hours
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

# Download URLhaus recent URLs
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"Ingested {len(indicators)} URLs from URLhaus")

# Filter for active threats only
active = [i for i in indicators if i["url_status"] == "online"]
print(f"Active threats: {len(active)}")

AlienVault OTX Pulse 情报源：

from OTXv2 import OTXv2, IndicatorTypes

otx = OTXv2("YOUR_OTX_API_KEY")

# Get subscribed pulses (last 24 hours)
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):
    """Convert raw IOC to STIX 2.1 indicator"""
    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}
    )

# Deduplicate across sources
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"Unique indicators: {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:
    # Extract IOC value from STIX pattern
    ioc_value = indicator.pattern.split("'")[1]

    # Upload to Splunk ES threat intel collection
    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=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
    )

推送到 MISP 进行集中管理：

from pymisp import PyMISP, MISPEvent, MISPAttribute

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

# Create event for feed batch
event = MISPEvent()
event.info = f"TI Feed Import - {datetime.now().strftime('%Y-%m-%d')}"
event.threat_level_id = 2  # Medium
event.analysis = 2  # Completed

# Add indicators as attributes
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 Event created: {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
IOC 过期	移除老化指标的生存时间策略（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 库

常见场景

• 新 Feed 接入：评估 Feed 质量，将字段映射到 STIX，配置自动化摄取流水线
• 多 SIEM 分发：将 MISP 中规范化后的 IOC 同时推送到 Splunk、Elastic 和 Sentinel
• 降低误报：按来源数量和时效性对 IOC 评分，自动过期陈旧指标
• Feed 质量审计：比较各 Feed 的检测命中率，找出价值最高的来源
• 事件 IOC 共享：将调查中的 IOC 打包为 STIX bundle，通过 TAXII 与 ISAC 共享

输出格式

THREAT INTEL FEED STATUS — Daily Report
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Date:         2024-03-15
Total IOCs:   45,892 active indicators

Feed Health:
  Feed                  IOCs    Matches  Match Rate  Status
  Abuse.ch URLhaus      12,340  47       0.38%       HEALTHY
  AlienVault OTX        18,567  23       0.12%       HEALTHY
  Abuse.ch Feodo        1,203   12       1.00%       HEALTHY
  CISA AIS              8,945   8        0.09%       HEALTHY
  CrowdStrike Intel     4,837   31       0.64%       HEALTHY

Actions Today:
  New IOCs ingested:    1,247
  IOCs expired:         892
  Duplicates removed:   156
  SIEM matches:         121 notable events generated
  False positives:      3 (CDN IPs removed from feed)