diff --git a/src/manus_agent/cli.py b/src/manus_agent/cli.py index e8442f2..bb9e1d7 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", + "version-range", } @@ -1935,6 +1936,67 @@ def _run_blast_radius(argv: list[str]) -> int: return 0 +# --------------------------------------------------------------------------- +# version-range subcommand +# --------------------------------------------------------------------------- + + +def _build_version_range_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="manus-agent version-range", + description=( + "Compute affected version ranges for a CVE by walking NVD CPE\n" + "configurations and cross-referencing OSV.dev ecosystem data." + ), + add_help=True, + ) + p.add_argument("cve_id", metavar="CVE-ID", help="CVE identifier, e.g. CVE-2021-44228") + p.add_argument( + "--ecosystem", + choices=["auto", "pypi", "npm", "maven", "go", "rust", "nuget", "rubygems", "packagist"], + default="auto", + help="Filter to a specific ecosystem (default: auto = all ecosystems)", + ) + p.add_argument( + "--output", + choices=["text", "json"], + default="text", + help="Output format (default: text)", + ) + return p + + +def _run_version_range(argv: list[str]) -> int: + import json as _json + + parser = _build_version_range_parser() + args = parser.parse_args(argv) + + try: + from manus_agent.tools.get_version_range import _format_text, fetch_version_range + except ImportError as exc: # pragma: no cover + print(f"Error: {exc}", file=sys.stderr) + return 1 + + cve_id = args.cve_id.strip() + if not cve_id.upper().startswith("CVE-"): + print("Error: CVE ID must start with 'CVE-'", file=sys.stderr) + return 1 + + try: + data = fetch_version_range(cve_id, ecosystem=args.ecosystem) + except Exception as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + + if args.output == "json": + print(_json.dumps(data, indent=2, default=str)) + else: + print(_format_text(data)) + + return 0 + + def _build_run_parser() -> argparse.ArgumentParser: """Build the top-level run/interactive parser.""" parser = argparse.ArgumentParser( @@ -2269,6 +2331,10 @@ def main() -> None: idx = argv.index("blast-radius") sys.exit(_run_blast_radius(argv[idx + 1 :])) + if first_positional == "version-range": + idx = argv.index("version-range") + sys.exit(_run_version_range(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/get_version_range.py b/src/manus_agent/tools/get_version_range.py new file mode 100644 index 0000000..d40902b --- /dev/null +++ b/src/manus_agent/tools/get_version_range.py @@ -0,0 +1,613 @@ +#!/usr/bin/env python3 +""" +Tool for computing affected version ranges for a CVE. + +Walks NVD CPE configurations and cross-references OSV.dev ecosystem data +(PyPI, npm, Maven, Go, RustSec, etc.) to produce structured vulnerable +version ranges, a list of affected releases, and the first patched release. + +Strategy (multi-source, best-effort merging): + +1. **NVD CPE configurations** — the ``configurations`` block in NVD 2.0 + contains CPE Match strings with optional ``versionStartIncluding``, + ``versionEndExcluding``, ``versionEndIncluding`` bounds. These are + vendor/product-level and not ecosystem-specific, but provide the + authoritative range boundaries from the CVE authority. + +2. **OSV.dev affected ranges** — OSV normalises advisories from GHSA, + PyPA/PyPI, npm, Go, RustSec, Maven etc. into concrete + package + version-range tuples with ``introduced`` / ``fixed`` / + ``last_affected`` events. These are the most actionable for + dependency-level triage. + +The tool merges both sources, deduplicates, and optionally filters by +ecosystem. Output includes: affected version ranges, first patched +version, affected package names, and ecosystem information. + +Public APIs used: +- NVD 2.0: GET https://services.nvd.nist.gov/rest/json/cves/2.0?cveId=... +- OSV.dev: GET https://api.osv.dev/v1/vulns/{id} +""" + +from __future__ import annotations + +import os +import time +from typing import Any + +import requests +from strands.types.tools import ToolResult, ToolUse + +from manus_agent.tools.tool_output_logger import log_tool_output_size + +# --------------------------------------------------------------------------- +# Retry / back-off configuration +# --------------------------------------------------------------------------- +_MAX_RETRIES: int = int(os.environ.get("VERSION_RANGE_MAX_RETRIES", "3")) +_RETRY_BASE_DELAY: float = float(os.environ.get("VERSION_RANGE_RETRY_BASE_DELAY", "1.0")) +_RETRYABLE_STATUSES: frozenset[int] = frozenset({429, 500, 502, 503, 504}) +_HTTP_TIMEOUT: int = 15 + +# --------------------------------------------------------------------------- +# API endpoints +# --------------------------------------------------------------------------- +_NVD_CVE_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0" +_OSV_VULN_URL = "https://api.osv.dev/v1/vulns/{osv_id}" + +# Cap on alias follows (same as get_osv_data) +_MAX_ALIAS_FOLLOWS = 8 + +# Ecosystem normalisation mapping +_ECOSYSTEM_ALIASES: dict[str, str] = { + "pypi": "PyPI", + "pip": "PyPI", + "python": "PyPI", + "npm": "npm", + "node": "npm", + "nodejs": "npm", + "maven": "Maven", + "java": "Maven", + "go": "Go", + "golang": "Go", + "rust": "crates.io", + "crates": "crates.io", + "nuget": "NuGet", + "rubygems": "RubyGems", + "gem": "RubyGems", + "packagist": "Packagist", + "php": "Packagist", + "hex": "Hex", + "erlang": "Hex", + "elixir": "Hex", + "pub": "Pub", + "dart": "Pub", + "swift": "SwiftURL", +} + +TOOL_SPEC = { + "name": "get_version_range", + "description": ( + "Computes affected version ranges for a CVE by walking NVD CPE configurations " + "and cross-referencing OSV.dev ecosystem data (PyPI, npm, Maven, Go, RustSec, etc.). " + "Returns structured vulnerable version ranges, affected releases, and the first " + "patched release per package. Use this to answer 'what versions are vulnerable?' " + "and 'what version fixes it?'. Optionally filter by ecosystem." + ), + "inputSchema": { + "json": { + "type": "object", + "properties": { + "cve_id": { + "type": "string", + "description": "The CVE identifier to look up (e.g., 'CVE-2021-44228').", + }, + "ecosystem": { + "type": "string", + "description": ( + "Filter results to a specific ecosystem. Accepts: " + "'auto' (all ecosystems), 'pypi', 'npm', 'maven', 'go', " + "'rust', 'nuget', 'rubygems', 'packagist'. Default: 'auto'." + ), + }, + }, + "required": ["cve_id"], + } + }, +} + + +# --------------------------------------------------------------------------- +# HTTP helpers +# --------------------------------------------------------------------------- + + +def _get_with_retry(url: str, *, headers: dict[str, str] | None = None) -> requests.Response: + """GET with exponential back-off retry on transient errors.""" + hdrs = headers or {} + last_exc: Exception | None = None + + for attempt in range(_MAX_RETRIES + 1): + if attempt > 0: + delay = _RETRY_BASE_DELAY * (2 ** (attempt - 1)) + time.sleep(delay) + try: + resp = requests.get(url, headers=hdrs, timeout=_HTTP_TIMEOUT) + if resp.status_code in _RETRYABLE_STATUSES: + last_exc = requests.exceptions.HTTPError(f"HTTP {resp.status_code}", response=resp) + if attempt < _MAX_RETRIES: + continue + raise last_exc + resp.raise_for_status() + return resp + except requests.exceptions.HTTPError as exc: + if exc.response is not None and exc.response.status_code not in _RETRYABLE_STATUSES: + raise + last_exc = exc + if attempt < _MAX_RETRIES: + continue + raise + except requests.exceptions.RequestException as exc: + last_exc = exc + if attempt < _MAX_RETRIES: + continue + raise + raise last_exc # type: ignore[misc] + + +def _build_nvd_headers() -> dict[str, str]: + """Return NVD request headers, injecting API key when available.""" + headers: dict[str, str] = {"Accept": "application/json"} + api_key = os.environ.get("NVD_API_KEY", "").strip() + if api_key: + headers["apiKey"] = api_key + return headers + + +# --------------------------------------------------------------------------- +# NVD CPE parsing +# --------------------------------------------------------------------------- + + +def _parse_cpe_uri(cpe_uri: str) -> dict[str, str]: + """Parse a CPE 2.3 URI into components. + + CPE format: cpe:2.3:part:vendor:product:version:update:edition:language:... + """ + parts = cpe_uri.split(":") + result: dict[str, str] = {} + if len(parts) >= 4: + result["vendor"] = parts[3] if parts[3] != "*" else "" + if len(parts) >= 5: + result["product"] = parts[4] if parts[4] != "*" else "" + if len(parts) >= 6: + result["version"] = parts[5] if parts[5] != "*" else "" + return result + + +def _extract_cpe_ranges(configurations: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Extract version range constraints from NVD CPE configurations. + + Each NVD configuration node contains CPE match criteria with optional + version bounds (versionStartIncluding, versionEndExcluding, etc.). + """ + ranges: list[dict[str, Any]] = [] + + for config in configurations or []: + nodes = config.get("nodes", []) + for node in nodes: + cpe_matches = node.get("cpeMatch", []) + for match in cpe_matches: + if not isinstance(match, dict): + continue + if not match.get("vulnerable", False): + continue + + criteria = match.get("criteria", "") + cpe_info = _parse_cpe_uri(criteria) + + range_entry: dict[str, Any] = { + "vendor": cpe_info.get("vendor", ""), + "product": cpe_info.get("product", ""), + "cpe_version": cpe_info.get("version", ""), + "version_start_including": match.get("versionStartIncluding"), + "version_start_excluding": match.get("versionStartExcluding"), + "version_end_including": match.get("versionEndIncluding"), + "version_end_excluding": match.get("versionEndExcluding"), + } + + # Build a human-readable range string + range_str = _build_range_string(range_entry) + range_entry["range_string"] = range_str + ranges.append(range_entry) + + return ranges + + +def _build_range_string(entry: dict[str, Any]) -> str: + """Build a human-readable version range string from CPE match bounds.""" + parts: list[str] = [] + + start_inc = entry.get("version_start_including") + start_exc = entry.get("version_start_excluding") + end_inc = entry.get("version_end_including") + end_exc = entry.get("version_end_excluding") + + if start_inc: + parts.append(f">={start_inc}") + elif start_exc: + parts.append(f">{start_exc}") + + if end_exc: + parts.append(f"<{end_exc}") + elif end_inc: + parts.append(f"<={end_inc}") + + if parts: + return ", ".join(parts) + + # Exact version match from CPE + cpe_ver = entry.get("cpe_version", "") + if cpe_ver and cpe_ver != "*": + return f"={cpe_ver}" + + return "all versions (no bounds specified)" + + +# --------------------------------------------------------------------------- +# OSV.dev parsing +# --------------------------------------------------------------------------- + + +def _fetch_osv_record(osv_id: str) -> dict[str, Any] | None: + """Fetch a single OSV record by ID. Returns None on 404 or error.""" + url = _OSV_VULN_URL.format(osv_id=osv_id) + try: + resp = _get_with_retry(url, headers={"Accept": "application/json"}) + return resp.json() + except (requests.exceptions.RequestException, ValueError): + return None + + +def _extract_osv_ranges(cve_id: str) -> list[dict[str, Any]]: + """Fetch OSV data for a CVE and extract per-package version ranges. + + Follows GHSA aliases when the CVE record itself has no package data. + """ + record = _fetch_osv_record(cve_id.upper()) + if record is None: + return [] + + # Collect records to process (CVE record + alias follows) + records_to_process: list[dict[str, Any]] = [record] + seen_ids: set[str] = {record.get("id", "")} + + # If CVE record has no affected data, follow GHSA aliases + if not record.get("affected"): + aliases = record.get("aliases", []) or [] + ghsa_aliases = [a for a in aliases if isinstance(a, str) and a.startswith("GHSA-")] + for alias in ghsa_aliases[:_MAX_ALIAS_FOLLOWS]: + if alias in seen_ids: + continue + alias_record = _fetch_osv_record(alias) + if alias_record and alias_record.get("affected"): + records_to_process.append(alias_record) + seen_ids.add(alias) + + # Parse all affected entries + packages: list[dict[str, Any]] = [] + seen_packages: set[str] = set() + + for rec in records_to_process: + for entry in rec.get("affected", []) or []: + if not isinstance(entry, dict): + continue + pkg = entry.get("package") or {} + ecosystem = pkg.get("ecosystem", "") if isinstance(pkg, dict) else "" + name = pkg.get("name", "") if isinstance(pkg, dict) else "" + + if not ecosystem and not name: + continue + + # Deduplicate by ecosystem+name + dedup_key = f"{ecosystem}:{name}" + if dedup_key in seen_packages: + continue + seen_packages.add(dedup_key) + + introduced: list[str] = [] + fixed: list[str] = [] + last_affected: list[str] = [] + + for rng in entry.get("ranges", []) or []: + if not isinstance(rng, dict): + continue + for ev in rng.get("events", []) or []: + if not isinstance(ev, dict): + continue + if "introduced" in ev: + introduced.append(str(ev["introduced"])) + if "fixed" in ev: + fixed.append(str(ev["fixed"])) + if "last_affected" in ev: + last_affected.append(str(ev["last_affected"])) + + versions = [str(v) for v in (entry.get("versions", []) or []) if v is not None] + + packages.append( + { + "source": "osv", + "ecosystem": ecosystem, + "package": name, + "introduced": introduced, + "fixed": fixed, + "last_affected": last_affected, + "affected_versions": versions[:20], + "affected_version_count": len(versions), + } + ) + + return packages + + +# --------------------------------------------------------------------------- +# NVD fetch +# --------------------------------------------------------------------------- + + +def _fetch_nvd_configurations(cve_id: str) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """Fetch NVD data and return (configurations, summary_info).""" + url = f"{_NVD_CVE_URL}?cveId={cve_id.upper()}" + headers = _build_nvd_headers() + + try: + resp = _get_with_retry(url, headers=headers) + data = resp.json() + except (requests.exceptions.RequestException, ValueError): + return [], {} + + vulns = data.get("vulnerabilities", []) + if not vulns: + return [], {} + + vuln = vulns[0].get("cve", {}) + configurations = vuln.get("configurations", []) + + # Extract CVSS for context + metrics = vuln.get("metrics", {}) + cvss_v31 = metrics.get("cvssMetricV31", [{}]) + cvss_data = cvss_v31[0].get("cvssData", {}) if cvss_v31 else {} + + summary = { + "published": vuln.get("published", ""), + "last_modified": vuln.get("lastModified", ""), + "base_score": cvss_data.get("baseScore"), + "base_severity": cvss_data.get("baseSeverity"), + } + + return configurations, summary + + +# --------------------------------------------------------------------------- +# Ecosystem filtering +# --------------------------------------------------------------------------- + + +def _normalise_ecosystem(eco: str) -> str: + """Normalise user-supplied ecosystem string to OSV canonical form.""" + if not eco: + return "" + lower = eco.lower().strip() + return _ECOSYSTEM_ALIASES.get(lower, eco) + + +def _filter_by_ecosystem(packages: list[dict[str, Any]], ecosystem: str) -> list[dict[str, Any]]: + """Filter package list to only those matching the requested ecosystem.""" + if not ecosystem or ecosystem.lower() == "auto": + return packages + target = _normalise_ecosystem(ecosystem) + return [p for p in packages if p.get("ecosystem", "").lower() == target.lower()] + + +# --------------------------------------------------------------------------- +# Core computation +# --------------------------------------------------------------------------- + + +def fetch_version_range(cve_id: str, ecosystem: str = "auto") -> dict[str, Any]: + """Compute version ranges for a CVE from NVD + OSV sources. + + Returns a structured dictionary with: + - cve_id: the CVE identifier + - nvd_ranges: CPE-based version constraints from NVD + - osv_packages: per-package version ranges from OSV.dev + - first_patched_version: the earliest fixed version found (if any) + - summary: CVSS and date context from NVD + """ + cve_id = cve_id.upper().strip() + + # Fetch from both sources + configurations, nvd_summary = _fetch_nvd_configurations(cve_id) + nvd_ranges = _extract_cpe_ranges(configurations) + osv_packages = _extract_osv_ranges(cve_id) + + # Apply ecosystem filter + filtered_osv = _filter_by_ecosystem(osv_packages, ecosystem) + + # Determine first patched version (prefer OSV fixed versions) + first_patched: str | None = None + all_fixed: list[str] = [] + for pkg in filtered_osv: + all_fixed.extend(pkg.get("fixed", [])) + if all_fixed: + # Take the first fixed version found (OSV usually lists them in order) + first_patched = all_fixed[0] + + # If no OSV fixed version, check NVD versionEndExcluding as a proxy + if not first_patched: + for rng in nvd_ranges: + end_exc = rng.get("version_end_excluding") + if end_exc: + first_patched = end_exc + break + + # Compute affected ecosystems + affected_ecosystems = sorted({p["ecosystem"] for p in filtered_osv if p.get("ecosystem")}) + + # Compute affected products from NVD + affected_products = [] + seen_products: set[str] = set() + for rng in nvd_ranges: + vendor = rng.get("vendor", "") + product = rng.get("product", "") + key = f"{vendor}/{product}" + if key not in seen_products and (vendor or product): + seen_products.add(key) + affected_products.append({"vendor": vendor, "product": product}) + + return { + "cve_id": cve_id, + "nvd_ranges": nvd_ranges, + "osv_packages": filtered_osv, + "first_patched_version": first_patched, + "affected_ecosystems": affected_ecosystems, + "affected_products": affected_products, + "ecosystem_filter": ecosystem if ecosystem.lower() != "auto" else None, + "summary": nvd_summary, + } + + +# --------------------------------------------------------------------------- +# Text formatting +# --------------------------------------------------------------------------- + + +def _format_text(result: dict[str, Any]) -> str: + """Format version range result as human-readable text.""" + lines: list[str] = [] + cve_id = result["cve_id"] + summary = result.get("summary", {}) + + lines.append(f"{'=' * 60}") + lines.append(f" VERSION RANGE ANALYSIS: {cve_id}") + lines.append(f"{'=' * 60}") + + # Summary + if summary: + score = summary.get("base_score") + severity = summary.get("base_severity") + if score: + lines.append(f"\n CVSS: {score} ({severity or 'N/A'})") + pub = summary.get("published", "")[:10] + if pub: + lines.append(f" Published: {pub}") + + # First patched version + first_patched = result.get("first_patched_version") + if first_patched: + lines.append(f"\n ⚡ First Patched Version: {first_patched}") + else: + lines.append("\n ⚠ No patched version identified") + + # Ecosystem filter + eco_filter = result.get("ecosystem_filter") + if eco_filter: + lines.append(f" Ecosystem Filter: {eco_filter}") + + # NVD CPE ranges + nvd_ranges = result.get("nvd_ranges", []) + if nvd_ranges: + lines.append(f"\n{'─' * 60}") + lines.append(" NVD CPE VERSION RANGES") + lines.append(f"{'─' * 60}") + for i, rng in enumerate(nvd_ranges, 1): + vendor = rng.get("vendor", "unknown") + product = rng.get("product", "unknown") + range_str = rng.get("range_string", "N/A") + lines.append(f"\n [{i}] {vendor}/{product}") + lines.append(f" Range: {range_str}") + else: + lines.append(f"\n{'─' * 60}") + lines.append(" NVD CPE VERSION RANGES: None found") + + # OSV packages + osv_packages = result.get("osv_packages", []) + if osv_packages: + lines.append(f"\n{'─' * 60}") + lines.append(" OSV.DEV ECOSYSTEM PACKAGES") + lines.append(f"{'─' * 60}") + for i, pkg in enumerate(osv_packages, 1): + eco = pkg.get("ecosystem", "unknown") + name = pkg.get("package", "unknown") + introduced = pkg.get("introduced", []) + fixed = pkg.get("fixed", []) + last_aff = pkg.get("last_affected", []) + ver_count = pkg.get("affected_version_count", 0) + + lines.append(f"\n [{i}] {eco}: {name}") + if introduced: + lines.append(f" Introduced: {', '.join(introduced)}") + if fixed: + lines.append(f" Fixed: {', '.join(fixed)}") + if last_aff: + lines.append(f" Last Affected: {', '.join(last_aff)}") + if ver_count: + lines.append(f" Affected Versions: {ver_count} known") + sample = pkg.get("affected_versions", []) + if sample: + lines.append(f" Sample: {', '.join(sample[:5])}") + else: + lines.append(f"\n{'─' * 60}") + lines.append(" OSV.DEV ECOSYSTEM PACKAGES: None found") + + # Affected ecosystems summary + ecosystems = result.get("affected_ecosystems", []) + if ecosystems: + lines.append(f"\n{'─' * 60}") + lines.append(f" Affected Ecosystems: {', '.join(ecosystems)}") + + lines.append(f"\n{'=' * 60}") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Strands tool handler +# --------------------------------------------------------------------------- + + +def get_version_range(tool: ToolUse, **kwargs: Any) -> ToolResult: + """Strands tool entry point for get_version_range.""" + 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.strip().upper().startswith("CVE-"): + result: ToolResult = { + "toolUseId": tool_use_id, + "status": "error", + "content": [{"text": "Invalid CVE ID format. Must be like 'CVE-YYYY-NNNN'."}], + } + log_tool_output_size("get_version_range", result) + return result + + ecosystem = tool_input.get("ecosystem", "auto") or "auto" + + try: + data = fetch_version_range(cve_id, ecosystem=ecosystem) + except Exception as exc: + result = { + "toolUseId": tool_use_id, + "status": "error", + "content": [{"text": f"Error computing version ranges: {exc}"}], + } + log_tool_output_size("get_version_range", result) + return result + + text_output = _format_text(data) + + result = { + "toolUseId": tool_use_id, + "status": "success", + "content": [{"text": text_output}], + } + log_tool_output_size("get_version_range", result) + return result diff --git a/tests/test_version_range.py b/tests/test_version_range.py new file mode 100644 index 0000000..461d2cb --- /dev/null +++ b/tests/test_version_range.py @@ -0,0 +1,1329 @@ +#!/usr/bin/env python3 +"""Comprehensive test suite for get_version_range tool and CLI subcommand.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from manus_agent.tools.get_version_range import ( # noqa: E402 + TOOL_SPEC, + _build_nvd_headers, + _build_range_string, + _extract_cpe_ranges, + _extract_osv_ranges, + _filter_by_ecosystem, + _format_text, + _get_with_retry, + _normalise_ecosystem, + _parse_cpe_uri, + fetch_version_range, + get_version_range, +) + + +# --------------------------------------------------------------------------- +# Module import (with retry env overrides for fast tests) +# --------------------------------------------------------------------------- +@pytest.fixture(autouse=True) +def _fast_retry(monkeypatch): + """Disable real HTTP retries/delays in all tests.""" + monkeypatch.setenv("VERSION_RANGE_MAX_RETRIES", "1") + monkeypatch.setenv("VERSION_RANGE_RETRY_BASE_DELAY", "0") + + +# --------------------------------------------------------------------------- +# TOOL_SPEC contract tests +# --------------------------------------------------------------------------- +class TestToolSpec: + def test_has_required_keys(self): + assert "name" in TOOL_SPEC + assert "description" in TOOL_SPEC + assert "inputSchema" in TOOL_SPEC + + def test_name_is_get_version_range(self): + assert TOOL_SPEC["name"] == "get_version_range" + + def test_input_schema_requires_cve_id(self): + schema = TOOL_SPEC["inputSchema"]["json"] + assert "cve_id" in schema["properties"] + assert "cve_id" in schema["required"] + + def test_input_schema_has_ecosystem(self): + schema = TOOL_SPEC["inputSchema"]["json"] + assert "ecosystem" in schema["properties"] + + def test_description_mentions_version_range(self): + assert "version" in TOOL_SPEC["description"].lower() + + +# --------------------------------------------------------------------------- +# Input validation tests +# --------------------------------------------------------------------------- +class TestInputValidation: + def test_invalid_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-1", "input": {"cve_id": "not-a-cve"}} + result = get_version_range(tool_use) + assert result["status"] == "error" + assert "Invalid CVE ID" in result["content"][0]["text"] + + def test_empty_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-2", "input": {"cve_id": ""}} + result = get_version_range(tool_use) + assert result["status"] == "error" + + def test_numeric_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-3", "input": {"cve_id": 12345}} + result = get_version_range(tool_use) + assert result["status"] == "error" + + def test_none_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-4", "input": {"cve_id": None}} + result = get_version_range(tool_use) + assert result["status"] == "error" + + def test_missing_cve_id_key_returns_error(self): + tool_use = {"toolUseId": "test-5", "input": {}} + result = get_version_range(tool_use) + assert result["status"] == "error" + + def test_valid_cve_format_accepted(self): + """A well-formed CVE should not fail on format validation.""" + tool_use = {"toolUseId": "test-6", "input": {"cve_id": "CVE-2021-44228"}} + with patch( + "manus_agent.tools.get_version_range.fetch_version_range", + return_value={ + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + }, + ): + result = get_version_range(tool_use) + assert result["status"] == "success" + + +# --------------------------------------------------------------------------- +# CPE parsing tests +# --------------------------------------------------------------------------- +class TestCpeParsing: + def test_parse_full_cpe_uri(self): + cpe = "cpe:2.3:a:apache:log4j:2.14.1:*:*:*:*:*:*:*" + result = _parse_cpe_uri(cpe) + assert result["vendor"] == "apache" + assert result["product"] == "log4j" + assert result["version"] == "2.14.1" + + def test_parse_wildcard_version(self): + cpe = "cpe:2.3:a:apache:log4j:*:*:*:*:*:*:*:*" + result = _parse_cpe_uri(cpe) + assert result["version"] == "" + + def test_parse_wildcard_vendor(self): + cpe = "cpe:2.3:a:*:product_name:1.0:*:*:*:*:*:*:*" + result = _parse_cpe_uri(cpe) + assert result["vendor"] == "" + assert result["product"] == "product_name" + + def test_parse_short_cpe(self): + cpe = "cpe:2.3:a" + result = _parse_cpe_uri(cpe) + assert "vendor" not in result or result.get("vendor") == "" + + def test_parse_minimum_fields(self): + cpe = "cpe:2.3:a:vendor" + result = _parse_cpe_uri(cpe) + assert result["vendor"] == "vendor" + + def test_parse_cpe_with_underscores(self): + cpe = "cpe:2.3:a:the_vendor:the_product:1.2.3:*:*:*:*:*:*:*" + result = _parse_cpe_uri(cpe) + assert result["vendor"] == "the_vendor" + assert result["product"] == "the_product" + + +# --------------------------------------------------------------------------- +# Range string building tests +# --------------------------------------------------------------------------- +class TestBuildRangeString: + def test_start_including_end_excluding(self): + entry = {"version_start_including": "2.0", "version_end_excluding": "2.15.0"} + assert _build_range_string(entry) == ">=2.0, <2.15.0" + + def test_start_excluding_end_including(self): + entry = {"version_start_excluding": "1.0", "version_end_including": "1.9.9"} + assert _build_range_string(entry) == ">1.0, <=1.9.9" + + def test_only_end_excluding(self): + entry = {"version_end_excluding": "3.0.0"} + assert _build_range_string(entry) == "<3.0.0" + + def test_only_start_including(self): + entry = {"version_start_including": "1.0.0"} + assert _build_range_string(entry) == ">=1.0.0" + + def test_exact_version_match(self): + entry = {"cpe_version": "2.14.1"} + assert _build_range_string(entry) == "=2.14.1" + + def test_no_bounds_no_version(self): + entry = {"cpe_version": ""} + assert "all versions" in _build_range_string(entry) + + def test_wildcard_version_no_bounds(self): + entry = {"cpe_version": "*"} + # Wildcard is treated as empty + assert "all versions" in _build_range_string(entry) + + def test_empty_entry(self): + entry = {} + assert "all versions" in _build_range_string(entry) + + def test_start_and_end_both_including(self): + entry = {"version_start_including": "1.0", "version_end_including": "2.0"} + assert _build_range_string(entry) == ">=1.0, <=2.0" + + +# --------------------------------------------------------------------------- +# CPE range extraction tests +# --------------------------------------------------------------------------- +class TestExtractCpeRanges: + def test_empty_configurations(self): + assert _extract_cpe_ranges([]) == [] + + def test_none_configurations(self): + assert _extract_cpe_ranges(None) == [] + + def test_single_vulnerable_match(self): + configs = [ + { + "nodes": [ + { + "cpeMatch": [ + { + "vulnerable": True, + "criteria": "cpe:2.3:a:apache:log4j:*:*:*:*:*:*:*:*", + "versionStartIncluding": "2.0", + "versionEndExcluding": "2.15.0", + } + ] + } + ] + } + ] + result = _extract_cpe_ranges(configs) + assert len(result) == 1 + assert result[0]["vendor"] == "apache" + assert result[0]["product"] == "log4j" + assert result[0]["version_start_including"] == "2.0" + assert result[0]["version_end_excluding"] == "2.15.0" + + def test_non_vulnerable_match_skipped(self): + configs = [ + { + "nodes": [ + { + "cpeMatch": [ + { + "vulnerable": False, + "criteria": "cpe:2.3:a:linux:kernel:*:*:*:*:*:*:*:*", + } + ] + } + ] + } + ] + assert _extract_cpe_ranges(configs) == [] + + def test_multiple_nodes(self): + configs = [ + { + "nodes": [ + { + "cpeMatch": [ + { + "vulnerable": True, + "criteria": "cpe:2.3:a:v1:p1:*:*:*:*:*:*:*:*", + "versionEndExcluding": "1.0", + } + ] + }, + { + "cpeMatch": [ + { + "vulnerable": True, + "criteria": "cpe:2.3:a:v2:p2:*:*:*:*:*:*:*:*", + "versionEndExcluding": "2.0", + } + ] + }, + ] + } + ] + result = _extract_cpe_ranges(configs) + assert len(result) == 2 + + def test_invalid_cpe_match_entry_skipped(self): + configs = [{"nodes": [{"cpeMatch": ["not a dict", None, 123]}]}] + assert _extract_cpe_ranges(configs) == [] + + def test_missing_nodes_key(self): + configs = [{"other_key": "value"}] + assert _extract_cpe_ranges(configs) == [] + + def test_range_string_included(self): + configs = [ + { + "nodes": [ + { + "cpeMatch": [ + { + "vulnerable": True, + "criteria": "cpe:2.3:a:vendor:product:*:*:*:*:*:*:*:*", + "versionStartIncluding": "1.0", + "versionEndExcluding": "2.0", + } + ] + } + ] + } + ] + result = _extract_cpe_ranges(configs) + assert result[0]["range_string"] == ">=1.0, <2.0" + + +# --------------------------------------------------------------------------- +# Ecosystem normalisation tests +# --------------------------------------------------------------------------- +class TestNormaliseEcosystem: + def test_pypi_variants(self): + assert _normalise_ecosystem("pypi") == "PyPI" + assert _normalise_ecosystem("pip") == "PyPI" + assert _normalise_ecosystem("python") == "PyPI" + + def test_npm_variants(self): + assert _normalise_ecosystem("npm") == "npm" + assert _normalise_ecosystem("node") == "npm" + assert _normalise_ecosystem("nodejs") == "npm" + + def test_maven_variants(self): + assert _normalise_ecosystem("maven") == "Maven" + assert _normalise_ecosystem("java") == "Maven" + + def test_go_variants(self): + assert _normalise_ecosystem("go") == "Go" + assert _normalise_ecosystem("golang") == "Go" + + def test_case_insensitive(self): + assert _normalise_ecosystem("PyPI") == "PyPI" + assert _normalise_ecosystem("NPM") == "npm" + + def test_unknown_ecosystem_passthrough(self): + assert _normalise_ecosystem("unknown_eco") == "unknown_eco" + + def test_empty_string(self): + assert _normalise_ecosystem("") == "" + + def test_rust_variants(self): + assert _normalise_ecosystem("rust") == "crates.io" + assert _normalise_ecosystem("crates") == "crates.io" + + +# --------------------------------------------------------------------------- +# Ecosystem filter tests +# --------------------------------------------------------------------------- +class TestFilterByEcosystem: + def test_auto_returns_all(self): + packages = [ + {"ecosystem": "PyPI", "package": "a"}, + {"ecosystem": "npm", "package": "b"}, + ] + assert len(_filter_by_ecosystem(packages, "auto")) == 2 + + def test_empty_ecosystem_returns_all(self): + packages = [{"ecosystem": "PyPI", "package": "a"}] + assert len(_filter_by_ecosystem(packages, "")) == 1 + + def test_filter_pypi(self): + packages = [ + {"ecosystem": "PyPI", "package": "a"}, + {"ecosystem": "npm", "package": "b"}, + {"ecosystem": "PyPI", "package": "c"}, + ] + result = _filter_by_ecosystem(packages, "pypi") + assert len(result) == 2 + assert all(p["ecosystem"] == "PyPI" for p in result) + + def test_filter_npm(self): + packages = [ + {"ecosystem": "PyPI", "package": "a"}, + {"ecosystem": "npm", "package": "b"}, + ] + result = _filter_by_ecosystem(packages, "npm") + assert len(result) == 1 + assert result[0]["package"] == "b" + + def test_filter_no_match(self): + packages = [{"ecosystem": "PyPI", "package": "a"}] + result = _filter_by_ecosystem(packages, "maven") + assert len(result) == 0 + + def test_case_insensitive_filter(self): + packages = [{"ecosystem": "Maven", "package": "a"}] + result = _filter_by_ecosystem(packages, "MAVEN") + assert len(result) == 1 + + +# --------------------------------------------------------------------------- +# NVD headers tests +# --------------------------------------------------------------------------- +class TestBuildNvdHeaders: + def test_no_api_key(self, monkeypatch): + monkeypatch.delenv("NVD_API_KEY", raising=False) + headers = _build_nvd_headers() + assert "apiKey" not in headers + assert "Accept" in headers + + def test_with_api_key(self, monkeypatch): + monkeypatch.setenv("NVD_API_KEY", "test-key-123") + headers = _build_nvd_headers() + assert headers["apiKey"] == "test-key-123" + + def test_empty_api_key_ignored(self, monkeypatch): + monkeypatch.setenv("NVD_API_KEY", " ") + headers = _build_nvd_headers() + assert "apiKey" not in headers + + +# --------------------------------------------------------------------------- +# HTTP retry tests +# --------------------------------------------------------------------------- +class TestGetWithRetry: + def test_success_on_first_try(self): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.raise_for_status = MagicMock() + with patch("manus_agent.tools.get_version_range.requests.get", return_value=mock_resp): + resp = _get_with_retry("http://example.com") + assert resp.status_code == 200 + + def test_retries_on_429(self, monkeypatch): + monkeypatch.setenv("VERSION_RANGE_MAX_RETRIES", "2") + monkeypatch.setenv("VERSION_RANGE_RETRY_BASE_DELAY", "0") + # Re-import to pick up new env values + import importlib + + import manus_agent.tools.get_version_range as mod + + importlib.reload(mod) + + mock_resp_429 = MagicMock() + mock_resp_429.status_code = 429 + mock_resp_429.raise_for_status = MagicMock(side_effect=Exception("429")) + + mock_resp_ok = MagicMock() + mock_resp_ok.status_code = 200 + mock_resp_ok.raise_for_status = MagicMock() + + with patch("manus_agent.tools.get_version_range.requests.get", side_effect=[mock_resp_429, mock_resp_ok]): + resp = mod._get_with_retry("http://example.com") + assert resp.status_code == 200 + + def test_raises_on_404(self): + mock_resp = MagicMock() + mock_resp.status_code = 404 + exc = requests.exceptions.HTTPError("404", response=mock_resp) + mock_resp.raise_for_status = MagicMock(side_effect=exc) + with patch("manus_agent.tools.get_version_range.requests.get", return_value=mock_resp): + with pytest.raises(requests.exceptions.HTTPError): + _get_with_retry("http://example.com") + + def test_raises_on_connection_error_after_retries(self): + with patch( + "manus_agent.tools.get_version_range.requests.get", + side_effect=requests.exceptions.ConnectionError("refused"), + ): + with pytest.raises(requests.exceptions.ConnectionError): + _get_with_retry("http://example.com") + + def test_raises_on_timeout_after_retries(self): + with patch( + "manus_agent.tools.get_version_range.requests.get", + side_effect=requests.exceptions.Timeout("timed out"), + ): + with pytest.raises(requests.exceptions.Timeout): + _get_with_retry("http://example.com") + + +# --------------------------------------------------------------------------- +# OSV range extraction tests +# --------------------------------------------------------------------------- +class TestExtractOsvRanges: + def _mock_osv_response(self, data): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = data + mock_resp.raise_for_status = MagicMock() + return mock_resp + + def test_no_osv_record(self): + with patch( + "manus_agent.tools.get_version_range.requests.get", + side_effect=requests.exceptions.HTTPError("404", response=MagicMock(status_code=404)), + ): + result = _extract_osv_ranges("CVE-2099-9999") + assert result == [] + + def test_single_package_with_ranges(self): + data = { + "id": "CVE-2021-44228", + "affected": [ + { + "package": {"ecosystem": "Maven", "name": "org.apache.logging.log4j:log4j-core"}, + "ranges": [{"events": [{"introduced": "2.0"}, {"fixed": "2.15.0"}]}], + "versions": ["2.0", "2.1", "2.14.1"], + } + ], + } + mock_resp = self._mock_osv_response(data) + with patch("manus_agent.tools.get_version_range.requests.get", return_value=mock_resp): + result = _extract_osv_ranges("CVE-2021-44228") + assert len(result) == 1 + assert result[0]["ecosystem"] == "Maven" + assert result[0]["package"] == "org.apache.logging.log4j:log4j-core" + assert "2.0" in result[0]["introduced"] + assert "2.15.0" in result[0]["fixed"] + + def test_multiple_packages(self): + data = { + "id": "CVE-2021-44228", + "affected": [ + { + "package": {"ecosystem": "Maven", "name": "log4j-core"}, + "ranges": [{"events": [{"introduced": "2.0"}, {"fixed": "2.15.0"}]}], + "versions": [], + }, + { + "package": {"ecosystem": "Maven", "name": "log4j-api"}, + "ranges": [{"events": [{"introduced": "2.0"}, {"fixed": "2.15.0"}]}], + "versions": [], + }, + ], + } + mock_resp = self._mock_osv_response(data) + with patch("manus_agent.tools.get_version_range.requests.get", return_value=mock_resp): + result = _extract_osv_ranges("CVE-2021-44228") + assert len(result) == 2 + + def test_follows_ghsa_aliases(self): + cve_record = { + "id": "CVE-2021-44228", + "aliases": ["GHSA-jfh8-c2jp-5v3q"], + "affected": [], + } + ghsa_record = { + "id": "GHSA-jfh8-c2jp-5v3q", + "affected": [ + { + "package": {"ecosystem": "Maven", "name": "log4j-core"}, + "ranges": [{"events": [{"introduced": "2.0"}, {"fixed": "2.15.0"}]}], + "versions": ["2.14.1"], + } + ], + } + + def side_effect(url, **kwargs): + resp = MagicMock() + resp.status_code = 200 + resp.raise_for_status = MagicMock() + if "CVE-2021-44228" in url: + resp.json.return_value = cve_record + else: + resp.json.return_value = ghsa_record + return resp + + with patch("manus_agent.tools.get_version_range.requests.get", side_effect=side_effect): + result = _extract_osv_ranges("CVE-2021-44228") + assert len(result) == 1 + assert result[0]["package"] == "log4j-core" + + def test_deduplicates_packages(self): + data = { + "id": "CVE-2021-44228", + "affected": [ + { + "package": {"ecosystem": "Maven", "name": "log4j"}, + "ranges": [{"events": [{"introduced": "2.0"}, {"fixed": "2.15.0"}]}], + "versions": [], + }, + { + "package": {"ecosystem": "Maven", "name": "log4j"}, + "ranges": [{"events": [{"introduced": "2.0"}, {"fixed": "2.16.0"}]}], + "versions": [], + }, + ], + } + mock_resp = self._mock_osv_response(data) + with patch("manus_agent.tools.get_version_range.requests.get", return_value=mock_resp): + result = _extract_osv_ranges("CVE-2021-44228") + assert len(result) == 1 + + def test_handles_last_affected(self): + data = { + "id": "CVE-2023-1234", + "affected": [ + { + "package": {"ecosystem": "PyPI", "name": "vulnerable-pkg"}, + "ranges": [{"events": [{"introduced": "0"}, {"last_affected": "1.2.3"}]}], + "versions": [], + } + ], + } + mock_resp = self._mock_osv_response(data) + with patch("manus_agent.tools.get_version_range.requests.get", return_value=mock_resp): + result = _extract_osv_ranges("CVE-2023-1234") + assert "1.2.3" in result[0]["last_affected"] + + def test_skips_entries_without_package(self): + data = { + "id": "CVE-2023-1234", + "affected": [ + {"package": {}, "ranges": [], "versions": []}, + {"ranges": [], "versions": []}, + ], + } + mock_resp = self._mock_osv_response(data) + with patch("manus_agent.tools.get_version_range.requests.get", return_value=mock_resp): + result = _extract_osv_ranges("CVE-2023-1234") + assert result == [] + + +# --------------------------------------------------------------------------- +# NVD fetch tests +# --------------------------------------------------------------------------- +class TestFetchNvdConfigurations: + def test_returns_configurations(self): + from manus_agent.tools.get_version_range import _fetch_nvd_configurations + + nvd_data = { + "vulnerabilities": [ + { + "cve": { + "configurations": [ + { + "nodes": [ + {"cpeMatch": [{"vulnerable": True, "criteria": "cpe:2.3:a:v:p:*:*:*:*:*:*:*:*"}]} + ] + } + ], + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 10.0, "baseSeverity": "CRITICAL"}}]}, + "published": "2021-12-10", + "lastModified": "2022-01-01", + } + } + ] + } + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = nvd_data + mock_resp.raise_for_status = MagicMock() + + with patch("manus_agent.tools.get_version_range._get_with_retry", return_value=mock_resp): + configs, summary = _fetch_nvd_configurations("CVE-2021-44228") + assert len(configs) == 1 + assert summary["base_score"] == 10.0 + + def test_empty_vulnerabilities(self): + from manus_agent.tools.get_version_range import _fetch_nvd_configurations + + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"vulnerabilities": []} + mock_resp.raise_for_status = MagicMock() + + with patch("manus_agent.tools.get_version_range._get_with_retry", return_value=mock_resp): + configs, summary = _fetch_nvd_configurations("CVE-2099-9999") + assert configs == [] + assert summary == {} + + def test_http_error_returns_empty(self): + from manus_agent.tools.get_version_range import _fetch_nvd_configurations + + with patch( + "manus_agent.tools.get_version_range._get_with_retry", + side_effect=requests.exceptions.ConnectionError("refused"), + ): + configs, summary = _fetch_nvd_configurations("CVE-2021-44228") + assert configs == [] + assert summary == {} + + def test_no_cvss_data(self): + from manus_agent.tools.get_version_range import _fetch_nvd_configurations + + nvd_data = { + "vulnerabilities": [ + { + "cve": { + "configurations": [], + "metrics": {}, + "published": "2021-12-10", + "lastModified": "2022-01-01", + } + } + ] + } + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = nvd_data + mock_resp.raise_for_status = MagicMock() + + with patch("manus_agent.tools.get_version_range._get_with_retry", return_value=mock_resp): + configs, summary = _fetch_nvd_configurations("CVE-2021-44228") + assert summary["base_score"] is None + + +# --------------------------------------------------------------------------- +# Core fetch_version_range tests +# --------------------------------------------------------------------------- +class TestFetchVersionRange: + def _mock_nvd_and_osv(self, nvd_configs=None, osv_packages=None, summary=None): + """Patch both NVD and OSV fetchers.""" + if nvd_configs is None: + nvd_configs = [] + if osv_packages is None: + osv_packages = [] + if summary is None: + summary = {} + + return ( + patch("manus_agent.tools.get_version_range._fetch_nvd_configurations", return_value=(nvd_configs, summary)), + patch("manus_agent.tools.get_version_range._extract_osv_ranges", return_value=osv_packages), + ) + + def test_basic_fetch(self): + nvd_configs = [ + { + "nodes": [ + { + "cpeMatch": [ + { + "vulnerable": True, + "criteria": "cpe:2.3:a:apache:log4j:*:*:*:*:*:*:*:*", + "versionStartIncluding": "2.0", + "versionEndExcluding": "2.15.0", + } + ] + } + ] + } + ] + osv_pkgs = [ + { + "source": "osv", + "ecosystem": "Maven", + "package": "log4j-core", + "introduced": ["2.0"], + "fixed": ["2.15.0"], + "last_affected": [], + "affected_versions": [], + "affected_version_count": 0, + } + ] + summary = { + "base_score": 10.0, + "base_severity": "CRITICAL", + "published": "2021-12-10", + "last_modified": "2022-01-01", + } + + p1, p2 = self._mock_nvd_and_osv(nvd_configs, osv_pkgs, summary) + with p1, p2: + result = fetch_version_range("CVE-2021-44228") + assert result["cve_id"] == "CVE-2021-44228" + assert result["first_patched_version"] == "2.15.0" + assert len(result["nvd_ranges"]) == 1 + assert len(result["osv_packages"]) == 1 + + def test_first_patched_from_nvd_fallback(self): + nvd_configs = [ + { + "nodes": [ + { + "cpeMatch": [ + { + "vulnerable": True, + "criteria": "cpe:2.3:a:v:p:*:*:*:*:*:*:*:*", + "versionEndExcluding": "3.0.0", + } + ] + } + ] + } + ] + p1, p2 = self._mock_nvd_and_osv(nvd_configs, [], {}) + with p1, p2: + result = fetch_version_range("CVE-2023-1234") + assert result["first_patched_version"] == "3.0.0" + + def test_no_patched_version(self): + p1, p2 = self._mock_nvd_and_osv([], [], {}) + with p1, p2: + result = fetch_version_range("CVE-2023-1234") + assert result["first_patched_version"] is None + + def test_ecosystem_filter_applied(self): + osv_pkgs = [ + { + "source": "osv", + "ecosystem": "PyPI", + "package": "a", + "introduced": ["0"], + "fixed": ["1.0"], + "last_affected": [], + "affected_versions": [], + "affected_version_count": 0, + }, + { + "source": "osv", + "ecosystem": "npm", + "package": "b", + "introduced": ["0"], + "fixed": ["2.0"], + "last_affected": [], + "affected_versions": [], + "affected_version_count": 0, + }, + ] + p1, p2 = self._mock_nvd_and_osv([], osv_pkgs, {}) + with p1, p2: + result = fetch_version_range("CVE-2023-1234", ecosystem="pypi") + assert len(result["osv_packages"]) == 1 + assert result["osv_packages"][0]["package"] == "a" + assert result["ecosystem_filter"] == "pypi" + + def test_cve_id_uppercased(self): + p1, p2 = self._mock_nvd_and_osv([], [], {}) + with p1, p2: + result = fetch_version_range("cve-2021-44228") + assert result["cve_id"] == "CVE-2021-44228" + + def test_affected_ecosystems_collected(self): + osv_pkgs = [ + { + "source": "osv", + "ecosystem": "PyPI", + "package": "a", + "introduced": [], + "fixed": [], + "last_affected": [], + "affected_versions": [], + "affected_version_count": 0, + }, + { + "source": "osv", + "ecosystem": "npm", + "package": "b", + "introduced": [], + "fixed": [], + "last_affected": [], + "affected_versions": [], + "affected_version_count": 0, + }, + ] + p1, p2 = self._mock_nvd_and_osv([], osv_pkgs, {}) + with p1, p2: + result = fetch_version_range("CVE-2023-1234") + assert "PyPI" in result["affected_ecosystems"] + assert "npm" in result["affected_ecosystems"] + + def test_affected_products_collected(self): + nvd_configs = [ + { + "nodes": [ + { + "cpeMatch": [ + { + "vulnerable": True, + "criteria": "cpe:2.3:a:apache:log4j:*:*:*:*:*:*:*:*", + "versionEndExcluding": "2.15.0", + }, + { + "vulnerable": True, + "criteria": "cpe:2.3:a:apache:log4j-api:*:*:*:*:*:*:*:*", + "versionEndExcluding": "2.15.0", + }, + ] + } + ] + } + ] + p1, p2 = self._mock_nvd_and_osv(nvd_configs, [], {}) + with p1, p2: + result = fetch_version_range("CVE-2021-44228") + assert len(result["affected_products"]) == 2 + + +# --------------------------------------------------------------------------- +# Text formatting tests +# --------------------------------------------------------------------------- +class TestFormatText: + def test_basic_format(self): + data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [{"vendor": "apache", "product": "log4j", "range_string": ">=2.0, <2.15.0"}], + "osv_packages": [ + { + "ecosystem": "Maven", + "package": "log4j-core", + "introduced": ["2.0"], + "fixed": ["2.15.0"], + "last_affected": [], + "affected_version_count": 50, + "affected_versions": ["2.0", "2.14.1"], + } + ], + "first_patched_version": "2.15.0", + "affected_ecosystems": ["Maven"], + "affected_products": [{"vendor": "apache", "product": "log4j"}], + "ecosystem_filter": None, + "summary": {"base_score": 10.0, "base_severity": "CRITICAL", "published": "2021-12-10T00:00:00"}, + } + text = _format_text(data) + assert "CVE-2021-44228" in text + assert "2.15.0" in text + assert "apache" in text + assert "log4j" in text + assert "Maven" in text + + def test_no_patched_version(self): + data = { + "cve_id": "CVE-2023-9999", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + text = _format_text(data) + assert "No patched version" in text + + def test_ecosystem_filter_shown(self): + data = { + "cve_id": "CVE-2023-1234", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": "pypi", + "summary": {}, + } + text = _format_text(data) + assert "pypi" in text + + def test_empty_nvd_ranges_message(self): + data = { + "cve_id": "CVE-2023-1234", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + text = _format_text(data) + assert "None found" in text + + def test_last_affected_shown(self): + data = { + "cve_id": "CVE-2023-1234", + "nvd_ranges": [], + "osv_packages": [ + { + "ecosystem": "PyPI", + "package": "pkg", + "introduced": ["0"], + "fixed": [], + "last_affected": ["1.9.9"], + "affected_version_count": 0, + "affected_versions": [], + } + ], + "first_patched_version": None, + "affected_ecosystems": ["PyPI"], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + text = _format_text(data) + assert "1.9.9" in text + + def test_cvss_shown(self): + data = { + "cve_id": "CVE-2023-1234", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {"base_score": 7.5, "base_severity": "HIGH", "published": "2023-01-01"}, + } + text = _format_text(data) + assert "7.5" in text + assert "HIGH" in text + + +# --------------------------------------------------------------------------- +# Strands handler tests +# --------------------------------------------------------------------------- +class TestStrandsHandler: + def test_success_result(self): + tool_use = {"toolUseId": "t-1", "input": {"cve_id": "CVE-2021-44228"}} + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data): + result = get_version_range(tool_use) + assert result["status"] == "success" + assert result["toolUseId"] == "t-1" + + def test_error_on_exception(self): + tool_use = {"toolUseId": "t-2", "input": {"cve_id": "CVE-2021-44228"}} + with patch( + "manus_agent.tools.get_version_range.fetch_version_range", + side_effect=RuntimeError("boom"), + ): + result = get_version_range(tool_use) + assert result["status"] == "error" + assert "boom" in result["content"][0]["text"] + + def test_ecosystem_param_passed(self): + tool_use = {"toolUseId": "t-3", "input": {"cve_id": "CVE-2021-44228", "ecosystem": "pypi"}} + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": "pypi", + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data) as mock_fn: + get_version_range(tool_use) + mock_fn.assert_called_once_with("CVE-2021-44228", ecosystem="pypi") + + def test_default_ecosystem_is_auto(self): + tool_use = {"toolUseId": "t-4", "input": {"cve_id": "CVE-2021-44228"}} + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data) as mock_fn: + get_version_range(tool_use) + mock_fn.assert_called_once_with("CVE-2021-44228", ecosystem="auto") + + def test_tool_use_id_preserved(self): + tool_use = {"toolUseId": "custom-id-123", "input": {"cve_id": "CVE-2021-44228"}} + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data): + result = get_version_range(tool_use) + assert result["toolUseId"] == "custom-id-123" + + +# --------------------------------------------------------------------------- +# CLI subcommand tests +# --------------------------------------------------------------------------- +class TestCliVersionRange: + def test_cli_text_output(self, capsys): + from manus_agent.cli import _run_version_range + + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [{"vendor": "apache", "product": "log4j", "range_string": ">=2.0, <2.15.0"}], + "osv_packages": [], + "first_patched_version": "2.15.0", + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {"base_score": 10.0, "base_severity": "CRITICAL", "published": "2021-12-10"}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data): + exit_code = _run_version_range(["CVE-2021-44228"]) + assert exit_code == 0 + out = capsys.readouterr().out + assert "CVE-2021-44228" in out + assert "2.15.0" in out + + def test_cli_json_output(self, capsys): + from manus_agent.cli import _run_version_range + + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": "2.15.0", + "affected_ecosystems": ["Maven"], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data): + exit_code = _run_version_range(["CVE-2021-44228", "--output", "json"]) + assert exit_code == 0 + out = capsys.readouterr().out + parsed = json.loads(out) + assert parsed["first_patched_version"] == "2.15.0" + + def test_cli_ecosystem_filter(self, capsys): + from manus_agent.cli import _run_version_range + + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": "pypi", + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data) as mock_fn: + exit_code = _run_version_range(["CVE-2021-44228", "--ecosystem", "pypi"]) + assert exit_code == 0 + mock_fn.assert_called_once_with("CVE-2021-44228", ecosystem="pypi") + + def test_cli_invalid_cve(self, capsys): + from manus_agent.cli import _run_version_range + + exit_code = _run_version_range(["not-a-cve"]) + assert exit_code == 1 + err = capsys.readouterr().err + assert "CVE-" in err + + def test_cli_error_handling(self, capsys): + from manus_agent.cli import _run_version_range + + with patch( + "manus_agent.tools.get_version_range.fetch_version_range", + side_effect=RuntimeError("network error"), + ): + exit_code = _run_version_range(["CVE-2021-44228"]) + assert exit_code == 1 + err = capsys.readouterr().err + assert "network error" in err + + def test_cli_registered_in_subcommands(self): + from manus_agent.cli import _SUBCOMMANDS + + assert "version-range" in _SUBCOMMANDS + + +# --------------------------------------------------------------------------- +# Edge cases / integration tests +# --------------------------------------------------------------------------- +class TestEdgeCases: + def test_cve_with_whitespace(self): + tool_use = {"toolUseId": "t-ws", "input": {"cve_id": " CVE-2021-44228 "}} + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data): + result = get_version_range(tool_use) + assert result["status"] == "success" + + def test_lowercase_cve_accepted(self): + tool_use = {"toolUseId": "t-lc", "input": {"cve_id": "cve-2021-44228"}} + mock_data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": None, + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + with patch("manus_agent.tools.get_version_range.fetch_version_range", return_value=mock_data): + result = get_version_range(tool_use) + assert result["status"] == "success" + + def test_multiple_fixed_versions_picks_first(self): + osv_pkgs = [ + { + "source": "osv", + "ecosystem": "PyPI", + "package": "a", + "introduced": ["0"], + "fixed": ["1.0", "2.0"], + "last_affected": [], + "affected_versions": [], + "affected_version_count": 0, + }, + ] + with ( + patch("manus_agent.tools.get_version_range._fetch_nvd_configurations", return_value=([], {})), + patch("manus_agent.tools.get_version_range._extract_osv_ranges", return_value=osv_pkgs), + ): + result = fetch_version_range("CVE-2023-1234") + assert result["first_patched_version"] == "1.0" + + def test_osv_connection_error_graceful(self): + """OSV failure should not crash — just return empty OSV data.""" + with ( + patch("manus_agent.tools.get_version_range._fetch_nvd_configurations", return_value=([], {})), + patch( + "manus_agent.tools.get_version_range.requests.get", + side_effect=requests.exceptions.ConnectionError("refused"), + ), + ): + result = fetch_version_range("CVE-2023-1234") + assert result["osv_packages"] == [] + + def test_nvd_no_configurations_key(self): + """NVD response without configurations key should not crash.""" + nvd_data = { + "vulnerabilities": [ + { + "cve": { + "metrics": {}, + "published": "2023-01-01", + "lastModified": "2023-01-01", + } + } + ] + } + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = nvd_data + mock_resp.raise_for_status = MagicMock() + + from manus_agent.tools.get_version_range import _fetch_nvd_configurations + + with patch("manus_agent.tools.get_version_range._get_with_retry", return_value=mock_resp): + configs, summary = _fetch_nvd_configurations("CVE-2023-1234") + assert configs == [] + + def test_affected_products_deduplication(self): + nvd_configs = [ + { + "nodes": [ + { + "cpeMatch": [ + {"vulnerable": True, "criteria": "cpe:2.3:a:vendor:prod:1.0:*:*:*:*:*:*:*"}, + {"vulnerable": True, "criteria": "cpe:2.3:a:vendor:prod:2.0:*:*:*:*:*:*:*"}, + ] + } + ] + } + ] + with ( + patch("manus_agent.tools.get_version_range._fetch_nvd_configurations", return_value=(nvd_configs, {})), + patch("manus_agent.tools.get_version_range._extract_osv_ranges", return_value=[]), + ): + result = fetch_version_range("CVE-2023-1234") + # Same vendor/product should be deduplicated + assert len(result["affected_products"]) == 1 + + +# --------------------------------------------------------------------------- +# JSON output tests +# --------------------------------------------------------------------------- +class TestJsonOutput: + def test_json_serialisable(self): + data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [ + { + "vendor": "apache", + "product": "log4j", + "range_string": ">=2.0, <2.15.0", + "version_start_including": "2.0", + "version_end_excluding": "2.15.0", + "cpe_version": "", + "version_start_excluding": None, + "version_end_including": None, + } + ], + "osv_packages": [ + { + "source": "osv", + "ecosystem": "Maven", + "package": "log4j-core", + "introduced": ["2.0"], + "fixed": ["2.15.0"], + "last_affected": [], + "affected_versions": ["2.14.1"], + "affected_version_count": 1, + } + ], + "first_patched_version": "2.15.0", + "affected_ecosystems": ["Maven"], + "affected_products": [{"vendor": "apache", "product": "log4j"}], + "ecosystem_filter": None, + "summary": { + "base_score": 10.0, + "base_severity": "CRITICAL", + "published": "2021-12-10", + "last_modified": "2022-01-01", + }, + } + # Should not raise + serialised = json.dumps(data, indent=2, default=str) + parsed = json.loads(serialised) + assert parsed["cve_id"] == "CVE-2021-44228" + + def test_first_patched_version_in_json(self): + data = { + "cve_id": "CVE-2021-44228", + "nvd_ranges": [], + "osv_packages": [], + "first_patched_version": "2.15.0", + "affected_ecosystems": [], + "affected_products": [], + "ecosystem_filter": None, + "summary": {}, + } + serialised = json.dumps(data) + parsed = json.loads(serialised) + assert parsed["first_patched_version"] == "2.15.0"