STIX (Structured Threat Information eXpression) and TAXII (Trusted Automated eXchange of Intelligence Information) are OASIS open standards for representing and transporting cyber threat intelligence.
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
STIX (Structured Threat Information eXpression) and TAXII (Trusted Automated eXchange of Intelligence Information) are OASIS open standards for representing and transporting cyber threat intelligence.
STIX (Structured Threat Information eXpression) and TAXII (Trusted Automated eXchange of Intelligence Information) are OASIS open standards for representing and transporting cyber threat intelligence. This skill covers implementing a STIX/TAXII 2.1 feed consumer and producer using Python, configuring TAXII server discovery, collection management, polling for new intelligence, parsing STIX 2.1 objects, and integrating feeds into SIEM and TIP platforms.
When to Use
When deploying or configuring implementing stix taxii feed integration capabilities in your environment
When establishing security controls aligned to compliance requirements
When building or improving security architecture for this domain
When conducting security assessments that require this implementation
Common Misconfigurations & Verification
Pagination dropped: calling get_objects() once instead of iterating as_pages() silently truncates large collections to the first page - always page until exhausted and log the running total.
Broken incremental polling: an added_after timestamp stored in local time or the wrong format re-fetches everything or skips objects. Persist the server-returned timestamp in UTC %Y-%m-%dT%H:%M:%S.000Z and track state per collection.
SCO vs Indicator confusion: extracting only type == "indicator" misses raw SCOs (ipv4-addr, domain-name, file) that carry the actual observables - handle both object families.
Collection auth/permissions: polling a collection where can_read is false, or pushing to one where can_write is false, returns empty results or 403. Check the capability flags during discovery.
TAXII version skew: pointing a 2.1 client at a 2.0 discovery endpoint mismatches media types and pagination semantics. Confirm the API root advertises application/taxii+json;version=2.1.
Verification: confirm discovery returns API roots and collections, validate that fetched indicator patterns parse via the stix2 library, and re-poll with the saved timestamp to prove incremental updates return only new objects.
Prerequisites
Python 3.9+ with taxii2-client, stix2, cti-taxii-client libraries
Understanding of STIX 2.1 data model (SDOs, SCOs, SROs)
Understanding of TAXII 2.1 protocol (discovery, API roots, collections)
Network access to TAXII servers (MITRE ATT&CK TAXII, Anomali STAXX)
Optional: medallion for running a local TAXII 2.1 server
Key Concepts
TAXII 2.1 Architecture
TAXII defines a RESTful API with three service types:
Discovery: Returns information about available API roots
API Root: Contains collections and serves as the main interaction point
Collection: A logical grouping of STIX objects accessible via GET/POST
Meta Objects: Marking Definition (TLP), Language Content, Extension Definition
STIX Bundle
A Bundle is a collection of STIX objects transmitted together. Bundles have a unique ID and contain an array of objects. TAXII collections serve bundles in response to GET requests.
Workflow
Step 1: TAXII Server Discovery
from taxii2client.v21 import Server, Collection, as_pages
# Connect to MITRE ATT&CK TAXII server
server = Server("https://cti-taxii.mitre.org/taxii2/", user="", password="")
print(f"Title: {server.title}")
print(f"Description: {server.description}")
# List API rootsfor api_root in server.api_roots:
print(f"\nAPI Root: {api_root.title}")
print(f" URL: {api_root.url}")
# List collectionsfor collection in api_root.collections:
print(f" Collection: {collection.title} (ID: {collection.id})")
print(f" Can Read: {collection.can_read}")
print(f" Can Write: {collection.can_write}")
Step 2: Fetch STIX Objects from Collection
from taxii2client.v21 import Collection, as_pages
import json
# Connect to Enterprise ATT&CK collection
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: {collection.title}")
# Fetch all objects (paginated)
all_objects = []
for envelope in as_pages(collection.get_objects, per_request=50):
objects = envelope.get("objects", [])
all_objects.extend(objects)
print(f" Fetched {len(objects)} objects (total: {len(all_objects)})")
print(f"\nTotal objects retrieved: {len(all_objects)}")
# Categorize by type
type_counts = {}
for obj in all_objects:
obj_type = obj.get("type", "unknown")
type_counts[obj_type] = type_counts.get(obj_type, 0) + 1for obj_type, count insorted(type_counts.items()):
print(f" {obj_type}: {count}")
Step 3: Parse STIX 2.1 Objects with stix2 Library
from stix2 import parse, Filter, MemoryStore
# Load objects into a MemoryStore for querying
store = MemoryStore(stix_data=all_objects)
# Query for all indicators
indicators = store.query([Filter("type", "=", "indicator")])
print(f"Indicators: {len(indicators)}")
for ind in indicators[:5]:
print(f" {ind.name}: {ind.pattern}")
# Query for malware
malware_list = store.query([Filter("type", "=", "malware")])
print(f"\nMalware families: {len(malware_list)}")
# Query for threat actors
actors = store.query([Filter("type", "=", "intrusion-set")])
print(f"Threat actors: {len(actors)}")
# Find relationships for a specific objectdefget_related(store, source_id):
relationships = store.query([
Filter("type", "=", "relationship"),
Filter("source_ref", "=", source_id),
])
return relationships
# Example: Get all techniques used by 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})")
Step 4: Implement Custom TAXII Consumer
from taxii2client.v21 import Collection, as_pages
from stix2 import parse, Bundle
from datetime import datetime, timedelta
import json
classTAXIIConsumer:
"""Consume STIX/TAXII 2.1 feeds and extract IOCs."""def__init__(self, collection_url, user="", password=""):
self.collection = Collection(collection_url, user=user, password=password)
self.last_poll = Nonedefpoll_new_objects(self, added_after=None):
"""Poll for objects added after a specific timestamp."""if added_after isNone:
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
defextract_indicators(self, objects):
"""Extract actionable indicators from STIX objects."""
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
defextract_observables(self, objects):
"""Extract STIX Cyber Observables."""
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
# Usage
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"New indicators: {len(indicators)}")
