/building-threat-intelligence-feed-integration

name: building-threat-intelligence-feed-integration
description: 'Builds automated threat intelligence feed integration pipelines connecting STIX/TAXII feeds, open-source threat
  intel, and commercial TI platforms into SIEM and security tools for real-time IOC matching and alerting. Use when SOC teams
  need to operationalize threat intelligence by automating feed ingestion, normalization, scoring, and distribution to detection
  systems.

  '
domain: cybersecurity
subdomain: soc-operations
tags:
- soc
- threat-intelligence
- stix
- taxii
- misp
- feeds
- ioc
- siem-integration
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- DE.CM-01
- DE.AE-02
- RS.MA-01
- DE.AE-06

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

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

Convert all feeds to STIX 2.1 format for standardization:

from stix2 import Indicator, Bundle
import hashlib

def create_stix_indicator(ioc_value, ioc_type, source, confidence=50):
    """Convert raw IOC to STIX 2.1 indicator"""
    pattern_map = {
        "ipv4": f"[ipv4-addr:value = '{ioc_value}']",
        "domain": f"[domain-name:value = '{ioc_value}']",
        "url": f"[url:value = '{ioc_value}']",
        "sha256": f"[file:hashes.'SHA-256' = '{ioc_value}']",
        "md5": f"[file:hashes.MD5 = '{ioc_value}']",
    }

    return Indicator(
        name=f"{ioc_type}: {ioc_value}",
        pattern=pattern_map[ioc_type],
        pattern_type="stix",
        valid_from="2024-03-15T00:00:00Z",
        confidence=confidence,
        labels=[source],
        custom_properties={"x_source_feed": source