building-threat-intelligence-feed-integration

Building Threat Intelligence Feed Integration

When to Use

Use this skill when:

• SOC teams need automated ingestion of threat intelligence feeds into SIEM platforms
• Multiple TI sources require normalization into a common format (STIX 2.1)
• Detection systems need real-time IOC matching against network and endpoint telemetry
• TI feed quality assessment and deduplication processes need to be established

Do not use for manual IOC lookup — use dedicated enrichment tools (VirusTotal, AbuseIPDB) for ad-hoc queries.

Prerequisites

• MISP instance or Threat Intelligence Platform (TIP) for feed aggregation
• STIX/TAXII client library (taxii2-client, stix2 Python packages)
• SIEM platform (Splunk ES, Elastic Security, or Sentinel) with TI framework configured
• API keys for commercial and open-source feeds (AlienVault OTX, Abuse.ch, CISA AIS)
• Python 3.8+ for feed processing automation

Workflow

Step 1: Identify and Catalog Intelligence Sources

Map available feeds by type, format, and update frequency:

Feed Source	Format	IOC Types	Update Freq	Cost
AlienVault OTX	STIX/JSON	IP, Domain, Hash, URL	Real-time	Free
Abuse.ch URLhaus	CSV/JSON	URL, Domain	Every 5 min	Free
Abuse.ch MalwareBazaar	JSON API	File Hash	Real-time	Free
CISA AIS	STIX/TAXII 2.1	All types	Daily	Free (US Gov)
CrowdStrike Intel	STIX/JSON	All types + Actor TTP	Real-time	Commercial
Mandiant Advantage	STIX 2.1	All types + Reports	Real-time	Commercial

Step 2: Ingest STIX/TAXII Feeds

Connect to a TAXII 2.1 server and download indicators:

from taxii2client.v21 import Server, Collection
from stix2 import parse

# Connect to TAXII server (example: CISA AIS)
server = Server(
    "https://taxii.cisa.gov/taxii2/",
    user="your_username",
    password="your_password"
)

# List available 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} (ID: {collection.id})")

# Fetch indicators from a collection
collection = Collection(
    "https://taxii.cisa.gov/taxii2/collections/COLLECTION_ID/",
    user="your_username",
    password="your_password"
)

# Get indicators added in last 24 hours
from datetime import datetime, timedelta
added_after = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S.000Z")

response = collection.get_objects(added_after=added_after, type=["indicator"])
for obj in response.get("objects", []):
    indicator = parse(obj)
    print(f"Type: {indicator.type}")
    print(f"Pattern: {indicator.pattern}")
    print(f"Valid Until: {indicator.valid_until}")
    print(f"Confidence: {indicator.confidence}")
    print("---")