DEV Community: David Moya

Building a Malware Intelligence Platform: 13 Public Feeds, Knowledge Graph, and Zero Budget

David Moya — Thu, 13 Aug 2026 09:43:59 +0000

Building a Malware Intelligence Platform: 13 Public Feeds, Knowledge Graph, and Zero Budget

I spent the last 6 months building MalwareIntel, an open CTI (Cyber Threat Intelligence) platform that aggregates data from 13 public feeds into a unified knowledge graph. Here is the architecture, what worked, and what almost broke everything.

The Problem

SOC analysts and threat hunters need context. When an alert fires on a SHA256 hash, they need to know: what malware family is this? What APT group uses it? What MITRE ATT&CK techniques does it employ? What are the defensive mitigations?

Commercial platforms (Recorded Future, ThreatConnect, Anomali) solve this for $30,000-100,000/year. Most small and mid-size security teams cannot afford that. They end up checking 5-10 different free tools manually.

MalwareIntel aggregates all of that into one platform with a unified data model, for free.

Architecture

13 Public Feeds
    |
    v
Celery Workers (scheduled ingestion)
    |
    v
Validation + Normalization Pipeline
    |
    v
PostgreSQL (Supabase) ──→ Qdrant (semantic search)
    |
    v
FastAPI REST API
    |
    v
Next.js Frontend (Cloudflare Pages)

Backend: FastAPI + Python 3.12, SQLAlchemy 2.0 async, Celery + Redis for scheduled ingestion.

Frontend: Next.js 14 App Router, TypeScript strict, Tailwind + shadcn/ui. Dark mode only (SOC analysts work in the dark).

Database: Supabase PostgreSQL with full-text search via GIN indexes. Qdrant for semantic search over malware family descriptions.

Hosting: Cloudflare Pages (frontend), Hetzner VPS (backend + workers), Supabase (managed Postgres).

The 13 Feeds

Feed	What it provides	Schedule
MITRE ATT&CK (TAXII)	TTPs, threat actors, software	Weekly
MITRE D3FEND	Defensive mitigations	Weekly
MalwareBazaar (abuse.ch)	Malware hashes + families	Every 6h
ThreatFox (abuse.ch)	C&C IOCs	Every 6h
URLhaus (abuse.ch)	Malware distribution URLs	Every 6h
Feodo Tracker (abuse.ch)	Botnet C&C IPs	Daily
OTX AlienVault	Pulses, IOCs, TTPs	Every 12h
MISP public feeds	IOCs, events, galaxies	Every 12h
CISA KEV	Exploited vulnerabilities	Daily
Ransomware.live	Ransomware victims + groups	Every 6h
ANY.RUN feeds	Sandbox behaviors	Daily
HoneyDB	Honeypot IOCs	Daily
GitHub (SigmaHQ)	Detection rules	Weekly

The Data Model

The core insight: everything in CTI is a graph. A malware family uses TTPs. TTPs have mitigations. Actors use families. Campaigns deploy families against targets.

MalwareFamily ──[uses]──────→ TTP (ATT&CK)
MalwareFamily ──[attributed]─→ ThreatActor
MalwareFamily ──[generates]──→ IOC (IP, hash, domain, URL)
MalwareFamily ──[exploits]───→ CVE
TTP ──[mitigated by]─────────→ D3FEND Mitigation
Campaign ──[uses]────────────→ MalwareFamily
Campaign ──[attributed]──────→ ThreatActor

We model this with 12 PostgreSQL tables and 5 M2M junction tables. Key tables:

malware_families: 4,000+ families with type taxonomy (ransomware, RAT, infostealer, loader...)
iocs: 1.7M+ indicators with type, confidence, severity, TLP classification
ttps: Full MITRE ATT&CK matrix with subtechniques
d3fend_mitigations: Defensive countermeasures mapped to TTPs

The Hardest Part: Normalization

Every feed has its own format. MalwareBazaar calls it "signature", MITRE calls it "software", Malpedia uses different naming. Emotet vs Heodo vs Geodo are the same family.

Our normalization pipeline:

Fetch raw data from feed API
Validate format (IPs via , hashes by length, domains via RFC 1035)
Reject private IPs (RFC 1918), localhost, future dates
Normalize family names against our canonical taxonomy
Deduplicate by unique index on
Link IOCs to families by matching signatures against aliases
Score confidence based on source reliability and corroboration count

@celery_app.task(bind=True, max_retries=3, rate_limit="10/m")
async def ingest_malwarebazaar(self, lookback_hours: int = 24):
    source = await get_or_create_source("MalwareBazaar")
    raw_data = await fetch_malwarebazaar_recent(lookback_hours)
    for sample in raw_data:
        validated = MalwareBazaarSample.model_validate(sample)
        normalized = normalize_to_ioc(validated)
        await ioc_repo.upsert(normalized, source_id=source.id)
        # NEVER download the binary, only metadata

The Knowledge Graph

The crown jewel. An interactive force-directed graph that visualizes relationships between families, actors, TTPs, IOCs, campaigns, and mitigations.

Built with Canvas API (not D3 force simulation, which is too slow for 500+ nodes). Custom physics engine:

Node repulsion: F = 14000 / dist²
Edge attraction: F = (dist - ideal) × 0.006 × strength
Central gravity: F = pos × 0.0008
Damping: 0.82 per frame

6 node types with semantic colors: purple (families), red (actors), green (TTPs), blue (IOCs), amber (campaigns), lime (mitigations).

Defense Kits

The most popular feature: downloadable defense kits per malware family. Each kit includes IOCs, YARA rules, Sigma rules, ATT&CK mappings, and D3FEND mitigations. One click to get everything you need to detect and defend against a specific threat.

4,251 families have defense kits available. Cobalt Strike alone generates a 621KB kit with 45 files.

What I Learned

Feed reliability varies wildly. abuse.ch feeds are rock solid. Some MISP feeds have 30% duplicate data. Build retry logic and dedup from day one.
Anti-prompt-injection in enrichment. When using LLMs to enrich family descriptions, IOC values can contain adversarial text. Sanitize everything before feeding to the model.
Redis cache saves your API. Without caching, the stats endpoints were hitting the database on every request. Redis with 5-minute TTL reduced DB load by 80%.
Spanish CTI content is an empty niche. 95% of threat intelligence is in English. The blog section with CTI articles in Spanish gets more organic traffic than any English content we tried.

Numbers (6 months in)

Metric	Value
Malware families indexed	4,251
IOCs in database	1.7M+
Public feeds integrated	13
Blog articles (ES)	36
Defense kits available	4,251
Monthly cost	~50 EUR
Revenue	Pre-revenue (subscription model launching)

Stack Summary

Layer	Tech
Frontend	Next.js 14, TypeScript, Tailwind, shadcn/ui
Backend	FastAPI, Python 3.12, SQLAlchemy 2.0 async
Workers	Celery + Redis
Database	Supabase PostgreSQL
Vector search	Qdrant
Hosting	Cloudflare Pages + Hetzner VPS
CI/CD	GitHub Actions (1027/1027 tests green)

Links

Building in public. If you work in security and want to contribute feeds or integrations, reach out on LinkedIn.

How We Built a 160-Article AI Education Platform with Next.js and Static HTML

David Moya — Thu, 13 Aug 2026 09:38:56 +0000

How We Built a 160-Article AI Education Platform with Next.js and Static HTML

Three months ago, I launched IAcademy — an AI education platform in Spanish with 160+ free guides covering everything from prompting basics to autonomous agents, LLM deployment, and MCP servers.

Here's what worked, what didn't, and the architecture behind it.

Why Spanish AI Education is Underserved

The AI education space is dominated by English content. Coursera, Udemy, DeepLearning.AI — all English-first. Spanish-speaking professionals (500M+ people) get translated scraps or nothing.

The opportunity: 0% competition on keywords like "agentes ia" (400 monthly searches), "herramientas ia" (400), "formación ia" (250). In English, these keywords have 30-50% competition. In Spanish, nobody's writing quality content.

Architecture: Why Static HTML, Not a CMS

Each blog post is a standalone index.html file. No WordPress, no Gatsby, no MDX compilation step.

site/blog/
├── agentes-ia-que-son/
│   └── index.html
├── herramientas-ia-guia/
│   └── index.html
├── formacion-ia/
│   └── index.html
└── ... (160+ directories)

Why this approach:

Zero build time. Adding an article = creating a directory + file. No compilation, no hydration errors, no framework upgrades breaking 160 pages.
Perfect SEO control. Every <title>, <meta>, JSON-LD schema, internal link, and heading hierarchy is hand-crafted per page. No CMS template imposing its structure.
Instant deploy. Push to GitHub → Cloudflare Pages deploys in ~30 seconds. No build queue.
No JavaScript required for content. Google indexes immediately. Core Web Vitals are perfect — there's nothing to load.

The dynamic parts (auth, course portal, labs) use Supabase + vanilla JS. But the blog — which is the SEO engine — is pure static HTML.

Content Strategy: Niche Prompts Beat Head Terms

After 3 months, here's what ranks and what doesn't:

What ranks (top 10 in Google):

prompts-ia-facturacion — prompts for accountants (position 8.4)
prompts-ia-logistica — prompts for logistics (position 4.3)
prompts-resumenes-ejecutivos — executive summary prompts (position 6.9)
rol-en-prompts — role prompting guide (position 7.2)

What doesn't rank yet:

herramientas-ia-guia — "AI tools" head term
agentes-ia-que-son — "AI agents" head term
ia-para-marketing — competitive keyword

The pattern: profession-specific prompt templates rank faster than generic guides. Nobody else writes "20 prompts for invoice reconciliation in Spanish." Everyone writes "best AI tools 2026."

SEO Schema: What Actually Matters

Every article includes:

// Article schema
{"@type": "Article", "headline": "...", "author": {...}, "wordCount": 2800}

// FAQ schema (drives rich snippets)
{"@type": "FAQPage", "mainEntity": [{...}, {...}, {...}]}

// Breadcrumb schema
{"@type": "BreadcrumbList", "itemListElement": [...]}

The FAQ schema is the highest-ROI structured data. Google shows your FAQ answers directly in search results → higher CTR → more clicks from the same position.

Honest Numbers (3 Months)

Metric	Value
Articles published	160+
Total clicks (3 months)	~120
Daily impressions	0 → 200 (growing)
Pages in top 10	5
Revenue	$0 (course just launched)
Domain Rating	~0 (new domain)

120 clicks in 3 months is painful. But the trajectory matters: impressions went from 0 to 200/day. The content exists, Google is indexing it, positions are slowly improving. The bottleneck is domain authority (zero backlinks) and domain age (Google sandbox).

What I'd Do Differently

Start with 20 niche articles, not 160. I over-invested in content before having any domain authority. 20 well-linked articles on a domain with 30 backlinks would rank faster than 160 articles on a domain with 0 backlinks.
Build backlinks from day one. Directory submissions, guest posts, GitHub profile links — these should start before the first article, not after the 160th.
Focus on profession-specific content. "Prompts for accountants" ranks. "AI tools guide" doesn't (yet). Double down on what works.

Stack

Blog: Static HTML on Cloudflare Pages
Auth + courses: Supabase (PostgreSQL + RLS)
Analytics: PostHog (EU)
Email: Brevo (transactional)
SEO tracking: Google Search Console + Sistrix

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

David Moya — Thu, 16 Jul 2026 13:20:35 +0000

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

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)

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)

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

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

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
    )

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);

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);

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]

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.