diff --git a/src/manus_agent/cli.py b/src/manus_agent/cli.py index e8442f2..404c1c8 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", + "cve-timeline", } @@ -1935,6 +1936,53 @@ def _run_blast_radius(argv: list[str]) -> int: return 0 +# --------------------------------------------------------------------------- +# cve-timeline subcommand +# --------------------------------------------------------------------------- + + +def _build_cve_timeline_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="manus-agent cve-timeline", + description=( + "Reconstruct the full event timeline for a CVE: NVD publish date, " + "EPSS history with spike detection, CISA KEV addition date, and " + "patch/advisory dates. Useful for understanding how quickly a " + "vulnerability was weaponised and fixed." + ), + 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_cve_timeline(argv: list[str]) -> int: + + from manus_agent.tools.cve_timeline import ( + build_timeline, + format_timeline_json, + format_timeline_text, + ) + + parser = _build_cve_timeline_parser() + args = parser.parse_args(argv) + + result = build_timeline(args.cve_id) + + if args.output == "json": + print(format_timeline_json(result)) + else: + print(format_timeline_text(result)) + + return 1 if "error" in result else 0 + + def _build_run_parser() -> argparse.ArgumentParser: """Build the top-level run/interactive parser.""" parser = argparse.ArgumentParser( @@ -2269,6 +2317,10 @@ def main() -> None: idx = argv.index("blast-radius") sys.exit(_run_blast_radius(argv[idx + 1 :])) + if first_positional == "cve-timeline": + idx = argv.index("cve-timeline") + sys.exit(_run_cve_timeline(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/cve_timeline.py b/src/manus_agent/tools/cve_timeline.py new file mode 100644 index 0000000..d73b505 --- /dev/null +++ b/src/manus_agent/tools/cve_timeline.py @@ -0,0 +1,532 @@ +""" +Tool for reconstructing the full event timeline of a CVE. + +Gathers dates from multiple sources — NVD (publish/modify), EPSS (score history), +CISA KEV (exploitation confirmation), and NVD references (patch/advisory dates) — +to produce a chronological narrative of the vulnerability lifecycle. +""" + +from __future__ import annotations + +import json as _json +import os +import re +import time +from datetime import datetime, timezone +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 constants +# --------------------------------------------------------------------------- +_MAX_RETRIES = int(os.environ.get("CVE_TIMELINE_MAX_RETRIES", "3")) +_RETRY_BASE_DELAY = float(os.environ.get("CVE_TIMELINE_RETRY_BASE_DELAY", "2")) +_RETRYABLE_STATUS = {429, 500, 502, 503, 504} + +# CVE ID pattern +_CVE_PATTERN = re.compile(r"^CVE-\d{4}-\d{4,}$", re.IGNORECASE) + +TOOL_SPEC = { + "name": "cve_timeline", + "description": ( + "Reconstructs the full event timeline for a CVE: NVD publish date, " + "modification date, EPSS score history with spike detection, CISA KEV " + "addition date, and patch/advisory dates extracted from NVD references. " + "Produces a chronological list of events showing how quickly a vulnerability " + "was weaponised and fixed. Use after get_nvd_data for deeper temporal context." + ), + "inputSchema": { + "json": { + "type": "object", + "properties": { + "cve_id": { + "type": "string", + "description": "The CVE identifier (e.g., 'CVE-2021-44228').", + }, + }, + "required": ["cve_id"], + } + }, +} + + +# --------------------------------------------------------------------------- +# HTTP helper with retry +# --------------------------------------------------------------------------- + + +def _get_with_retry( + url: str, + *, + headers: dict[str, str] | None = None, + params: dict[str, Any] | None = None, + timeout: int = 20, +) -> requests.Response: + """GET with exponential back-off retry on transient errors.""" + 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=headers, params=params, timeout=timeout) + if resp.status_code in _RETRYABLE_STATUS: + 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: + raise + except requests.exceptions.RequestException as exc: + last_exc = exc + if attempt < _MAX_RETRIES: + continue + + raise last_exc # type: ignore[misc] + + +# --------------------------------------------------------------------------- +# Data fetchers +# --------------------------------------------------------------------------- + + +def _fetch_nvd_dates(cve_id: str) -> dict[str, Any]: + """Fetch publish/modify dates and references from NVD.""" + url = f"https://services.nvd.nist.gov/rest/json/cves/2.0?cveId={cve_id.upper()}" + headers: dict[str, str] = {} + api_key = os.environ.get("NVD_API_KEY", "").strip() + if api_key: + headers["apiKey"] = api_key + + try: + resp = _get_with_retry(url, headers=headers) + data = resp.json() + except Exception as exc: + return {"error": str(exc)} + + vulns = data.get("vulnerabilities", []) + if not vulns: + return {"error": "CVE not found in NVD"} + + cve_data = vulns[0].get("cve", {}) + result: dict[str, Any] = {} + + # Publication and modification dates + published = cve_data.get("published") + if published: + result["published"] = published[:10] # YYYY-MM-DD + + last_modified = cve_data.get("lastModified") + if last_modified: + result["last_modified"] = last_modified[:10] + + # CVSS score for context + metrics = cve_data.get("metrics", {}) + for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"): + metric_list = metrics.get(key, []) + if metric_list: + cvss_data = metric_list[0].get("cvssData", {}) + result["cvss_score"] = cvss_data.get("baseScore") + result["cvss_severity"] = cvss_data.get("baseSeverity") + break + + # Extract reference dates (patches, advisories) + references = cve_data.get("references", []) + ref_events: list[dict[str, str]] = [] + for ref in references: + tags = ref.get("tags", []) + url_str = ref.get("url", "") + # Look for patch and advisory references + if "Patch" in tags or "Vendor Advisory" in tags or "Third Party Advisory" in tags: + ref_type = "patch" if "Patch" in tags else "advisory" + ref_events.append({"type": ref_type, "url": url_str, "tags": tags}) + + result["references"] = ref_events + return result + + +def _fetch_epss_history(cve_id: str, days: int = 90) -> dict[str, Any]: + """Fetch EPSS score history from FIRST.org.""" + url = "https://api.first.org/data/v1/epss" + params: dict[str, Any] = { + "cve": cve_id.upper(), + "scope": "time-series", + "limit": min(days, 365), + } + + try: + resp = _get_with_retry(url, params=params) + data = resp.json() + except Exception as exc: + return {"error": str(exc)} + + epss_data = data.get("data", []) + if not epss_data: + return {"error": "No EPSS data available"} + + cve_entry = epss_data[0] if epss_data else {} + time_series = cve_entry.get("time-series", cve_entry.get("timeSeries", [])) + + if not time_series: + # Might be a single-point response + epss_val = cve_entry.get("epss") + date_val = cve_entry.get("date") + if epss_val and date_val: + return { + "current_score": float(epss_val), + "first_seen": date_val, + "history": [{"date": date_val, "epss": float(epss_val)}], + } + return {"error": "No EPSS time-series data"} + + # Sort oldest first + points = sorted(time_series, key=lambda p: p.get("date", "")) + history = [{"date": p["date"], "epss": float(p["epss"])} for p in points] + + # Detect spikes (>= 0.10 jump between consecutive days) + spikes: list[dict[str, Any]] = [] + for i in range(1, len(history)): + jump = history[i]["epss"] - history[i - 1]["epss"] + if jump >= 0.10: + spikes.append( + { + "date": history[i]["date"], + "from_score": history[i - 1]["epss"], + "to_score": history[i]["epss"], + "jump": round(jump, 4), + } + ) + + result: dict[str, Any] = { + "current_score": history[-1]["epss"] if history else None, + "first_seen": history[0]["date"] if history else None, + "last_seen": history[-1]["date"] if history else None, + "history_points": len(history), + "spikes": spikes, + } + return result + + +def _fetch_kev_date(cve_id: str) -> dict[str, Any]: + """Check CISA KEV catalog for the CVE and return the date added.""" + url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json" + + try: + resp = _get_with_retry(url) + data = resp.json() + except Exception as exc: + return {"error": str(exc)} + + cve_upper = cve_id.upper() + for vuln in data.get("vulnerabilities", []): + if vuln.get("cveID", "").upper() == cve_upper: + return { + "in_kev": True, + "date_added": vuln.get("dateAdded"), + "due_date": vuln.get("dueDate"), + "known_ransomware": vuln.get("knownRansomwareCampaignUse", "Unknown"), + "vendor": vuln.get("vendorProject"), + "product": vuln.get("product"), + "short_description": vuln.get("shortDescription"), + } + + return {"in_kev": False} + + +# --------------------------------------------------------------------------- +# Timeline assembly +# --------------------------------------------------------------------------- + + +def _parse_date(date_str: str | None) -> datetime | None: + """Parse a YYYY-MM-DD date string to datetime.""" + if not date_str: + return None + try: + return datetime.strptime(date_str[:10], "%Y-%m-%d").replace(tzinfo=timezone.utc) + except (ValueError, TypeError): + return None + + +def _days_between(date1: str | None, date2: str | None) -> int | None: + """Calculate days between two date strings.""" + d1 = _parse_date(date1) + d2 = _parse_date(date2) + if d1 and d2: + return abs((d2 - d1).days) + return None + + +def build_timeline(cve_id: str) -> dict[str, Any]: + """ + Assemble the full CVE timeline from multiple sources. + + Returns a dict with: + - events: list of {date, event_type, description} sorted chronologically + - summary: high-level stats (time-to-exploit, time-to-patch, etc.) + - sources: which data sources were successfully queried + """ + cve_id = cve_id.upper().strip() + + if not _CVE_PATTERN.match(cve_id): + return {"error": f"Invalid CVE ID format: {cve_id}"} + + events: list[dict[str, Any]] = [] + sources: dict[str, str] = {} + summary: dict[str, Any] = {"cve_id": cve_id} + + # --- NVD data --- + nvd = _fetch_nvd_dates(cve_id) + if "error" in nvd: + sources["nvd"] = f"error: {nvd['error']}" + else: + sources["nvd"] = "ok" + if nvd.get("published"): + events.append( + { + "date": nvd["published"], + "event_type": "nvd_published", + "description": "CVE published in NVD", + } + ) + summary["nvd_published"] = nvd["published"] + + if nvd.get("last_modified") and nvd.get("last_modified") != nvd.get("published"): + events.append( + { + "date": nvd["last_modified"], + "event_type": "nvd_modified", + "description": "CVE record last modified in NVD", + } + ) + + if nvd.get("cvss_score"): + summary["cvss_score"] = nvd["cvss_score"] + summary["cvss_severity"] = nvd.get("cvss_severity") + + # Reference-based events (patches/advisories) + for ref in nvd.get("references", []): + ref_type = ref.get("type", "reference") + url_str = ref.get("url", "") + # Try to extract date from GitHub commit URLs or common patterns + date_from_url = _extract_date_from_url(url_str) + if date_from_url: + label = "Patch published" if ref_type == "patch" else "Advisory published" + events.append( + { + "date": date_from_url, + "event_type": f"{ref_type}_released", + "description": f"{label}: {url_str[:80]}", + } + ) + + # --- EPSS history --- + epss = _fetch_epss_history(cve_id) + if "error" in epss: + sources["epss"] = f"error: {epss['error']}" + else: + sources["epss"] = "ok" + if epss.get("first_seen"): + events.append( + { + "date": epss["first_seen"], + "event_type": "epss_first_score", + "description": f"First EPSS score recorded ({epss.get('current_score', 'N/A')} current)", + } + ) + summary["epss_current"] = epss.get("current_score") + + # EPSS spikes are significant timeline events + for spike in epss.get("spikes", []): + events.append( + { + "date": spike["date"], + "event_type": "epss_spike", + "description": ( + f"EPSS spike: {spike['from_score']:.4f} → {spike['to_score']:.4f} (+{spike['jump']:.4f})" + ), + } + ) + + # --- CISA KEV --- + kev = _fetch_kev_date(cve_id) + if "error" in kev: + sources["kev"] = f"error: {kev['error']}" + else: + sources["kev"] = "ok" + summary["in_kev"] = kev.get("in_kev", False) + if kev.get("in_kev") and kev.get("date_added"): + events.append( + { + "date": kev["date_added"], + "event_type": "kev_added", + "description": "Added to CISA KEV (confirmed exploitation in the wild)", + } + ) + summary["kev_date_added"] = kev["date_added"] + summary["kev_due_date"] = kev.get("due_date") + summary["kev_ransomware"] = kev.get("known_ransomware") + + # --- Sort events chronologically --- + events.sort(key=lambda e: e.get("date", "9999-99-99")) + + # --- Compute time deltas --- + nvd_pub = summary.get("nvd_published") + kev_added = summary.get("kev_date_added") + if nvd_pub and kev_added: + days = _days_between(nvd_pub, kev_added) + if days is not None: + summary["days_publish_to_exploit"] = days + + # Time from publish to first EPSS spike + if nvd_pub and epss.get("spikes"): + first_spike_date = epss["spikes"][0]["date"] + days = _days_between(nvd_pub, first_spike_date) + if days is not None: + summary["days_publish_to_epss_spike"] = days + + return { + "cve_id": cve_id, + "events": events, + "summary": summary, + "sources": sources, + "event_count": len(events), + } + + +def _extract_date_from_url(url: str) -> str | None: + """ + Attempt to extract a date from common URL patterns. + Supports GitHub commit/release URLs and advisory publication URLs. + """ + # GitHub release/tag URLs often contain dates or version timestamps + # e.g., https://github.com/org/repo/commit/abc123 + # We can't reliably get dates from these without an API call, + # so we only extract from URLs with explicit date patterns. + + # Pattern: YYYY-MM-DD in URL path + match = re.search(r"(\d{4}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12]\d|3[01]))", url) + if match: + return match.group(1) + + # Pattern: YYYY/MM/DD in URL path + match = re.search(r"/(\d{4})/(0[1-9]|1[0-2])/(0[1-9]|[12]\d|3[01])/", url) + if match: + return f"{match.group(1)}-{match.group(2)}-{match.group(3)}" + + return None + + +# --------------------------------------------------------------------------- +# Formatting +# --------------------------------------------------------------------------- + + +def format_timeline_text(result: dict[str, Any]) -> str: + """Format timeline as human-readable text.""" + if "error" in result: + return f"Error: {result['error']}" + + lines: list[str] = [] + cve_id = result["cve_id"] + summary = result.get("summary", {}) + events = result.get("events", []) + + lines.append(f"CVE Timeline: {cve_id}") + lines.append("=" * (len(f"CVE Timeline: {cve_id}"))) + lines.append("") + + # Summary section + if summary.get("cvss_score"): + lines.append(f"CVSS: {summary['cvss_score']} ({summary.get('cvss_severity', 'N/A')})") + if summary.get("epss_current") is not None: + lines.append(f"EPSS (current): {summary['epss_current']:.4f}") + if summary.get("in_kev"): + lines.append("KEV Status: ⚠️ IN CISA KEV (actively exploited)") + else: + lines.append("KEV Status: Not in CISA KEV") + lines.append("") + + # Events + if events: + lines.append("Timeline Events:") + lines.append("-" * 50) + for event in events: + date = event.get("date", "Unknown") + etype = event.get("event_type", "") + desc = event.get("description", "") + icon = _event_icon(etype) + lines.append(f" {date} {icon} {desc}") + lines.append("") + else: + lines.append("No timeline events found.") + lines.append("") + + # Time deltas + deltas: list[str] = [] + if summary.get("days_publish_to_exploit") is not None: + deltas.append(f" Publish → KEV exploit: {summary['days_publish_to_exploit']} days") + if summary.get("days_publish_to_epss_spike") is not None: + deltas.append(f" Publish → EPSS spike: {summary['days_publish_to_epss_spike']} days") + if deltas: + lines.append("Key Intervals:") + lines.extend(deltas) + lines.append("") + + # Sources + sources = result.get("sources", {}) + lines.append("Data Sources:") + for src, status in sources.items(): + icon = "✓" if status == "ok" else "✗" + lines.append(f" {icon} {src}: {status}") + + return "\n".join(lines) + + +def format_timeline_json(result: dict[str, Any]) -> str: + """Format timeline as JSON string.""" + return _json.dumps(result, indent=2, default=str) + + +def _event_icon(event_type: str) -> str: + """Return an icon for the event type.""" + icons = { + "nvd_published": "📋", + "nvd_modified": "✏️", + "epss_first_score": "📊", + "epss_spike": "📈", + "kev_added": "🚨", + "patch_released": "🩹", + "advisory_released": "📢", + } + return icons.get(event_type, "•") + + +# --------------------------------------------------------------------------- +# Strands tool handler +# --------------------------------------------------------------------------- + + +def handler(tool: ToolUse, **kwargs: Any) -> ToolResult: + """Strands-compatible tool handler.""" + tool_input = tool["input"] + cve_id = tool_input.get("cve_id", "") + + if not cve_id: + content = "Error: cve_id is required" + log_tool_output_size("cve_timeline", content) + return {"status": "error", "content": [{"text": content}]} + + result = build_timeline(cve_id) + output = format_timeline_json(result) + log_tool_output_size("cve_timeline", output) + + status = "error" if "error" in result else "success" + return {"status": status, "content": [{"text": output}]} diff --git a/tests/test_cve_timeline.py b/tests/test_cve_timeline.py new file mode 100644 index 0000000..c247431 --- /dev/null +++ b/tests/test_cve_timeline.py @@ -0,0 +1,1120 @@ +""" +Comprehensive test suite for the cve_timeline tool. + +All tests are fully mocked — no real HTTP calls. +""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from manus_agent.tools.cve_timeline import ( + TOOL_SPEC, + _days_between, + _extract_date_from_url, + _fetch_epss_history, + _fetch_kev_date, + _fetch_nvd_dates, + _parse_date, + build_timeline, + format_timeline_json, + format_timeline_text, + handler, +) + +ToolUse = dict # type alias for test typing + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def nvd_response_log4j(): + """NVD API response fixture for CVE-2021-44228.""" + return { + "vulnerabilities": [ + { + "cve": { + "id": "CVE-2021-44228", + "published": "2021-12-10T10:15:00.000", + "lastModified": "2023-11-07T03:39:00.000", + "metrics": { + "cvssMetricV31": [ + { + "cvssData": { + "baseScore": 10.0, + "baseSeverity": "CRITICAL", + } + } + ] + }, + "references": [ + { + "url": "https://github.com/apache/logging-log4j2/commit/abc123", + "tags": ["Patch"], + }, + { + "url": "https://logging.apache.org/log4j/2.x/security.html", + "tags": ["Vendor Advisory"], + }, + { + "url": "https://www.kb.cert.org/vuls/id/930724", + "tags": ["Third Party Advisory"], + }, + ], + } + } + ] + } + + +@pytest.fixture +def epss_response(): + """EPSS time-series response fixture.""" + return { + "data": [ + { + "cve": "CVE-2021-44228", + "time-series": [ + {"date": "2021-12-10", "epss": "0.0100", "percentile": "0.50"}, + {"date": "2021-12-11", "epss": "0.0500", "percentile": "0.70"}, + {"date": "2021-12-12", "epss": "0.2000", "percentile": "0.90"}, + {"date": "2021-12-13", "epss": "0.5000", "percentile": "0.95"}, + {"date": "2021-12-14", "epss": "0.9700", "percentile": "0.99"}, + ], + } + ] + } + + +@pytest.fixture +def kev_response(): + """CISA KEV catalog response fixture.""" + return { + "vulnerabilities": [ + { + "cveID": "CVE-2021-44228", + "dateAdded": "2021-12-10", + "dueDate": "2021-12-24", + "knownRansomwareCampaignUse": "Known", + "vendorProject": "Apache", + "product": "Log4j", + "shortDescription": "Apache Log4j2 RCE", + }, + { + "cveID": "CVE-2022-99999", + "dateAdded": "2022-05-01", + "dueDate": "2022-05-15", + "knownRansomwareCampaignUse": "Unknown", + "vendorProject": "Example", + "product": "Widget", + "shortDescription": "Example vuln", + }, + ] + } + + +# --------------------------------------------------------------------------- +# TOOL_SPEC contract tests +# --------------------------------------------------------------------------- + + +class TestToolSpec: + """Verify TOOL_SPEC follows Strands conventions.""" + + def test_has_name(self): + assert TOOL_SPEC["name"] == "cve_timeline" + + def test_has_description(self): + assert isinstance(TOOL_SPEC["description"], str) + assert len(TOOL_SPEC["description"]) > 20 + + def test_has_input_schema(self): + schema = TOOL_SPEC["inputSchema"]["json"] + assert schema["type"] == "object" + assert "cve_id" in schema["properties"] + assert "cve_id" in schema["required"] + + def test_cve_id_property(self): + prop = TOOL_SPEC["inputSchema"]["json"]["properties"]["cve_id"] + assert prop["type"] == "string" + assert "description" in prop + + +# --------------------------------------------------------------------------- +# Input validation tests +# --------------------------------------------------------------------------- + + +class TestInputValidation: + """Test input validation in build_timeline.""" + + def test_invalid_cve_format_no_prefix(self): + result = build_timeline("2021-44228") + assert "error" in result + assert "Invalid CVE ID" in result["error"] + + def test_invalid_cve_format_wrong_prefix(self): + result = build_timeline("VUL-2021-44228") + assert "error" in result + + def test_invalid_cve_format_short_number(self): + result = build_timeline("CVE-2021-12") + assert "error" in result + + def test_empty_cve_id(self): + result = build_timeline("") + assert "error" in result + + def test_valid_cve_format_accepted(self): + """Valid format should not return a format error (may fail on network).""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_get.return_value = mock_resp + result = build_timeline("CVE-2021-44228") + # Should not have a format error + assert "Invalid CVE ID" not in result.get("error", "") + + def test_case_insensitive(self): + """CVE ID should be normalized to uppercase.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_get.return_value = mock_resp + result = build_timeline("cve-2021-44228") + assert result["cve_id"] == "CVE-2021-44228" + + def test_whitespace_stripped(self): + """CVE ID with whitespace should be cleaned.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_get.return_value = mock_resp + result = build_timeline(" CVE-2021-44228 ") + assert result["cve_id"] == "CVE-2021-44228" + + +# --------------------------------------------------------------------------- +# NVD fetch tests +# --------------------------------------------------------------------------- + + +class TestFetchNvdDates: + """Test _fetch_nvd_dates function.""" + + def test_successful_fetch(self, nvd_response_log4j): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = nvd_response_log4j + mock_get.return_value = mock_resp + + result = _fetch_nvd_dates("CVE-2021-44228") + assert result["published"] == "2021-12-10" + assert result["last_modified"] == "2023-11-07" + assert result["cvss_score"] == 10.0 + assert result["cvss_severity"] == "CRITICAL" + + def test_references_extracted(self, nvd_response_log4j): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = nvd_response_log4j + mock_get.return_value = mock_resp + + result = _fetch_nvd_dates("CVE-2021-44228") + assert len(result["references"]) == 3 + assert result["references"][0]["type"] == "patch" + assert result["references"][1]["type"] == "advisory" + + def test_cve_not_found(self): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_get.return_value = mock_resp + + result = _fetch_nvd_dates("CVE-9999-99999") + assert "error" in result + assert "not found" in result["error"] + + def test_network_error(self): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + import requests + + mock_get.side_effect = requests.exceptions.ConnectionError("timeout") + result = _fetch_nvd_dates("CVE-2021-44228") + assert "error" in result + + def test_nvd_api_key_used(self, nvd_response_log4j): + with ( + patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get, + patch.dict("os.environ", {"NVD_API_KEY": "test-key-123"}), + ): + mock_resp = MagicMock() + mock_resp.json.return_value = nvd_response_log4j + mock_get.return_value = mock_resp + + _fetch_nvd_dates("CVE-2021-44228") + call_kwargs = mock_get.call_args[1] + assert call_kwargs["headers"]["apiKey"] == "test-key-123" + + def test_cvss_v2_fallback(self): + """Falls back to CVSS v2 when v3.1 is not available.""" + nvd_data = { + "vulnerabilities": [ + { + "cve": { + "published": "2020-01-01T00:00:00.000", + "lastModified": "2020-01-02T00:00:00.000", + "metrics": { + "cvssMetricV2": [ + { + "cvssData": { + "baseScore": 7.5, + "baseSeverity": "HIGH", + } + } + ] + }, + "references": [], + } + } + ] + } + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = nvd_data + mock_get.return_value = mock_resp + + result = _fetch_nvd_dates("CVE-2020-0001") + assert result["cvss_score"] == 7.5 + + +# --------------------------------------------------------------------------- +# EPSS fetch tests +# --------------------------------------------------------------------------- + + +class TestFetchEpssHistory: + """Test _fetch_epss_history function.""" + + def test_successful_fetch(self, epss_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = epss_response + mock_get.return_value = mock_resp + + result = _fetch_epss_history("CVE-2021-44228") + assert result["current_score"] == 0.97 + assert result["first_seen"] == "2021-12-10" + assert result["last_seen"] == "2021-12-14" + assert result["history_points"] == 5 + + def test_spike_detection(self, epss_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = epss_response + mock_get.return_value = mock_resp + + result = _fetch_epss_history("CVE-2021-44228") + # 0.01→0.05 = +0.04 (no spike) + # 0.05→0.20 = +0.15 (spike!) + # 0.20→0.50 = +0.30 (spike!) + # 0.50→0.97 = +0.47 (spike!) + assert len(result["spikes"]) == 3 + assert result["spikes"][0]["date"] == "2021-12-12" + assert result["spikes"][0]["jump"] == 0.15 + + def test_no_data(self): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"data": []} + mock_get.return_value = mock_resp + + result = _fetch_epss_history("CVE-9999-99999") + assert "error" in result + + def test_single_point_response(self): + """Handle case where API returns a single data point (no time-series).""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"data": [{"cve": "CVE-2024-1234", "epss": "0.05", "date": "2024-06-01"}]} + mock_get.return_value = mock_resp + + result = _fetch_epss_history("CVE-2024-1234") + assert result["current_score"] == 0.05 + assert result["first_seen"] == "2024-06-01" + + def test_network_error(self): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + import requests + + mock_get.side_effect = requests.exceptions.Timeout("timeout") + result = _fetch_epss_history("CVE-2021-44228") + assert "error" in result + + def test_no_spikes_in_flat_series(self): + """No spikes when EPSS scores are stable.""" + flat_data = { + "data": [ + { + "cve": "CVE-2024-0001", + "time-series": [ + {"date": "2024-01-01", "epss": "0.01", "percentile": "0.30"}, + {"date": "2024-01-02", "epss": "0.02", "percentile": "0.31"}, + {"date": "2024-01-03", "epss": "0.02", "percentile": "0.31"}, + ], + } + ] + } + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = flat_data + mock_get.return_value = mock_resp + + result = _fetch_epss_history("CVE-2024-0001") + assert result["spikes"] == [] + + +# --------------------------------------------------------------------------- +# KEV fetch tests +# --------------------------------------------------------------------------- + + +class TestFetchKevDate: + """Test _fetch_kev_date function.""" + + def test_cve_in_kev(self, kev_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = kev_response + mock_get.return_value = mock_resp + + result = _fetch_kev_date("CVE-2021-44228") + assert result["in_kev"] is True + assert result["date_added"] == "2021-12-10" + assert result["due_date"] == "2021-12-24" + assert result["known_ransomware"] == "Known" + assert result["vendor"] == "Apache" + assert result["product"] == "Log4j" + + def test_cve_not_in_kev(self, kev_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = kev_response + mock_get.return_value = mock_resp + + result = _fetch_kev_date("CVE-2024-0001") + assert result["in_kev"] is False + + def test_case_insensitive_match(self, kev_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = kev_response + mock_get.return_value = mock_resp + + result = _fetch_kev_date("cve-2021-44228") + assert result["in_kev"] is True + + def test_network_error(self): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + import requests + + mock_get.side_effect = requests.exceptions.ConnectionError("fail") + result = _fetch_kev_date("CVE-2021-44228") + assert "error" in result + + +# --------------------------------------------------------------------------- +# HTTP retry tests +# --------------------------------------------------------------------------- + + +class TestRetry: + """Test _get_with_retry retry logic.""" + + def test_retry_on_429(self): + from manus_agent.tools.cve_timeline import _get_with_retry + + mock_resp_429 = MagicMock() + mock_resp_429.status_code = 429 + mock_resp_success = MagicMock() + mock_resp_success.status_code = 200 + mock_resp_success.raise_for_status = MagicMock() + + with patch("manus_agent.tools.cve_timeline.requests.get") as mock_get: + with patch("manus_agent.tools.cve_timeline.time.sleep"): + mock_get.side_effect = [mock_resp_429, mock_resp_success] + result = _get_with_retry("https://example.com") + assert result == mock_resp_success + assert mock_get.call_count == 2 + + def test_retry_exhausted(self): + import requests + + from manus_agent.tools.cve_timeline import _get_with_retry + + mock_resp_500 = MagicMock() + mock_resp_500.status_code = 500 + + with patch("manus_agent.tools.cve_timeline.requests.get") as mock_get: + with patch("manus_agent.tools.cve_timeline.time.sleep"): + mock_get.return_value = mock_resp_500 + with pytest.raises(requests.exceptions.HTTPError): + _get_with_retry("https://example.com") + assert mock_get.call_count == 4 # 1 + 3 retries + + def test_no_retry_on_404(self): + import requests + + from manus_agent.tools.cve_timeline import _get_with_retry + + mock_resp_404 = MagicMock() + mock_resp_404.status_code = 404 + mock_resp_404.raise_for_status.side_effect = requests.exceptions.HTTPError("404") + + with patch("manus_agent.tools.cve_timeline.requests.get") as mock_get: + mock_get.return_value = mock_resp_404 + with pytest.raises(requests.exceptions.HTTPError): + _get_with_retry("https://example.com") + assert mock_get.call_count == 1 + + def test_retry_on_connection_error(self): + import requests + + from manus_agent.tools.cve_timeline import _get_with_retry + + mock_resp_success = MagicMock() + mock_resp_success.status_code = 200 + mock_resp_success.raise_for_status = MagicMock() + + with patch("manus_agent.tools.cve_timeline.requests.get") as mock_get: + with patch("manus_agent.tools.cve_timeline.time.sleep"): + mock_get.side_effect = [ + requests.exceptions.ConnectionError("fail"), + mock_resp_success, + ] + result = _get_with_retry("https://example.com") + assert result == mock_resp_success + + +# --------------------------------------------------------------------------- +# Date utility tests +# --------------------------------------------------------------------------- + + +class TestDateUtils: + """Test date parsing and calculation utilities.""" + + def test_parse_date_valid(self): + dt = _parse_date("2021-12-10") + assert dt is not None + assert dt.year == 2021 + assert dt.month == 12 + assert dt.day == 10 + + def test_parse_date_none(self): + assert _parse_date(None) is None + + def test_parse_date_invalid(self): + assert _parse_date("not-a-date") is None + + def test_parse_date_empty(self): + assert _parse_date("") is None + + def test_parse_date_truncates(self): + """Handles full ISO timestamps by taking first 10 chars.""" + dt = _parse_date("2021-12-10T10:15:00.000") + assert dt is not None + assert dt.day == 10 + + def test_days_between_same_day(self): + assert _days_between("2021-12-10", "2021-12-10") == 0 + + def test_days_between_normal(self): + assert _days_between("2021-12-10", "2021-12-24") == 14 + + def test_days_between_reversed(self): + """Order doesn't matter — returns absolute difference.""" + assert _days_between("2021-12-24", "2021-12-10") == 14 + + def test_days_between_none_input(self): + assert _days_between(None, "2021-12-10") is None + assert _days_between("2021-12-10", None) is None + + def test_days_between_invalid(self): + assert _days_between("invalid", "2021-12-10") is None + + +# --------------------------------------------------------------------------- +# URL date extraction tests +# --------------------------------------------------------------------------- + + +class TestExtractDateFromUrl: + """Test _extract_date_from_url function.""" + + def test_iso_date_in_url(self): + url = "https://advisory.example.com/2021-12-10/fix" + assert _extract_date_from_url(url) == "2021-12-10" + + def test_slash_date_in_url(self): + url = "https://blog.example.com/2021/12/10/security-advisory" + assert _extract_date_from_url(url) == "2021-12-10" + + def test_no_date_in_url(self): + url = "https://github.com/apache/logging-log4j2/commit/abc123" + assert _extract_date_from_url(url) is None + + def test_invalid_month_rejected(self): + url = "https://example.com/2021-13-10/fix" + assert _extract_date_from_url(url) is None + + def test_invalid_day_rejected(self): + url = "https://example.com/2021-12-32/fix" + assert _extract_date_from_url(url) is None + + +# --------------------------------------------------------------------------- +# Timeline assembly tests +# --------------------------------------------------------------------------- + + +class TestBuildTimeline: + """Test the full build_timeline assembly function.""" + + def test_full_timeline(self, nvd_response_log4j, epss_response, kev_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + # NVD → EPSS → KEV (three calls) + mock_nvd = MagicMock() + mock_nvd.json.return_value = nvd_response_log4j + mock_epss = MagicMock() + mock_epss.json.return_value = epss_response + mock_kev = MagicMock() + mock_kev.json.return_value = kev_response + mock_get.side_effect = [mock_nvd, mock_epss, mock_kev] + + result = build_timeline("CVE-2021-44228") + + assert result["cve_id"] == "CVE-2021-44228" + assert result["event_count"] > 0 + assert "events" in result + assert "summary" in result + assert "sources" in result + + def test_events_sorted_chronologically(self, nvd_response_log4j, epss_response, kev_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_nvd = MagicMock() + mock_nvd.json.return_value = nvd_response_log4j + mock_epss = MagicMock() + mock_epss.json.return_value = epss_response + mock_kev = MagicMock() + mock_kev.json.return_value = kev_response + mock_get.side_effect = [mock_nvd, mock_epss, mock_kev] + + result = build_timeline("CVE-2021-44228") + dates = [e["date"] for e in result["events"]] + assert dates == sorted(dates) + + def test_summary_includes_time_deltas(self, nvd_response_log4j, epss_response, kev_response): + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_nvd = MagicMock() + mock_nvd.json.return_value = nvd_response_log4j + mock_epss = MagicMock() + mock_epss.json.return_value = epss_response + mock_kev = MagicMock() + mock_kev.json.return_value = kev_response + mock_get.side_effect = [mock_nvd, mock_epss, mock_kev] + + result = build_timeline("CVE-2021-44228") + summary = result["summary"] + # Published 2021-12-10, KEV added 2021-12-10 → 0 days + assert summary["days_publish_to_exploit"] == 0 + + def test_graceful_degradation_nvd_fails(self, epss_response, kev_response): + """Timeline still works if NVD fails.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + import requests + + mock_epss = MagicMock() + mock_epss.json.return_value = epss_response + mock_kev = MagicMock() + mock_kev.json.return_value = kev_response + mock_get.side_effect = [ + requests.exceptions.Timeout("nvd timeout"), + mock_epss, + mock_kev, + ] + + result = build_timeline("CVE-2021-44228") + assert "error" not in result + assert result["sources"]["nvd"].startswith("error:") + assert result["sources"]["epss"] == "ok" + assert result["sources"]["kev"] == "ok" + + def test_graceful_degradation_epss_fails(self, nvd_response_log4j, kev_response): + """Timeline still works if EPSS fails.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + import requests + + mock_nvd = MagicMock() + mock_nvd.json.return_value = nvd_response_log4j + mock_kev = MagicMock() + mock_kev.json.return_value = kev_response + mock_get.side_effect = [ + mock_nvd, + requests.exceptions.ConnectionError("epss fail"), + mock_kev, + ] + + result = build_timeline("CVE-2021-44228") + assert "error" not in result + assert result["sources"]["nvd"] == "ok" + assert result["sources"]["epss"].startswith("error:") + + def test_graceful_degradation_kev_fails(self, nvd_response_log4j, epss_response): + """Timeline still works if KEV fails.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + import requests + + mock_nvd = MagicMock() + mock_nvd.json.return_value = nvd_response_log4j + mock_epss = MagicMock() + mock_epss.json.return_value = epss_response + mock_get.side_effect = [ + mock_nvd, + mock_epss, + requests.exceptions.Timeout("kev timeout"), + ] + + result = build_timeline("CVE-2021-44228") + assert "error" not in result + assert result["sources"]["kev"].startswith("error:") + + def test_all_sources_fail(self): + """Timeline returns partial result even when all sources fail.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + import requests + + mock_get.side_effect = requests.exceptions.Timeout("all fail") + + result = build_timeline("CVE-2021-44228") + assert "error" not in result # Not a fatal error, just empty + assert result["event_count"] == 0 + assert all(v.startswith("error:") for v in result["sources"].values()) + + def test_kev_not_present(self, nvd_response_log4j, epss_response): + """CVE not in KEV should reflect in summary.""" + kev_data = {"vulnerabilities": []} + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_nvd = MagicMock() + mock_nvd.json.return_value = nvd_response_log4j + mock_epss = MagicMock() + mock_epss.json.return_value = epss_response + mock_kev = MagicMock() + mock_kev.json.return_value = kev_data + mock_get.side_effect = [mock_nvd, mock_epss, mock_kev] + + result = build_timeline("CVE-2021-44228") + assert result["summary"]["in_kev"] is False + + +# --------------------------------------------------------------------------- +# Text formatting tests +# --------------------------------------------------------------------------- + + +class TestFormatTimelineText: + """Test text output formatting.""" + + def test_basic_formatting(self): + result = { + "cve_id": "CVE-2021-44228", + "events": [ + { + "date": "2021-12-10", + "event_type": "nvd_published", + "description": "CVE published in NVD", + }, + ], + "summary": { + "cve_id": "CVE-2021-44228", + "cvss_score": 10.0, + "cvss_severity": "CRITICAL", + "in_kev": True, + "epss_current": 0.97, + }, + "sources": {"nvd": "ok", "epss": "ok", "kev": "ok"}, + "event_count": 1, + } + text = format_timeline_text(result) + assert "CVE-2021-44228" in text + assert "10.0" in text + assert "CRITICAL" in text + assert "CISA KEV" in text + assert "2021-12-10" in text + + def test_error_formatting(self): + result = {"error": "Something went wrong"} + text = format_timeline_text(result) + assert "Error:" in text + assert "Something went wrong" in text + + def test_no_events_formatting(self): + result = { + "cve_id": "CVE-2024-0001", + "events": [], + "summary": {"cve_id": "CVE-2024-0001", "in_kev": False}, + "sources": {"nvd": "error: not found"}, + "event_count": 0, + } + text = format_timeline_text(result) + assert "No timeline events found" in text + + def test_key_intervals_shown(self): + result = { + "cve_id": "CVE-2021-44228", + "events": [], + "summary": { + "cve_id": "CVE-2021-44228", + "days_publish_to_exploit": 0, + "days_publish_to_epss_spike": 2, + "in_kev": True, + }, + "sources": {}, + "event_count": 0, + } + text = format_timeline_text(result) + assert "Key Intervals" in text + assert "0 days" in text + assert "2 days" in text + + def test_not_in_kev_text(self): + result = { + "cve_id": "CVE-2024-0001", + "events": [], + "summary": {"cve_id": "CVE-2024-0001", "in_kev": False}, + "sources": {}, + "event_count": 0, + } + text = format_timeline_text(result) + assert "Not in CISA KEV" in text + + +# --------------------------------------------------------------------------- +# JSON formatting tests +# --------------------------------------------------------------------------- + + +class TestFormatTimelineJson: + """Test JSON output formatting.""" + + def test_valid_json(self): + result = { + "cve_id": "CVE-2021-44228", + "events": [], + "summary": {}, + "sources": {}, + "event_count": 0, + } + output = format_timeline_json(result) + parsed = json.loads(output) + assert parsed["cve_id"] == "CVE-2021-44228" + + def test_preserves_all_fields(self): + result = { + "cve_id": "CVE-2021-44228", + "events": [{"date": "2021-12-10", "event_type": "nvd_published", "description": "x"}], + "summary": {"cvss_score": 10.0}, + "sources": {"nvd": "ok"}, + "event_count": 1, + } + output = format_timeline_json(result) + parsed = json.loads(output) + assert parsed["events"][0]["event_type"] == "nvd_published" + assert parsed["summary"]["cvss_score"] == 10.0 + + +# --------------------------------------------------------------------------- +# Strands handler tests +# --------------------------------------------------------------------------- + + +class TestHandler: + """Test the Strands tool handler.""" + + def test_handler_success(self): + with patch("manus_agent.tools.cve_timeline.build_timeline") as mock_build: + mock_build.return_value = { + "cve_id": "CVE-2021-44228", + "events": [], + "summary": {}, + "sources": {}, + "event_count": 0, + } + tool_use: ToolUse = { + "toolUseId": "test-1", + "name": "cve_timeline", + "input": {"cve_id": "CVE-2021-44228"}, + } + result = handler(tool_use) + assert result["status"] == "success" + assert len(result["content"]) == 1 + + def test_handler_error_missing_cve_id(self): + tool_use: ToolUse = { + "toolUseId": "test-2", + "name": "cve_timeline", + "input": {}, + } + result = handler(tool_use) + assert result["status"] == "error" + assert "required" in result["content"][0]["text"].lower() + + def test_handler_error_invalid_cve(self): + tool_use: ToolUse = { + "toolUseId": "test-3", + "name": "cve_timeline", + "input": {"cve_id": "INVALID"}, + } + result = handler(tool_use) + assert result["status"] == "error" + + def test_handler_returns_json(self): + with patch("manus_agent.tools.cve_timeline.build_timeline") as mock_build: + mock_build.return_value = { + "cve_id": "CVE-2021-44228", + "events": [ + { + "date": "2021-12-10", + "event_type": "nvd_published", + "description": "test", + } + ], + "summary": {}, + "sources": {}, + "event_count": 1, + } + tool_use: ToolUse = { + "toolUseId": "test-4", + "name": "cve_timeline", + "input": {"cve_id": "CVE-2021-44228"}, + } + result = handler(tool_use) + content_text = result["content"][0]["text"] + # Should be valid JSON + parsed = json.loads(content_text) + assert parsed["event_count"] == 1 + + +# --------------------------------------------------------------------------- +# CLI subcommand tests +# --------------------------------------------------------------------------- + + +class TestCliSubcommand: + """Test CLI dispatch for cve-timeline.""" + + def test_parser_creation(self): + from manus_agent.cli import _build_cve_timeline_parser + + parser = _build_cve_timeline_parser() + args = parser.parse_args(["CVE-2021-44228"]) + assert args.cve_id == "CVE-2021-44228" + assert args.output == "text" + + def test_parser_json_output(self): + from manus_agent.cli import _build_cve_timeline_parser + + parser = _build_cve_timeline_parser() + args = parser.parse_args(["CVE-2021-44228", "--output", "json"]) + assert args.output == "json" + + def test_run_text_output(self, capsys): + from manus_agent.cli import _run_cve_timeline + + with patch("manus_agent.tools.cve_timeline.build_timeline") as mock_build: + mock_build.return_value = { + "cve_id": "CVE-2021-44228", + "events": [ + { + "date": "2021-12-10", + "event_type": "nvd_published", + "description": "CVE published in NVD", + } + ], + "summary": {"cve_id": "CVE-2021-44228", "in_kev": False}, + "sources": {"nvd": "ok"}, + "event_count": 1, + } + exit_code = _run_cve_timeline(["CVE-2021-44228"]) + assert exit_code == 0 + captured = capsys.readouterr() + assert "CVE-2021-44228" in captured.out + + def test_run_json_output(self, capsys): + from manus_agent.cli import _run_cve_timeline + + with patch("manus_agent.tools.cve_timeline.build_timeline") as mock_build: + mock_build.return_value = { + "cve_id": "CVE-2021-44228", + "events": [], + "summary": {}, + "sources": {}, + "event_count": 0, + } + exit_code = _run_cve_timeline(["CVE-2021-44228", "--output", "json"]) + assert exit_code == 0 + captured = capsys.readouterr() + parsed = json.loads(captured.out) + assert parsed["cve_id"] == "CVE-2021-44228" + + def test_run_error_exit_code(self, capsys): + from manus_agent.cli import _run_cve_timeline + + with patch("manus_agent.tools.cve_timeline.build_timeline") as mock_build: + mock_build.return_value = {"error": "Invalid CVE ID format: BAD"} + exit_code = _run_cve_timeline(["BAD"]) + assert exit_code == 1 + + def test_subcommand_registered(self): + from manus_agent.cli import _SUBCOMMANDS + + assert "cve-timeline" in _SUBCOMMANDS + + def test_dispatch_in_main(self): + """Verify cve-timeline is dispatched in main().""" + import inspect + + import manus_agent.cli as cli_module + + source = inspect.getsource(cli_module.main) + assert "cve-timeline" in source + assert "_run_cve_timeline" in source + + +# --------------------------------------------------------------------------- +# Event icon tests +# --------------------------------------------------------------------------- + + +class TestEventIcons: + """Test _event_icon function.""" + + def test_known_icons(self): + from manus_agent.tools.cve_timeline import _event_icon + + assert _event_icon("nvd_published") == "📋" + assert _event_icon("kev_added") == "🚨" + assert _event_icon("epss_spike") == "📈" + assert _event_icon("patch_released") == "🩹" + + def test_unknown_icon_fallback(self): + from manus_agent.tools.cve_timeline import _event_icon + + assert _event_icon("unknown_event") == "•" + + +# --------------------------------------------------------------------------- +# Edge case tests +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + """Test edge cases and boundary conditions.""" + + def test_cve_with_five_digit_number(self): + """CVE-2024-12345 format should be valid.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_get.return_value = mock_resp + result = build_timeline("CVE-2024-12345") + assert result["cve_id"] == "CVE-2024-12345" + + def test_nvd_no_metrics(self): + """NVD entry with no CVSS metrics should still work.""" + nvd_data = { + "vulnerabilities": [ + { + "cve": { + "published": "2024-01-01T00:00:00.000", + "lastModified": "2024-01-01T00:00:00.000", + "metrics": {}, + "references": [], + } + } + ] + } + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = nvd_data + mock_get.return_value = mock_resp + result = _fetch_nvd_dates("CVE-2024-0001") + assert "cvss_score" not in result + assert result["published"] == "2024-01-01" + + def test_same_publish_and_modify_date(self): + """Same publish/modify date shouldn't create duplicate events.""" + nvd_data = { + "vulnerabilities": [ + { + "cve": { + "published": "2024-01-01T00:00:00.000", + "lastModified": "2024-01-01T00:00:00.000", + "metrics": {}, + "references": [], + } + } + ] + } + epss_data = {"data": []} + kev_data = {"vulnerabilities": []} + + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_nvd = MagicMock() + mock_nvd.json.return_value = nvd_data + mock_epss = MagicMock() + mock_epss.json.return_value = epss_data + mock_kev = MagicMock() + mock_kev.json.return_value = kev_data + mock_get.side_effect = [mock_nvd, mock_epss, mock_kev] + + result = build_timeline("CVE-2024-0001") + # Should only have one event (published), not two + nvd_events = [e for e in result["events"] if e["event_type"] in ("nvd_published", "nvd_modified")] + assert len(nvd_events) == 1 + + def test_epss_alternative_key_format(self): + """Handle EPSS API using 'timeSeries' key instead of 'time-series'.""" + epss_data = { + "data": [ + { + "cve": "CVE-2024-0001", + "timeSeries": [ + {"date": "2024-01-01", "epss": "0.05", "percentile": "0.60"}, + {"date": "2024-01-02", "epss": "0.06", "percentile": "0.61"}, + ], + } + ] + } + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = epss_data + mock_get.return_value = mock_resp + + result = _fetch_epss_history("CVE-2024-0001") + assert result["current_score"] == 0.06 + assert result["history_points"] == 2 + + def test_large_cve_number(self): + """CVE with very long number should be accepted.""" + with patch("manus_agent.tools.cve_timeline._get_with_retry") as mock_get: + mock_resp = MagicMock() + mock_resp.json.return_value = {"vulnerabilities": []} + mock_get.return_value = mock_resp + result = build_timeline("CVE-2024-1234567") + assert result["cve_id"] == "CVE-2024-1234567"