What it does. Implements a STIX 2.1/TAXII 2.1 threat-intelligence feed consumer and Part of mukul975/Anthropic-Cybersecurity-Skills (817 security skills) (mukul975/Anthropic-Cybersecurity-Skills).
Install
npx skills add mukul975/Anthropic-Cybersecurity-Skills --skill implementing-stix-taxii-feed-integration, or copy the skill folder into ~/.claude/skills/implementing-stix-taxii-feed-integration/.- Raw file:
curl -sL https://raw.githubusercontent.com/mukul975/Anthropic-Cybersecurity-Skills/HEAD/skills/implementing-stix-taxii-feed-integration/SKILL.md
SKILL.md (verbatim)
name: implementing-stix-taxii-feed-integration
description: Implements a STIX 2.1/TAXII 2.1 threat-intelligence feed consumer and
producer in Python, covering TAXII server discovery, collection polling, parsing
STIX bundles with the stix2 library, and standing up a local TAXII server with Medallion.
Use when integrating a STIX/TAXII CTI feed into a SIEM or TIP, writing a TAXII client
to poll for new indicators, or setting up TAXII collections for indicator exchange.
domain: cybersecurity
subdomain: threat-intelligence
tags:
- threat-intelligence
- cti
- ioc
- mitre-attack
- stix
- taxii
- feed-integration
- oasis
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- ID.RA-01
- ID.RA-05
- DE.CM-01
- DE.AE-02
mitre_attack:
- T1591
- T1592
- T1593
- T1589
Implementing STIX/TAXII Feed Integration
Overview
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
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
STIX 2.1 Object Model
STIX objects are categorized as:
- SDOs (STIX Domain Objects): Indicator, Malware, Threat Actor, Campaign, Attack Pattern, Tool, Infrastructure, Vulnerability, Identity, Location, Note, Opinion, Report, Grouping
- SCOs (STIX Cyber Observables): IPv4-Addr, Domain-Name, URL, File, Email-Addr, Process, Network-Traffic, Artifact
- SROs (STIX Relationship Objects): Relationship, Sighting
- 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 roots
for api_root in server.api_roots:
print(f"\nAPI Root: {api_root.title}")
print(f" URL: {api_root.url}")
# List collections
for 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) + 1
for obj_type, count in sorted(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 object
def get_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
class TAXIIConsumer:
"""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 = None
def poll_new_objects(self, added_after=None):
"""Poll for objects added after a specific timestamp."""
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):
"""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
def extract_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)}")
Step 5: Set Up Local TAXII Server with Medallion
# medallion configuration (medallion.conf)
TAXII_CONFIG = {
"backend": {
"module_class": "MemoryBackend",
},
"users": {
"admin": "admin_password",
"readonly": "readonly_password",
},
"taxii": {
"max_content_length": 10485760,
},
}
# Run medallion server:
# pip install medallion
# python -m medallion --config medallion.conf --port 5000
# Add objects to local TAXII server
import requests
def push_to_taxii(server_url, collection_id, stix_bundle, user, password):
"""Push STIX bundle to a TAXII 2.1 collection."""
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()
Validation Criteria
- TAXII server discovery returns valid API roots and collections
- STIX objects fetched and parsed correctly from TAXII collections
- Indicators extracted with valid STIX patterns
- Pagination handled correctly for large collections
- Consumer tracks polling state for incremental updates
- Local TAXII server accepts and serves STIX bundles
References
Other files in this skill
assets/template.md (verbatim)
STIX/TAXII Feed Integration Report Template
Feed Configuration
| Field |
Value |
| TAXII Server |
|
| API Root |
|
| Collection ID |
|
| Collection Name |
|
| Last Poll Timestamp |
|
| Authentication |
API Key / Basic / Certificate |
Discovery Summary
| Metric |
Count |
| API Roots |
|
| Collections (Read) |
|
| Collections (Write) |
|
Poll Results
| Metric |
Count |
| Total Objects Fetched |
|
| New Since Last Poll |
|
| Indicators |
|
| Malware |
|
| Threat Actors |
|
| Attack Patterns |
|
| Relationships |
|
| Observables |
|
Network Indicators
| Type |
Value |
Confidence |
Valid From |
Valid Until |
| IP |
|
|
|
|
| Domain |
|
|
|
|
| URL |
|
|
|
|
File Indicators
| Hash Type |
Value |
Confidence |
Associated Malware |
| SHA-256 |
|
|
|
| MD5 |
|
|
|
Integration Status
| Downstream System |
Status |
Records Pushed |
| SIEM (Splunk/Elastic) |
Success/Failed |
|
| MISP |
Success/Failed |
|
| OpenCTI |
Success/Failed |
|
| Firewall Blocklist |
Success/Failed |
|
references/api-reference.md (verbatim)
API Reference: STIX/TAXII Threat Intelligence Feed Integration
Libraries Used
| Library |
Purpose |
taxii2-client |
TAXII 2.0/2.1 client for fetching CTI collections |
stix2 |
Parse and create STIX 2.1 objects (indicators, malware, etc.) |
requests |
HTTP fallback for custom TAXII endpoints |
json |
Serialize and filter STIX bundles |
Installation
pip install taxii2-client stix2 requests
Authentication
TAXII Server with HTTP Basic Auth
from taxii2client.v21 import Server, Collection
import os
TAXII_URL = os.environ["TAXII_URL"] # e.g., "https://cti-taxii.mitre.org/taxii2/"
server = Server(
TAXII_URL,
user=os.environ.get("TAXII_USER"),
password=os.environ.get("TAXII_PASS"),
)
TAXII Server with API Key
from taxii2client.v21 import Server as Server21
server = Server21(
url=TAXII_URL,
headers={"Authorization": f"Bearer {os.environ['TAXII_TOKEN']}"},
)
TAXII 2.1 Endpoints
| Endpoint |
Description |
GET /taxii2/ |
Server discovery — returns API roots |
GET /{api-root}/ |
API root information |
GET /{api-root}/collections/ |
List available collections |
GET /{api-root}/collections/{id}/ |
Get collection details |
GET /{api-root}/collections/{id}/objects/ |
Get STIX objects from collection |
GET /{api-root}/collections/{id}/manifest/ |
Object manifest (metadata only) |
POST /{api-root}/collections/{id}/objects/ |
Add objects to a collection |
GET /{api-root}/status/{id}/ |
Check status of a POST operation |
Core Operations
Discover 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} ({collection.id})")
print(f" Can read: {collection.can_read}, Can write: {collection.can_write}")
Fetch STIX Objects from a Collection
from taxii2client.v21 import Collection
collection = Collection(
f"{TAXII_URL}collections/{collection_id}/",
user=os.environ.get("TAXII_USER"),
password=os.environ.get("TAXII_PASS"),
)
# Get all objects
stix_bundle = collection.get_objects()
# Filter by STIX type
indicators = collection.get_objects(type=["indicator"])
# Filter by time range
from datetime import datetime
recent = collection.get_objects(
added_after=datetime(2025, 1, 1).strftime("%Y-%m-%dT%H:%M:%SZ")
)
Parse STIX Objects
import stix2
bundle = stix2.parse(stix_bundle, allow_custom=True)
for obj in bundle.objects:
if obj.type == "indicator":
print(f"Indicator: {obj.name}")
print(f" Pattern: {obj.pattern}")
print(f" Valid: {obj.valid_from} — {getattr(obj, 'valid_until', 'N/A')}")
elif obj.type == "malware":
print(f"Malware: {obj.name} — {obj.malware_types}")
elif obj.type == "attack-pattern":
print(f"TTP: {obj.name}")
import re
def extract_iocs(stix_objects):
iocs = {"ipv4": [], "domain": [], "url": [], "sha256": [], "md5": []}
for obj in stix_objects:
if obj.get("type") != "indicator":
continue
pattern = obj.get("pattern", "")
# IPv4
for ip in re.findall(r"ipv4-addr:value\s*=\s*'([^']+)'", pattern):
iocs["ipv4"].append(ip)
# Domain
for domain in re.findall(r"domain-name:value\s*=\s*'([^']+)'", pattern):
iocs["domain"].append(domain)
# SHA-256
for sha in re.findall(r"file:hashes\.'SHA-256'\s*=\s*'([^']+)'", pattern):
iocs["sha256"].append(sha)
return iocs
Create and Push STIX Objects
indicator = stix2.Indicator(
name="Malicious IP",
pattern="[ipv4-addr:value = '198.51.100.42']",
pattern_type="stix",
valid_from=datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
labels=["malicious-activity"],
)
bundle = stix2.Bundle(objects=[indicator])
collection.add_objects(bundle.serialize())
Public TAXII Feeds
| Provider |
URL |
Content |
| MITRE ATT&CK |
https://cti-taxii.mitre.org/taxii2/ |
ATT&CK Enterprise, Mobile, ICS |
| AlienVault OTX |
OTX API + STIX export |
Community threat intel |
| Anomali STAXX |
STAXX TAXII endpoint |
Curated threat feeds |
{
"type": "bundle",
"id": "bundle--a1b2c3d4",
"objects": [
{
"type": "indicator",
"id": "indicator--e5f6a7b8",
"created": "2025-01-15T10:30:00Z",
"name": "Malicious C2 IP",
"pattern": "[ipv4-addr:value = '198.51.100.42']",
"pattern_type": "stix",
"valid_from": "2025-01-15T10:30:00Z",
"labels": ["malicious-activity"]
}
]
}
references/standards.md (verbatim)
Standards and Frameworks Reference
STIX 2.1 Standard (OASIS)
Core Concepts
- STIX Bundle: Top-level container for STIX objects (type: "bundle")
- STIX ID Format:
type--uuid (e.g., indicator--a1b2c3d4-...)
- Versioning: Objects use
modified timestamp for version tracking
- Confidence: 0-100 scale for reliability assessment
STIX Domain Objects (SDOs)
| Type |
Purpose |
Key Properties |
| attack-pattern |
ATT&CK technique |
name, kill_chain_phases |
| campaign |
Related intrusion activity |
name, first_seen, objective |
| identity |
Individuals/orgs |
name, identity_class, sectors |
| indicator |
Detection pattern |
pattern, pattern_type, valid_from |
| infrastructure |
Adversary systems |
name, infrastructure_types |
| intrusion-set |
Threat group |
name, aliases, goals |
| malware |
Malware family |
name, malware_types, is_family |
| note |
Analyst annotation |
content, object_refs |
| report |
CTI document |
name, published, object_refs |
| threat-actor |
Human adversary |
name, threat_actor_types, roles |
| tool |
Legitimate software |
name, tool_types |
| vulnerability |
CVE/weakness |
name, external_references |
STIX Patterning Language
[file:hashes.'SHA-256' = 'abc...']
[ipv4-addr:value = '198.51.100.1']
[domain-name:value = 'malware.example.com']
[network-traffic:dst_ref.type = 'ipv4-addr' AND network-traffic:dst_port = 443]
[process:name = 'cmd.exe' AND process:command_line MATCHES '.*powershell.*']
TAXII 2.1 Standard (OASIS)
Endpoints
| Endpoint |
Method |
Purpose |
| /taxii2/ |
GET |
Server discovery |
| /{api-root}/ |
GET |
API root information |
| /{api-root}/collections/ |
GET |
List collections |
| /{api-root}/collections/{id}/ |
GET |
Collection details |
| /{api-root}/collections/{id}/objects/ |
GET/POST |
Get/add objects |
| /{api-root}/collections/{id}/manifest/ |
GET |
Object manifest |
| /{api-root}/status/{id}/ |
GET |
Status of add operation |
- Content-Type:
application/stix+json;version=2.1
- Accept:
application/taxii+json;version=2.1
limit: Maximum number of objects per responsenext: Cursor for next pageadded_after: Filter objects by timestamp
Marking Definitions (TLP)
{"definition_type": "tlp", "definition": {"tlp": "clear"}}
{"definition_type": "tlp", "definition": {"tlp": "green"}}
{"definition_type": "tlp", "definition": {"tlp": "amber"}}
{"definition_type": "tlp", "definition": {"tlp": "amber+strict"}}
{"definition_type": "tlp", "definition": {"tlp": "red"}}
References
references/workflows.md (verbatim)
STIX/TAXII Feed Integration Workflows
Workflow 1: TAXII Feed Consumption
[TAXII Discovery] --> [API Root Enumeration] --> [Collection Selection] --> [Object Polling]
|
v
[STIX Parsing] --> [IOC Extraction]
|
v
[SIEM/TIP Ingestion]
Steps:
- Discovery: Query TAXII server discovery endpoint for available API roots
- Root Enumeration: List available API roots and their supported features
- Collection Listing: Enumerate collections with read/write permissions
- Incremental Polling: Fetch new objects using added_after timestamp filter
- STIX Parsing: Deserialize JSON into typed STIX objects
- IOC Extraction: Extract indicators, observables, and relationships
- Platform Ingestion: Push to SIEM, MISP, or OpenCTI
Workflow 2: STIX Bundle Production
[IOC Sources] --> [Normalization] --> [STIX Object Creation] --> [Bundle Assembly]
|
v
[TAXII Publication]
Steps:
- Source Collection: Gather IOCs from internal analysis, feeds, incident response
- Normalization: Standardize IOC formats and remove duplicates
- Object Creation: Create STIX Indicators, Observables, and Relationships
- TLP Marking: Apply appropriate TLP marking definitions
- Bundle Assembly: Package objects into STIX 2.1 bundles
- TAXII Push: POST bundles to writable TAXII collections
Workflow 3: Multi-Feed Aggregation
[TAXII Feed A] --+
|--> [Deduplication] --> [Correlation] --> [Unified Store]
[TAXII Feed B] --+ |
| v
[STIX File C] ---+ [Dashboard/Alerts]
Steps:
- Feed Registration: Configure multiple TAXII and file-based STIX sources
- Parallel Polling: Poll all feeds concurrently with rate limiting
- Deduplication: Remove duplicate objects by STIX ID and modified timestamp
- Correlation: Link related objects across feeds via relationships
- Unified Storage: Store in MemoryStore, FileSystemStore, or database-backed store
- Output: Generate alerts, dashboards, or exports for downstream consumers
Back to mukul975/Anthropic-Cybersecurity-Skills (817 security skills) or Agent skills.