building-threat-intelligence-feed-integration
构建自动化威胁情报(Threat Intelligence)源集成管道,将 STIX/TAXII 源、 开源威胁情报和商业 TI 平台接入 SIEM 和安全工具,实现实时 IOC 匹配和告警。 适用于 SOC 团队需要通过自动化源接入、标准化、评分和分发到检测系统来 将威胁情报付诸实践的场景。
Best use case
building-threat-intelligence-feed-integration is best used when you need a repeatable AI agent workflow instead of a one-off prompt.
构建自动化威胁情报(Threat Intelligence)源集成管道,将 STIX/TAXII 源、 开源威胁情报和商业 TI 平台接入 SIEM 和安全工具,实现实时 IOC 匹配和告警。 适用于 SOC 团队需要通过自动化源接入、标准化、评分和分发到检测系统来 将威胁情报付诸实践的场景。
Teams using building-threat-intelligence-feed-integration should expect a more consistent output, faster repeated execution, less prompt rewriting.
When to use this skill
- You want a reusable workflow that can be run more than once with consistent structure.
When not to use this skill
- You only need a quick one-off answer and do not need a reusable workflow.
- You cannot install or maintain the underlying files, dependencies, or repository context.
Installation
Claude Code / Cursor / Codex
Manual Installation
- Download SKILL.md from GitHub
- Place it in
.claude/skills/building-threat-intelligence-feed-integration/SKILL.mdinside your project - Restart your AI agent — it will auto-discover the skill
How building-threat-intelligence-feed-integration Compares
| Feature / Agent | building-threat-intelligence-feed-integration | Standard Approach |
|---|---|---|
| Platform Support | Not specified | Limited / Varies |
| Context Awareness | High | Baseline |
| Installation Complexity | Unknown | N/A |
Frequently Asked Questions
What does this skill do?
构建自动化威胁情报(Threat Intelligence)源集成管道,将 STIX/TAXII 源、 开源威胁情报和商业 TI 平台接入 SIEM 和安全工具,实现实时 IOC 匹配和告警。 适用于 SOC 团队需要通过自动化源接入、标准化、评分和分发到检测系统来 将威胁情报付诸实践的场景。
Where can I find the source code?
You can find the source code on GitHub using the link provided at the top of the page.
SKILL.md Source
# 构建威胁情报源集成
## 适用场景
以下情况使用本技能:
- 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 已从源移除)
```Related Skills
tracking-threat-actor-infrastructure
威胁行为者基础设施追踪涉及使用被动 DNS、证书透明度日志、Shodan/Censys 扫描、WHOIS 分析和网络指纹技术,对对手控制的 C2 服务器、钓鱼域名和暂存服务器等资产进行监控、映射和持续追踪
profiling-threat-actor-groups
通过聚合 TTP 文档、历史活动数据、工具指纹和来自多个情报源的归因指标,为 APT 组织、犯罪组织和黑客活动组织开发全面的威胁行为者画像。适用于就行业特定威胁向管理层汇报、更新威胁模型假设,或针对特定对手优先部署防御控制措施。当涉及 MITRE ATT&CK 组织、Mandiant APT 画像、CrowdStrike 对手命名或行业特定威胁简报时激活。
processing-stix-taxii-feeds
处理通过 TAXII 2.1 服务器分发的 STIX 2.1 威胁情报包,将对象规范化为平台原生模式并路由到相应消费系统。适用于接入新的 TAXII 集合端点、自动化与 ISAC 的双向情报共享,或为格式错误的 STIX 包构建管道验证。当涉及 OASIS STIX、TAXII 服务器配置、MISP TAXII 或 Cortex XSOAR 情报源集成时激活。
performing-threat-modeling-with-owasp-threat-dragon
使用 OWASP Threat Dragon 创建数据流图,运用 STRIDE 和 LINDDUN 方法论识别威胁,并生成威胁模型报告用于安全设计审查。
performing-threat-landscape-assessment-for-sector
通过分析威胁行为者定向攻击模式、常见攻击向量和行业特定漏洞,开展行业特定威胁态势评估,为组织风险管理提供决策依据
performing-threat-intelligence-sharing-with-misp
使用 PyMISP 在 MISP 平台上创建、丰富和共享威胁情报事件,包括 IOC 管理、情报源集成、STIX 导出及社区共享工作流
performing-threat-hunting-with-yara-rules
使用 YARA 模式匹配规则在文件系统和内存转储中狩猎恶意软件、可疑文件和入侵指标。 涵盖规则编写、yara-python 扫描以及与威胁情报源的集成。
performing-threat-hunting-with-elastic-siem
使用 KQL/EQL 查询、检测规则和 Timeline 调查在 Elastic Security SIEM 中执行主动威胁狩猎, 识别绕过自动检测的威胁。适用于 SOC 团队针对特定 ATT&CK 技术进行狩猎、调查异常行为, 或使用 Elasticsearch 和 Kibana Security 验证检测覆盖缺口。
performing-threat-emulation-with-atomic-red-team
使用 atomic-operator Python 框架执行 Atomic Red Team 测试,进行 MITRE ATT&CK 技术验证。 从 YAML 原子测试加载测试定义、运行攻击模拟并验证检测覆盖率。适用于测试 SIEM 检测规则、 验证 EDR 覆盖率或开展紫队演练。
performing-open-source-intelligence-gathering
开源情报(OSINT)收集是红队演练的第一个主动阶段,操作员收集关于目标组织的公开可用信息,以识别攻击面、社会工程学目标、技术栈和凭据泄露情况。
performing-insider-threat-investigation
调查内部威胁事件,涉及滥用授权访问权限窃取数据、破坏系统或违反安全策略的员工、承包商或受信任合作伙伴。 结合数字取证、用户行为分析以及 HR/法务协调,构建基于证据的案例。适用于内部威胁调查、 员工数据盗窃、权限滥用、用户行为异常或内部威胁检测等请求场景。
performing-hardware-security-module-integration
使用 PKCS#11 接口集成硬件安全模块(HSM),通过 python-pkcs11、AWS CloudHSM 和 YubiHSM2 实现密码学密钥管理、签名操作和安全密钥存储。