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:

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("---")

Step 3: Ingest Open-Source Feeds

Abuse.ch URLhaus Feed:

import requests
import csv
from io import StringIO

# Download URLhaus recent URLs
response = requests.get("https://urlhaus.abuse.ch/downloads/csv_recent/")
reader = csv.reader(StringIO(response.text), delimiter=',')

indicators = []
for row in reader:
    if row[0].startswith("#"):
        continue
    indicators.append({
        "id": row[0],
        "dateadded": row[1],
        "url": row[2],
        "url_status": row[3],
        "threat": row[5],
        "tags": row[6]
    })

print(f"Ingested {len(indicators)} URLs from URLhaus")

# Filter for active threats only
active = [i for i in indicators if i["url_status"] == "online"]
print(f"Active threats: {len(active)}")

AlienVault OTX Pulse Feed:

from OTXv2 import OTXv2, IndicatorTypes

otx = OTXv2("YOUR_OTX_API_KEY")

# Get subscribed pulses (last 24 hours)
pulses = otx.getall(modified_since="2024-03-14T00:00:00")

for pulse in pulses:
    print(f"Pulse: {pulse['name']}")
    print(f"Tags: {pulse['tags']}")
    for indicator in pulse["indicators"]:
        print(f"  IOC: {indicator['indicator']} ({indicator['type']})")

Abuse.ch Feodo Tracker (C2 IPs):

response = requests.get("https://feodotracker.abuse.ch/downloads/ipblocklist_recommended.json")
c2_data = response.json()

for entry in c2_data:
    print(f"IP: {entry['ip_address']}:{entry['port']}")
    print(f"Malware: {entry['malware']}")
    print(f"First Seen: {entry['first_seen']}")
    print(f"Last Online: {entry['last_online']}")

Step 4: Normalize and Deduplicate

…