| name | processing-stix-taxii-feeds |
| description | Processes STIX 2.1 threat intelligence bundles delivered via TAXII 2.1 servers, normalizing objects into platform-native schemas and routing them to appropriate consuming systems. Use when onboarding new TAXII collection endpoints, automating bi-directional intelligence sharing with ISACs, or building pipeline validation for malformed STIX bundles. Activates for requests involving OASIS STIX, TAXII server configuration, MISP TAXII, or Cortex XSOAR feed integrations.
|
| domain | cybersecurity |
| subdomain | threat-intelligence |
| tags | ["STIX-2.1","TAXII-2.1","OASIS","MISP","CTI","IOC","threat-intelligence","NIST-SP-800-150"] |
| version | 1.0.0 |
| author | mahipal |
| license | Apache-2.0 |
Processing STIX/TAXII Feeds
When to Use
Use this skill when:
- Onboarding a new TAXII 2.1 collection from a government feed (CISA AIS, FS-ISAC) or commercial provider
- Validating that ingested STIX bundles conform to the OASIS STIX 2.1 specification before import
- Building automated pipelines that parse STIX relationship objects to reconstruct campaign context
Do not use this skill for proprietary vendor feed formats (Recorded Future JSON, CrowdStrike IOC lists) that require vendor-specific parsers rather than STIX processing.
Prerequisites
- Python 3.9+ with
stix2 library (pip install stix2) and taxii2-client library
- Network access to TAXII 2.1 server endpoint with valid credentials
- Target TIP or SIEM with import API (MISP, OpenCTI, or Splunk ES)
Workflow
Step 1: Discover TAXII Server Collections
from taxii2client.v21 import Server, as_pages
server = Server("https://cti.example.com/taxii/",
user="apiuser", password="apikey")
api_root = server.api_roots[0]
for collection in api_root.collections:
print(collection.id, collection.title, collection.can_read)
Select collections relevant to your threat profile. CISA AIS provides collections segmented by sector (financial, energy, healthcare).
Step 2: Fetch STIX Bundles with Pagination
from taxii2client.v21 import Collection
from datetime import datetime, timedelta, timezone
collection = Collection(
"https://cti.example.com/taxii/api1/collections/<id>/objects/",
user="apiuser", password="apikey")
added_after = datetime.now(timezone.utc) - timedelta(hours=24)
for bundle_page in as_pages(collection.get_objects,
added_after=added_after, per_request=):
process_bundle(bundle_page)
