Observability & Detection intermediate 13 min read

Integrating CISA KEV into Your SIEM for Real-Time Exploitation Alerts

cisa kev siem cve alerting threat-intelligence vulnerability-management

Integrating CISA KEV into Your SIEM for Real-Time Exploitation Alerts

Problem

CISA’s Known Exploited Vulnerabilities (KEV) catalog is the most operationally actionable CVE intelligence source available. Unlike CVSS scores (which measure theoretical severity) or EPSS scores (which predict exploitation probability), KEV lists CVEs that have been confirmed by CISA to be actively exploited in the wild. Every CVE in the KEV catalog has real attackers using it against real targets.

The KEV catalog currently contains over 1,100 CVEs (as of early 2026) and grows by several entries per week. CISA publishes addition dates and remediation deadlines. Federal agencies are required to patch KEV CVEs within the published deadline. For private sector organisations, the KEV catalog is the most reliable signal for “patch this now.”

The integration gap. Most security teams are aware of the KEV catalog but do not have it integrated into their operational tooling. The typical workflow is: a security engineer checks the KEV website manually, cross-references it against their asset inventory by hand, and emails a list of affected systems to the operations team. This happens once a month if it happens at all. By the time a newly added KEV CVE is noticed, it may have been actively exploited for weeks.

What integration enables. When a new CVE is added to the KEV catalog, a properly integrated pipeline can:

  1. Immediately query the asset inventory for systems running affected software
  2. Create high-priority tickets for the affected systems
  3. Alert on-call security if any affected system is internet-facing
  4. Check whether the CVE is already being exploited against your infrastructure (via SIEM correlation against existing log data)
  5. Trigger the vulnerability scanner to run a targeted scan against affected hosts

Without this integration, the KEV catalog is a document that gets read; with it, the KEV catalog is an alerting feed.

Target systems: any organisation with a SIEM or log aggregation platform; security teams responsible for CVE remediation; organisations using Splunk, Elastic SIEM, Chronicle, QRadar, or Sentinel.

Threat Model

Risk 1 — KEV CVE exploited before discovery. A CVE is added to KEV because threat actors are actively using it. The organisation has the vulnerable software. Without KEV integration, the security team does not discover the addition for two weeks. During those two weeks, the organisation’s systems are targeted and compromised.

Risk 2 — Compliance deadline missed. CISA publishes remediation deadlines alongside KEV additions. Federal contractors are required to meet these deadlines. Without automated tracking, deadline adherence is measured manually and may be missed.

Risk 3 — Historical exploitation in existing logs. When a CVE enters KEV, it means exploitation is already occurring. An organisation may have logs showing attack patterns matching the CVE from before the KEV addition was noticed. Without SIEM integration, the historical indicators are not searched.

Configuration / Implementation

Step 1 — Fetch and monitor the KEV catalog

#!/bin/bash
# scripts/kev-monitor.sh
# Fetch the CISA KEV catalog and detect new additions since last run

KEV_URL="https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
STATE_FILE="/var/lib/kev-monitor/known-kev-cves.txt"
ALERT_WEBHOOK="${SLACK_WEBHOOK:-}"

mkdir -p "$(dirname "$STATE_FILE")"

# Fetch current KEV catalog
TEMP_FILE=$(mktemp)
if ! curl -s --max-time 30 "$KEV_URL" -o "$TEMP_FILE"; then
    echo "ERROR: Failed to fetch KEV catalog"
    exit 1
fi

# Extract CVE IDs and metadata for new additions
NEW_ENTRIES=$(python3 - << 'PYEOF'
import json, sys

with open("$TEMP_FILE") as f:
    data = json.load(f)

try:
    with open("$STATE_FILE") as f:
        known = set(f.read().splitlines())
except FileNotFoundError:
    known = set()

new_vulns = [v for v in data["vulnerabilities"] if v["cveID"] not in known]

for v in new_vulns:
    print(f"{v['cveID']}|{v['vendorProject']}|{v['product']}|{v['shortDescription'][:100]}|{v.get('dueDate', 'N/A')}")
PYEOF
)

if [[ -n "$NEW_ENTRIES" ]]; then
    echo "=== NEW KEV ADDITIONS ==="
    echo "$NEW_ENTRIES" | while IFS='|' read -r cve vendor product desc due; do
        echo ""
        echo "CVE: $cve"
        echo "Vendor/Product: $vendor / $product"
        echo "Description: $desc"
        echo "CISA Remediation Due: $due"
        
        # Send alert
        if [[ -n "$ALERT_WEBHOOK" ]]; then
            curl -s -X POST "$ALERT_WEBHOOK" \
                -H "Content-Type: application/json" \
                -d "{\"text\": \":red_circle: *New CISA KEV Entry*\n*$cve*: $vendor/$product\n$desc\nDue: $due\"}"
        fi
    done
    
    # Update known CVEs file
    python3 -c "
import json
with open('$TEMP_FILE') as f:
    data = json.load(f)
with open('$STATE_FILE', 'w') as f:
    for v in data['vulnerabilities']:
        f.write(v['cveID'] + '\n')
"
else
    echo "No new KEV additions since last check"
fi

rm -f "$TEMP_FILE"

Step 2 — Cross-reference KEV against asset inventory

#!/usr/bin/env python3
# scripts/kev-asset-correlation.py
# Cross-reference KEV entries against your asset inventory

import json
import urllib.request
import subprocess
from dataclasses import dataclass
from typing import Optional

@dataclass
class KEVEntry:
    cve_id: str
    vendor: str
    product: str
    description: str
    due_date: str
    notes: str

@dataclass
class AffectedAsset:
    hostname: str
    ip: str
    service: str
    version: str
    cve_id: str
    internet_facing: bool

def fetch_kev_catalog() -> list[KEVEntry]:
    url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json"
    with urllib.request.urlopen(url, timeout=30) as resp:
        data = json.loads(resp.read())
    return [
        KEVEntry(
            cve_id=v["cveID"],
            vendor=v["vendorProject"],
            product=v["product"],
            description=v.get("shortDescription", ""),
            due_date=v.get("dueDate", ""),
            notes=v.get("notes", "")
        )
        for v in data.get("vulnerabilities", [])
    ]

def query_asset_inventory(product_name: str) -> list[dict]:
    """
    Query your CMDB/asset inventory for hosts running a specific product.
    Replace this with your actual CMDB API call.
    """
    # Example: query via a simple REST API
    # In practice, integrate with your CMDB (ServiceNow, Snipe-IT, Netbox, etc.)
    
    # Placeholder: query Trivy scan results database
    result = subprocess.run(
        ["trivy", "db", "query", "--format", "json", f"package:{product_name}"],
        capture_output=True, text=True
    )
    
    # Parse results and return host/service mappings
    return []  # Replace with actual CMDB query

def cross_reference_kev(
    kev_entries: list[KEVEntry],
    asset_inventory: dict  # hostname -> [{"product": ..., "version": ..., "internet_facing": bool}]
) -> list[AffectedAsset]:
    """
    Match KEV entries against asset inventory.
    Returns list of affected assets with KEV CVE context.
    """
    affected = []
    
    # Build a lookup of product name variations to KEV entries
    product_to_kev: dict[str, list[KEVEntry]] = {}
    for entry in kev_entries:
        key = entry.product.lower().replace(" ", "-")
        product_to_kev.setdefault(key, []).append(entry)
        # Also index by vendor/product combination
        vendor_product = f"{entry.vendor}-{entry.product}".lower().replace(" ", "-")
        product_to_kev.setdefault(vendor_product, []).append(entry)
    
    for hostname, services in asset_inventory.items():
        for service in services:
            product_key = service["product"].lower().replace(" ", "-")
            if product_key in product_to_kev:
                for kev_entry in product_to_kev[product_key]:
                    affected.append(AffectedAsset(
                        hostname=hostname,
                        ip=service.get("ip", ""),
                        service=service["product"],
                        version=service.get("version", "unknown"),
                        cve_id=kev_entry.cve_id,
                        internet_facing=service.get("internet_facing", False)
                    ))
    
    return affected

def generate_kev_report(affected: list[AffectedAsset]) -> str:
    """Generate a prioritized remediation report."""
    internet_facing = [a for a in affected if a.internet_facing]
    internal_only = [a for a in affected if not a.internet_facing]
    
    report = f"KEV Asset Correlation Report\n{'='*40}\n"
    report += f"Total affected assets: {len(affected)}\n"
    report += f"Internet-facing (IMMEDIATE): {len(internet_facing)}\n"
    report += f"Internal only: {len(internal_only)}\n\n"
    
    if internet_facing:
        report += "IMMEDIATE ACTION REQUIRED — Internet-facing affected assets:\n"
        for asset in internet_facing:
            report += f"  [{asset.cve_id}] {asset.hostname} ({asset.ip}) — {asset.service}\n"
    
    return report

#!/usr/bin/env python3
# scripts/kev-retroactive-search.py
# When a CVE enters KEV, search historical logs for exploitation indicators

import json
import urllib.request
from datetime import datetime, timedelta, timezone

# Map CVEs to log search patterns
# These are example exploitation indicators — customize for your environment
CVE_LOG_PATTERNS = {
    "CVE-2024-7347": [
        # nginx mp4 module exploit indicators
        "ngx_http_mp4_module",
        r"\.mp4\?start=\d{10,}",
        "nginx.*worker.*exited.*signal 11",
    ],
    "CVE-2025-23419": [
        # mTLS session resumption bypass indicators
        "ssl_verify_result.*0.*session.*resumed",
        "TLS session resumed without client cert",
    ],
    # Add patterns for CVEs relevant to your stack
}

def search_elastic_siem(cve_id: str, patterns: list[str], days_back: int = 30) -> dict:
    """Search Elasticsearch for historical exploitation indicators."""
    from_time = (datetime.now(timezone.utc) - timedelta(days=days_back)).isoformat()
    
    query = {
        "bool": {
            "must": [
                {"range": {"@timestamp": {"gte": from_time}}},
                {"bool": {
                    "should": [
                        {"match": {"message": pattern}} for pattern in patterns
                    ],
                    "minimum_should_match": 1
                }}
            ]
        }
    }
    
    # Execute against Elasticsearch
    # Replace with your Elasticsearch endpoint and credentials
    url = "https://elasticsearch:9200/logs-*/_search"
    payload = json.dumps({"query": query, "size": 100}).encode()
    
    req = urllib.request.Request(
        url, data=payload,
        headers={"Content-Type": "application/json"}
    )
    
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            results = json.loads(resp.read())
            hits = results.get("hits", {}).get("hits", [])
            return {
                "cve_id": cve_id,
                "hit_count": len(hits),
                "earliest_match": hits[0]["_source"].get("@timestamp") if hits else None,
                "sample_matches": [h["_source"].get("message", "")[:200] for h in hits[:3]]
            }
    except Exception as e:
        return {"cve_id": cve_id, "error": str(e)}

def trigger_retroactive_search_for_kev_additions(new_kev_cves: list[str]):
    """Run retroactive log search for newly added KEV CVEs."""
    results = []
    for cve_id in new_kev_cves:
        patterns = CVE_LOG_PATTERNS.get(cve_id, [])
        if patterns:
            result = search_elastic_siem(cve_id, patterns, days_back=90)
            results.append(result)
            
            if result.get("hit_count", 0) > 0:
                print(f"ALERT: {cve_id} exploitation indicators found in historical logs!")
                print(f"  Hit count: {result['hit_count']}")
                print(f"  Earliest match: {result.get('earliest_match')}")
                print(f"  Sample: {result['sample_matches'][0] if result['sample_matches'] else 'N/A'}")
        else:
            print(f"No log patterns defined for {cve_id} — add to CVE_LOG_PATTERNS")
    
    return results

Step 4 — Prometheus alerting on KEV catalog freshness

# Prometheus rules for KEV monitoring

groups:
- name: cisa_kev_monitoring
  rules:
  # Alert when KEV catalog hasn't been checked recently
  - alert: KEVCatalogStale
    expr: |
      time() - kev_last_fetch_timestamp > 86400
    labels:
      severity: warning
    annotations:
      summary: "CISA KEV catalog not checked in >24 hours"
      description: "KEV feed integration may be broken. New exploited CVEs will not be detected."

  # Alert when new KEV entries are found
  - alert: NewKEVEntries
    expr: |
      increase(kev_new_entries_total[1h]) > 0
    labels:
      severity: critical
    annotations:
      summary: "{{ $value }} new CISA KEV entries detected"
      description: "Cross-reference new KEV CVEs against asset inventory immediately."

  # Alert when a KEV CVE affects internet-facing assets
  - alert: KEVInternetFacingAsset
    expr: |
      kev_internet_facing_assets_affected > 0
    labels:
      severity: critical
    annotations:
      summary: "{{ $value }} internet-facing assets affected by KEV CVE"
      description: "Immediate patching required — CVE is known to be exploited in the wild."

  # Alert when KEV remediation deadline is approaching
  - alert: KEVRemediationDeadlineApproaching
    expr: |
      kev_deadline_days_remaining < 7
    labels:
      severity: warning
    annotations:
      summary: "KEV remediation deadline in {{ $value }} days"
      description: "Review affected assets and confirm patch status."

Step 5 — Automate ticket creation for KEV CVEs

#!/bin/bash
# scripts/kev-ticket-creation.sh
# Create high-priority tickets for new KEV entries affecting your environment

create_jira_ticket() {
    local cve_id="$1"
    local vendor="$2"
    local product="$3"
    local due_date="$4"
    local affected_hosts="$5"
    
    local payload
    payload=$(jq -n \
        --arg cve "$cve_id" \
        --arg vendor "$vendor" \
        --arg product "$product" \
        --arg due "$due_date" \
        --arg hosts "$affected_hosts" \
        '{
            "fields": {
                "project": {"key": "SEC"},
                "summary": ("CISA KEV: " + $cve + " — " + $vendor + " " + $product),
                "issuetype": {"name": "Security Incident"},
                "priority": {"name": "Critical"},
                "description": {
                    "type": "doc",
                    "version": 1,
                    "content": [{
                        "type": "paragraph",
                        "content": [{
                            "type": "text",
                            "text": ("CVE: " + $cve + "\nVendor/Product: " + $vendor + "/" + $product + "\nCISA Deadline: " + $due + "\nAffected Hosts:\n" + $hosts)
                        }]
                    }]
                },
                "duedate": $due,
                "labels": ["cve", "kev", "security-critical"]
            }
        }')
    
    curl -s -X POST \
        -H "Authorization: Bearer ${JIRA_TOKEN}" \
        -H "Content-Type: application/json" \
        -d "$payload" \
        "${JIRA_BASE_URL}/rest/api/3/issue"
}

# Called when new KEV entries are detected
# Process each new KEV addition
while IFS='|' read -r cve_id vendor product due_date; do
    # Query asset inventory for affected hosts
    affected_hosts=$(python3 scripts/kev-asset-correlation.py --cve "$cve_id" --format csv 2>/dev/null || echo "Unknown — manual inventory check required")
    
    if [[ -n "$affected_hosts" ]]; then
        echo "Creating ticket for $cve_id affecting: $affected_hosts"
        create_jira_ticket "$cve_id" "$vendor" "$product" "$due_date" "$affected_hosts"
    fi
done < /tmp/new-kev-entries.txt

Expected Behaviour

Event Without KEV integration With KEV integration
New CVE added to KEV catalog Discovered manually; days to weeks later Alert fires within 1 hour of CISA publication
KEV CVE affects internet-facing asset Unknown until manual audit Critical alert fires; on-call paged; ticket created automatically
KEV CVE with historical exploitation Logs not searched Retroactive search runs; historical compromise indicators surfaced
CISA remediation deadline approaching No tracking Prometheus alert fires at 7-day mark
KEV catalog fetch fails Not detected Staleness alert fires after 24 hours

Trade-offs

Aspect Benefit Cost Mitigation
Automated retroactive log search Detects historical exploitation Log search may produce false positives on generic patterns Start with narrow, high-confidence patterns; expand as tuned
Asset inventory dependency Enables targeted alerts Inaccurate inventory produces false positives or missed alerts Keep CMDB current; run reconciliation between inventory and actual hosts
CISA KEV as primary severity signal Operationally precise; confirmed exploitation KEV lags real-world exploitation by days; early adopter attacks missed Supplement KEV with threat intelligence feeds for faster detection
Jira ticket auto-creation Ensures no KEV CVE is missed High-volume KEV additions can flood ticket queue Throttle ticket creation; group multiple KEV additions affecting same product

Failure Modes

Failure Symptom Detection Recovery
CISA KEV feed URL changes Fetch fails silently Staleness alert after 24h; KEV count stops growing Update URL; check CISA website for feed migration notice
False negative: affected asset not in CMDB KEV CVE affects host not in inventory Post-incident review finds uncatalogued host Regular reconciliation: compare CMDB against network scan results
Retroactive search overloads SIEM High-volume log queries spike SIEM load SIEM performance metrics; query timeout errors Rate-limit retroactive searches; run during off-peak hours
Alert fatigue from KEV additions affecting EOL software Team ignores KEV alerts for unsupported software KEV hits grow without tickets being resolved Tag EOL software in CMDB; suppress KEV alerts for EOL assets (with documented acceptance)