Build a CVE Alert System in Python: Slack, Email, and Webhook Notifications for New Critical Vulnerabilities

Security teams consistently learn about critical CVEs too late. The CVE drops, exploits start circulating on GitHub and Telegram, threat actors spin up scanning infrastructure — and your team hears about it the next morning via a newsletter or a Slack message from a colleague who happened to be scrolling Twitter. By that point, the window for low-noise remediation has already closed.

This guide builds a complete CVE alerting pipeline: a Python script that polls the ValtersIT API every 6 hours using the changed_since parameter, evaluates new CVEs against configurable alert criteria, and routes notifications to Slack, email, or any webhook endpoint (PagerDuty, OpsGenie, generic HTTP). Everything is stateful — no duplicate alerts, no missed CVEs. Deploy it on a server with cron, as a Docker container, or as a GitHub Actions scheduled workflow.

Founding pricing — ends 31 Dec 2026

Standard €19/mo (list €29) · 50,000 credits/mo — sufficient for continuous polling of 20–30 vendors. Pro €65/mo (list €99) · 250,000 credits/mo. PAYG €0.08/credit. Full pricing →

Why polling changed_since beats RSS and email newsletters

RSS feeds and vulnerability newsletters give you a CVE ID and a severity string. What they do not give you:

• Exploit status — whether a public PoC or Nuclei template exists for this specific CVE
• EPSS score — empirical probability that this CVE will be exploited in the next 30 days
• CISA KEV membership — CISA’s confirmed-exploited-in-the-wild list
• Vendor filtering — alerts scoped to the exact software stack you run
• Structured data — patch status, affected version string, Shodan dorks, in machine-readable form you can act on programmatically

The ValtersIT API GET /cve endpoint with changed_since returns exactly those fields for every CVE that was added or updated after your last poll timestamp. You get structured data you can route, filter, and act on — not a human-readable digest you have to parse.

Base URL:   https://api.valtersit.com/api/v1
Auth:       Authorization: Bearer <your key>
Sandbox:    vit_sandbox_free_demo_key_2026  (CVE lookups only, demo data)

Key GET /cve parameters for an alerter

Parameter	Type	Description
changed_since	ISO date string	Returns CVEs updated after this timestamp — the core polling mechanism
vendor	string	Filter to a specific vendor name (e.g., Cisco, Apache, Microsoft)
severity	string	CRITICAL, HIGH, MEDIUM, LOW
cvss_min	float	Minimum CVSS score (0.0–10.0)
epss_min	float	Minimum EPSS score (0.0–1.0)
has_exploit	bool	true returns only CVEs with public PoC or Nuclei template
published_since	ISO date string	Filter by NVD publish date
limit	int	Max 100 per page
page	int	Pagination

Response fields you will use in alerts

{
  "cve_id": "CVE-2026-12345",
  "vendor": "Cisco",
  "cvss_score": 9.8,
  "epss": 0.9421,
  "patch_status": "UNKNOWN / UNPATCHED",
  "cisa_kev": "TRUE",
  "poc_status": "Public",
  "nuclei_template": "http/cves/2026/CVE-2026-12345.yaml",
  "affected_version": "IOS XE 17.3.x through 17.9.x",
  "publish_date": "2026-09-20"
}

Alert criteria to configure

Not every CVE warrants a page at 3 AM. Define three tiers before writing code:

Tier 1 — Immediate alert (Slack + PagerDuty):

• Severity CRITICAL AND poc_status = "Public" or nuclei_template is set
• Any CVE where cisa_kev = "TRUE" (confirmed exploited in the wild)
• EPSS ≥ 0.7 (high exploitation probability, regardless of severity label)

Tier 2 — Same-day alert (Slack only):

• Severity HIGH AND cisa_kev = "TRUE"
• Severity CRITICAL (even without a public exploit — patch window is short)
• Any vendor in your watchlist with cvss_score ≥ 7.0

Tier 3 — Daily digest (email summary):

• Any vendor in your watchlist at any severity
• Severity MEDIUM with epss_min ≥ 0.3

These tiers map directly to the script’s routing logic below.

Project structure

cve-alerter/
├── alerter.py          # Main polling and routing script
├── config.yaml         # Vendor watchlist and thresholds
├── state.json          # Last run timestamp + alerted CVE IDs (auto-created)
├── requirements.txt
├── Dockerfile
└── docker-compose.yml

Configuration file

config.yaml

api:
  key: "vit_v1_your_key_here"
  base_url: "https://api.valtersit.com/api/v1"
  poll_interval_hours: 6
  page_size: 100

vendors:
  # Software your organisation actually runs
  - Cisco
  - Palo Alto Networks
  - Fortinet
  - Apache
  - Microsoft
  - VMware
  - GitLab
  - Nginx
  - Linux Kernel
  - OpenSSL

thresholds:
  tier1_cvss_min: 9.0        # Immediate alert floor
  tier2_cvss_min: 7.0        # Same-day alert floor
  tier3_epss_min: 0.3        # Digest inclusion floor

slack:
  tier1_webhook: "https://hooks.slack.com/services/T.../B.../..."
  tier1_channel: "#security-critical"
  tier2_webhook: "https://hooks.slack.com/services/T.../B.../..."
  tier2_channel: "#security-alerts"

email:
  enabled: true
  smtp_host: "smtp.gmail.com"
  smtp_port: 587
  smtp_user: "alerts@example.com"
  smtp_pass: "your-app-password"
  recipients:
    - "security-team@example.com"

pagerduty:
  enabled: false
  integration_key: "your-pagerduty-routing-key"

opsgenie:
  enabled: false
  api_key: "your-opsgenie-api-key"
  team: "security-oncall"

generic_webhook:
  enabled: false
  url: "https://your-internal-system/webhook/cve"
  headers:
    X-Auth-Token: "your-token"

Requirements

requirements.txt

requests>=2.31.0
PyYAML>=6.0

No exotic dependencies. The script uses only requests for HTTP and PyYAML for config parsing. Email is handled by stdlib smtplib.

Complete alerter script

#!/usr/bin/env python3
"""
CVE Alert Pipeline — polls ValtersIT API, routes notifications to Slack,
email, PagerDuty, OpsGenie, and generic webhooks.

Usage:
    python alerter.py [--config config.yaml] [--state state.json] [--dry-run]
"""

import argparse
import json
import logging
import smtplib
import sys
import time
from datetime import datetime, timedelta, timezone
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from pathlib import Path

import requests
import yaml

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%SZ",
)
log = logging.getLogger("cve-alerter")


# ---------------------------------------------------------------------------
# Config and state helpers
# ---------------------------------------------------------------------------

def load_config(path: str) -> dict:
    with open(path) as f:
        return yaml.safe_load(f)


def load_state(path: str) -> dict:
    p = Path(path)
    if p.exists():
        with open(p) as f:
            return json.load(f)
    # First run: look back 6 hours so we don't flood on startup
    return {
        "last_run": (datetime.now(timezone.utc) - timedelta(hours=6)).isoformat(),
        "alerted_ids": [],
    }


def save_state(path: str, state: dict) -> None:
    with open(path, "w") as f:
        json.dump(state, f, indent=2)


# ---------------------------------------------------------------------------
# API polling
# ---------------------------------------------------------------------------

def fetch_changed_cves(cfg: dict, changed_since: str) -> list[dict]:
    """
    Fetches all CVEs updated since `changed_since`, paginating automatically.
    Applies vendor filter per configured watchlist.
    """
    base = cfg["api"]["base_url"]
    headers = {"Authorization": f"Bearer {cfg['api']['key']}"}
    page_size = cfg["api"].get("page_size", 100)
    results = []

    # Fetch once per vendor to leverage server-side filtering.
    # For large watchlists (30+ vendors) this is more efficient than a single
    # broad query that you then filter client-side.
    vendors = cfg.get("vendors", [])
    if not vendors:
        # No watchlist — fetch all recent CVEs
        results.extend(_paginate(base, headers, {
            "changed_since": changed_since,
            "limit": page_size,
        }))
    else:
        for vendor in vendors:
            vendor_results = _paginate(base, headers, {
                "vendor": vendor,
                "changed_since": changed_since,
                "limit": page_size,
            })
            results.extend(vendor_results)
            if vendor_results:
                log.info("  %s: %d CVE(s) changed", vendor, len(vendor_results))

    return results


def _paginate(base: str, headers: dict, params: dict) -> list[dict]:
    """Fetches all pages for a given query."""
    items = []
    page = 1
    while True:
        p = {**params, "page": page}
        try:
            r = requests.get(f"{base}/cve", headers=headers, params=p, timeout=30)
            r.raise_for_status()
        except requests.RequestException as exc:
            log.error("API request failed: %s", exc)
            break

        data = r.json()
        batch = data.get("data") or data.get("results") or []
        if not batch:
            break
        items.extend(batch)
        if len(batch) < params.get("limit", 100):
            break  # Last page
        page += 1
        time.sleep(0.2)  # Polite pacing

    return items


# ---------------------------------------------------------------------------
# Alert routing
# ---------------------------------------------------------------------------

def classify_cve(cve: dict, cfg: dict) -> str | None:
    """
    Returns 'tier1', 'tier2', 'tier3', or None (below all thresholds).
    """
    cvss = cve.get("cvss_score") or 0.0
    epss = cve.get("epss") or 0.0
    kev = cve.get("cisa_kev", "").upper() == "TRUE"
    has_exploit = (
        cve.get("poc_status", "").lower() == "public"
        or bool(cve.get("nuclei_template"))
    )

    t1_floor = cfg["thresholds"].get("tier1_cvss_min", 9.0)
    t2_floor = cfg["thresholds"].get("tier2_cvss_min", 7.0)
    t3_epss = cfg["thresholds"].get("tier3_epss_min", 0.3)

    # Tier 1: immediate
    if kev:
        return "tier1"
    if cvss >= t1_floor and has_exploit:
        return "tier1"
    if epss >= 0.7:
        return "tier1"

    # Tier 2: same-day
    if cvss >= t1_floor:  # CRITICAL without exploit still warrants same-day
        return "tier2"
    if cvss >= t2_floor and kev:
        return "tier2"
    if cvss >= t2_floor and cve.get("vendor") in cfg.get("vendors", []):
        return "tier2"

    # Tier 3: daily digest
    if cve.get("vendor") in cfg.get("vendors", []):
        return "tier3"
    if epss >= t3_epss:
        return "tier3"

    return None


def cve_url(cve: dict) -> str:
    slug = cve.get("slug") or cve["cve_id"].lower().replace("-", "-")
    return f"https://www.valtersit.com/cve/{slug}/"


# ---------------------------------------------------------------------------
# Slack
# ---------------------------------------------------------------------------

PATCH_EMOJI = {
    "PATCHED / FIX AVAILABLE": ":white_check_mark:",
    "PATCH UNDER REVIEW": ":large_orange_circle:",
    "UNKNOWN / UNPATCHED": ":red_circle:",
}


def build_slack_blocks(cve: dict, tier: str) -> list[dict]:
    cvss = cve.get("cvss_score")
    epss = cve.get("epss")
    kev = cve.get("cisa_kev", "").upper() == "TRUE"
    patch = cve.get("patch_status", "UNKNOWN / UNPATCHED")
    patch_icon = PATCH_EMOJI.get(patch, ":white_circle:")
    has_exploit = (
        cve.get("poc_status", "").lower() == "public"
        or bool(cve.get("nuclei_template"))
    )

    tier_label = {
        "tier1": ":rotating_light: *CRITICAL ALERT*",
        "tier2": ":warning: *HIGH SEVERITY ALERT*",
        "tier3": ":information_source: *Watchlist CVE*",
    }[tier]

    fields = [
        {"type": "mrkdwn", "text": f"*Vendor*\n{cve.get('vendor', 'Unknown')}"},
        {"type": "mrkdwn", "text": f"*CVSS*\n{cvss if cvss is not None else 'Pending'}"},
        {"type": "mrkdwn", "text": f"*EPSS*\n{epss:.1%} exploitation probability" if epss is not None else "*EPSS*\nN/A"},
        {"type": "mrkdwn", "text": f"*Patch Status*\n{patch_icon} {patch}"},
    ]

    if kev:
        fields.append({"type": "mrkdwn", "text": "*CISA KEV*\n:fire: Confirmed exploited in the wild"})
    if has_exploit:
        exploit_detail = "Nuclei template available" if cve.get("nuclei_template") else "Public PoC exists"
        fields.append({"type": "mrkdwn", "text": f"*Exploit*\n:skull: {exploit_detail}"})
    if cve.get("affected_version"):
        fields.append({"type": "mrkdwn", "text": f"*Affected Versions*\n{cve['affected_version'][:80]}"})

    return [
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"{tier_label}\n*<{cve_url(cve)}|{cve['cve_id']}>*",
            },
        },
        {"type": "section", "fields": fields},
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "View Full Report"},
                    "url": cve_url(cve),
                    "style": "primary" if tier == "tier1" else "default",
                }
            ],
        },
        {"type": "divider"},
    ]


def send_slack(webhook_url: str, blocks: list[dict], dry_run: bool = False) -> bool:
    if dry_run:
        log.info("[DRY RUN] Would POST to Slack: %d blocks", len(blocks))
        return True
    try:
        r = requests.post(
            webhook_url,
            json={"blocks": blocks},
            timeout=10,
        )
        return r.status_code == 200
    except requests.RequestException as exc:
        log.error("Slack send failed: %s", exc)
        return False


# ---------------------------------------------------------------------------
# Email
# ---------------------------------------------------------------------------

def send_email(cfg: dict, subject: str, body_html: str, dry_run: bool = False) -> bool:
    email_cfg = cfg.get("email", {})
    if not email_cfg.get("enabled"):
        return False
    if dry_run:
        log.info("[DRY RUN] Would send email: %s", subject)
        return True

    msg = MIMEMultipart("alternative")
    msg["Subject"] = subject
    msg["From"] = email_cfg["smtp_user"]
    msg["To"] = ", ".join(email_cfg["recipients"])
    msg.attach(MIMEText(body_html, "html"))

    try:
        with smtplib.SMTP(email_cfg["smtp_host"], email_cfg["smtp_port"]) as server:
            server.ehlo()
            server.starttls()
            server.login(email_cfg["smtp_user"], email_cfg["smtp_pass"])
            server.sendmail(
                email_cfg["smtp_user"],
                email_cfg["recipients"],
                msg.as_string(),
            )
        return True
    except Exception as exc:
        log.error("Email send failed: %s", exc)
        return False


def build_email_digest(cves: list[dict]) -> str:
    rows = ""
    for c in cves:
        cvss = c.get("cvss_score", "—")
        epss = f"{c['epss']:.1%}" if c.get("epss") else "—"
        kev = "YES" if c.get("cisa_kev", "").upper() == "TRUE" else "no"
        rows += f"""
        <tr>
          <td><a href="{cve_url(c)}">{c['cve_id']}</a></td>
          <td>{c.get('vendor', '—')}</td>
          <td>{cvss}</td>
          <td>{epss}</td>
          <td>{kev}</td>
          <td>{c.get('patch_status', '—')}</td>
        </tr>"""

    return f"""
    <html><body style="font-family: monospace; background: #0a0a0a; color: #e0e0e0; padding: 20px;">
    <h2 style="color: #ff6600;">CVE Daily Digest — {datetime.now(timezone.utc).strftime('%Y-%m-%d')}</h2>
    <p>{len(cves)} new/updated CVEs matching your watchlist in the last 24 hours.</p>
    <table border="1" cellpadding="6" style="border-collapse:collapse; width:100%;">
      <thead style="background:#1a1a1a;">
        <tr><th>CVE ID</th><th>Vendor</th><th>CVSS</th><th>EPSS</th><th>KEV</th><th>Patch</th></tr>
      </thead>
      <tbody>{rows}</tbody>
    </table>
    <p style="margin-top:20px; font-size:12px; color:#666;">
      Powered by <a href="https://api.valtersit.com" style="color:#ff6600;">ValtersIT CVE API</a>
    </p>
    </body></html>"""


# ---------------------------------------------------------------------------
# PagerDuty
# ---------------------------------------------------------------------------

def send_pagerduty(cfg: dict, cve: dict, dry_run: bool = False) -> bool:
    pd_cfg = cfg.get("pagerduty", {})
    if not pd_cfg.get("enabled"):
        return False
    if dry_run:
        log.info("[DRY RUN] Would trigger PagerDuty for %s", cve["cve_id"])
        return True

    cvss = cve.get("cvss_score", 0)
    payload = {
        "routing_key": pd_cfg["integration_key"],
        "event_action": "trigger",
        "payload": {
            "summary": f"{cve['cve_id']} ({cve.get('vendor', 'Unknown')}) CVSS {cvss} — {cve.get('patch_status', 'Unknown')}",
            "severity": "critical" if (cvss or 0) >= 9.0 else "error",
            "source": "ValtersIT CVE API",
            "component": cve.get("vendor", "Unknown"),
            "custom_details": {
                "cve_id": cve["cve_id"],
                "cvss_score": cvss,
                "epss": cve.get("epss"),
                "cisa_kev": cve.get("cisa_kev"),
                "poc_status": cve.get("poc_status"),
                "patch_status": cve.get("patch_status"),
                "affected_version": cve.get("affected_version"),
                "report_url": cve_url(cve),
            },
        },
        "links": [{"href": cve_url(cve), "text": "Full CVE Report"}],
    }
    try:
        r = requests.post(
            "https://events.pagerduty.com/v2/enqueue",
            json=payload,
            timeout=10,
        )
        return r.status_code in (200, 202)
    except requests.RequestException as exc:
        log.error("PagerDuty send failed: %s", exc)
        return False


# ---------------------------------------------------------------------------
# OpsGenie
# ---------------------------------------------------------------------------

def send_opsgenie(cfg: dict, cve: dict, dry_run: bool = False) -> bool:
    og_cfg = cfg.get("opsgenie", {})
    if not og_cfg.get("enabled"):
        return False
    if dry_run:
        log.info("[DRY RUN] Would create OpsGenie alert for %s", cve["cve_id"])
        return True

    cvss = cve.get("cvss_score", 0)
    payload = {
        "message": f"{cve['cve_id']} — {cve.get('vendor', 'Unknown')} CVSS {cvss}",
        "alias": cve["cve_id"],  # Prevents duplicate alerts
        "description": (
            f"Patch status: {cve.get('patch_status', 'Unknown')}\n"
            f"EPSS: {cve.get('epss', 'N/A')}\n"
            f"CISA KEV: {cve.get('cisa_kev', 'FALSE')}\n"
            f"Report: {cve_url(cve)}"
        ),
        "priority": "P1" if (cvss or 0) >= 9.0 else "P2",
        "tags": ["cve", "security", cve.get("vendor", "unknown").lower().replace(" ", "-")],
    }
    if og_cfg.get("team"):
        payload["responders"] = [{"name": og_cfg["team"], "type": "team"}]

    try:
        r = requests.post(
            "https://api.opsgenie.com/v2/alerts",
            json=payload,
            headers={"Authorization": f"GenieKey {og_cfg['api_key']}"},
            timeout=10,
        )
        return r.status_code == 202
    except requests.RequestException as exc:
        log.error("OpsGenie send failed: %s", exc)
        return False


# ---------------------------------------------------------------------------
# Generic webhook
# ---------------------------------------------------------------------------

def send_generic_webhook(cfg: dict, cve: dict, tier: str, dry_run: bool = False) -> bool:
    wh_cfg = cfg.get("generic_webhook", {})
    if not wh_cfg.get("enabled"):
        return False
    if dry_run:
        log.info("[DRY RUN] Would POST generic webhook for %s", cve["cve_id"])
        return True

    payload = {
        "cve_id": cve["cve_id"],
        "vendor": cve.get("vendor"),
        "cvss_score": cve.get("cvss_score"),
        "epss": cve.get("epss"),
        "patch_status": cve.get("patch_status"),
        "cisa_kev": cve.get("cisa_kev"),
        "poc_status": cve.get("poc_status"),
        "nuclei_template": cve.get("nuclei_template"),
        "affected_version": cve.get("affected_version"),
        "publish_date": cve.get("publish_date"),
        "alert_tier": tier,
        "report_url": cve_url(cve),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    try:
        r = requests.post(
            wh_cfg["url"],
            json=payload,
            headers=wh_cfg.get("headers", {}),
            timeout=10,
        )
        return r.ok
    except requests.RequestException as exc:
        log.error("Generic webhook failed: %s", exc)
        return False


# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------

def run(config_path: str, state_path: str, dry_run: bool) -> None:
    cfg = load_config(config_path)
    state = load_state(state_path)

    changed_since = state["last_run"]
    alerted_ids = set(state.get("alerted_ids", []))

    log.info("Polling API for CVEs changed since %s", changed_since)

    cves = fetch_changed_cves(cfg, changed_since)
    log.info("Fetched %d CVE(s) from API", len(cves))

    tier1_cves, tier2_cves, tier3_cves = [], [], []

    for cve in cves:
        cve_id = cve.get("cve_id")
        if not cve_id or cve_id in alerted_ids:
            continue

        tier = classify_cve(cve, cfg)
        if tier == "tier1":
            tier1_cves.append(cve)
        elif tier == "tier2":
            tier2_cves.append(cve)
        elif tier == "tier3":
            tier3_cves.append(cve)

    log.info(
        "Alert tiers — T1: %d, T2: %d, T3: %d",
        len(tier1_cves), len(tier2_cves), len(tier3_cves),
    )

    slack_cfg = cfg.get("slack", {})

    # Tier 1: Slack + PagerDuty/OpsGenie + generic webhook
    for cve in tier1_cves:
        log.info("[T1] %s — %s CVSS %s", cve["cve_id"], cve.get("vendor"), cve.get("cvss_score"))
        blocks = build_slack_blocks(cve, "tier1")
        if slack_cfg.get("tier1_webhook"):
            send_slack(slack_cfg["tier1_webhook"], blocks, dry_run)
        send_pagerduty(cfg, cve, dry_run)
        send_opsgenie(cfg, cve, dry_run)
        send_generic_webhook(cfg, cve, "tier1", dry_run)
        alerted_ids.add(cve["cve_id"])

    # Tier 2: Slack only
    for cve in tier2_cves:
        log.info("[T2] %s — %s CVSS %s", cve["cve_id"], cve.get("vendor"), cve.get("cvss_score"))
        blocks = build_slack_blocks(cve, "tier2")
        if slack_cfg.get("tier2_webhook"):
            send_slack(slack_cfg["tier2_webhook"], blocks, dry_run)
        send_generic_webhook(cfg, cve, "tier2", dry_run)
        alerted_ids.add(cve["cve_id"])

    # Tier 3: accumulate for email digest (send once per run, not per CVE)
    for cve in tier3_cves:
        alerted_ids.add(cve["cve_id"])

    if tier3_cves:
        subject = f"CVE Digest: {len(tier3_cves)} new CVE(s) matching your watchlist"
        html = build_email_digest(tier3_cves)
        send_email(cfg, subject, html, dry_run)

    # Keep alerted_ids from growing unboundedly — cap at 10,000
    if len(alerted_ids) > 10_000:
        alerted_ids = set(list(alerted_ids)[-5_000:])

    # Save state
    state["last_run"] = datetime.now(timezone.utc).isoformat()
    state["alerted_ids"] = list(alerted_ids)
    if not dry_run:
        save_state(state_path, state)
    else:
        log.info("[DRY RUN] State not saved")


def main() -> None:
    parser = argparse.ArgumentParser(description="CVE Alert Pipeline")
    parser.add_argument("--config", default="config.yaml")
    parser.add_argument("--state", default="state.json")
    parser.add_argument("--dry-run", action="store_true", help="Log actions without sending")
    args = parser.parse_args()

    try:
        run(args.config, args.state, args.dry_run)
    except Exception as exc:
        log.exception("Alerter failed: %s", exc)
        sys.exit(1)


if __name__ == "__main__":
    main()

Cron setup

Run the alerter every 6 hours. Add this to your crontab (crontab -e):

# CVE alert pipeline — every 6 hours
0 */6 * * * cd /opt/cve-alerter && /usr/bin/python3 alerter.py --config config.yaml >> /var/log/cve-alerter.log 2>&1

Log rotation to prevent unbounded growth:

/etc/logrotate.d/cve-alerter

/var/log/cve-alerter.log {
    daily
    rotate 14
    compress
    missingok
    notifempty
}

First run test without sending anything:

python3 alerter.py --dry-run

Docker deployment

Wrapping the alerter in a container gives you a clean environment, easy secret injection, and portable deployment.

# Dockerfile
FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY alerter.py .

# State file lives on a mounted volume so it survives container restarts
VOLUME ["/data"]
ENV STATE_PATH=/data/state.json

# Run every 6 hours via internal loop (simpler than container cron)
CMD ["sh", "-c", "while true; do python3 alerter.py --config /config/config.yaml --state /data/state.json; sleep 21600; done"]

docker-compose.yml

services:
  cve-alerter:
    build: .
    restart: unless-stopped
    volumes:
      - ./config.yaml:/config/config.yaml:ro   # Mount config read-only
      - alerter_state:/data                     # Persistent state
    environment:
      - TZ=UTC
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "5"

volumes:
  alerter_state:

Deploy:

docker compose up -d
docker compose logs -f cve-alerter

To update the config without restarting the container — just edit config.yaml. The script reads it fresh on every poll cycle.

Secret injection

Do not put API keys in config.yaml when using Docker. Use environment variable substitution instead:

# config.yaml — reference env vars
api:
  key: "${VALTERSIT_API_KEY}"

# In load_config(), expand env vars after loading:
import os, re

def load_config(path: str) -> dict:
    with open(path) as f:
        raw = f.read()
    # Expand ${VAR_NAME} patterns
    raw = re.sub(r'\$\{(\w+)\}', lambda m: os.environ.get(m.group(1), m.group(0)), raw)
    return yaml.safe_load(raw)

Pass the key at runtime:

docker run -e VALTERSIT_API_KEY=vit_v1_yourkey ... cve-alerter

GitHub Actions variant

If you already run your security tooling through GitHub Actions, this drops in as a scheduled workflow with zero additional infrastructure.

.github/workflows/cve-alerter.yml

name: CVE Alert Pipeline

on:
  schedule:
    # Every 6 hours: 00:00, 06:00, 12:00, 18:00 UTC
    - cron: "0 */6 * * *"
  workflow_dispatch:  # Allow manual trigger from UI

jobs:
  alert:
    runs-on: ubuntu-latest
    timeout-minutes: 10

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip

      - name: Install dependencies
        run: pip install -r cve-alerter/requirements.txt

      - name: Restore alerter state
        uses: actions/cache@v4
        with:
          path: cve-alerter/state.json
          key: cve-alerter-state-${{ github.run_number }}
          restore-keys: |
            cve-alerter-state-

      - name: Write config from secrets
        run: |
          cat > cve-alerter/config.yaml << 'EOF'
          api:
            key: "${{ secrets.VALTERSIT_API_KEY }}"
            base_url: "https://api.valtersit.com/api/v1"
            poll_interval_hours: 6
            page_size: 100
          vendors:
            - Cisco
            - Palo Alto Networks
            - Apache
            - Linux Kernel
            - OpenSSL
          thresholds:
            tier1_cvss_min: 9.0
            tier2_cvss_min: 7.0
            tier3_epss_min: 0.3
          slack:
            tier1_webhook: "${{ secrets.SLACK_CRITICAL_WEBHOOK }}"
            tier1_channel: "#security-critical"
            tier2_webhook: "${{ secrets.SLACK_ALERTS_WEBHOOK }}"
            tier2_channel: "#security-alerts"
          email:
            enabled: true
            smtp_host: "smtp.gmail.com"
            smtp_port: 587
            smtp_user: "${{ secrets.SMTP_USER }}"
            smtp_pass: "${{ secrets.SMTP_PASS }}"
            recipients:
              - "${{ secrets.SECURITY_EMAIL }}"
          pagerduty:
            enabled: true
            integration_key: "${{ secrets.PAGERDUTY_KEY }}"
          EOF

      - name: Run CVE alerter
        working-directory: cve-alerter
        run: python3 alerter.py --state state.json

      - name: Save alerter state
        uses: actions/cache@v4
        if: always()
        with:
          path: cve-alerter/state.json
          key: cve-alerter-state-${{ github.run_number }}

Required GitHub secrets (Settings → Secrets → Actions):

Secret	Value
VALTERSIT_API_KEY	Your API key from the dashboard
SLACK_CRITICAL_WEBHOOK	Slack Incoming Webhook URL for #security-critical
SLACK_ALERTS_WEBHOOK	Slack Incoming Webhook URL for #security-alerts
SMTP_USER	Gmail or SMTP account address
SMTP_PASS	Gmail App Password (16 characters, no spaces)
SECURITY_EMAIL	Email recipient for digests
PAGERDUTY_KEY	PagerDuty Events API v2 integration key

State is persisted across workflow runs using the Actions cache. The restore-keys prefix ensures the previous run’s state is always loaded even when the exact cache key changes.

GitHub Actions cache limits

Actions cache has a 10 GB limit per repository and entries expire after 7 days of inactivity. For production use where continuity matters, prefer the Docker deployment with a persistent volume — the state file is tiny (~50 KB) but needs to survive past the cache TTL.

Severity routing summary

Condition	Tier	Destinations
CISA KEV	T1	Slack #security-critical + PagerDuty + OpsGenie + webhook
CVSS ≥ 9.0 + public exploit or Nuclei template	T1	Same as above
EPSS ≥ 0.70	T1	Same as above
CVSS ≥ 9.0 (no exploit yet)	T2	Slack #security-alerts + webhook
Watchlist vendor + CVSS ≥ 7.0	T2	Slack #security-alerts + webhook
Watchlist vendor, any severity	T3	Daily email digest
EPSS ≥ 0.30	T3	Daily email digest

Avoiding alert fatigue

A few configuration habits that keep the pipeline signal-to-noise ratio high:

Keep your vendor watchlist specific. “Microsoft” generates hundreds of CVEs per month. If you only run Exchange and Azure AD, add those as vendor strings instead of the parent company name. The vendor filter is an exact string match against the ValtersIT vendor field.

Use epss_min to suppress noisy LOW/MEDIUM CVEs. Setting tier3_epss_min: 0.5 means a MEDIUM severity CVE only shows up in your digest if the empirical exploitation probability is above 50% — a strong signal worth tracking even at lower CVSS scores.

Let the alerted_ids set do its job. The script tracks every CVE it has already dispatched an alert for. If a CVE gets updated (new CVSS, exploit added), it will appear in the changed_since response again — but the alerter will skip it since it already sent a notification. If you want re-alerts on status changes (e.g., PATCHED status updated), remove the CVE from alerted_ids in state.json manually or add logic to re-classify on field changes.

Test with the sandbox key first:

# Override the config key for a dry run
VALTERSIT_API_KEY=vit_sandbox_free_demo_key_2026 python3 alerter.py --dry-run

The sandbox key returns fixed demo CVE data and logs what would have been sent without dispatching any actual alerts.

Credit usage estimate

Each GET /cve call with a vendor filter costs 1 credit. With a 30-vendor watchlist polled every 6 hours:

30 vendors × 4 polls/day × 30 days = 3,600 credits/month

Well within the Standard plan’s 50,000 credits/month. Even with pagination (multiple pages per vendor on busy periods), continuous monitoring of a 30-vendor watchlist costs under 10,000 credits/month in practice.

Standard plan at €19/mo (founding price, locked for life while your subscription stays active) is the right tier for this use case. Full pricing and plan comparison →