From 0ed9cd17ae544e6ba52cbd6d32a2af53273d3f12 Mon Sep 17 00:00:00 2001 From: manusjs Date: Mon, 20 Jul 2026 16:06:46 +0000 Subject: [PATCH] feat(cli): add manus-agent cluster-variants subcommand for CVE variant clustering --- src/manus_agent/cli.py | 57 ++ src/manus_agent/tools/cluster_variants.py | 474 +++++++++++++++ tests/test_cli_cluster_variants.py | 667 ++++++++++++++++++++++ 3 files changed, 1198 insertions(+) create mode 100644 src/manus_agent/tools/cluster_variants.py create mode 100644 tests/test_cli_cluster_variants.py diff --git a/src/manus_agent/cli.py b/src/manus_agent/cli.py index e8442f2..6d4de29 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", + "cluster-variants", } @@ -1935,6 +1936,58 @@ def _run_blast_radius(argv: list[str]) -> int: return 0 +# --------------------------------------------------------------------------- +# cluster-variants subcommand +# --------------------------------------------------------------------------- + + +def _build_cluster_variants_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="manus-agent cluster-variants", + description=( + "Group CVEs related to an input CVE across three cluster dimensions:\n" + "same component/vendor, same CWE weakness class, and same\n" + "researcher/disclosure source." + ), + add_help=True, + ) + p.add_argument("cve_id", metavar="CVE-ID", help="CVE identifier, e.g. CVE-2021-44228") + p.add_argument( + "--output", + choices=["text", "json"], + default="text", + help="Output format (default: text)", + ) + return p + + +def _run_cluster_variants(argv: list[str]) -> int: + import json as _json + + parser = _build_cluster_variants_parser() + args = parser.parse_args(argv) + cve_id = args.cve_id.strip().upper() + + if not cve_id.startswith("CVE-"): + print("[error] CVE-ID must start with 'CVE-'", file=sys.stderr) + return 1 + + from manus_agent.tools.cluster_variants import _render_text, cluster_variants + + result = cluster_variants(cve_id) + + if "error" in result: + print(f"[error] {result['error']}", file=sys.stderr) + return 1 + + if args.output == "json": + print(_json.dumps(result, indent=2)) + else: + print(_render_text(result)) + + return 0 + + def _build_run_parser() -> argparse.ArgumentParser: """Build the top-level run/interactive parser.""" parser = argparse.ArgumentParser( @@ -2269,6 +2322,10 @@ def main() -> None: idx = argv.index("blast-radius") sys.exit(_run_blast_radius(argv[idx + 1 :])) + if first_positional == "cluster-variants": + idx = argv.index("cluster-variants") + sys.exit(_run_cluster_variants(argv[idx + 1 :])) + if first_positional == "discover": idx = argv.index("discover") discover_args = _build_discover_parser().parse_args(argv[idx + 1 :]) diff --git a/src/manus_agent/tools/cluster_variants.py b/src/manus_agent/tools/cluster_variants.py new file mode 100644 index 0000000..5b5bac7 --- /dev/null +++ b/src/manus_agent/tools/cluster_variants.py @@ -0,0 +1,474 @@ +#!/usr/bin/env python3 +""" +CVE variant clustering tool. + +Groups CVEs related to an input CVE across three cluster dimensions: +1. Same component/vendor (CPE match) +2. Same CWE weakness class +3. Same researcher/disclosure source + +Useful for finding the full attack surface when one CVE is confirmed exploited. +""" + +from __future__ import annotations + +import re +from typing import Any + +import requests +from strands.types.tools import ToolResult, ToolUse + +from manus_agent.tools.get_nvd_data import _nvd_get_with_retry +from manus_agent.tools.tool_output_logger import log_tool_output_size + +TOOL_SPEC = { + "name": "cluster_variants", + "description": ( + "Groups CVEs related to an input CVE across three cluster dimensions: " + "same component/vendor (CPE match), same CWE weakness class, and same " + "researcher/disclosure source. Useful for finding the full attack surface " + "when one CVE is confirmed exploited." + ), + "inputSchema": { + "json": { + "type": "object", + "properties": { + "cve_id": { + "type": "string", + "description": "The CVE identifier to cluster around (e.g., 'CVE-2021-44228').", + } + }, + "required": ["cve_id"], + } + }, +} + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- +_NVD_BASE = "https://services.nvd.nist.gov/rest/json/cves/2.0" +_MAX_RESULTS_PER_CLUSTER = 20 +_REQUEST_TIMEOUT = 15 + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _extract_cpe_info(vuln_data: dict) -> list[dict[str, str]]: + """Extract vendor/product pairs from CPE configurations.""" + cpe_pairs: list[dict[str, str]] = [] + configurations = vuln_data.get("cve", {}).get("configurations", []) + for config in configurations: + for node in config.get("nodes", []): + for match in node.get("cpeMatch", []): + cpe_uri = match.get("criteria", "") + # CPE 2.3 format: cpe:2.3:part:vendor:product:version:... + parts = cpe_uri.split(":") + if len(parts) >= 5: + vendor = parts[3] + product = parts[4] + if vendor != "*" and product != "*": + pair = {"vendor": vendor, "product": product} + if pair not in cpe_pairs: + cpe_pairs.append(pair) + return cpe_pairs + + +def _extract_cwes(vuln_data: dict) -> list[str]: + """Extract CWE IDs from the vulnerability data.""" + cwes: list[str] = [] + weaknesses = vuln_data.get("cve", {}).get("weaknesses", []) + for weakness in weaknesses: + for desc in weakness.get("description", []): + cwe_id = desc.get("value", "") + if cwe_id.startswith("CWE-") and cwe_id != "CWE-noinfo" and cwe_id not in cwes: + cwes.append(cwe_id) + return cwes + + +def _extract_sources(vuln_data: dict) -> list[str]: + """Extract researcher/disclosure source organizations from references.""" + sources: list[str] = [] + references = vuln_data.get("cve", {}).get("references", []) + for ref in references: + source = ref.get("source", "") + if source and source not in sources: + sources.append(source) + return sources + + +def _extract_source_domains(vuln_data: dict) -> list[str]: + """Extract unique domains from reference URLs as disclosure sources.""" + domains: list[str] = [] + references = vuln_data.get("cve", {}).get("references", []) + for ref in references: + url = ref.get("url", "") + match = re.match(r"https?://([^/]+)", url) + if match: + domain = match.group(1).lower() + # Skip generic domains + if ( + domain + not in ( + "nvd.nist.gov", + "cve.org", + "cve.mitre.org", + "www.cve.org", + "web.nvd.nist.gov", + ) + and domain not in domains + ): + domains.append(domain) + return domains + + +def _extract_cvss_score(vuln_data: dict) -> float | None: + """Extract the best available CVSS base score.""" + metrics = vuln_data.get("cve", {}).get("metrics", {}) + # Try CVSS 3.1 first, then 3.0, then 2.0 + for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"): + metric_list = metrics.get(key, []) + if metric_list: + return metric_list[0].get("cvssData", {}).get("baseScore") + return None + + +def _extract_description(vuln_data: dict) -> str: + """Extract the English description.""" + descriptions = vuln_data.get("cve", {}).get("descriptions", []) + for desc in descriptions: + if desc.get("lang") == "en": + return desc.get("value", "") + if descriptions: + return descriptions[0].get("value", "") + return "" + + +def _fetch_cves_by_cpe(vendor: str, product: str, exclude_cve: str) -> list[dict]: + """Fetch CVEs affecting the same vendor/product via NVD keyword search.""" + # Use cpeName parameter for precise matching + cpe_name = f"cpe:2.3:*:{vendor}:{product}:*:*:*:*:*:*:*:*" + url = f"{_NVD_BASE}?cpeName={cpe_name}&resultsPerPage={_MAX_RESULTS_PER_CLUSTER}" + try: + resp = _nvd_get_with_retry(url, timeout=_REQUEST_TIMEOUT) + data = resp.json() + results = [] + for vuln in data.get("vulnerabilities", []): + cve_id = vuln.get("cve", {}).get("id", "") + if cve_id and cve_id.upper() != exclude_cve.upper(): + results.append(vuln) + return results[:_MAX_RESULTS_PER_CLUSTER] + except (requests.exceptions.RequestException, ValueError): + return [] + + +def _fetch_cves_by_keyword(keyword: str, exclude_cve: str) -> list[dict]: + """Fetch CVEs by keyword search (fallback for component matching).""" + url = f"{_NVD_BASE}?keywordSearch={keyword}&resultsPerPage={_MAX_RESULTS_PER_CLUSTER}" + try: + resp = _nvd_get_with_retry(url, timeout=_REQUEST_TIMEOUT) + data = resp.json() + results = [] + for vuln in data.get("vulnerabilities", []): + cve_id = vuln.get("cve", {}).get("id", "") + if cve_id and cve_id.upper() != exclude_cve.upper(): + results.append(vuln) + return results[:_MAX_RESULTS_PER_CLUSTER] + except (requests.exceptions.RequestException, ValueError): + return [] + + +def _fetch_cves_by_cwe(cwe_id: str, exclude_cve: str) -> list[dict]: + """Fetch CVEs sharing the same CWE weakness.""" + url = f"{_NVD_BASE}?cweId={cwe_id}&resultsPerPage={_MAX_RESULTS_PER_CLUSTER}" + try: + resp = _nvd_get_with_retry(url, timeout=_REQUEST_TIMEOUT) + data = resp.json() + results = [] + for vuln in data.get("vulnerabilities", []): + cve_id_str = vuln.get("cve", {}).get("id", "") + if cve_id_str and cve_id_str.upper() != exclude_cve.upper(): + results.append(vuln) + return results[:_MAX_RESULTS_PER_CLUSTER] + except (requests.exceptions.RequestException, ValueError): + return [] + + +def _fetch_cves_by_source(source_domain: str, exclude_cve: str) -> list[dict]: + """Fetch CVEs sharing the same disclosure source via keyword search on source domain.""" + url = f"{_NVD_BASE}?sourceIdentifier={source_domain}&resultsPerPage={_MAX_RESULTS_PER_CLUSTER}" + try: + resp = _nvd_get_with_retry(url, timeout=_REQUEST_TIMEOUT) + data = resp.json() + results = [] + for vuln in data.get("vulnerabilities", []): + cve_id_str = vuln.get("cve", {}).get("id", "") + if cve_id_str and cve_id_str.upper() != exclude_cve.upper(): + results.append(vuln) + return results[:_MAX_RESULTS_PER_CLUSTER] + except (requests.exceptions.RequestException, ValueError): + return [] + + +def _summarize_cve(vuln: dict) -> dict[str, Any]: + """Create a compact summary of a CVE for cluster output.""" + cve_data = vuln.get("cve", {}) + cve_id = cve_data.get("id", "unknown") + desc = _extract_description(vuln) + score = _extract_cvss_score(vuln) + cwes = _extract_cwes(vuln) + published = cve_data.get("published", "")[:10] # YYYY-MM-DD + + return { + "cve_id": cve_id, + "description": desc[:200] + ("..." if len(desc) > 200 else ""), + "cvss_score": score, + "cwes": cwes, + "published": published, + } + + +# --------------------------------------------------------------------------- +# Main clustering logic +# --------------------------------------------------------------------------- + + +def cluster_variants(cve_id: str) -> dict[str, Any]: + """ + Cluster CVEs related to the given CVE across three dimensions. + + Returns a dict with: + - input_cve: info about the queried CVE + - clusters: dict with component, cwe, source keys + - summary: counts per cluster + """ + cve_id = cve_id.strip().upper() + + # Fetch the seed CVE + url = f"{_NVD_BASE}?cveId={cve_id}" + try: + resp = _nvd_get_with_retry(url, timeout=_REQUEST_TIMEOUT) + seed_data = resp.json() + except requests.exceptions.RequestException as exc: + return {"error": f"Failed to fetch NVD data for {cve_id}: {exc}"} + + vulns = seed_data.get("vulnerabilities", []) + if not vulns: + return {"error": f"No vulnerability data found for {cve_id}"} + + seed_vuln = vulns[0] + cpe_pairs = _extract_cpe_info(seed_vuln) + cwes = _extract_cwes(seed_vuln) + sources = _extract_sources(seed_vuln) + + # Seed CVE summary + input_cve_info = { + "cve_id": cve_id, + "description": _extract_description(seed_vuln), + "cvss_score": _extract_cvss_score(seed_vuln), + "cwes": cwes, + "cpe_vendors": [f"{p['vendor']}/{p['product']}" for p in cpe_pairs], + "sources": sources, + } + + # --- Cluster 1: Same Component/Vendor --- + component_cluster: list[dict] = [] + seen_cves: set[str] = set() + for pair in cpe_pairs[:3]: # Limit API calls + related = _fetch_cves_by_cpe(pair["vendor"], pair["product"], cve_id) + for vuln in related: + vid = vuln.get("cve", {}).get("id", "") + if vid not in seen_cves: + seen_cves.add(vid) + component_cluster.append(_summarize_cve(vuln)) + # Fallback: keyword search on product name if no CPE hits + if not component_cluster and cpe_pairs: + product_name = cpe_pairs[0]["product"].replace("_", " ") + related = _fetch_cves_by_keyword(product_name, cve_id) + for vuln in related: + vid = vuln.get("cve", {}).get("id", "") + if vid not in seen_cves: + seen_cves.add(vid) + component_cluster.append(_summarize_cve(vuln)) + + # --- Cluster 2: Same CWE Weakness Class --- + cwe_cluster: list[dict] = [] + seen_cves_cwe: set[str] = set() + for cwe in cwes[:2]: # Limit API calls + related = _fetch_cves_by_cwe(cwe, cve_id) + for vuln in related: + vid = vuln.get("cve", {}).get("id", "") + if vid not in seen_cves_cwe: + seen_cves_cwe.add(vid) + cwe_cluster.append(_summarize_cve(vuln)) + + # --- Cluster 3: Same Source/Researcher --- + source_cluster: list[dict] = [] + seen_cves_src: set[str] = set() + # Use sourceIdentifier (the reporting CNA email/org) + for source in sources[:2]: # Limit API calls + related = _fetch_cves_by_source(source, cve_id) + for vuln in related: + vid = vuln.get("cve", {}).get("id", "") + if vid not in seen_cves_src: + seen_cves_src.add(vid) + source_cluster.append(_summarize_cve(vuln)) + + clusters = { + "component": { + "dimension": "Same Component/Vendor", + "criteria": [f"{p['vendor']}/{p['product']}" for p in cpe_pairs[:3]], + "cves": component_cluster[:_MAX_RESULTS_PER_CLUSTER], + }, + "cwe": { + "dimension": "Same CWE Weakness Class", + "criteria": cwes[:2], + "cves": cwe_cluster[:_MAX_RESULTS_PER_CLUSTER], + }, + "source": { + "dimension": "Same Researcher/Disclosure Source", + "criteria": sources[:2], + "cves": source_cluster[:_MAX_RESULTS_PER_CLUSTER], + }, + } + + summary = { + "component_count": len(clusters["component"]["cves"]), + "cwe_count": len(clusters["cwe"]["cves"]), + "source_count": len(clusters["source"]["cves"]), + "total_unique": len({c["cve_id"] for cluster in clusters.values() for c in cluster["cves"]}), + } + + return { + "input_cve": input_cve_info, + "clusters": clusters, + "summary": summary, + } + + +# --------------------------------------------------------------------------- +# Text rendering +# --------------------------------------------------------------------------- + + +def _render_text(result: dict[str, Any]) -> str: + """Render cluster result as human-readable text.""" + if "error" in result: + return f"❌ {result['error']}" + + lines: list[str] = [] + input_cve = result["input_cve"] + lines.append(f"╔══ CVE Variant Clustering: {input_cve['cve_id']} ══╗") + lines.append("") + + if input_cve.get("description"): + desc = input_cve["description"] + if len(desc) > 120: + desc = desc[:117] + "..." + lines.append(f" {desc}") + lines.append("") + + if input_cve.get("cvss_score"): + lines.append(f" CVSS: {input_cve['cvss_score']}") + if input_cve.get("cwes"): + lines.append(f" CWEs: {', '.join(input_cve['cwes'])}") + if input_cve.get("cpe_vendors"): + lines.append(f" Components: {', '.join(input_cve['cpe_vendors'][:5])}") + if input_cve.get("sources"): + lines.append(f" Sources: {', '.join(input_cve['sources'][:3])}") + lines.append("") + + summary = result["summary"] + lines.append(f" 📊 Found {summary['total_unique']} unique related CVEs across 3 dimensions") + lines.append("") + + clusters = result["clusters"] + + # Component cluster + comp = clusters["component"] + lines.append(f"┌── Cluster 1: {comp['dimension']} ({len(comp['cves'])} CVEs) ──┐") + if comp["criteria"]: + lines.append(f" Criteria: {', '.join(comp['criteria'])}") + lines.append("") + for cve in comp["cves"][:10]: + score_str = f"CVSS {cve['cvss_score']}" if cve.get("cvss_score") else "no score" + lines.append(f" • {cve['cve_id']} [{score_str}] {cve.get('published', '')}") + if cve.get("description"): + lines.append(f" {cve['description'][:100]}") + if len(comp["cves"]) > 10: + lines.append(f" ... and {len(comp['cves']) - 10} more") + lines.append("") + + # CWE cluster + cwe = clusters["cwe"] + lines.append(f"┌── Cluster 2: {cwe['dimension']} ({len(cwe['cves'])} CVEs) ──┐") + if cwe["criteria"]: + lines.append(f" Criteria: {', '.join(cwe['criteria'])}") + lines.append("") + for cve in cwe["cves"][:10]: + score_str = f"CVSS {cve['cvss_score']}" if cve.get("cvss_score") else "no score" + lines.append(f" • {cve['cve_id']} [{score_str}] {cve.get('published', '')}") + if cve.get("description"): + lines.append(f" {cve['description'][:100]}") + if len(cwe["cves"]) > 10: + lines.append(f" ... and {len(cwe['cves']) - 10} more") + lines.append("") + + # Source cluster + src = clusters["source"] + lines.append(f"┌── Cluster 3: {src['dimension']} ({len(src['cves'])} CVEs) ──┐") + if src["criteria"]: + lines.append(f" Criteria: {', '.join(src['criteria'])}") + lines.append("") + for cve in src["cves"][:10]: + score_str = f"CVSS {cve['cvss_score']}" if cve.get("cvss_score") else "no score" + lines.append(f" • {cve['cve_id']} [{score_str}] {cve.get('published', '')}") + if cve.get("description"): + lines.append(f" {cve['description'][:100]}") + if len(src["cves"]) > 10: + lines.append(f" ... and {len(src['cves']) - 10} more") + lines.append("") + lines.append("╚══════════════════════════════════════════════════════════╝") + + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Strands tool entry point +# --------------------------------------------------------------------------- + + +def cluster_variants_tool(tool: ToolUse, **kwargs: Any) -> ToolResult: + """Strands SDK entry point for cluster_variants.""" + tool_use_id = tool["toolUseId"] + tool_input = tool["input"] + cve_id = tool_input.get("cve_id", "") + + if not isinstance(cve_id, str) or not cve_id.upper().startswith("CVE-"): + result: ToolResult = { + "toolUseId": tool_use_id, + "status": "error", + "content": [{"text": "Invalid CVE ID format. Must be like 'CVE-2021-44228'."}], + } + log_tool_output_size("cluster_variants", result) + return result + + data = cluster_variants(cve_id) + + if "error" in data: + result = { + "toolUseId": tool_use_id, + "status": "error", + "content": [{"text": data["error"]}], + } + else: + result = { + "toolUseId": tool_use_id, + "status": "success", + "content": [{"json": data}], + } + + log_tool_output_size("cluster_variants", result) + return result diff --git a/tests/test_cli_cluster_variants.py b/tests/test_cli_cluster_variants.py new file mode 100644 index 0000000..ac8d7da --- /dev/null +++ b/tests/test_cli_cluster_variants.py @@ -0,0 +1,667 @@ +"""Comprehensive tests for cluster_variants tool and CLI subcommand.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +from manus_agent.tools.cluster_variants import ( + _extract_cpe_info, + _extract_cvss_score, + _extract_cwes, + _extract_description, + _extract_source_domains, + _extract_sources, + _fetch_cves_by_cpe, + _fetch_cves_by_cwe, + _fetch_cves_by_keyword, + _fetch_cves_by_source, + _render_text, + _summarize_cve, + cluster_variants, + cluster_variants_tool, +) + +# --------------------------------------------------------------------------- +# Fixtures / sample data +# --------------------------------------------------------------------------- + +SAMPLE_VULN = { + "cve": { + "id": "CVE-2021-44228", + "descriptions": [{"lang": "en", "value": "Apache Log4j2 JNDI features used in configuration allow RCE."}], + "configurations": [ + { + "nodes": [ + { + "cpeMatch": [ + { + "criteria": "cpe:2.3:a:apache:log4j:2.0:*:*:*:*:*:*:*", + "vulnerable": True, + }, + { + "criteria": "cpe:2.3:a:apache:log4j:2.14.1:*:*:*:*:*:*:*", + "vulnerable": True, + }, + ] + } + ] + } + ], + "weaknesses": [ + { + "description": [ + {"lang": "en", "value": "CWE-917"}, + {"lang": "en", "value": "CWE-502"}, + ] + } + ], + "references": [ + {"url": "https://logging.apache.org/log4j/2.x/security.html", "source": "security@apache.org"}, + {"url": "https://github.com/advisories/GHSA-jfh8-c2jp-5v3q", "source": "security@apache.org"}, + ], + "metrics": { + "cvssMetricV31": [ + {"cvssData": {"baseScore": 10.0, "vectorString": "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H"}} + ] + }, + "published": "2021-12-10T10:15:00.000", + } +} + +SAMPLE_RELATED_VULN = { + "cve": { + "id": "CVE-2021-45046", + "descriptions": [{"lang": "en", "value": "Apache Log4j2 Thread Context DoS."}], + "configurations": [], + "weaknesses": [{"description": [{"lang": "en", "value": "CWE-917"}]}], + "references": [{"url": "https://logging.apache.org/", "source": "security@apache.org"}], + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 9.0}}]}, + "published": "2021-12-14T00:00:00.000", + } +} + +SAMPLE_RELATED_VULN_2 = { + "cve": { + "id": "CVE-2021-45105", + "descriptions": [{"lang": "en", "value": "Apache Log4j2 does not protect from uncontrolled recursion."}], + "configurations": [], + "weaknesses": [{"description": [{"lang": "en", "value": "CWE-674"}]}], + "references": [], + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 7.5}}]}, + "published": "2021-12-18T00:00:00.000", + } +} + + +# --------------------------------------------------------------------------- +# Unit tests: extraction helpers +# --------------------------------------------------------------------------- + + +class TestExtractCpeInfo: + def test_extracts_vendor_product(self): + pairs = _extract_cpe_info(SAMPLE_VULN) + assert len(pairs) == 1 # deduplicated (same vendor/product) + assert pairs[0] == {"vendor": "apache", "product": "log4j"} + + def test_skips_wildcard_entries(self): + vuln = { + "cve": {"configurations": [{"nodes": [{"cpeMatch": [{"criteria": "cpe:2.3:a:*:*:1.0:*:*:*:*:*:*:*"}]}]}]} + } + assert _extract_cpe_info(vuln) == [] + + def test_empty_configurations(self): + vuln = {"cve": {"configurations": []}} + assert _extract_cpe_info(vuln) == [] + + def test_no_configurations_key(self): + vuln = {"cve": {}} + assert _extract_cpe_info(vuln) == [] + + def test_multiple_products(self): + vuln = { + "cve": { + "configurations": [ + { + "nodes": [ + { + "cpeMatch": [ + {"criteria": "cpe:2.3:a:vendor1:product1:1.0:*:*:*:*:*:*:*"}, + {"criteria": "cpe:2.3:a:vendor2:product2:2.0:*:*:*:*:*:*:*"}, + ] + } + ] + } + ] + } + } + pairs = _extract_cpe_info(vuln) + assert len(pairs) == 2 + assert {"vendor": "vendor1", "product": "product1"} in pairs + assert {"vendor": "vendor2", "product": "product2"} in pairs + + +class TestExtractCwes: + def test_extracts_cwes(self): + cwes = _extract_cwes(SAMPLE_VULN) + assert "CWE-917" in cwes + assert "CWE-502" in cwes + + def test_skips_noinfo(self): + vuln = {"cve": {"weaknesses": [{"description": [{"lang": "en", "value": "CWE-noinfo"}]}]}} + assert _extract_cwes(vuln) == [] + + def test_empty_weaknesses(self): + vuln = {"cve": {"weaknesses": []}} + assert _extract_cwes(vuln) == [] + + def test_no_weaknesses_key(self): + vuln = {"cve": {}} + assert _extract_cwes(vuln) == [] + + +class TestExtractSources: + def test_extracts_source_orgs(self): + sources = _extract_sources(SAMPLE_VULN) + assert "security@apache.org" in sources + + def test_deduplicates(self): + sources = _extract_sources(SAMPLE_VULN) + # Same source appears twice in refs but should be deduplicated + assert sources.count("security@apache.org") == 1 + + def test_empty_references(self): + vuln = {"cve": {"references": []}} + assert _extract_sources(vuln) == [] + + +class TestExtractSourceDomains: + def test_extracts_domains(self): + domains = _extract_source_domains(SAMPLE_VULN) + assert "logging.apache.org" in domains + assert "github.com" in domains + + def test_skips_nvd_domains(self): + vuln = { + "cve": { + "references": [ + {"url": "https://nvd.nist.gov/vuln/detail/CVE-2021-44228"}, + {"url": "https://cve.mitre.org/something"}, + ] + } + } + domains = _extract_source_domains(vuln) + assert "nvd.nist.gov" not in domains + assert "cve.mitre.org" not in domains + + +class TestExtractCvssScore: + def test_cvss31(self): + assert _extract_cvss_score(SAMPLE_VULN) == 10.0 + + def test_cvss30_fallback(self): + vuln = {"cve": {"metrics": {"cvssMetricV30": [{"cvssData": {"baseScore": 8.5}}]}}} + assert _extract_cvss_score(vuln) == 8.5 + + def test_cvss2_fallback(self): + vuln = {"cve": {"metrics": {"cvssMetricV2": [{"cvssData": {"baseScore": 7.0}}]}}} + assert _extract_cvss_score(vuln) == 7.0 + + def test_no_metrics(self): + vuln = {"cve": {"metrics": {}}} + assert _extract_cvss_score(vuln) is None + + def test_no_metrics_key(self): + vuln = {"cve": {}} + assert _extract_cvss_score(vuln) is None + + +class TestExtractDescription: + def test_english_description(self): + desc = _extract_description(SAMPLE_VULN) + assert "Log4j2" in desc + + def test_fallback_to_first(self): + vuln = {"cve": {"descriptions": [{"lang": "fr", "value": "Description en français"}]}} + assert _extract_description(vuln) == "Description en français" + + def test_empty_descriptions(self): + vuln = {"cve": {"descriptions": []}} + assert _extract_description(vuln) == "" + + +class TestSummarizeCve: + def test_summarizes_fields(self): + summary = _summarize_cve(SAMPLE_VULN) + assert summary["cve_id"] == "CVE-2021-44228" + assert summary["cvss_score"] == 10.0 + assert "CWE-917" in summary["cwes"] + assert summary["published"] == "2021-12-10" + + def test_truncates_long_description(self): + vuln = { + "cve": { + "id": "CVE-TEST-0001", + "descriptions": [{"lang": "en", "value": "A" * 300}], + "weaknesses": [], + "metrics": {}, + "published": "2024-01-01", + } + } + summary = _summarize_cve(vuln) + assert len(summary["description"]) <= 203 # 200 + "..." + + +# --------------------------------------------------------------------------- +# Unit tests: fetch helpers (mocked HTTP) +# --------------------------------------------------------------------------- + + +class TestFetchCvesByCpe: + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_returns_related_cves(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_RELATED_VULN]} + mock_get.return_value = mock_resp + + results = _fetch_cves_by_cpe("apache", "log4j", "CVE-2021-44228") + assert len(results) == 1 + assert results[0]["cve"]["id"] == "CVE-2021-45046" + + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_excludes_input_cve(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_VULN]} + mock_get.return_value = mock_resp + + results = _fetch_cves_by_cpe("apache", "log4j", "CVE-2021-44228") + assert len(results) == 0 + + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_handles_request_exception(self, mock_get): + import requests + + mock_get.side_effect = requests.exceptions.Timeout("timeout") + results = _fetch_cves_by_cpe("apache", "log4j", "CVE-2021-44228") + assert results == [] + + +class TestFetchCvesByKeyword: + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_returns_results(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_RELATED_VULN]} + mock_get.return_value = mock_resp + + results = _fetch_cves_by_keyword("log4j", "CVE-2021-44228") + assert len(results) == 1 + + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_handles_json_error(self, mock_get): + mock_get.side_effect = ValueError("bad json") + results = _fetch_cves_by_keyword("log4j", "CVE-2021-44228") + assert results == [] + + +class TestFetchCvesByCwe: + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_returns_results(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_RELATED_VULN, SAMPLE_RELATED_VULN_2]} + mock_get.return_value = mock_resp + + results = _fetch_cves_by_cwe("CWE-917", "CVE-2021-44228") + assert len(results) == 2 + + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_excludes_input(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_VULN]} + mock_get.return_value = mock_resp + + results = _fetch_cves_by_cwe("CWE-917", "CVE-2021-44228") + assert len(results) == 0 + + +class TestFetchCvesBySource: + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_returns_results(self, mock_get): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_RELATED_VULN]} + mock_get.return_value = mock_resp + + results = _fetch_cves_by_source("security@apache.org", "CVE-2021-44228") + assert len(results) == 1 + + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_handles_exception(self, mock_get): + import requests + + mock_get.side_effect = requests.exceptions.ConnectionError("refused") + results = _fetch_cves_by_source("security@apache.org", "CVE-2021-44228") + assert results == [] + + +# --------------------------------------------------------------------------- +# Integration tests: cluster_variants main function +# --------------------------------------------------------------------------- + + +class TestClusterVariants: + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_source") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cwe") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cpe") + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_full_clustering(self, mock_nvd, mock_cpe, mock_cwe, mock_src): + # Seed CVE fetch + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_VULN]} + mock_nvd.return_value = mock_resp + + # Cluster fetches + mock_cpe.return_value = [SAMPLE_RELATED_VULN] + mock_cwe.return_value = [SAMPLE_RELATED_VULN_2] + mock_src.return_value = [SAMPLE_RELATED_VULN] + + result = cluster_variants("CVE-2021-44228") + + assert "error" not in result + assert result["input_cve"]["cve_id"] == "CVE-2021-44228" + assert result["input_cve"]["cvss_score"] == 10.0 + assert "CWE-917" in result["input_cve"]["cwes"] + assert "apache/log4j" in result["input_cve"]["cpe_vendors"] + + # Clusters populated + assert len(result["clusters"]["component"]["cves"]) == 1 + assert len(result["clusters"]["cwe"]["cves"]) == 1 + assert len(result["clusters"]["source"]["cves"]) == 1 + + # Summary + assert result["summary"]["total_unique"] >= 1 + + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_cve_not_found(self, mock_nvd): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_nvd.return_value = mock_resp + + result = cluster_variants("CVE-9999-99999") + assert "error" in result + assert "No vulnerability data" in result["error"] + + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_network_error(self, mock_nvd): + import requests + + mock_nvd.side_effect = requests.exceptions.Timeout("timeout") + + result = cluster_variants("CVE-2021-44228") + assert "error" in result + assert "Failed to fetch" in result["error"] + + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_source") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cwe") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cpe") + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_handles_cve_with_no_cpe(self, mock_nvd, mock_cpe, mock_cwe, mock_src): + vuln_no_cpe = { + "cve": { + "id": "CVE-2024-0001", + "descriptions": [{"lang": "en", "value": "Test vuln no CPE."}], + "configurations": [], + "weaknesses": [{"description": [{"lang": "en", "value": "CWE-79"}]}], + "references": [{"url": "https://example.com", "source": "test@example.com"}], + "metrics": {}, + "published": "2024-01-01", + } + } + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [vuln_no_cpe]} + mock_nvd.return_value = mock_resp + + mock_cpe.return_value = [] + mock_cwe.return_value = [] + mock_src.return_value = [] + + result = cluster_variants("CVE-2024-0001") + assert "error" not in result + assert result["clusters"]["component"]["cves"] == [] + + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_source") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cwe") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cpe") + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_lowercase_cve_id_normalized(self, mock_nvd, mock_cpe, mock_cwe, mock_src): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_VULN]} + mock_nvd.return_value = mock_resp + mock_cpe.return_value = [] + mock_cwe.return_value = [] + mock_src.return_value = [] + + result = cluster_variants("cve-2021-44228") + assert result["input_cve"]["cve_id"] == "CVE-2021-44228" + + +# --------------------------------------------------------------------------- +# Tests: text rendering +# --------------------------------------------------------------------------- + + +class TestRenderText: + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_source") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cwe") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cpe") + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_renders_full_output(self, mock_nvd, mock_cpe, mock_cwe, mock_src): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_VULN]} + mock_nvd.return_value = mock_resp + mock_cpe.return_value = [SAMPLE_RELATED_VULN] + mock_cwe.return_value = [SAMPLE_RELATED_VULN_2] + mock_src.return_value = [] + + result = cluster_variants("CVE-2021-44228") + text = _render_text(result) + + assert "CVE-2021-44228" in text + assert "Cluster 1" in text + assert "Cluster 2" in text + assert "Cluster 3" in text + assert "CVE-2021-45046" in text + assert "CVE-2021-45105" in text + + def test_renders_error(self): + text = _render_text({"error": "Something went wrong"}) + assert "❌" in text + assert "Something went wrong" in text + + +# --------------------------------------------------------------------------- +# Tests: Strands tool entry point +# --------------------------------------------------------------------------- + + +class TestClusterVariantsTool: + @patch("manus_agent.tools.cluster_variants.cluster_variants") + def test_success(self, mock_cluster): + mock_cluster.return_value = { + "input_cve": {"cve_id": "CVE-2021-44228"}, + "clusters": {"component": {"cves": []}, "cwe": {"cves": []}, "source": {"cves": []}}, + "summary": {"total_unique": 0}, + } + + tool_use = {"toolUseId": "test-123", "input": {"cve_id": "CVE-2021-44228"}} + result = cluster_variants_tool(tool_use) + + assert result["status"] == "success" + assert result["toolUseId"] == "test-123" + + @patch("manus_agent.tools.cluster_variants.cluster_variants") + def test_error_result(self, mock_cluster): + mock_cluster.return_value = {"error": "CVE not found"} + + tool_use = {"toolUseId": "test-456", "input": {"cve_id": "CVE-9999-99999"}} + result = cluster_variants_tool(tool_use) + + assert result["status"] == "error" + assert "CVE not found" in result["content"][0]["text"] + + def test_invalid_cve_format(self): + tool_use = {"toolUseId": "test-789", "input": {"cve_id": "not-a-cve"}} + result = cluster_variants_tool(tool_use) + + assert result["status"] == "error" + assert "Invalid CVE ID" in result["content"][0]["text"] + + def test_empty_cve_id(self): + tool_use = {"toolUseId": "test-000", "input": {"cve_id": ""}} + result = cluster_variants_tool(tool_use) + + assert result["status"] == "error" + + +# --------------------------------------------------------------------------- +# Tests: CLI subcommand +# --------------------------------------------------------------------------- + + +class TestCliClusterVariants: + @patch("manus_agent.tools.cluster_variants.cluster_variants") + def test_text_output(self, mock_cluster, capsys): + mock_cluster.return_value = { + "input_cve": { + "cve_id": "CVE-2021-44228", + "description": "Log4Shell", + "cvss_score": 10.0, + "cwes": ["CWE-917"], + "cpe_vendors": ["apache/log4j"], + "sources": ["security@apache.org"], + }, + "clusters": { + "component": {"dimension": "Same Component/Vendor", "criteria": ["apache/log4j"], "cves": []}, + "cwe": {"dimension": "Same CWE Weakness Class", "criteria": ["CWE-917"], "cves": []}, + "source": { + "dimension": "Same Researcher/Disclosure Source", + "criteria": ["security@apache.org"], + "cves": [], + }, + }, + "summary": {"component_count": 0, "cwe_count": 0, "source_count": 0, "total_unique": 0}, + } + + from manus_agent.cli import _run_cluster_variants + + exit_code = _run_cluster_variants(["CVE-2021-44228"]) + assert exit_code == 0 + + captured = capsys.readouterr() + assert "CVE-2021-44228" in captured.out + + @patch("manus_agent.tools.cluster_variants.cluster_variants") + def test_json_output(self, mock_cluster, capsys): + mock_cluster.return_value = { + "input_cve": {"cve_id": "CVE-2021-44228"}, + "clusters": { + "component": {"dimension": "Same Component/Vendor", "criteria": [], "cves": []}, + "cwe": {"dimension": "Same CWE Weakness Class", "criteria": [], "cves": []}, + "source": {"dimension": "Same Researcher/Disclosure Source", "criteria": [], "cves": []}, + }, + "summary": {"component_count": 0, "cwe_count": 0, "source_count": 0, "total_unique": 0}, + } + + from manus_agent.cli import _run_cluster_variants + + exit_code = _run_cluster_variants(["CVE-2021-44228", "--output", "json"]) + assert exit_code == 0 + + captured = capsys.readouterr() + data = json.loads(captured.out) + assert data["input_cve"]["cve_id"] == "CVE-2021-44228" + + @patch("manus_agent.tools.cluster_variants.cluster_variants") + def test_error_exits_nonzero(self, mock_cluster, capsys): + mock_cluster.return_value = {"error": "NVD unavailable"} + + from manus_agent.cli import _run_cluster_variants + + exit_code = _run_cluster_variants(["CVE-2021-44228"]) + assert exit_code == 1 + + def test_invalid_cve_exits_nonzero(self, capsys): + from manus_agent.cli import _run_cluster_variants + + exit_code = _run_cluster_variants(["not-a-cve"]) + assert exit_code == 1 + + def test_help_flag(self): + from manus_agent.cli import _build_cluster_variants_parser + + parser = _build_cluster_variants_parser() + assert parser.prog == "manus-agent cluster-variants" + + +# --------------------------------------------------------------------------- +# Edge case tests +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_source") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cwe") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cpe") + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_deduplicates_across_cpe_queries(self, mock_nvd, mock_cpe, mock_cwe, mock_src): + """When same CVE appears for multiple CPE vendor/product pairs, it's deduplicated.""" + vuln_multi_cpe = { + "cve": { + "id": "CVE-2024-MULTI", + "descriptions": [{"lang": "en", "value": "Multi CPE vuln"}], + "configurations": [ + { + "nodes": [ + { + "cpeMatch": [ + {"criteria": "cpe:2.3:a:vendorA:productA:1.0:*:*:*:*:*:*:*"}, + {"criteria": "cpe:2.3:a:vendorB:productB:2.0:*:*:*:*:*:*:*"}, + ] + } + ] + } + ], + "weaknesses": [], + "references": [], + "metrics": {}, + "published": "2024-01-01", + } + } + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [vuln_multi_cpe]} + mock_nvd.return_value = mock_resp + + # Same CVE returned for both CPE queries + mock_cpe.return_value = [SAMPLE_RELATED_VULN] + mock_cwe.return_value = [] + mock_src.return_value = [] + + result = cluster_variants("CVE-2024-MULTI") + # CPE fetch called twice (2 pairs), but the same result deduplicated + assert mock_cpe.call_count == 2 + # Only one unique entry in component cluster + assert len(result["clusters"]["component"]["cves"]) == 1 + + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_source") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cwe") + @patch("manus_agent.tools.cluster_variants._fetch_cves_by_cpe") + @patch("manus_agent.tools.cluster_variants._nvd_get_with_retry") + def test_whitespace_cve_id_handled(self, mock_nvd, mock_cpe, mock_cwe, mock_src): + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": [SAMPLE_VULN]} + mock_nvd.return_value = mock_resp + mock_cpe.return_value = [] + mock_cwe.return_value = [] + mock_src.return_value = [] + + result = cluster_variants(" CVE-2021-44228 ") + assert result["input_cve"]["cve_id"] == "CVE-2021-44228"