Use this skill when:
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.
stix2 library (pip install stix2) and taxii2-client libraryfrom 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).
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")
# Fetch only objects added in the last 24 hours
added_after = datetime.now(timezone.utc) - timedelta(hours=24)
for bundle_page in as_pages(collection.get_objects,
added_after=added_after, per_request=100):
process_bundle(bundle_page)
import stix2
def process_bundle(bundle_dict):
bundle = stix2.parse(bundle_dict, allow_custom=True)
for obj in bundle.objects:
if obj.type == "indicator":
validate_indicator(obj)
elif obj.type == "threat-actor":
upsert_threat_actor(obj)
elif obj.type == "relationship":
link_objects(obj)
def validate_indicator(indicator):
required = ["id", "type", "spec_version", "created",
"modified", "pattern", "pattern_type", "valid_from"]
for field in required:
if not hasattr(indicator, field):
raise ValueError(f"Missing required field: {field}")
# Check confidence range
if hasattr(indicator, "confidence"):
assert 0 <= indicator.confidence <= 100
Map STIX object types to destination systems:
indicator objects → SIEM lookup tables and firewall blocklistsmalware objects → EDR threat intelligence librarythreat-actor / campaign objects → TIP for analyst contextcourse-of-action objects → Security team wiki or SOAR playbook triggersUse TLP marking definitions to enforce sharing restrictions:
for marking in obj.get("object_marking_refs", []):
if "tlp-red" in marking:
route_to_restricted_platform_only(obj)
# Add validated local intelligence back to shared collection
new_indicator = stix2.Indicator(
name="Malicious C2 Domain",
pattern="[domain-name:value = 'evil-c2.example.com']",
pattern_type="stix",
valid_from="2025-01-15T00:00:00Z",
confidence=80,
labels=["malicious-activity"],
object_marking_refs=["marking-definition--34098fce-860f-479c-ae..."] # TLP:GREEN
)
collection.add_objects(stix2.Bundle(new_indicator))
| Term | Definition |
|---|---|
| STIX Bundle | Top-level STIX container object (type: "bundle") holding any number of STIX Domain Objects (SDOs) and STIX Relationship Objects (SROs) |
| SDO | STIX Domain Object — core intelligence types: indicator, threat-actor, malware, campaign, attack-pattern, course-of-action |
| SRO | STIX Relationship Object — links two SDOs with a labeled relationship (e.g., "uses", "attributed-to", "indicates") |
| Pattern Language | STIX pattern syntax for indicator conditions: [network-traffic:dst_port = 443 AND ipv4-addr:value = '10.0.0.1'] |
| Marking Definition | STIX object encoding TLP or statement restrictions on intelligence sharing |
| added_after | TAXII 2.1 filter parameter (RFC 3339 timestamp) for incremental polling of new objects |
spec_version field: STIX 2.0 and 2.1 have incompatible schemas (2.1 adds confidence, object_marking_refs at bundle level). Always check spec_version before parsing.next link header) causes silent data loss.added_after: Server and client time misalignment causes missed objects at interval boundaries. Use UTC exclusively and add 5-minute overlap windows.