# Implementing Stix Taxii Feed Integration > STIX（结构化威胁信息表达式）和 TAXII（可信自动化情报信息交换）是 OASIS 开放标准，用于表示和传输网络威胁情报。 - Skill: `killvxk/implementing-stix-taxii-feed-integration` (Agent Skill, multi-file: 8 files) - Install (CLI): `npx skillmds@latest add killvxk/implementing-stix-taxii-feed-integration` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/implementing-stix-taxii-feed-integration/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/implementing-stix-taxii-feed-integration --- # 实现 STIX/TAXII Feed 集成 ## 概述 STIX（结构化威胁信息表达式）和 TAXII（可信自动化情报信息交换）是 OASIS 开放标准，用于表示和传输网络威胁情报。本技能涵盖使用 Python 实现 STIX/TAXII 2.1 Feed 消费者和生产者，配置 TAXII 服务器发现，集合管理，轮询新情报，解析 STIX 2.1 对象，以及将 Feed 集成到 SIEM 和 TIP 平台。 ## 前置条件 - Python 3.9+ 及 `taxii2-client`、`stix2`、`cti-taxii-client` 库 - 理解 STIX 2.1 数据模型（SDO、SCO、SRO） - 理解 TAXII 2.1 协议（发现、API 根、集合） - 可访问 TAXII 服务器（MITRE ATT&CK TAXII、Anomali STAXX） - 可选：medallion 用于运行本地 TAXII 2.1 服务器 ## 核心概念 ### TAXII 2.1 架构 TAXII 定义了三种服务类型的 RESTful API： - **发现（Discovery）**：返回可用 API 根的信息 - **API 根（API Root）**：包含集合并作为主要交互点 - **集合（Collection）**：通过 GET/POST 访问的 STIX 对象逻辑分组 ### STIX 2.1 对象模型 STIX 对象分为以下类别： - **SDO（STIX 领域对象）**：Indicator、Malware、Threat Actor、Campaign、Attack Pattern、Tool、Infrastructure、Vulnerability、Identity、Location、Note、Opinion、Report、Grouping - **SCO（STIX 网络可观测对象）**：IPv4-Addr、Domain-Name、URL、File、Email-Addr、Process、Network-Traffic、Artifact - **SRO（STIX 关系对象）**：Relationship、Sighting - **元对象**：标记定义（TLP）、语言内容、扩展定义 ### STIX Bundle Bundle 是一组一起传输的 STIX 对象集合。Bundle 具有唯一 ID，包含对象数组。TAXII 集合响应 GET 请求时提供 Bundle。 ## 实践步骤 ### 步骤 1：TAXII 服务器发现 ```python from taxii2client.v21 import Server, Collection, as_pages # 连接到 MITRE ATT&CK TAXII 服务器 server = Server("https://cti-taxii.mitre.org/taxii2/", user="", password="") print(f"标题：{server.title}") print(f"描述：{server.description}") # 列出 API 根 for api_root in server.api_roots: print(f"\nAPI 根：{api_root.title}") print(f" URL：{api_root.url}") # 列出集合 for collection in api_root.collections: print(f" 集合：{collection.title}（ID：{collection.id}）") print(f" 可读：{collection.can_read}") print(f" 可写：{collection.can_write}") ``` ### 步骤 2：从集合获取 STIX 对象 ```python from taxii2client.v21 import Collection, as_pages import json # 连接到 Enterprise ATT&CK 集合 ENTERPRISE_ATTACK_ID = "95ecc380-afe9-11e4-9b6c-751b66dd541e" collection = Collection( f"https://cti-taxii.mitre.org/stix/collections/{ENTERPRISE_ATTACK_ID}/", user="", password="", ) print(f"集合：{collection.title}") # 获取所有对象（分页） all_objects = [] for envelope in as_pages(collection.get_objects, per_request=50): objects = envelope.get("objects", []) all_objects.extend(objects) print(f" 已获取 {len(objects)} 个对象（总计：{len(all_objects)}）") print(f"\n总共获取对象：{len(all_objects)}") # 按类型分类 type_counts = {} for obj in all_objects: obj_type = obj.get("type", "unknown") type_counts[obj_type] = type_counts.get(obj_type, 0) + 1 for obj_type, count in sorted(type_counts.items()): print(f" {obj_type}: {count}") ``` ### 步骤 3：使用 stix2 库解析 STIX 2.1 对象 ```python from stix2 import parse, Filter, MemoryStore # 将对象加载到 MemoryStore 中查询 store = MemoryStore(stix_data=all_objects) # 查询所有指标 indicators = store.query([Filter("type", "=", "indicator")]) print(f"指标：{len(indicators)}") for ind in indicators[:5]: print(f" {ind.name}: {ind.pattern}") # 查询恶意软件 malware_list = store.query([Filter("type", "=", "malware")]) print(f"\n恶意软件家族：{len(malware_list)}") # 查询威胁行为者 actors = store.query([Filter("type", "=", "intrusion-set")]) print(f"威胁行为者：{len(actors)}") # 查找特定对象的关系 def get_related(store, source_id): relationships = store.query([ Filter("type", "=", "relationship"), Filter("source_ref", "=", source_id), ]) return relationships # 示例：获取 APT28 使用的所有技术 apt28 = store.query([ Filter("type", "=", "intrusion-set"), Filter("name", "=", "APT28"), ]) if apt28: rels = get_related(store, apt28[0].id) for rel in rels: target = store.get(rel.target_ref) if target: print(f" {rel.relationship_type} -> {target.name}（{target.type}）") ``` ### 步骤 4：实现自定义 TAXII 消费者 ```python from taxii2client.v21 import Collection, as_pages from stix2 import parse, Bundle from datetime import datetime, timedelta import json class TAXIIConsumer: """消费 STIX/TAXII 2.1 Feed 并提取 IOC。""" def __init__(self, collection_url, user="", password=""): self.collection = Collection(collection_url, user=user, password=password) self.last_poll = None def poll_new_objects(self, added_after=None): """轮询特定时间戳之后添加的对象。""" if added_after is None: added_after = ( self.last_poll or (datetime.utcnow() - timedelta(days=1)).strftime( "%Y-%m-%dT%H:%M:%S.000Z" ) ) all_objects = [] kwargs = {"added_after": added_after} for envelope in as_pages( self.collection.get_objects, per_request=100, **kwargs ): objects = envelope.get("objects", []) all_objects.extend(objects) self.last_poll = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z") return all_objects def extract_indicators(self, objects): """从 STIX 对象中提取可操作指标。""" indicators = [] for obj in objects: if obj.get("type") == "indicator": indicators.append({ "id": obj.get("id"), "name": obj.get("name", ""), "pattern": obj.get("pattern", ""), "pattern_type": obj.get("pattern_type", ""), "valid_from": obj.get("valid_from", ""), "valid_until": obj.get("valid_until", ""), "indicator_types": obj.get("indicator_types", []), "confidence": obj.get("confidence", 0), "labels": obj.get("labels", []), }) return indicators def extract_observables(self, objects): """提取 STIX 网络可观测对象。""" observables = [] observable_types = { "ipv4-addr", "ipv6-addr", "domain-name", "url", "file", "email-addr", "network-traffic", } for obj in objects: if obj.get("type") in observable_types: observables.append({ "type": obj["type"], "value": obj.get("value", ""), "id": obj.get("id"), }) return observables # 使用示例 consumer = TAXIIConsumer( f"https://cti-taxii.mitre.org/stix/collections/{ENTERPRISE_ATTACK_ID}/" ) new_objects = consumer.poll_new_objects() indicators = consumer.extract_indicators(new_objects) print(f"新指标：{len(indicators)}") ``` ### 步骤 5：使用 Medallion 设置本地 TAXII 服务器 ```python # medallion 配置（medallion.conf） TAXII_CONFIG = { "backend": { "module_class": "MemoryBackend", }, "users": { "admin": "admin_password", "readonly": "readonly_password", }, "taxii": { "max_content_length": 10485760, }, } # 运行 medallion 服务器： # pip install medallion # python -m medallion --config medallion.conf --port 5000 # 向本地 TAXII 服务器添加对象 import requests def push_to_taxii(server_url, collection_id, stix_bundle, user, password): """将 STIX Bundle 推送到 TAXII 2.1 集合。""" url = f"{server_url}/collections/{collection_id}/objects/" headers = { "Content-Type": "application/stix+json;version=2.1", "Accept": "application/taxii+json;version=2.1", } response = requests.post( url, json=stix_bundle, headers=headers, auth=(user, password), timeout=30, ) return response.json() ``` ## 验证标准 - TAXII 服务器发现返回有效的 API 根和集合 - STIX 对象从 TAXII 集合正确获取和解析 - 指标提取包含有效的 STIX 模式 - 正确处理大型集合的分页 - 消费者跟踪轮询状态以进行增量更新 - 本地 TAXII 服务器接受并提供 STIX Bundle ## 参考资料 - [STIX 2.1 规范](https://docs.oasis-open.org/cti/stix/v2.1/stix-v2.1.html) - [TAXII 2.1 规范](https://docs.oasis-open.org/cti/taxii/v2.1/taxii-v2.1.html) - [taxii2-client PyPI](https://pypi.org/project/taxii2-client/) - [stix2 Python 库](https://stix2.readthedocs.io/) - [MITRE ATT&CK TAXII 服务器](https://cti-taxii.mitre.org/taxii2/) - [Medallion TAXII 服务器](https://github.com/oasis-open/cti-taxii-server)