Threat Intelligence Ingestion: STIX/TAXII Feed Processor
Threat intelligence is the fuel of modern security operations: without up-to-date data on actors, campaigns, techniques, and Indicators of Compromise (IOCs), a SOC operates reactively rather than proactively. But the challenge is not the availability of intelligence - it is its systematic processing.
STIX/TAXII feeds (Structured Threat Information eXpression / Trusted Automated eXchange of Intelligence Information) are the international standard for automated sharing of threat intelligence in machine-readable format. The combination of STIX 2.1 for data structure and TAXII 2.1 for transport allows building fully automated ingestion pipelines that update SIEMs, EDRs, and firewalls with new IOCs in near real-time.
This article builds a complete threat intelligence processor: from connecting to TAXII servers, to parsing STIX 2.1 objects, to IOC enrichment, to correlation with SIEM alerts, and automatic routing to appropriate security controls.
What You Will Learn
- STIX 2.1 object structure and TI modeling patterns
- Connecting to and polling TAXII 2.1 servers with Python taxii2-client
- Parsing and normalizing IOCs from STIX bundles
- IOC scoring and deduplication
- Automatic routing to SIEM, EDR, and firewall
- Real-time IOC-alert correlation and automatic alert enrichment generation
STIX 2.1: The Data Model
STIX 2.1 defines a vocabulary of objects to represent practically every aspect of threat intelligence. Understanding the structure of the main objects is fundamental before building the processor.
| STIX Object | Purpose | Key Fields |
|---|---|---|
indicator |
IOC with detection pattern | pattern, valid_from, valid_until, labels |
malware |
Malware family details | name, malware_types, aliases, capabilities |
threat-actor |
Threat actor profile | name, aliases, threat_actor_types, goals |
attack-pattern |
Attack technique (ATT&CK) | name, external_references (ATT&CK ID) |
campaign |
Attack campaign | name, aliases, first_seen, last_seen |
relationship |
Connections between objects | relationship_type, source_ref, target_ref |
# STIX Pattern Language examples
# MD5 file hash
[file:hashes.'MD5' = 'd41d8cd98f00b204e9800998ecf8427e']
# Malicious URL
[url:value = 'http://malicious-c2.com/beacon']
# IP with port
[network-traffic:dst_ref.type = 'ipv4-addr'
AND network-traffic:dst_ref.value = '192.168.1.1'
AND network-traffic:dst_port = 443]
# Domain
[domain-name:value = 'evil-c2.com']Connecting to TAXII 2.1 Servers
TAXII 2.1 defines a REST API for discovery and polling of threat intelligence collections.
The Python taxii2-client library provides a complete client that handles authentication,
pagination, and error management.
# TAXII 2.1 Client
from taxii2client.v21 import Server
from taxii2client.common import TokenAuth, BasicAuth
from datetime import datetime, timedelta
from typing import Iterator
import logging
class TaxiiIngestor:
def __init__(self, configs: list):
self.configs = configs
self.logger = logging.getLogger(__name__)
def poll_collection(
self,
config,
collection_id: str,
added_after: datetime = None,
limit: int = 1000
) -> Iterator[dict]:
"""Polls a TAXII collection and yields STIX bundles."""
if added_after is None:
added_after = datetime.utcnow() - timedelta(hours=24)
auth = TokenAuth(key=config.auth_params.get('token'))
server = Server(config.url, auth=auth)
for api_root in server.api_roots:
for collection in api_root.collections:
if collection.id != collection_id:
continue
next_page = None
total_fetched = 0
while True:
try:
kwargs = {
'added_after': added_after.isoformat() + 'Z',
'limit': min(limit, 500)
}
if next_page:
kwargs['next'] = next_page
response = collection.get_objects(**kwargs)
objects = response.get('objects', [])
total_fetched += len(objects)
if objects:
yield response
next_page = response.get('next')
if not next_page or total_fetched >= limit:
break
except Exception as e:
self.logger.error(f"Polling error: {e}")
breakParsing and Normalizing STIX IOCs
STIX objects must be transformed into normalized structures that security systems (SIEM, EDR, firewall) can consume directly. Parsing indicators requires interpreting the STIX Pattern Language.
# STIX IOC Parser
import re
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class NormalizedIOC:
raw_id: str
ioc_type: str # 'ip', 'domain', 'url', 'hash_md5', 'hash_sha256'
value: str
source: str
confidence: int # 0-100
severity: str # 'low', 'medium', 'high', 'critical'
valid_from: datetime
valid_until: Optional[datetime]
tags: list[str]
related_malware: list[str]
related_threat_actors: list[str]
mitre_techniques: list[str]
class STIXParser:
PATTERN_EXTRACTORS = {
'ip': re.compile(
r"network-traffic:(?:dst|src)_ref\.value\s*=\s*'([0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3})'"
),
'domain': re.compile(
r"domain-name:value\s*=\s*'([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})'"
),
'url': re.compile(r"url:value\s*=\s*'(https?://[^']+)'"),
'hash_md5': re.compile(r"file:hashes\.'MD5'\s*=\s*'([a-fA-F0-9]{32})'"),
'hash_sha256': re.compile(r"file:hashes\.'SHA-256'\s*=\s*'([a-fA-F0-9]{64})'"),
}
def parse_bundle(self, stix_bundle: dict) -> list[NormalizedIOC]:
objects = stix_bundle.get('objects', [])
obj_index = {obj['id']: obj for obj in objects}
relationships = self._index_relationships(objects)
iocs = []
for obj in objects:
if obj.get('type') == 'indicator':
try:
ioc = self._parse_indicator(obj, obj_index, relationships)
if ioc:
iocs.append(ioc)
except Exception as e:
pass # Log warning in production
return iocs
def _extract_ioc_from_pattern(self, pattern: str) -> tuple[str, str]:
clean = pattern.strip('[]').strip()
for ioc_type, regex in self.PATTERN_EXTRACTORS.items():
match = regex.search(clean)
if match:
return ioc_type, match.group(1)
return None, None
def _index_relationships(self, objects: list) -> dict:
rels = {}
for obj in objects:
if obj.get('type') == 'relationship':
src = obj.get('source_ref', '')
rels.setdefault(src, []).append(obj)
return relsDeduplication and IOC Scoring
Multiple feeds may contain the same IOC with different confidence levels. The deduplication system must intelligently aggregate information, increasing confidence when multiple independent sources report the same indicator.
# IOC Deduplication
from collections import defaultdict
class IOCDeduplicator:
def __init__(self, min_confidence: int = 30):
self.min_confidence = min_confidence
def deduplicate_and_score(self, iocs: list[NormalizedIOC]) -> list[NormalizedIOC]:
"""
Deduplicates IOCs and increases confidence when
multiple sources report the same value.
"""
grouped = defaultdict(list)
for ioc in iocs:
grouped[(ioc.ioc_type, ioc.value.lower())].append(ioc)
result = []
for (ioc_type, value), group in grouped.items():
ioc = group[0] if len(group) == 1 else self._merge_iocs(group)
if ioc.confidence >= self.min_confidence:
result.append(ioc)
return sorted(result, key=lambda x: x.confidence, reverse=True)
def _merge_iocs(self, iocs: list[NormalizedIOC]) -> NormalizedIOC:
"""Merges duplicate IOCs from different sources."""
base = max(iocs, key=lambda x: x.valid_from)
base_confidence = max(i.confidence for i in iocs)
# Confidence increases with number of sources (diminishing returns)
source_boost = min(len(iocs) - 1, 3) * 5 # Max +15
merged_confidence = min(base_confidence + source_boost, 100)
return NormalizedIOC(
raw_id=base.raw_id,
ioc_type=base.ioc_type,
value=base.value,
source=",".join(set(i.source for i in iocs)),
confidence=merged_confidence,
severity=base.severity,
valid_from=base.valid_from,
valid_until=base.valid_until,
tags=list(set(t for i in iocs for t in i.tags)),
related_malware=list(set(m for i in iocs for m in i.related_malware)),
related_threat_actors=list(set(a for i in iocs for a in i.related_threat_actors)),
mitre_techniques=list(set(t for i in iocs for t in i.mitre_techniques))
)Routing to Security Controls
A complete TI processor does not stop at parsing: it distributes normalized IOCs to the systems that can actually enact defense. Routing is based on IOC type and confidence.
# IOC Router to SIEM, EDR, Firewall
import httpx
class IOCRouter:
CONFIDENCE_THRESHOLDS = {
'firewall_block': 80, # Block in firewall only at high confidence
'edr_detection': 60, # EDR alert from medium+ confidence
'siem_enrichment': 30, # SIEM enrichment for all IOCs
}
def route(self, ioc: NormalizedIOC) -> list[dict]:
"""Distributes an IOC to appropriate systems."""
results = []
# All IOCs go to SIEM for enrichment
if ioc.confidence >= self.CONFIDENCE_THRESHOLDS['siem_enrichment']:
results.append({'destination': 'siem', **self._push_to_siem(ioc)})
# High confidence network IOCs go to firewall
if (ioc.ioc_type in ('ip', 'domain', 'url') and
ioc.confidence >= self.CONFIDENCE_THRESHOLDS['firewall_block']):
results.append({'destination': 'firewall', **self._push_to_firewall(ioc)})
# Hashes and IPs go to EDR
if (ioc.ioc_type in ('hash_md5', 'hash_sha256', 'ip') and
ioc.confidence >= self.CONFIDENCE_THRESHOLDS['edr_detection']):
results.append({'destination': 'edr', **self._push_to_edr(ioc)})
return resultsReal-Time Alert Correlation
The true value of an operational TI pipeline is real-time correlation: when a new IOC is ingested, the system immediately checks if recent alerts contain that value, automatically enriching them with threat context.
# Real-time IOC-Alert correlation
async def _correlate_with_recent_alerts(
self,
iocs: list[NormalizedIOC],
lookback_hours: int = 48
) -> list[dict]:
"""Searches recent alerts matching new IOCs."""
correlations = []
recent_alerts = await self.siem.search_recent_alerts(
hours_back=lookback_hours,
limit=10000
)
# Build IOC index for efficient lookup
ioc_index = {(ioc.ioc_type, ioc.value.lower()): ioc for ioc in iocs}
for alert in recent_alerts:
fields_to_check = [
('ip', alert.get('src_ip', '')),
('ip', alert.get('dst_ip', '')),
('domain', alert.get('dns_query', '')),
('hash_md5', alert.get('file_hash_md5', '')),
('hash_sha256', alert.get('file_hash_sha256', '')),
]
for ioc_type, value in fields_to_check:
if not value:
continue
ioc = ioc_index.get((ioc_type, value.lower()))
if ioc:
correlation = {
'alert_id': alert.get('id'),
'ioc_type': ioc_type,
'ioc_value': value,
'ioc_confidence': ioc.confidence,
'related_malware': ioc.related_malware,
'related_actors': ioc.related_threat_actors
}
correlations.append(correlation)
await self.siem.enrich_alert(alert.get('id'), correlation)
return correlationsRecommended Free TAXII Feeds
- MITRE ATT&CK TAXII:
https://cti-taxii.mitre.org/taxii/- Groups and software - Anomali LIMO: Free tier with community-curated IOCs
- AlienVault OTX: Community-driven threat intelligence feed
- MISP: Self-hosted with community feeds
- AbuseIPDB: Malicious IP feeds (REST API)
Feed Quality Warning
Not all TAXII feeds have the same quality. Feeds with low confidence and high false positive rate (e.g., generic IPs, CDNs, Tor exit nodes) can block legitimate traffic if inserted into the firewall without validation. Always implement a whitelisting system (e.g., known cloud provider CIDRs, organization IPs) before blocking.
Conclusions and Key Takeaways
A mature STIX/TAXII processor transforms threat intelligence from a human resource (analysts reading reports) to an automated operational resource (IOCs that automatically propagate to security controls within minutes of publication).
Related deep dive: 04 - SOAR Playbook in Python: Automazione Incident Response.
Related deep dive: 06 - Anomaly Detection Comportamentale: ML su Log Data.
Key Takeaways
- STIX 2.1 is the standard language for modeling TI; TAXII 2.1 is the transport protocol
- The STIX parser must handle relationships between objects to enrich IOCs with malware/actor context
- Multi-source deduplication increases confidence proportionally to concordant sources
- Routing must be based on IOC type and confidence (do not block everything in the firewall)
- Real-time IOC-alert correlation is the true added value: turns passive IOCs into active alerts
- Always implement whitelisting before automatic blocking to prevent self-inflicted outages
Related Articles
- AI-Assisted Detection: LLMs for Sigma Rule Generation
- SOAR Playbook in Python: Incident Response Automation
- MITRE ATT&CK Integration: Mapping Coverage Gaps
Frequently asked questions
What is the purpose of using STIX/TAXII feeds for threat intelligence?
The purpose of using STIX/TAXII feeds is to enable automated sharing and processing of threat intelligence in machine-readable format, allowing security systems to stay up-to-date with new Indicators of Compromise (IOCs) in near real-time.
What is the difference between STIX 2.1 and TAXII 2.1?
STIX 2.1 defines a vocabulary of objects for representing threat intelligence data, while TAXII 2.1 provides a REST API for discovery and polling of threat intelligence collections.
How does the deduplication system handle multiple feeds with the same IOC but different confidence levels?
The deduplication system intelligently aggregates information from multiple independent sources, increasing confidence when multiple sources report the same indicator.
Why is it necessary to implement a whitelisting system before blocking IOCs from unknown TAXII feeds?
It is necessary to implement a whitelisting system to prevent legitimate traffic from being blocked due to low-confidence or high-false-positive-rate indicators, such as generic IPs or CDNs.
Commenti
Caricamento commenti...
Sign in to comment