Integrate Volerion CVE Intelligence

.env.example

@@ -50,3 +50,19 @@ GREYNOISE_API_KEY=
 # Contact https://www.circl.lu/contact/ for access
 CIRCL_PDNS_USER=
 CIRCL_PDNS_PASS=
+
+# ── Volerion CVE Intelligence ────────────────────────────────────────────────
+
+# Volerion API key — paid plan required, get access at https://volerion.com
+# Pre-encoded Base64 "user:pass" string provided by Volerion
+VOLERION_API_KEY=
+
+# Volerion API base URL — override if using a custom endpoint or proxy
+# Default: https://api.volerion.com
+VOLERION_BASE=
+
+# Rate limiting (requests per second and burst capacity)
+# Defaults are conservative; increase if your Volerion plan allows higher throughput
+VOLERION_RATE=2.0
+VOLERION_BURST=5
+

CHANGELOG.md

@@ -6,7 +6,14 @@ This project follows [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 ### Added
-- (Next release items go here)
+- **Volerion CVE Intelligence (3 tools):**
+  - `get_volerion_intel`: Fetch deep AI-written CVE analysis, CVSS v3/v4 scores, remediation guidance, structured attack patterns/flows, affected products, and graphical links.
+  - `get_volerion_scores`: Retrieve proprietary risk scores across 8 dimensions (`spread`, `impact`, `exploitability`, `remediation`, `threat`, `urgency`, `incentive`, and composite `score`) sorted by a specific profile.
+  - `lookup_volerion_product`: Query the Volerion CPE catalog for vendor/product details, sw-targets, deprecation status, and platform labels.
+- Volerion independent cache TTLs (`TTL_VOLERION_INTEL` at 12 hours, `TTL_VOLERION_SCORES` at 1 hour, `TTL_VOLERION_PRODUCTS` at 24 hours).
+- Custom `volerion_limiter` rate limiting block added to `AppContext` and `app_lifespan` to decouple from NVD's.
+- Structured Pydantic validation models for all Volerion API endpoints in `src/cve_mcp/models.py`.
+- Subprocess environment isolation (`SERVER_PARAMS_NO_KEY`) in tests to robustly test no-key paths even when local `.env` contains valid credentials.
 
 ## [0.1.0] — 2026-04-14
 

README.md

@@ -97,7 +97,7 @@ All traffic is **outbound HTTPS only** — no inbound ports are opened. API keys
 
 ---
 
-## 🔍 Tool catalog (27 tools)
+## 🔍 Tool catalog (30 tools)
 
 ### Core Vulnerability Intelligence (8 tools)
 
@@ -156,6 +156,16 @@ All traffic is **outbound HTTPS only** — no inbound ports are opened. API keys
 | `scan_github_advisories` | Search GitHub Security Advisories by ecosystem, package, or severity | `GITHUB_TOKEN` (optional) | `scan_github_advisories(ecosystem="pip", package="django")` |
 | `urlscan_check` | Submit a URL for scanning or retrieve previous scan results from URLScan.io | `URLSCAN_API_KEY` | `urlscan_check("https://suspicious-site.com")` |
 
+### Volerion CVE Intelligence (3 tools)
+
+Requires `VOLERION_API_KEY` (paid plan — [volerion.com](https://volerion.com)).
+
+| Tool | Description | API Key Required | Example Usage |
+|------|-------------|-----------------|---------------|
+| `get_volerion_intel` | Deep AI-written CVE analysis: CVSS v3/v4 scores, structured attack flow, remediation guidance, affected products, and a link to the Volerion graph explorer | `VOLERION_API_KEY` | `get_volerion_intel("CVE-2021-44228")` |
+| `get_volerion_scores` | Proprietary risk scores across 8 dimensions. Profile options: `spread`, `impact`, `exploitability`, `remediation`, `threat`, `prioritization`, `risk` | `VOLERION_API_KEY` | `get_volerion_scores(profile="risk", cve_ids="CVE-2021-44228")` |
+| `lookup_volerion_product` | Search the Volerion CPE product catalog by vendor, product name, platform type (`a`/`o`/`h`), or title | `VOLERION_API_KEY` | `lookup_volerion_product(vendor="apache")` |
+
 ---
 
 ## 📦 Installation
@@ -243,6 +253,7 @@ API keys are organized by priority — get the **Tier 1** keys first for maximum
 | `URLSCAN_KEY` | URL scanning and website analysis | [Sign up at urlscan.io](https://urlscan.io/user/signup) | **5,000 public scans/day** | Optional |
 | `CIRCL_PDNS_USER` | CIRCL Passive DNS lookups | [Request access at circl.lu](https://www.circl.lu/services/passive-dns/) | Partner access only | Optional |
 | `CIRCL_PDNS_PASS` | CIRCL Passive DNS authentication | Provided with CIRCL registration | Partner access only | Optional |
+| `VOLERION_API_KEY` | Volerion deep CVE intel and risk scoring | [Sign up at volerion.com](https://volerion.com) | None (Paid plan required) | Optional |
 
 > **⚡ Zero-key start:** Eight tools work without any API key — EPSS, CISA KEV, OSV.dev, MITRE ATT&CK, CWE lookups, CVSS parsing, Ransomwhere, and NVD (at reduced rate). You can start using the server immediately and add keys progressively.
 
@@ -273,6 +284,12 @@ GREYNOISE_API_KEY=   # https://viz.greynoise.io/signup
 CIRCL_PDNS_USER=
 CIRCL_PDNS_PASS=
 
+# Volerion CVE Intelligence (paid plan required — volerion.com)
+VOLERION_API_KEY=
+VOLERION_BASE=       # optional base URL override
+VOLERION_RATE=2.0    # optional rate limit (requests per second)
+VOLERION_BURST=5     # optional burst capacity
+
 # Optional overrides
 CACHE_DB_PATH=       # defaults to ~/.cve-mcp/cache.db
 AUDIT_LOG_PATH=      # defaults to ~/.cve-mcp/audit.log
@@ -285,6 +302,22 @@ MAX_RETRIES=3        # retries on transient errors
 **macOS:** `~/Library/Application Support/Claude/claude_desktop_config.json`
 **Windows:** `%APPDATA%\Claude\claude_desktop_config.json`
 
+Using **uv** (recommended for automatic `.env` loading and fast execution):
+
+```json
+{
+  "mcpServers": {
+    "cve-mcp": {
+      "command": "uv",
+      "args": ["run", "cve-mcp"],
+      "cwd": "/absolute/path/to/cve-mcp-server"
+    }
+  }
+}
+```
+
+Or using standard Python (requires explicit environment variables or manual venv python path):
+
 ```json
 {
   "mcpServers": {
@@ -297,15 +330,15 @@ MAX_RETRIES=3        # retries on transient errors
         "GITHUB_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxx",
         "ABUSEIPDB_KEY": "your-abuseipdb-key",
         "GREYNOISE_API_KEY": "your-greynoise-key",
-        "SHODAN_KEY": "your-shodan-key"
+        "SHODAN_KEY": "your-shodan-key",
+        "VOLERION_API_KEY": "your-volerion-key"
       }
     }
   }
 }
 ```
 
-> ⚠️ **Important:** Always use **absolute paths**. Fully quit Claude Desktop (Cmd+Q / Alt+F4) after changing the config — reloading is not enough.
-
+> ⚠️ **Important:** Always use **absolute paths** for `cwd`. When using the `uv` setup, it automatically loads `.env` variables from the specified `cwd` so you don't need to specify keys in the `env` dictionary! Fully quit Claude Desktop (Cmd+Q / Alt+F4) after changing the config.
 ### Claude Code configuration
 
 ```bash

src/cve_mcp/api/volerion_client.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+
+import httpx
+
+from cve_mcp.api.rate_limiter import TokenBucketRateLimiter
+from cve_mcp.cache.sqlite_cache import (
+    TTL_VOLERION_INTEL,
+    TTL_VOLERION_PRODUCTS,
+    TTL_VOLERION_SCORES,
+    VulnCache,
+)
+from cve_mcp.config import MAX_RESPONSE_BYTES, MAX_RETRIES, VOLERION_API_KEY, VOLERION_BASE
+from cve_mcp.models import VolerionCPE, VolerionIntel, VolerionScore
+
+logger = logging.getLogger(__name__)
+
+# NOTE: validate_url_host / ALLOWED_HOSTS in validators.py is dead code — never
+# called at any HTTP call site in this codebase. We follow the established pattern.
+
+_RETRY_DELAYS = [6.0, 12.0, 24.0]
+_VALID_PROFILES = frozenset(
+    {"spread", "impact", "exploitability", "remediation", "threat", "prioritization", "risk"}
+)
+
+
+def _auth_header() -> dict[str, str]:
+    return {
+        "Authorization": f"Basic {VOLERION_API_KEY}",
+        "Accept": "application/json",
+    }
+
+
+async def _get_with_retry(
+    client: httpx.AsyncClient,
+    limiter: TokenBucketRateLimiter,
+    url: str,
+    params: dict,
+) -> httpx.Response:
+    await limiter.acquire()
+    for attempt in range(MAX_RETRIES):
+        try:
+            resp = await client.get(url, params=params, headers=_auth_header())
+            if resp.status_code in (429, 500, 502, 503, 504):
+                delay = _RETRY_DELAYS[min(attempt, len(_RETRY_DELAYS) - 1)]
+                logger.warning(
+                    "Volerion HTTP %d, retry %d/%d after %.0fs",
+                    resp.status_code,
+                    attempt + 1,
+                    MAX_RETRIES,
+                    delay,
+                )
+                if attempt < MAX_RETRIES - 1:
+                    await asyncio.sleep(delay)
+                    continue
+                raise httpx.HTTPStatusError(
+                    f"Volerion transient error (HTTP {resp.status_code})",
+                    request=resp.request,
+                    response=resp,
+                )
+            resp.raise_for_status()
+            content_length = resp.headers.get("content-length")
+            if content_length and int(content_length) > MAX_RESPONSE_BYTES:
+                raise ValueError("Volerion response too large (Content-Length)")
+            if len(resp.content) > MAX_RESPONSE_BYTES:
+                raise ValueError("Volerion response too large (Actual Body)")
+            return resp
+        except (httpx.HTTPStatusError, httpx.RequestError, ValueError) as exc:
+            if attempt < MAX_RETRIES - 1 and isinstance(exc, httpx.RequestError):
+                delay = _RETRY_DELAYS[attempt]
+                logger.warning(
+                    "Volerion request error, retry %d/%d: %s",
+                    attempt + 1,
+                    MAX_RETRIES,
+                    exc,
+                )
+                await asyncio.sleep(delay)
+                continue
+            raise
+    raise RuntimeError("Volerion retry loop exited unexpectedly")  # unreachable
+
+
+def _unwrap(data: object) -> list:
+    if isinstance(data, list):
+        return data
+    if isinstance(data, dict):
+        return data.get("data", [])
+    logger.warning("Volerion: unexpected response shape %s, returning empty list", type(data).__name__)
+    return []
+
+
+async def fetch_volerion_intel(
+    cve_ids: list[str],
+    client: httpx.AsyncClient,
+    limiter: TokenBucketRateLimiter,
+    cache: VulnCache,
+) -> list[VolerionIntel]:
+    if not VOLERION_API_KEY:
+        return []
+
+    results: list[VolerionIntel] = []
+    uncached: list[str] = []
+
+    for cve_id in cve_ids:
+        cached = await cache.get(f"volerion:intel:{cve_id}")
+        if cached:
+            results.append(VolerionIntel.model_validate(cached))
+        else:
+            uncached.append(cve_id)
+
+    for i in range(0, len(uncached), 100):
+        chunk = uncached[i : i + 100]
+        resp = await _get_with_retry(
+            client,
+            limiter,
+            f"{VOLERION_BASE}/v1/cve/intel",
+            {"id": ",".join(chunk), "limit": len(chunk), "page": 0},
+        )
+        for item in _unwrap(resp.json()):
+            intel = VolerionIntel.model_validate(item)
+            await cache.set(f"volerion:intel:{intel.id}", intel.model_dump(), TTL_VOLERION_INTEL)
+            results.append(intel)
+
+    return results
+
+
+async def fetch_volerion_scores(
+    profile: str,
+    cve_ids: list[str],
+    client: httpx.AsyncClient,
+    limiter: TokenBucketRateLimiter,
+    cache: VulnCache,
+    above: float | None = None,
+    below: float | None = None,
+) -> list[VolerionScore]:
+    if not VOLERION_API_KEY:
+        return []
+
+    use_cache = above is None and below is None
+    results: list[VolerionScore] = []
+    uncached: list[str] = []
+
+    if use_cache:
+        for cve_id in cve_ids:
+            cached = await cache.get(f"volerion:score:{profile}:{cve_id}")
+            if cached:
+                results.append(VolerionScore.model_validate(cached))
+            else:
+                uncached.append(cve_id)
+    else:
+        uncached = list(cve_ids)
+
+    if not uncached:
+        return results
+
+    for i in range(0, len(uncached), 100):
+        chunk = uncached[i : i + 100]
+        params: dict = {"profile": profile, "id": ",".join(chunk)}
+        if above is not None:
+            params["above"] = above
+        if below is not None:
+            params["below"] = below
+
+        resp = await _get_with_retry(
+            client, limiter, f"{VOLERION_BASE}/v1/cve/intel/scores", params
+        )
+        for item in _unwrap(resp.json()):
+            score = VolerionScore.model_validate(item)
+            if use_cache:
+                await cache.set(
+                    f"volerion:score:{profile}:{score.id}", score.model_dump(), TTL_VOLERION_SCORES
+                )
+            results.append(score)
+
+    return results
+
+
+async def fetch_volerion_products(
+    client: httpx.AsyncClient,
+    limiter: TokenBucketRateLimiter,
+    cache: VulnCache,
+    vendor: str = "",
+    product: str = "",
+    part: str = "",
+    title: str = "",
+    limit: int = 50,
+    page: int = 0,
+) -> list[VolerionCPE]:
+    if not VOLERION_API_KEY:
+        return []
+
+    limit = min(limit, 2000)
+    cache_key = f"volerion:products:{vendor}:{product}:{part}:{title}:{limit}:{page}"
+    cached = await cache.get(cache_key)
+    if cached:
+        return [VolerionCPE.model_validate(item) for item in cached]
+
+    params: dict = {"limit": limit, "page": page}
+    if vendor:
+        params["vendor"] = vendor
+    if product:
+        params["product"] = product
+    if part:
+        params["part"] = part
+    if title:
+        params["title"] = title
+
+    resp = await _get_with_retry(
+        client, limiter, f"{VOLERION_BASE}/v1/products", params
+    )
+    cpes = [VolerionCPE.model_validate(item) for item in _unwrap(resp.json())]
+    await cache.set(cache_key, [c.model_dump() for c in cpes], TTL_VOLERION_PRODUCTS)
+    return cpes

src/cve_mcp/cache/sqlite_cache.py

@@ -22,6 +22,9 @@
 TTL_EPSS = 3600  # 1 hour
 TTL_KEV = 21600  # 6 hours
 TTL_CVSS = 86400  # 24 hours
+TTL_VOLERION_INTEL = 43200    # 12 hours
+TTL_VOLERION_SCORES = 3600    # 1 hour
+TTL_VOLERION_PRODUCTS = 86400 # 24 hours
 
 _MAX_ROWS = 10_000
 

src/cve_mcp/config.py

@@ -60,3 +60,9 @@
 MITRE_ATTACK_BASE: str = (
     "https://raw.githubusercontent.com/mitre-attack/attack-stix-data/master/enterprise-attack"
 )
+
+# ── Volerion API ─────────────────────────────────────────────────────────────
+VOLERION_API_KEY: str = os.getenv("VOLERION_API_KEY", "")
+VOLERION_BASE: str = os.getenv("VOLERION_BASE", "https://api.volerion.com")
+VOLERION_RATE: float = float(os.getenv("VOLERION_RATE", "2.0"))
+VOLERION_BURST: int = int(os.getenv("VOLERION_BURST", "5"))