From 8c4f7e1d1e203301ccdd4a36f9b623d42d105774 Mon Sep 17 00:00:00 2001 From: manusjs Date: Sun, 2 Aug 2026 08:13:48 +0000 Subject: [PATCH] feat: add cve-enrich tool and CLI subcommand MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a lightweight multi-source CVE enrichment tool that fetches data from NVD, EPSS, CISA KEV, OSV.dev, and VulnCheck KEV **in parallel** and returns a unified risk snapshot. No LLM agent required — pure API aggregation. Features: - Parallel fetching via ThreadPoolExecutor (5 sources, ~15s worst case) - Composite risk scoring from CVSS, EPSS, CISA KEV, VulnCheck signals - CVE ID validation and case normalization - Graceful partial failure: individual source errors don't break the pipeline - Strands @tool interface for agent use - CLI subcommand: manus-agent enrich CVE-XXXX-YYYY [--output json|text] [--no-vulncheck] - Text output with color-coded risk levels and structured sections - JSON output for programmatic consumption Test coverage: 43 new tests covering all fetcher functions, risk computation, integration scenarios (partial failure, exceptions, missing APIs), CLI text/json output modes, and the Strands tool interface. --- src/manus_agent/cli.py | 161 +++++++ src/manus_agent/tools/cve_enrich.py | 380 +++++++++++++++ tests/test_cve_enrich.py | 690 ++++++++++++++++++++++++++++ 3 files changed, 1231 insertions(+) create mode 100644 src/manus_agent/tools/cve_enrich.py create mode 100644 tests/test_cve_enrich.py diff --git a/src/manus_agent/cli.py b/src/manus_agent/cli.py index e8442f2..dca1078 100644 --- a/src/manus_agent/cli.py +++ b/src/manus_agent/cli.py @@ -1057,6 +1057,7 @@ def _run_variants(argv: list[str]) -> int: "poc-search", "changelog", "blast-radius", + "enrich", } @@ -1763,6 +1764,162 @@ def _git(*cmd: str) -> str: return 0 +# --------------------------------------------------------------------------- +# enrich subcommand +# --------------------------------------------------------------------------- + + +def _build_enrich_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="manus-agent enrich", + description=( + "Quick multi-source CVE enrichment. Fetches NVD, EPSS, CISA KEV, OSV,\n" + "and VulnCheck data in parallel and returns a unified risk snapshot.\n" + "No LLM agent required — pure API aggregation." + ), + add_help=True, + ) + p.add_argument( + "cve_id", + metavar="CVE-ID", + help="CVE identifier to enrich (e.g. CVE-2024-3094)", + ) + p.add_argument( + "--no-vulncheck", + action="store_true", + default=False, + help="Skip VulnCheck KEV lookup (useful when no API key is set)", + ) + p.add_argument( + "--output", + choices=["text", "json"], + default="text", + help="Output format (default: text)", + ) + return p + + +def _run_enrich(argv: list[str]) -> int: + """Run the CVE enrichment pipeline and display results.""" + parser = _build_enrich_parser() + args = parser.parse_args(argv) + cve_id = args.cve_id.strip().upper() + + if not re.match(r"^CVE-\d{4}-\d{4,}$", cve_id): + console.print(f"[red]\u2717 Invalid CVE ID format: {cve_id}[/red]") + return 1 + + try: + from manus_agent.tools.cve_enrich import enrich_cve + except ImportError as exc: + console.print(f"[red]\u2717 Failed to import enrich module: {exc}[/red]") + return 1 + + include_vc = not args.no_vulncheck + + with console.status(f"Enriching {cve_id}\u2026", spinner="dots"): + result = enrich_cve(cve_id, include_vulncheck=include_vc) + + if "error" in result: + console.print(f"[red]\u2717 {result['error']}[/red]") + return 1 + + if args.output == "json": + console.print_json(json.dumps(result, default=str)) + return 0 + + # Text output + risk = result.get("risk_assessment", {}) + nvd = result.get("nvd", {}) + epss = result.get("epss", {}) + cisa = result.get("cisa_kev", {}) + osv = result.get("osv", {}) + vc = result.get("vulncheck_kev", {}) + + # Risk banner + level = risk.get("level", "unknown").upper() + level_colors = { + "CRITICAL": "bold red", + "HIGH": "red", + "MEDIUM": "yellow", + "LOW": "green", + "UNKNOWN": "dim", + } + color = level_colors.get(level, "dim") + console.print(f"\n[{color}]\u25cf Risk: {level}[/{color}] (score: {risk.get('score', '?')})") + for sig in risk.get("signals", []): + console.print(f" \u2022 {sig}") + + # NVD section + console.print("\n[bold]NVD[/bold]") + if nvd.get("error"): + console.print(f" [red]\u2717 {nvd['error']}[/red]") + else: + console.print(f" CVSS: {nvd.get('cvss_score', 'N/A')} ({nvd.get('cvss_severity', 'N/A')})") + console.print(f" CWE: {', '.join(nvd.get('cwe_ids', [])) or 'N/A'}") + console.print(f" Published: {nvd.get('published', 'N/A')}") + console.print(f" Status: {nvd.get('status', 'N/A')}") + desc = nvd.get("description", "") + if desc: + truncated = desc[:200] + "\u2026" if len(desc) > 200 else desc + console.print(f" Description: {truncated}") + + # EPSS section + console.print("\n[bold]EPSS[/bold]") + if epss.get("error"): + console.print(f" [red]\u2717 {epss['error']}[/red]") + elif epss.get("score") is not None: + console.print(f" Score: {epss['score']:.4f} (percentile: {epss.get('percentile', 0):.2%})") + else: + console.print(" [dim]No EPSS data available[/dim]") + + # CISA KEV section + console.print("\n[bold]CISA KEV[/bold]") + if cisa.get("error"): + console.print(f" [red]\u2717 {cisa['error']}[/red]") + elif cisa.get("in_kev"): + console.print(" [bold red]\u26a0 ACTIVELY EXPLOITED[/bold red]") + console.print(f" Vendor: {cisa.get('vendor', 'N/A')} | Product: {cisa.get('product', 'N/A')}") + console.print(f" Added: {cisa.get('date_added', 'N/A')} | Due: {cisa.get('due_date', 'N/A')}") + if cisa.get("known_ransomware_use", "Unknown") != "Unknown": + console.print(f" Ransomware: {cisa['known_ransomware_use']}") + else: + console.print(" [green]Not in CISA KEV[/green]") + + # OSV section + console.print("\n[bold]OSV (affected packages)[/bold]") + if osv.get("error"): + console.print(f" [red]\u2717 {osv['error']}[/red]") + elif not osv.get("found"): + console.print(" [dim]No OSV record found[/dim]") + else: + pkgs = osv.get("affected_packages", []) + if not pkgs: + console.print(" [dim]No package-level data in OSV[/dim]") + else: + for pkg in pkgs[:10]: + fixed = ", ".join(pkg.get("fixed_versions", [])) or "none listed" + console.print(f" \u2022 {pkg.get('ecosystem', '?')}:{pkg.get('name', '?')} (fixed: {fixed})") + if len(pkgs) > 10: + console.print(f" \u2026 and {len(pkgs) - 10} more") + + # VulnCheck section + if include_vc: + console.print("\n[bold]VulnCheck KEV[/bold]") + if vc.get("error"): + console.print(f" [red]\u2717 {vc['error']}[/red]") + elif not vc.get("available"): + console.print(" [dim]Skipped (no API key)[/dim]") + elif vc.get("in_kev"): + console.print(" [bold red]\u26a0 In VulnCheck KEV[/bold red]") + console.print(f" Maturity: {vc.get('exploit_maturity', 'N/A')}") + else: + console.print(" [green]Not in VulnCheck KEV[/green]") + + console.print() + return 0 + + # --------------------------------------------------------------------------- # blast-radius subcommand # --------------------------------------------------------------------------- @@ -2265,6 +2422,10 @@ def main() -> None: idx = argv.index("changelog") sys.exit(_run_changelog(argv[idx + 1 :])) + if first_positional == "enrich": + idx = argv.index("enrich") + sys.exit(_run_enrich(argv[idx + 1 :])) + if first_positional == "blast-radius": idx = argv.index("blast-radius") sys.exit(_run_blast_radius(argv[idx + 1 :])) diff --git a/src/manus_agent/tools/cve_enrich.py b/src/manus_agent/tools/cve_enrich.py new file mode 100644 index 0000000..b7743ac --- /dev/null +++ b/src/manus_agent/tools/cve_enrich.py @@ -0,0 +1,380 @@ +""" +Tool: cve_enrich + +Lightweight, non-agent CVE enrichment that fetches data from multiple public +sources **in parallel** and returns a single unified snapshot. No LLM required. + +Sources queried (all public, no API key required for basic operation): + + 1. **NVD** — CVSS scores, CWE, affected CPE, references, published/modified dates + 2. **EPSS** — current exploitation probability score + percentile + 3. **CISA KEV** — active exploitation flag + remediation deadline + 4. **OSV.dev** — affected packages with version ranges + first-fixed versions + +Optional (requires VULNCHECK_API_KEY): + 5. **VulnCheck KEV** — multi-source exploitation signal + +The output is a flat, structured dict suitable for JSON serialization, CLI +display, or programmatic consumption by other tools/agents. + +CLI: ``manus-agent enrich CVE-XXXX-YYYY`` +""" + +from __future__ import annotations + +import os +import re +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Any + +import requests +from strands import tool + +__all__ = ["cve_enrich"] + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_CVE_RE = re.compile(r"^CVE-\d{4}-\d{4,}$", re.IGNORECASE) +_REQUEST_TIMEOUT = 15 # seconds per HTTP call +_NVD_API = "https://services.nvd.nist.gov/rest/json/cves/2.0" +_EPSS_API = "https://api.first.org/data/v1/epss" +_CISA_KEV_API = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json" +_OSV_API = "https://api.osv.dev/v1/vulns" +_VULNCHECK_KEV_API = "https://api.vulncheck.com/v3/index/vulncheck-kev" + + +# --------------------------------------------------------------------------- +# Individual source fetchers +# --------------------------------------------------------------------------- + + +def _fetch_nvd(cve_id: str) -> dict[str, Any]: + """Fetch NVD record for a CVE. Returns structured data or error.""" + url = f"{_NVD_API}?cveId={cve_id}" + headers: dict[str, str] = {} + api_key = os.environ.get("NVD_API_KEY", "").strip() + if api_key: + headers["apiKey"] = api_key + + try: + resp = requests.get(url, headers=headers, timeout=_REQUEST_TIMEOUT) + resp.raise_for_status() + data = resp.json() + vulns = data.get("vulnerabilities", []) + if not vulns: + return {"error": f"No NVD record for {cve_id}"} + + cve_item = vulns[0].get("cve", {}) + + # Extract CVSS v3.1 or v3.0 + metrics = cve_item.get("metrics", {}) + cvss_v31 = metrics.get("cvssMetricV31", []) + cvss_v30 = metrics.get("cvssMetricV30", []) + cvss_data = (cvss_v31[0] if cvss_v31 else cvss_v30[0] if cvss_v30 else {}).get("cvssData", {}) + + # Extract CWE + weaknesses = cve_item.get("weaknesses", []) + cwe_ids = [] + for w in weaknesses: + for desc in w.get("description", []): + val = desc.get("value", "") + if val.startswith("CWE-"): + cwe_ids.append(val) + + # Extract description + descriptions = cve_item.get("descriptions", []) + description = "" + for d in descriptions: + if d.get("lang") == "en": + description = d.get("value", "") + break + if not description and descriptions: + description = descriptions[0].get("value", "") + + # Extract references + references = [ + {"url": ref.get("url", ""), "source": ref.get("source", ""), "tags": ref.get("tags", [])} + for ref in cve_item.get("references", []) + ] + + return { + "source": "nvd", + "cve_id": cve_id, + "published": cve_item.get("published"), + "last_modified": cve_item.get("lastModified"), + "status": cve_item.get("vulnStatus"), + "description": description, + "cvss_score": cvss_data.get("baseScore"), + "cvss_severity": cvss_data.get("baseSeverity"), + "cvss_vector": cvss_data.get("vectorString"), + "cwe_ids": cwe_ids, + "reference_count": len(references), + "references": references[:10], # Cap to avoid huge output + } + except requests.exceptions.RequestException as exc: + return {"source": "nvd", "error": str(exc)} + + +def _fetch_epss(cve_id: str) -> dict[str, Any]: + """Fetch current EPSS score for a CVE.""" + url = f"{_EPSS_API}?cve={cve_id}" + try: + resp = requests.get(url, timeout=_REQUEST_TIMEOUT) + resp.raise_for_status() + data = resp.json() + results = data.get("data", []) + if not results: + return {"source": "epss", "score": None, "percentile": None} + + entry = results[0] + return { + "source": "epss", + "score": float(entry.get("epss", 0)), + "percentile": float(entry.get("percentile", 0)), + "date": entry.get("date"), + } + except requests.exceptions.RequestException as exc: + return {"source": "epss", "error": str(exc)} + + +def _fetch_cisa_kev(cve_id: str) -> dict[str, Any]: + """Check if CVE is in CISA KEV catalog.""" + try: + resp = requests.get(_CISA_KEV_API, timeout=_REQUEST_TIMEOUT) + resp.raise_for_status() + data = resp.json() + for vuln in data.get("vulnerabilities", []): + if vuln.get("cveID", "").upper() == cve_id.upper(): + return { + "source": "cisa_kev", + "in_kev": True, + "vendor": vuln.get("vendorProject"), + "product": vuln.get("product"), + "date_added": vuln.get("dateAdded"), + "due_date": vuln.get("dueDate"), + "required_action": vuln.get("requiredAction"), + "known_ransomware_use": vuln.get("knownRansomwareCampaignUse", "Unknown"), + } + return {"source": "cisa_kev", "in_kev": False} + except requests.exceptions.RequestException as exc: + return {"source": "cisa_kev", "error": str(exc)} + + +def _fetch_osv(cve_id: str) -> dict[str, Any]: + """Fetch OSV.dev record for affected packages.""" + url = f"{_OSV_API}/{cve_id}" + try: + resp = requests.get(url, timeout=_REQUEST_TIMEOUT) + if resp.status_code == 404: + return {"source": "osv", "found": False, "affected_packages": []} + resp.raise_for_status() + data = resp.json() + + packages = [] + for affected in data.get("affected", []): + pkg = affected.get("package", {}) + ranges = affected.get("ranges", []) + fixed_versions = [] + for r in ranges: + for event in r.get("events", []): + if "fixed" in event: + fixed_versions.append(event["fixed"]) + + packages.append( + { + "ecosystem": pkg.get("ecosystem", ""), + "name": pkg.get("name", ""), + "fixed_versions": fixed_versions, + } + ) + + aliases = data.get("aliases", []) + return { + "source": "osv", + "found": True, + "aliases": aliases, + "affected_packages": packages, + } + except requests.exceptions.RequestException as exc: + return {"source": "osv", "error": str(exc)} + + +def _fetch_vulncheck_kev(cve_id: str) -> dict[str, Any]: + """Fetch VulnCheck KEV data (requires API key).""" + api_key = os.environ.get("VULNCHECK_API_KEY", "").strip() + if not api_key: + return {"source": "vulncheck_kev", "available": False, "reason": "no_api_key"} + + url = f"{_VULNCHECK_KEV_API}?cve={cve_id}" + headers = {"Authorization": f"Bearer {api_key}", "Accept": "application/json"} + try: + resp = requests.get(url, headers=headers, timeout=_REQUEST_TIMEOUT) + resp.raise_for_status() + data = resp.json() + entries = data.get("data", []) + if not entries: + return {"source": "vulncheck_kev", "available": True, "in_kev": False} + + entry = entries[0] + return { + "source": "vulncheck_kev", + "available": True, + "in_kev": True, + "date_added": entry.get("date_added"), + "exploit_maturity": entry.get("exploit_maturity"), + "reporting_source": entry.get("reporting_source"), + } + except requests.exceptions.RequestException as exc: + return {"source": "vulncheck_kev", "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Enrichment orchestrator +# --------------------------------------------------------------------------- + + +def _compute_risk_level( + cvss_score: float | None, + epss_score: float | None, + in_cisa_kev: bool, + in_vulncheck_kev: bool, +) -> dict[str, Any]: + """Compute a composite risk level from enrichment data. + + Returns a risk level (critical/high/medium/low/unknown) with rationale. + """ + signals: list[str] = [] + score = 0.0 + + if in_cisa_kev: + score += 40 + signals.append("CISA KEV (active exploitation)") + if in_vulncheck_kev: + score += 20 + signals.append("VulnCheck KEV (multi-source exploitation signal)") + + if cvss_score is not None: + if cvss_score >= 9.0: + score += 25 + signals.append(f"CVSS {cvss_score} (critical)") + elif cvss_score >= 7.0: + score += 15 + signals.append(f"CVSS {cvss_score} (high)") + elif cvss_score >= 4.0: + score += 8 + signals.append(f"CVSS {cvss_score} (medium)") + else: + score += 3 + signals.append(f"CVSS {cvss_score} (low)") + + if epss_score is not None: + if epss_score >= 0.5: + score += 25 + signals.append(f"EPSS {epss_score:.4f} (very high exploitation probability)") + elif epss_score >= 0.1: + score += 15 + signals.append(f"EPSS {epss_score:.4f} (elevated exploitation probability)") + elif epss_score >= 0.01: + score += 5 + signals.append(f"EPSS {epss_score:.4f} (moderate)") + else: + score += 1 + signals.append(f"EPSS {epss_score:.4f} (low)") + + if score >= 60: + level = "critical" + elif score >= 35: + level = "high" + elif score >= 15: + level = "medium" + elif score > 0: + level = "low" + else: + level = "unknown" + + return {"level": level, "score": round(score, 1), "signals": signals} + + +def enrich_cve(cve_id: str, *, include_vulncheck: bool = True) -> dict[str, Any]: + """Fetch enrichment data from all sources in parallel and return unified snapshot. + + Args: + cve_id: CVE identifier (e.g. 'CVE-2024-3094'). + include_vulncheck: Whether to query VulnCheck (requires API key). + + Returns: + Structured dict with data from each source plus a composite risk assessment. + """ + cve_id = cve_id.strip().upper() + if not _CVE_RE.match(cve_id): + return {"error": f"Invalid CVE ID format: {cve_id!r}. Expected CVE-YYYY-NNNNN."} + + # Dispatch all fetchers in parallel + fetchers: dict[str, Any] = { + "nvd": (_fetch_nvd, cve_id), + "epss": (_fetch_epss, cve_id), + "cisa_kev": (_fetch_cisa_kev, cve_id), + "osv": (_fetch_osv, cve_id), + } + if include_vulncheck: + fetchers["vulncheck_kev"] = (_fetch_vulncheck_kev, cve_id) + + results: dict[str, Any] = {} + with ThreadPoolExecutor(max_workers=5) as pool: + futures = {pool.submit(fn, arg): key for key, (fn, arg) in fetchers.items()} + for future in as_completed(futures): + key = futures[future] + try: + results[key] = future.result() + except Exception as exc: + results[key] = {"source": key, "error": str(exc)} + + # Extract key fields for risk computation + nvd = results.get("nvd", {}) + epss = results.get("epss", {}) + cisa = results.get("cisa_kev", {}) + vc = results.get("vulncheck_kev", {}) + + cvss_score = nvd.get("cvss_score") + epss_score = epss.get("score") + in_cisa_kev = cisa.get("in_kev", False) + in_vulncheck_kev = vc.get("in_kev", False) + + risk = _compute_risk_level(cvss_score, epss_score, in_cisa_kev, in_vulncheck_kev) + + return { + "cve_id": cve_id, + "risk_assessment": risk, + "nvd": results.get("nvd", {}), + "epss": results.get("epss", {}), + "cisa_kev": results.get("cisa_kev", {}), + "osv": results.get("osv", {}), + "vulncheck_kev": results.get("vulncheck_kev", {}), + } + + +# --------------------------------------------------------------------------- +# Strands @tool interface +# --------------------------------------------------------------------------- + + +@tool +def cve_enrich(cve_id: str, include_vulncheck: str = "true") -> dict[str, Any]: + """Enrich a CVE with data from NVD, EPSS, CISA KEV, OSV, and VulnCheck in parallel. + + Returns a unified snapshot with risk assessment, CVSS score, EPSS probability, + KEV status, affected packages, and version-range data — all from a single call. + + Args: + cve_id: The CVE identifier to enrich (e.g. 'CVE-2024-3094'). + include_vulncheck: Whether to include VulnCheck KEV lookup ('true'/'false'). + Requires VULNCHECK_API_KEY env var. Default: 'true'. + + Returns: + Structured enrichment result with risk_assessment, nvd, epss, cisa_kev, + osv, and vulncheck_kev sections. + """ + vc = include_vulncheck.lower().strip() in ("true", "1", "yes") + return enrich_cve(cve_id, include_vulncheck=vc) diff --git a/tests/test_cve_enrich.py b/tests/test_cve_enrich.py new file mode 100644 index 0000000..1874072 --- /dev/null +++ b/tests/test_cve_enrich.py @@ -0,0 +1,690 @@ +"""Comprehensive test suite for cve_enrich tool and CLI subcommand. + +All HTTP calls are fully mocked — no real network access. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from manus_agent.tools.cve_enrich import ( + _compute_risk_level, + _fetch_cisa_kev, + _fetch_epss, + _fetch_nvd, + _fetch_osv, + _fetch_vulncheck_kev, + enrich_cve, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +_SAMPLE_NVD_RESPONSE = { + "vulnerabilities": [ + { + "cve": { + "id": "CVE-2024-3094", + "published": "2024-03-29T17:15:00.000", + "lastModified": "2024-04-12T12:00:00.000", + "vulnStatus": "Analyzed", + "descriptions": [{"lang": "en", "value": "XZ Utils backdoor via build process manipulation."}], + "metrics": { + "cvssMetricV31": [ + { + "cvssData": { + "baseScore": 10.0, + "baseSeverity": "CRITICAL", + "vectorString": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H", + } + } + ] + }, + "weaknesses": [{"description": [{"lang": "en", "value": "CWE-506"}]}], + "references": [ + {"url": "https://example.com/advisory", "source": "cve@mitre.org", "tags": ["Advisory"]}, + {"url": "https://github.com/example/commit/abc123", "source": "cve@mitre.org", "tags": ["Patch"]}, + ], + } + } + ] +} + +_SAMPLE_EPSS_RESPONSE = { + "data": [{"cve": "CVE-2024-3094", "epss": "0.9752", "percentile": "0.9998", "date": "2024-04-01"}] +} + +_SAMPLE_KEV_RESPONSE = { + "vulnerabilities": [ + { + "cveID": "CVE-2024-3094", + "vendorProject": "XZ Utils", + "product": "xz", + "dateAdded": "2024-03-30", + "dueDate": "2024-04-15", + "requiredAction": "Apply mitigations per vendor instructions.", + "knownRansomwareCampaignUse": "Unknown", + } + ] +} + +_SAMPLE_OSV_RESPONSE = { + "aliases": ["CVE-2024-3094", "GHSA-jq36-prxc-m8vr"], + "affected": [ + { + "package": {"ecosystem": "Debian", "name": "xz-utils"}, + "ranges": [ + { + "type": "ECOSYSTEM", + "events": [{"introduced": "0"}, {"fixed": "5.6.1+really5.4.5-1"}], + } + ], + } + ], +} + +_SAMPLE_VULNCHECK_KEV_RESPONSE = { + "data": [ + { + "date_added": "2024-03-30", + "exploit_maturity": "active", + "reporting_source": "multi-source", + } + ] +} + + +# --------------------------------------------------------------------------- +# _fetch_nvd tests +# --------------------------------------------------------------------------- + + +class TestFetchNvd: + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_success(self, mock_get): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = _SAMPLE_NVD_RESPONSE + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_nvd("CVE-2024-3094") + assert result["source"] == "nvd" + assert result["cvss_score"] == 10.0 + assert result["cvss_severity"] == "CRITICAL" + assert result["cwe_ids"] == ["CWE-506"] + assert "XZ Utils" in result["description"] + assert result["published"] == "2024-03-29T17:15:00.000" + assert result["reference_count"] == 2 + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_no_vulnerabilities(self, mock_get): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"vulnerabilities": []} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_nvd("CVE-9999-99999") + assert "error" in result + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_network_error(self, mock_get): + import requests + + mock_get.side_effect = requests.exceptions.ConnectionError("timeout") + result = _fetch_nvd("CVE-2024-3094") + assert result["source"] == "nvd" + assert "error" in result + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_cvss_v30_fallback(self, mock_get): + """Uses CVSS v3.0 when v3.1 is absent.""" + response = { + "vulnerabilities": [ + { + "cve": { + "id": "CVE-2020-1234", + "published": "2020-01-01", + "lastModified": "2020-02-01", + "vulnStatus": "Analyzed", + "descriptions": [{"lang": "en", "value": "Test vuln"}], + "metrics": { + "cvssMetricV30": [ + { + "cvssData": { + "baseScore": 7.5, + "baseSeverity": "HIGH", + "vectorString": "CVSS:3.0/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N", + } + } + ] + }, + "weaknesses": [], + "references": [], + } + } + ] + } + mock_resp = MagicMock() + mock_resp.json.return_value = response + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_nvd("CVE-2020-1234") + assert result["cvss_score"] == 7.5 + assert result["cvss_severity"] == "HIGH" + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_no_cvss_data(self, mock_get): + """Handles missing CVSS gracefully.""" + response = { + "vulnerabilities": [ + { + "cve": { + "id": "CVE-2024-0001", + "published": "2024-01-01", + "lastModified": "2024-01-02", + "vulnStatus": "Awaiting Analysis", + "descriptions": [{"lang": "en", "value": "Pending"}], + "metrics": {}, + "weaknesses": [], + "references": [], + } + } + ] + } + mock_resp = MagicMock() + mock_resp.json.return_value = response + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_nvd("CVE-2024-0001") + assert result["cvss_score"] is None + assert result["cvss_severity"] is None + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_nvd_api_key_header(self, mock_get, monkeypatch): + """NVD_API_KEY is sent as a header.""" + monkeypatch.setenv("NVD_API_KEY", "test-key-123") + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + _fetch_nvd("CVE-2024-3094") + call_kwargs = mock_get.call_args + assert call_kwargs[1]["headers"]["apiKey"] == "test-key-123" + + +# --------------------------------------------------------------------------- +# _fetch_epss tests +# --------------------------------------------------------------------------- + + +class TestFetchEpss: + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_success(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = _SAMPLE_EPSS_RESPONSE + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_epss("CVE-2024-3094") + assert result["source"] == "epss" + assert result["score"] == pytest.approx(0.9752) + assert result["percentile"] == pytest.approx(0.9998) + assert result["date"] == "2024-04-01" + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_no_data(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"data": []} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_epss("CVE-9999-99999") + assert result["score"] is None + assert result["percentile"] is None + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_network_error(self, mock_get): + import requests + + mock_get.side_effect = requests.exceptions.Timeout("timeout") + result = _fetch_epss("CVE-2024-3094") + assert "error" in result + + +# --------------------------------------------------------------------------- +# _fetch_cisa_kev tests +# --------------------------------------------------------------------------- + + +class TestFetchCisaKev: + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_found_in_kev(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = _SAMPLE_KEV_RESPONSE + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_cisa_kev("CVE-2024-3094") + assert result["in_kev"] is True + assert result["vendor"] == "XZ Utils" + assert result["date_added"] == "2024-03-30" + assert result["due_date"] == "2024-04-15" + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_not_in_kev(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [{"cveID": "CVE-OTHER-1234"}]} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_cisa_kev("CVE-2024-3094") + assert result["in_kev"] is False + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_case_insensitive_match(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = _SAMPLE_KEV_RESPONSE + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_cisa_kev("cve-2024-3094") + assert result["in_kev"] is True + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_network_error(self, mock_get): + import requests + + mock_get.side_effect = requests.exceptions.ConnectionError("failed") + result = _fetch_cisa_kev("CVE-2024-3094") + assert "error" in result + + +# --------------------------------------------------------------------------- +# _fetch_osv tests +# --------------------------------------------------------------------------- + + +class TestFetchOsv: + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_found(self, mock_get): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = _SAMPLE_OSV_RESPONSE + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_osv("CVE-2024-3094") + assert result["found"] is True + assert len(result["affected_packages"]) == 1 + assert result["affected_packages"][0]["ecosystem"] == "Debian" + assert result["affected_packages"][0]["name"] == "xz-utils" + assert "5.6.1+really5.4.5-1" in result["affected_packages"][0]["fixed_versions"] + assert "GHSA-jq36-prxc-m8vr" in result["aliases"] + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_not_found_404(self, mock_get): + mock_resp = MagicMock() + mock_resp.status_code = 404 + mock_get.return_value = mock_resp + + result = _fetch_osv("CVE-9999-99999") + assert result["found"] is False + assert result["affected_packages"] == [] + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_no_affected_field(self, mock_get): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"aliases": ["CVE-2024-0001"]} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_osv("CVE-2024-0001") + assert result["found"] is True + assert result["affected_packages"] == [] + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_network_error(self, mock_get): + import requests + + mock_get.side_effect = requests.exceptions.Timeout("slow") + result = _fetch_osv("CVE-2024-3094") + assert "error" in result + + +# --------------------------------------------------------------------------- +# _fetch_vulncheck_kev tests +# --------------------------------------------------------------------------- + + +class TestFetchVulncheckKev: + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_found(self, mock_get, monkeypatch): + monkeypatch.setenv("VULNCHECK_API_KEY", "vc-test-key") + mock_resp = MagicMock() + mock_resp.json.return_value = _SAMPLE_VULNCHECK_KEV_RESPONSE + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_vulncheck_kev("CVE-2024-3094") + assert result["available"] is True + assert result["in_kev"] is True + assert result["exploit_maturity"] == "active" + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_not_found(self, mock_get, monkeypatch): + monkeypatch.setenv("VULNCHECK_API_KEY", "vc-test-key") + mock_resp = MagicMock() + mock_resp.json.return_value = {"data": []} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _fetch_vulncheck_kev("CVE-2024-3094") + assert result["available"] is True + assert result["in_kev"] is False + + def test_no_api_key(self, monkeypatch): + monkeypatch.delenv("VULNCHECK_API_KEY", raising=False) + result = _fetch_vulncheck_kev("CVE-2024-3094") + assert result["available"] is False + assert result["reason"] == "no_api_key" + + @patch("manus_agent.tools.cve_enrich.requests.get") + def test_auth_header_sent(self, mock_get, monkeypatch): + monkeypatch.setenv("VULNCHECK_API_KEY", "my-secret-key") + mock_resp = MagicMock() + mock_resp.json.return_value = {"data": []} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + _fetch_vulncheck_kev("CVE-2024-3094") + call_kwargs = mock_get.call_args + assert "Bearer my-secret-key" in call_kwargs[1]["headers"]["Authorization"] + + +# --------------------------------------------------------------------------- +# _compute_risk_level tests +# --------------------------------------------------------------------------- + + +class TestComputeRiskLevel: + def test_critical_all_signals(self): + result = _compute_risk_level(10.0, 0.95, True, True) + assert result["level"] == "critical" + assert result["score"] >= 60 + assert len(result["signals"]) == 4 + + def test_high_kev_plus_medium_cvss(self): + result = _compute_risk_level(5.0, 0.05, True, False) + assert result["level"] == "high" + + def test_medium_high_cvss_no_exploitation(self): + result = _compute_risk_level(8.5, 0.02, False, False) + assert result["level"] == "medium" + + def test_low_score(self): + result = _compute_risk_level(3.0, 0.001, False, False) + assert result["level"] == "low" + + def test_unknown_no_data(self): + result = _compute_risk_level(None, None, False, False) + assert result["level"] == "unknown" + assert result["score"] == 0.0 + + def test_epss_high_alone_is_high(self): + result = _compute_risk_level(None, 0.8, False, False) + assert result["level"] in ("high", "medium") + assert any("EPSS" in s for s in result["signals"]) + + def test_vulncheck_kev_adds_signal(self): + result = _compute_risk_level(7.0, 0.05, False, True) + assert any("VulnCheck" in s for s in result["signals"]) + + +# --------------------------------------------------------------------------- +# enrich_cve integration tests (all sources mocked) +# --------------------------------------------------------------------------- + + +class TestEnrichCve: + @patch("manus_agent.tools.cve_enrich._fetch_vulncheck_kev") + @patch("manus_agent.tools.cve_enrich._fetch_osv") + @patch("manus_agent.tools.cve_enrich._fetch_cisa_kev") + @patch("manus_agent.tools.cve_enrich._fetch_epss") + @patch("manus_agent.tools.cve_enrich._fetch_nvd") + def test_full_enrichment(self, mock_nvd, mock_epss, mock_kev, mock_osv, mock_vc): + mock_nvd.return_value = { + "source": "nvd", + "cvss_score": 10.0, + "cvss_severity": "CRITICAL", + "cwe_ids": ["CWE-506"], + "description": "XZ backdoor", + "published": "2024-03-29", + } + mock_epss.return_value = {"source": "epss", "score": 0.975, "percentile": 0.999} + mock_kev.return_value = {"source": "cisa_kev", "in_kev": True, "vendor": "XZ"} + mock_osv.return_value = { + "source": "osv", + "found": True, + "affected_packages": [{"ecosystem": "Debian", "name": "xz-utils", "fixed_versions": ["5.4.5"]}], + } + mock_vc.return_value = {"source": "vulncheck_kev", "available": True, "in_kev": True} + + result = enrich_cve("CVE-2024-3094") + assert result["cve_id"] == "CVE-2024-3094" + assert result["risk_assessment"]["level"] == "critical" + assert result["nvd"]["cvss_score"] == 10.0 + assert result["epss"]["score"] == 0.975 + assert result["cisa_kev"]["in_kev"] is True + assert result["osv"]["found"] is True + assert result["vulncheck_kev"]["in_kev"] is True + + def test_invalid_cve_format(self): + result = enrich_cve("not-a-cve") + assert "error" in result + assert "Invalid CVE ID" in result["error"] + + def test_invalid_cve_short(self): + result = enrich_cve("CVE-2024-12") + assert "error" in result + + @patch("manus_agent.tools.cve_enrich._fetch_vulncheck_kev") + @patch("manus_agent.tools.cve_enrich._fetch_osv") + @patch("manus_agent.tools.cve_enrich._fetch_cisa_kev") + @patch("manus_agent.tools.cve_enrich._fetch_epss") + @patch("manus_agent.tools.cve_enrich._fetch_nvd") + def test_without_vulncheck(self, mock_nvd, mock_epss, mock_kev, mock_osv, mock_vc): + mock_nvd.return_value = {"source": "nvd", "cvss_score": 5.0} + mock_epss.return_value = {"source": "epss", "score": 0.01} + mock_kev.return_value = {"source": "cisa_kev", "in_kev": False} + mock_osv.return_value = {"source": "osv", "found": False, "affected_packages": []} + + result = enrich_cve("CVE-2024-3094", include_vulncheck=False) + mock_vc.assert_not_called() + # vulncheck_kev key will be empty dict or missing + assert result.get("vulncheck_kev") == {} or "vulncheck_kev" not in result + + @patch("manus_agent.tools.cve_enrich._fetch_vulncheck_kev") + @patch("manus_agent.tools.cve_enrich._fetch_osv") + @patch("manus_agent.tools.cve_enrich._fetch_cisa_kev") + @patch("manus_agent.tools.cve_enrich._fetch_epss") + @patch("manus_agent.tools.cve_enrich._fetch_nvd") + def test_case_normalization(self, mock_nvd, mock_epss, mock_kev, mock_osv, mock_vc): + mock_nvd.return_value = {"source": "nvd", "cvss_score": 5.0} + mock_epss.return_value = {"source": "epss", "score": 0.01} + mock_kev.return_value = {"source": "cisa_kev", "in_kev": False} + mock_osv.return_value = {"source": "osv", "found": False, "affected_packages": []} + mock_vc.return_value = {"source": "vulncheck_kev", "available": False} + + result = enrich_cve("cve-2024-3094") + assert result["cve_id"] == "CVE-2024-3094" + + @patch("manus_agent.tools.cve_enrich._fetch_vulncheck_kev") + @patch("manus_agent.tools.cve_enrich._fetch_osv") + @patch("manus_agent.tools.cve_enrich._fetch_cisa_kev") + @patch("manus_agent.tools.cve_enrich._fetch_epss") + @patch("manus_agent.tools.cve_enrich._fetch_nvd") + def test_partial_failure_graceful(self, mock_nvd, mock_epss, mock_kev, mock_osv, mock_vc): + """If one source fails, others still return data.""" + mock_nvd.return_value = {"source": "nvd", "error": "timeout"} + mock_epss.return_value = {"source": "epss", "score": 0.5, "percentile": 0.9} + mock_kev.return_value = {"source": "cisa_kev", "in_kev": False} + mock_osv.return_value = {"source": "osv", "found": True, "affected_packages": []} + mock_vc.return_value = {"source": "vulncheck_kev", "available": False} + + result = enrich_cve("CVE-2024-3094") + # Should still have all keys + assert "nvd" in result + assert "epss" in result + assert result["nvd"]["error"] == "timeout" + assert result["epss"]["score"] == 0.5 + + @patch("manus_agent.tools.cve_enrich._fetch_vulncheck_kev") + @patch("manus_agent.tools.cve_enrich._fetch_osv") + @patch("manus_agent.tools.cve_enrich._fetch_cisa_kev") + @patch("manus_agent.tools.cve_enrich._fetch_epss") + @patch("manus_agent.tools.cve_enrich._fetch_nvd") + def test_exception_in_fetcher_caught(self, mock_nvd, mock_epss, mock_kev, mock_osv, mock_vc): + """An unhandled exception in a fetcher is caught and reported.""" + mock_nvd.side_effect = RuntimeError("unexpected crash") + mock_epss.return_value = {"source": "epss", "score": 0.01} + mock_kev.return_value = {"source": "cisa_kev", "in_kev": False} + mock_osv.return_value = {"source": "osv", "found": False, "affected_packages": []} + mock_vc.return_value = {"source": "vulncheck_kev", "available": False} + + result = enrich_cve("CVE-2024-3094") + assert "error" in result["nvd"] + assert "unexpected crash" in result["nvd"]["error"] + + @patch("manus_agent.tools.cve_enrich._fetch_vulncheck_kev") + @patch("manus_agent.tools.cve_enrich._fetch_osv") + @patch("manus_agent.tools.cve_enrich._fetch_cisa_kev") + @patch("manus_agent.tools.cve_enrich._fetch_epss") + @patch("manus_agent.tools.cve_enrich._fetch_nvd") + def test_whitespace_in_cve_id(self, mock_nvd, mock_epss, mock_kev, mock_osv, mock_vc): + mock_nvd.return_value = {"source": "nvd", "cvss_score": 7.0} + mock_epss.return_value = {"source": "epss", "score": 0.1} + mock_kev.return_value = {"source": "cisa_kev", "in_kev": False} + mock_osv.return_value = {"source": "osv", "found": False, "affected_packages": []} + mock_vc.return_value = {"source": "vulncheck_kev", "available": False} + + result = enrich_cve(" CVE-2024-3094 ") + assert result["cve_id"] == "CVE-2024-3094" + + +# --------------------------------------------------------------------------- +# CLI _run_enrich tests +# --------------------------------------------------------------------------- + + +class TestRunEnrichCli: + @patch("manus_agent.tools.cve_enrich.enrich_cve") + def test_json_output(self, mock_enrich, capsys): + mock_enrich.return_value = { + "cve_id": "CVE-2024-3094", + "risk_assessment": {"level": "critical", "score": 90, "signals": ["CISA KEV"]}, + "nvd": {"cvss_score": 10.0}, + "epss": {"score": 0.95}, + "cisa_kev": {"in_kev": True}, + "osv": {"found": True, "affected_packages": []}, + "vulncheck_kev": {"available": True, "in_kev": True}, + } + + from manus_agent.cli import _run_enrich + + exit_code = _run_enrich(["CVE-2024-3094", "--output", "json"]) + assert exit_code == 0 + + @patch("manus_agent.tools.cve_enrich.enrich_cve") + def test_text_output(self, mock_enrich, capsys): + mock_enrich.return_value = { + "cve_id": "CVE-2024-3094", + "risk_assessment": {"level": "high", "score": 50, "signals": ["CVSS 8.0 (high)"]}, + "nvd": { + "cvss_score": 8.0, + "cvss_severity": "HIGH", + "cwe_ids": ["CWE-79"], + "published": "2024-01-01", + "status": "Analyzed", + "description": "Test vuln", + }, + "epss": {"score": 0.15, "percentile": 0.85}, + "cisa_kev": {"in_kev": False}, + "osv": { + "found": True, + "affected_packages": [{"ecosystem": "PyPI", "name": "flask", "fixed_versions": ["2.3.3"]}], + }, + "vulncheck_kev": {"available": True, "in_kev": False}, + } + + from manus_agent.cli import _run_enrich + + exit_code = _run_enrich(["CVE-2024-3094"]) + assert exit_code == 0 + + def test_invalid_cve_id(self): + from manus_agent.cli import _run_enrich + + exit_code = _run_enrich(["not-a-cve"]) + assert exit_code == 1 + + @patch("manus_agent.tools.cve_enrich.enrich_cve") + def test_enrich_error_result(self, mock_enrich): + mock_enrich.return_value = {"error": "Invalid CVE ID format"} + + from manus_agent.cli import _run_enrich + + exit_code = _run_enrich(["CVE-2024-3094"]) + assert exit_code == 1 + + @patch("manus_agent.tools.cve_enrich.enrich_cve") + def test_no_vulncheck_flag(self, mock_enrich): + mock_enrich.return_value = { + "cve_id": "CVE-2024-3094", + "risk_assessment": {"level": "low", "score": 5, "signals": []}, + "nvd": {"cvss_score": 3.0}, + "epss": {"score": 0.001}, + "cisa_kev": {"in_kev": False}, + "osv": {"found": False, "affected_packages": []}, + "vulncheck_kev": {}, + } + + from manus_agent.cli import _run_enrich + + exit_code = _run_enrich(["CVE-2024-3094", "--no-vulncheck"]) + assert exit_code == 0 + # Verify include_vulncheck=False was passed + mock_enrich.assert_called_once_with("CVE-2024-3094", include_vulncheck=False) + + +# --------------------------------------------------------------------------- +# Strands @tool interface test +# --------------------------------------------------------------------------- + + +class TestCveEnrichTool: + @patch("manus_agent.tools.cve_enrich.enrich_cve") + def test_tool_interface(self, mock_enrich): + mock_enrich.return_value = {"cve_id": "CVE-2024-3094", "risk_assessment": {"level": "high"}} + + from manus_agent.tools.cve_enrich import cve_enrich + + cve_enrich("CVE-2024-3094") + mock_enrich.assert_called_once_with("CVE-2024-3094", include_vulncheck=True) + + @patch("manus_agent.tools.cve_enrich.enrich_cve") + def test_tool_interface_no_vulncheck(self, mock_enrich): + mock_enrich.return_value = {"cve_id": "CVE-2024-3094", "risk_assessment": {"level": "low"}} + + from manus_agent.tools.cve_enrich import cve_enrich + + cve_enrich("CVE-2024-3094", include_vulncheck="false") + mock_enrich.assert_called_once_with("CVE-2024-3094", include_vulncheck=False)