DEV Community

David Moya
David Moya

Posted on

How I Aggregated 13 Public CTI Feeds into a Free Threat Intelligence Platform

#cybersecurity #python #opensource #webdev

TL;DR: I built a free threat intelligence aggregation platform by pulling from 13 public CTI feeds (MISP, OTX, Abuse.ch, etc.), normalizing the data into a common schema, and exposing it through a REST API. This post walks through the architecture, the normalization pipeline, deduplication strategy, and a few gotchas I hit along the way.

Why Build Another Threat Intel Tool?

Commercial threat intelligence platforms are expensive. Recorded Future, ThreatConnect, and Anomali are excellent products, but their price tags put them out of reach for small security teams, indie researchers, and hobbyist defenders. At the same time, there are dozens of high-quality public feeds sitting out there—completely free—that most people never integrate properly because aggregating them is genuinely annoying.

The pain points are real:

  • Every feed has a different format (STIX, CSV, JSON, plain text)
  • Update frequencies vary wildly (every 5 minutes to once a week)
  • Duplicate indicators across feeds create noise
  • Confidence scoring is inconsistent or absent entirely

I spent about three months building a pipeline to solve exactly this. The result is what became MalwareIntel, a platform that aggregates, deduplicates, and scores threat indicators from 13 public sources. In this article I'll show you the technical pieces that make it work.

The 13 Feeds and What They Provide

Before writing a single line of code, I catalogued every feed and what data it actually provides. Here's the breakdown:

Feed Format Type Update Frequency
Abuse.ch URLhaus CSV/JSON URLs, domains ~15 min
Abuse.ch MalwareBazaar JSON File hashes ~1 hour
Abuse.ch ThreatFox JSON IPs, domains, URLs, hashes ~1 hour
AlienVault OTX JSON (STIX) All IOC types Varies
MISP Default Feeds MISP JSON All IOC types Varies
Feodo Tracker CSV IPs (C2) ~1 hour
PhishTank JSON/CSV URLs ~1 hour
OpenPhish Plain text URLs ~12 hours
Emerging Threats (rules) Snort rules IPs, domains Daily
Spamhaus DROP Plain text IP blocks (CIDR) Daily
Spamhaus EDROP Plain text IP blocks (CIDR) Daily
CISA KEV JSON CVEs As published
Botvrij.eu Plain text sets Domains, IPs, hashes Daily

The diversity in formats is the core challenge. Let's look at how to handle it.

Building the Normalization Pipeline

The most important architectural decision I made was defining a common indicator schema before touching any feed. Every indicator, regardless of source, gets mapped to this structure:

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List
from enum import Enum

class IndicatorType(Enum):
    IP = "ip"
    DOMAIN = "domain"
    URL = "url"
    FILE_HASH_MD5 = "md5"
    FILE_HASH_SHA1 = "sha1"
    FILE_HASH_SHA256 = "sha256"
    CVE = "cve"
    CIDR = "cidr"

@dataclass
class ThreatIndicator:
    value: str
    itype: IndicatorType
    source: str
    first_seen: datetime
    last_seen: datetime
    confidence: float          # 0.0 - 1.0
    tags: List[str] = field(default_factory=list)
    malware_family: Optional[str] = None
    threat_type: Optional[str] = None
    raw: Optional[dict] = None # original payload for debugging
Enter fullscreen mode Exit fullscreen mode

Each feed gets its own parser class that inherits from a base:

from abc import ABC, abstractmethod
from typing import Iterator
import httpx

class BaseFeedParser(ABC):
    def __init__(self, feed_url: str, feed_name: str):
        self.feed_url = feed_url
        self.feed_name = feed_name
        self.client = httpx.Client(timeout=30.0)

    def fetch(self) -> bytes:
        response = self.client.get(self.feed_url)
        response.raise_for_status()
        return response.content

    @abstractmethod
    def parse(self, raw_data: bytes) -> Iterator[ThreatIndicator]:
        """Yield normalized ThreatIndicator objects."""
        pass

    def run(self) -> Iterator[ThreatIndicator]:
        data = self.fetch()
        yield from self.parse(data)
Enter fullscreen mode Exit fullscreen mode

Here's a concrete example for Abuse.ch ThreatFox, one of the richer feeds:

import json
from datetime import datetime

class ThreatFoxParser(BaseFeedParser):
    CONFIDENCE_MAP = {
        "high": 0.9,
        "medium": 0.6,
        "low": 0.3,
    }

    def __init__(self):
        super().__init__(
            feed_url="https://threatfox-api.abuse.ch/api/v1/",
            feed_name="threatfox"
        )

    def fetch(self) -> bytes:
        payload = {"query": "get_iocs", "days": 1}
        response = self.client.post(self.feed_url, json=payload)
        response.raise_for_status()
        return response.content

    def parse(self, raw_data: bytes) -> Iterator[ThreatIndicator]:
        data = json.loads(raw_data)

        if data.get("query_status") != "ok":
            return

        for entry in data.get("data", []):
            ioc_type = self._map_type(entry.get("ioc_type", ""))
            if ioc_type is None:
                continue

            yield ThreatIndicator(
                value=entry["ioc"].split(":")[0],  # strip port if present
                itype=ioc_type,
                source=self.feed_name,
                first_seen=datetime.fromisoformat(entry["first_seen"].replace(" ", "T")),
                last_seen=datetime.fromisoformat(entry["last_seen"].replace(" ", "T")),
                confidence=self.CONFIDENCE_MAP.get(entry.get("confidence_level", "low"), 0.3),
                tags=entry.get("tags") or [],
                malware_family=entry.get("malware"),
                threat_type=entry.get("threat_type"),
                raw=entry
            )

    def _map_type(self, raw_type: str) -> Optional[IndicatorType]:
        mapping = {
            "ip:port": IndicatorType.IP,
            "domain": IndicatorType.DOMAIN,
            "url": IndicatorType.URL,
            "md5_hash": IndicatorType.FILE_HASH_MD5,
            "sha256_hash": IndicatorType.FILE_HASH_SHA256,
        }
        return mapping.get(raw_type)
Enter fullscreen mode Exit fullscreen mode

The Spamhaus DROP feed is a totally different animal—plain text with comment lines:

import ipaddress

class SpamhausDropParser(BaseFeedParser):
    def __init__(self):
        super().__init__(
            feed_url="https://www.spamhaus.org/drop/drop.txt",
            feed_name="spamhaus_drop"
        )

    def parse(self, raw_data: bytes) -> Iterator[ThreatIndicator]:
        now = datetime.utcnow()
        for line in raw_data.decode("utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith(";"):
                continue

            # Format: "1.2.3.0/24 ; SBL123456"
            cidr = line.split(";")[0].strip()

            try:
                ipaddress.ip_network(cidr, strict=False)
            except ValueError:
                continue

            yield ThreatIndicator(
                value=cidr,
                itype=IndicatorType.CIDR,
                source=self.feed_name,
                first_seen=now,
                last_seen=now,
                confidence=0.85,  # Spamhaus has high reputation
                tags=["spam", "botnet"],
                threat_type="spam_infrastructure"
            )
Enter fullscreen mode Exit fullscreen mode

Deduplication and Confidence Scoring

This is where things get interesting. A single malicious IP might appear in Feodo Tracker, ThreatFox, AND OTX simultaneously. Naively storing all three entries creates noise. Instead, I merge duplicate indicators and use multi-source sightings to boost confidence.

The deduplication key is (value, itype). When the same indicator appears in multiple sources, I merge them:

from collections import defaultdict

class IndicatorAggregator:
    def __init__(self):
        self.store: dict[tuple, ThreatIndicator] = {}
        self.source_counts: dict[tuple, set] = defaultdict(set)

    def ingest(self, indicator: ThreatIndicator) -> None:
        key = (indicator.value, indicator.itype)
        self.source_counts[key].add(indicator.source)

        if key not in self.store:
            self.store[key] = indicator
        else:
            existing = self.store[key]
            # Update temporal fields
            existing.first_seen = min(existing.first_seen, indicator.first_seen)
            existing.last_seen = max(existing.last_seen, indicator.last_seen)
            # Merge tags
            existing.tags = list(set(existing.tags + indicator.tags))
            # Take highest individual confidence, then boost for multi-source
            existing.confidence = max(existing.confidence, indicator.confidence)

        # Recalculate multi-source boost
        self._apply_source_boost(key)

    def _apply_source_boost(self, key: tuple) -> None:
        indicator = self.store[key]
        n_sources = len(self.source_counts[key])

        # Each additional source adds a 5% boost, capped at 0.99
        boost = min((n_sources - 1) * 0.05, 0.2)
        indicator.confidence = min(indicator.confidence + boost, 0.99)

        # Track source count as metadata
        indicator.tags.append(f"sources:{n_sources}")
Enter fullscreen mode Exit fullscreen mode

The multi-source boost is a simple heuristic but it works well in practice. An IP seen in three independent feeds is almost certainly malicious, and a confidence of 0.95+ reflects that appropriately.

Scheduling, Storage, and the API Layer

Running all 13 parsers sequentially every hour would be inefficient. I use APScheduler with per-feed intervals based on their actual update frequencies:

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger

scheduler = BlockingScheduler()

feed_config = [
    (URLhausParser(), 15),        # minutes
    (ThreatFoxParser(), 60),
    (MalwareBazaarParser(), 60),
    (FeodoTrackerParser(), 60),
    (PhishTankParser(), 60),
    (SpamhausDropParser(), 1440), # daily
    (SpamhausEDropParser(), 1440),
    (OpenPhishParser(), 720),
    # ... etc
]

for parser, interval_minutes in feed_config:
    scheduler.add_job(
        func=run_feed_pipeline,
        args=[parser],
        trigger=IntervalTrigger(minutes=interval_minutes),
        id=parser.feed_name,
        max_instances=1,
        coalesce=True
    )
Enter fullscreen mode Exit fullscreen mode

For storage, I went with PostgreSQL with a GIN index on the tags array and a B-tree index on (value, itype) for fast lookups:

CREATE TABLE indicators (
    id          BIGSERIAL PRIMARY KEY,
    value       TEXT NOT NULL,
    itype       VARCHAR(20) NOT NULL,
    confidence  NUMERIC(4,3) NOT NULL,
    first_seen  TIMESTAMPTZ NOT NULL,
    last_seen   TIMESTAMPTZ NOT NULL,
    sources     TEXT[] NOT NULL DEFAULT '{}',
    tags        TEXT[] NOT NULL DEFAULT '{}',
    malware_family VARCHAR(100),
    threat_type VARCHAR(100),
    UNIQUE (value, itype)
);

CREATE INDEX idx_indicators_value_type ON indicators (value, itype);
CREATE INDEX idx_indicators_tags ON indicators USING GIN (tags);
CREATE INDEX idx_indicators_last_seen ON indicators (last_seen DESC);
CREATE INDEX idx_indicators_confidence ON indicators (confidence DESC);
Enter fullscreen mode Exit fullscreen mode

The upsert operation handles the merge logic at the database level:

INSERT INTO indicators (value, itype, confidence, first_seen, last_seen, sources, tags, malware_family, threat_type)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (value, itype) DO UPDATE SET
    confidence    = GREATEST(indicators.confidence, EXCLUDED.confidence),
    first_seen    = LEAST(indicators.first_seen, EXCLUDED.first_seen),
    last_seen     = GREATEST(indicators.last_seen, EXCLUDED.last_seen),
    sources       = ARRAY(SELECT DISTINCT UNNEST(indicators.sources || EXCLUDED.sources)),
    tags          = ARRAY(SELECT DISTINCT UNNEST(indicators.tags || EXCLUDED.tags)),
    malware_family = COALESCE(EXCLUDED.malware_family, indicators.malware_family),
    threat_type   = COALESCE(EXCLUDED.threat_type, indicators.threat_type);
Enter fullscreen mode Exit fullscreen mode

The REST API is built with FastAPI and exposes a simple lookup endpoint:

from fastapi import FastAPI, HTTPException, Query
import asyncpg

app = FastAPI(title="CTI Aggregator API")

@app.get("/v1/indicator/{value}")
async def lookup_indicator(
    value: str,
    min_confidence: float = Query(default=0.0, ge=0.0, le=1.0)
):
    async with app.state.db_pool.acquire() as conn:
        row = await conn.fetchrow(
            """
            SELECT value, itype, confidence, first_seen, last_seen,
                   sources, tags, malware_family, threat_type
            FROM indicators
            WHERE value = $1 AND confidence >= $2
            """,
            value, min_confidence
        )

    if not row:
        raise HTTPException(status_code=404, detail="Indicator not found")

    return dict(row)


@app.get("/v1/search")
async def search_indicators(
    tag: Optional[str] = None,
    malware_family: Optional[str] = None,
    itype: Optional[str] = None,
    min_confidence: float = Query(default=0.5, ge=0.0, le=1.0),
    limit: int = Query(default=100, le=1000)
):
    conditions = ["confidence >= $1"]
    params = [min_confidence]
    i = 2

    if tag:
        conditions.append(f"${ i} = ANY(tags)")
        params.append(tag)
        i += 1
    if malware_family:
        conditions.append(f"malware_family ILIKE ${i}")
        params.append(f"%{malware_family}%")
        i += 1
    if itype:
        conditions.append(f"itype = ${i}")
        params.append(itype)
        i += 1

    query = f"""
        SELECT value, itype, confidence, last_seen, sources, tags, malware_family
        FROM indicators
        WHERE {' AND '.join(conditions)}
        ORDER BY last_seen DESC, confidence DESC
        LIMIT {limit}
    """

    async with app.state.db_pool.acquire() as conn:
        rows = await conn.fetch(query, *params)

    return [dict(r) for r in rows]
Enter fullscreen mode Exit fullscreen mode

What I Learned (and Would Do Differently)

After running this pipeline for several months, a few lessons stand out:

1. Start with fewer feeds. I launched with all 13 simultaneously. That was a mistake. Each feed has its own quirks: rate limits, downtime patterns, format changes. Starting with 3-4 high-quality feeds (Abuse.ch ecosystem + OTX + CISA KEV) and adding incrementally would have saved debugging time.

2. Deduplication at ingestion is non-negotiable. Without the upsert-based merge, my database would have 10x more rows with no additional intelligence value. The multi-source confidence boost turned out to be one of the most useful features for downstream consumers.

3. Observability matters from day one. I added structured logging and feed health monitoring after the third "why did this feed stop updating?" incident. Each feed now has a heartbeat check and alerts when it misses two consecutive update cycles.

4. The Knowledge Graph was the surprise hit. I initially built MalwareIntel as a simple indicator lookup tool. Adding the relationship graph between malware families, threat actors, and TTPs turned it from a search engine into an analysis platform. If I were starting over, I'd build the graph data model first.

If you want to dive deeper into threat intelligence operations, I've written extensively about CTI feeds, SIEM integration, and SOC automation on the Riskitera cybersecurity blog.

Try It

MalwareIntel is free to use. Search any IOC, explore malware families, generate Sigma/KQL/SPL detection rules. The API is documented at /docs.

If you build something similar or have questions about the architecture, find me on LinkedIn or open an issue on GitHub.

Top comments (0)