Automating a MISP Vulnerability Attack Feed

This started as a capstone project and earned a permanent spot in the lab. The NVD firehose is useless as-is: thousands of CVEs a week, almost none relevant to me. The goal was a self-curated intel feed scoped to my actual attack surface, published into MISP so the rest of the lab (Wazuh especially) can act on it automatically.

The problem with raw CVE feeds

Three things make the raw feeds hard to use directly:

1. Volume. The NVD publishes far more than any one environment cares about.
2. No inventory awareness. A CVE in software you don't run is noise.
3. Thin correlation. A CVE ID alone isn't an indicator: you want the associated IPs, domains, and exploit references that turn "a vuln exists" into "watch for this."

MISP solves the storage, correlation, and sharing side. The pipeline solves the filtering and enrichment side.

Architecture

  NVD API ─┐
  CISA KEV ─┼──▶  collector.py  ──▶  filter (inventory)  ──▶  enrich  ──▶  MISP /events
  vendor   ─┘        (dedupe)          (CPE match)          (CVSS,KEV)       (REST API)
  advisories                                                     │
                                                                 ▼
                                                     Wazuh CDB list export
                                                     (malicious IPs / domains)

Everything runs as a scheduled container. It's idempotent: re-running never creates duplicate MISP events, because I key on the CVE ID and update in place.

Step 1: Collect from multiple sources

The NVD 2.0 API is the base, but I merge in CISA's Known Exploited Vulnerabilities (KEV) catalog, because "is this being exploited in the wild right now" is the single most useful prioritization signal there is.

import requests
from datetime import datetime, timedelta

NVD = "https://services.nvd.nist.gov/rest/json/cves/2.0"

def recent_cves(hours=24):
    window_start = (datetime.utcnow() - timedelta(hours=hours)).isoformat()
    params = {"lastModStartDate": window_start + "Z",
              "lastModEndDate": datetime.utcnow().isoformat() + "Z"}
    r = requests.get(NVD, params=params, timeout=30)
    r.raise_for_status()
    return r.json().get("vulnerabilities", [])

Gotcha #1: the NVD rate limit. Without an API key you get a handful of requests per rolling 30 seconds and you will get throttled mid-run. Request a free key, set it in the header, and add backoff. My first version hammered the endpoint and got silently rate-limited into returning partial data, which looked exactly like "no new CVEs," the worst kind of bug.

Step 2: Scope to inventory (the whole point)

I keep a small inventory of the products the lab runs, expressed as CPE prefixes. Every CVE is checked against it; anything that doesn't match is dropped before it ever reaches MISP.

INVENTORY_CPES = [
    "cpe:2.3:a:openbsd:openssh",
    "cpe:2.3:a:docker:docker",
    "cpe:2.3:o:linux:linux_kernel",
    "cpe:2.3:a:netgate:pfsense",
    "cpe:2.3:a:wazuh:wazuh",
    # ... the rest of what I actually run
]

def relevant(cve):
    configs = cve["cve"].get("configurations", [])
    for node in configs:
        for match in node.get("nodes", []):
            for cpe in match.get("cpeMatch", []):
                if any(cpe["criteria"].startswith(p) for p in INVENTORY_CPES):
                    return True
    return False

This one filter is the difference between a feed I read and a feed I ignore. It cuts the volume by well over 95% and everything that survives is, by definition, something I need to care about.

Step 3: Enrich and prioritize

Surviving CVEs get scored. CVSS base score sets a floor, but KEV membership overrides everything: a medium-CVSS bug that's actively exploited outranks a theoretical critical.

def priority(cve, kev_ids):
    cid = cve["cve"]["id"]
    metrics = cve["cve"].get("metrics", {})
    cvss = 0.0
    for key in ("cvssMetricV31", "cvssMetricV30"):
        if key in metrics:
            cvss = metrics[key][0]["cvssData"]["baseScore"]
            break
    if cid in kev_ids:
        return "critical"          # actively exploited: top of the stack
    return ("high" if cvss >= 7 else "medium" if cvss >= 4 else "low")

Step 4: Publish to MISP

Each relevant CVE becomes (or updates) a MISP event with attributes for the CVE ID, CVSS, KEV status, and any associated indicators. I use PyMISP rather than hand-rolling the REST calls. It handles the object model correctly:

from pymisp import PyMISP, MISPEvent, MISPAttribute

misp = PyMISP(MISP_URL, MISP_KEY, ssl=True)

def upsert_event(cve, level):
    cid = cve["cve"]["id"]
    existing = misp.search(controller="events", eventinfo=cid, pythonify=True)
    event = existing[0] if existing else MISPEvent()
    if not existing:
        event.info = cid
        event.threat_level_id = {"critical": 1, "high": 2, "medium": 3}.get(level, 4)
        event.analysis = 1
        event.add_attribute("vulnerability", cid, comment=f"priority={level}")
    misp.add_event(event) if not existing else misp.update_event(event)

Gotcha #2: MISP correlation can melt your instance. MISP auto-correlates every attribute against every other. Publish a few thousand events with shared attributes and correlation grinds the UI to a halt. The fix: disable correlation on high-cardinality attributes (disable_correlation=True) and be selective about what you actually promote to a correlating IOC.

Step 5: Close the loop with Wazuh

The last stage exports MISP's malicious IPs and domains into a CDB list that Wazuh reloads. Intel collected here becomes a live detection there:

attrs = misp.search(controller="attributes", type_attribute="ip-dst",
                    to_ids=True, pythonify=True)
with open("/export/misp-malicious-ips", "w") as f:
    for a in attrs:
        f.write(f"{a.value}:malicious\n")

A cron job on the Wazuh side pulls that file into etc/lists/ and runs wazuh-control restart on a schedule. Now a connection from any MISP-flagged IP raises a Wazuh alert, no human in the loop.

Design decisions I'd defend

• Inventory-scoped, not comprehensive. A feed that tries to cover everything covers nothing usefully. Narrow and relevant beats broad and ignored.
• KEV as the top signal. "Exploited in the wild" is worth more than any static severity score for a small environment that has to prioritize hard.
• Idempotent by CVE ID. The pipeline can run every hour forever without creating a mess, which is what makes it safe to automate and forget.

What's next

• Pull in EPSS (exploit prediction scoring) alongside KEV for probabilistic ranking.
• Auto-open a tracking task in my Notion admin board for anything scored critical.
• Add vendor advisory RSS (pfSense, Docker) as first-class sources, not just NVD.