diff --git a/src/manus_agent/cli.py b/src/manus_agent/cli.py index e8442f2..00893a4 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", + "temporal-priority", } @@ -1935,6 +1936,108 @@ def _run_blast_radius(argv: list[str]) -> int: return 0 +# --------------------------------------------------------------------------- +# temporal-priority subcommand +# --------------------------------------------------------------------------- + + +def _build_temporal_priority_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="manus-agent temporal-priority", + description=( + "Compute a temporal priority score (0–100) for a CVE combining CVSS base score, " + "current EPSS, EPSS spike recency, CISA KEV membership, patch availability, and " + "CVE age. Answers: 'given everything I know today, how urgent is this?'" + ), + add_help=True, + ) + p.add_argument("cve_id", metavar="CVE-ID", help="CVE identifier, e.g. CVE-2024-3094") + p.add_argument( + "--output", + choices=["text", "json"], + default="text", + help="Output format (default: text)", + ) + return p + + +def _run_temporal_priority(argv: list[str]) -> int: + import json as _json + + parser = _build_temporal_priority_parser() + args = parser.parse_args(argv) + cve_id = args.cve_id.strip() + if not cve_id: + parser.error("CVE-ID is required") + + try: + from manus_agent.tools.temporal_priority import compute_temporal_priority + except ImportError as exc: # pragma: no cover + print(f"[error] missing dependencies: {exc}", file=sys.stderr) + return 1 + + try: + data = compute_temporal_priority(cve_id) + except Exception as exc: + print(f"[error] Failed to compute temporal priority: {exc}", file=sys.stderr) + return 1 + + if args.output == "json": + print(_json.dumps(data, indent=2)) + return 0 + + # Text output + print() + print(f"Temporal Priority — {data['cve_id']}") + print("=" * 60) + print(f" Score: {data['score']}/100 ({data['label']})") + print() + print("Signal Breakdown:") + print("-" * 60) + + signals = data.get("signals", {}) + + # CVSS + cvss = signals.get("cvss", {}) + raw_cvss = cvss.get("raw_score") + cvss_str = f"{raw_cvss}/10" if raw_cvss is not None else "N/A" + print(f" CVSS Base Score: {cvss_str:>10} (weight {cvss.get('weight', 0):.0%})") + + # EPSS current + epss = signals.get("epss_current", {}) + raw_epss = epss.get("raw_epss") + epss_str = f"{raw_epss:.4f}" if raw_epss is not None else "N/A" + print(f" EPSS Current: {epss_str:>10} (weight {epss.get('weight', 0):.0%})") + + # EPSS spike + spike = signals.get("epss_spike", {}) + if spike.get("spike_detected"): + spike_str = f"+{spike.get('max_jump', 0):.4f} ({spike.get('days_ago', '?')}d ago)" + else: + spike_str = "none detected" + print(f" EPSS Spike: {spike_str:>10} (weight {spike.get('weight', 0):.0%})") + + # KEV + kev = signals.get("cisa_kev", {}) + kev_str = "YES ⚠" if kev.get("in_kev") else "no" + print(f" CISA KEV: {kev_str:>10} (weight {kev.get('weight', 0):.0%})") + + # Patch + patch = signals.get("patch_availability", {}) + patch_str = "available" if patch.get("has_patch") else "NOT FOUND" + print(f" Patch: {patch_str:>10} (weight {patch.get('weight', 0):.0%})") + + # Age + age = signals.get("age", {}) + age_days = age.get("age_days") + age_str = f"{age_days}d" if age_days is not None else "N/A" + print(f" CVE Age: {age_str:>10} (weight {age.get('weight', 0):.0%})") + + print() + print(f"Interpretation: {data.get('interpretation', '')}") + return 0 + + def _build_run_parser() -> argparse.ArgumentParser: """Build the top-level run/interactive parser.""" parser = argparse.ArgumentParser( @@ -2269,6 +2372,10 @@ def main() -> None: idx = argv.index("blast-radius") sys.exit(_run_blast_radius(argv[idx + 1 :])) + if first_positional == "temporal-priority": + idx = argv.index("temporal-priority") + sys.exit(_run_temporal_priority(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/temporal_priority.py b/src/manus_agent/tools/temporal_priority.py new file mode 100644 index 0000000..2ea0350 --- /dev/null +++ b/src/manus_agent/tools/temporal_priority.py @@ -0,0 +1,460 @@ +""" +Tool for computing a temporal priority score (0–100) for a CVE. + +Combines multiple signals to answer: "given everything I know today, how urgent +is this vulnerability?" + +Signals +------- +1. CVSS base score (from NVD) +2. Current EPSS score (from FIRST.org) +3. EPSS spike recency (recent jump → higher urgency) +4. CISA KEV membership (active exploitation) +5. Patch availability (NVD references tagged "Patch") +6. CVE age (newer CVEs get a recency boost) + +Each signal is normalised to 0–1 and combined via configurable weights to produce +a final 0–100 urgency score with a human-readable urgency label. +""" + +from __future__ import annotations + +import math +import os +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 + +# --------------------------------------------------------------------------- +# TOOL_SPEC — Strands SDK interface +# --------------------------------------------------------------------------- + +TOOL_SPEC = { + "name": "temporal_priority", + "description": ( + "Computes a temporal priority score (0–100) for a CVE combining CVSS base score, " + "current EPSS, EPSS spike recency, CISA KEV membership, patch availability, and " + "CVE age. Answers: 'given everything I know today, how urgent is this?' " + "Use after get_nvd_data and get_epss_trend for a single actionable urgency number." + ), + "inputSchema": { + "json": { + "type": "object", + "properties": { + "cve_id": { + "type": "string", + "description": "The CVE identifier to score (e.g., 'CVE-2024-3094').", + }, + }, + "required": ["cve_id"], + } + }, +} + +# --------------------------------------------------------------------------- +# Weights (sum to 1.0) — override via environment for tuning/testing +# --------------------------------------------------------------------------- + +_W_CVSS = float(os.environ.get("TP_W_CVSS", "0.25")) +_W_EPSS = float(os.environ.get("TP_W_EPSS", "0.25")) +_W_SPIKE = float(os.environ.get("TP_W_SPIKE", "0.15")) +_W_KEV = float(os.environ.get("TP_W_KEV", "0.20")) +_W_PATCH = float(os.environ.get("TP_W_PATCH", "0.05")) +_W_AGE = float(os.environ.get("TP_W_AGE", "0.10")) + +# --------------------------------------------------------------------------- +# Retry / back-off constants +# --------------------------------------------------------------------------- + +_MAX_RETRIES = int(os.environ.get("TP_MAX_RETRIES", "3")) +_RETRY_BASE_DELAY = float(os.environ.get("TP_RETRY_BASE_DELAY", "1.5")) +_RETRYABLE_STATUS = {429, 500, 502, 503, 504} + +# EPSS spike thresholds +_SPIKE_THRESHOLD = 0.05 # absolute jump considered a spike +_SPIKE_RECENCY_HALFLIFE_DAYS = 14 # exponential decay half-life + + +# --------------------------------------------------------------------------- +# HTTP helper with retry/back-off +# --------------------------------------------------------------------------- + + +def _get_with_retry(url: str, params: dict | None = None, headers: dict | None = None, timeout: int = 20) -> dict: + """HTTP GET with exponential back-off on retryable failures.""" + last_exc: Exception | None = None + for attempt in range(_MAX_RETRIES): + try: + resp = requests.get(url, params=params, headers=headers, timeout=timeout) + if resp.status_code in _RETRYABLE_STATUS and attempt < _MAX_RETRIES - 1: + time.sleep(_RETRY_BASE_DELAY * (2**attempt)) + continue + resp.raise_for_status() + return resp.json() + except (requests.exceptions.RequestException, ValueError) as exc: + last_exc = exc + if attempt < _MAX_RETRIES - 1: + time.sleep(_RETRY_BASE_DELAY * (2**attempt)) + raise last_exc # type: ignore[misc] + + +# --------------------------------------------------------------------------- +# Signal collectors +# --------------------------------------------------------------------------- + + +def _fetch_nvd(cve_id: str) -> dict[str, Any]: + """Fetch NVD CVE data. Returns parsed JSON or empty dict on failure.""" + url = "https://services.nvd.nist.gov/rest/json/cves/2.0" + headers: dict[str, str] = {} + api_key = os.environ.get("NVD_API_KEY", "") + if api_key: + headers["apiKey"] = api_key + params = {"cveId": cve_id.upper()} + try: + return _get_with_retry(url, params=params, headers=headers) + except Exception: + return {} + + +def _fetch_epss(cve_id: str) -> dict[str, Any]: + """Fetch current EPSS score and time-series. Returns parsed JSON or empty dict.""" + url = "https://api.first.org/data/v1/epss" + params = {"cve": cve_id.upper(), "scope": "time-series", "limit": "30"} + try: + return _get_with_retry(url, params=params) + except Exception: + return {} + + +def _fetch_kev() -> set[str]: + """Fetch the CISA KEV catalog and return a set of CVE IDs.""" + url = "https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json" + try: + data = _get_with_retry(url, timeout=15) + vulns = data.get("vulnerabilities", []) + return {v.get("cveID", "").upper() for v in vulns} + except Exception: + return set() + + +# --------------------------------------------------------------------------- +# Signal scoring functions (each returns 0.0–1.0) +# --------------------------------------------------------------------------- + + +def score_cvss(nvd_data: dict) -> tuple[float, float | None]: + """Extract best CVSS score and normalise to 0–1. Returns (normalised, raw_score).""" + vulns = nvd_data.get("vulnerabilities", []) + if not vulns: + return 0.0, None + cve_item = vulns[0].get("cve", {}) + metrics = cve_item.get("metrics", {}) + + # Try CVSS 3.1 first, then 3.0, then 2.0 + best_score: float | None = None + for key in ("cvssMetricV31", "cvssMetricV30", "cvssMetricV2"): + entries = metrics.get(key, []) + if entries: + cvss_data = entries[0].get("cvssData", {}) + score = cvss_data.get("baseScore") + if score is not None: + best_score = float(score) + break + + if best_score is None: + return 0.0, None + + # CVSS 2.0 is out of 10, same as 3.x + return best_score / 10.0, best_score + + +def score_epss_current(epss_data: dict) -> tuple[float, float | None]: + """Extract latest EPSS score (already 0–1). Returns (score, raw_epss).""" + data_entries = epss_data.get("data", []) + if not data_entries: + return 0.0, None + entry = data_entries[0] + # Current score is top-level epss field + epss_val = entry.get("epss") + if epss_val is not None: + val = float(epss_val) + return val, val + # Fallback: first time-series point + ts = entry.get("time-series", []) + if ts: + val = float(ts[0].get("epss", 0)) + return val, val + return 0.0, None + + +def score_epss_spike(epss_data: dict) -> tuple[float, dict[str, Any]]: + """ + Detect the largest recent spike in EPSS and score its urgency via exponential decay. + Returns (spike_score_0_1, details_dict). + """ + data_entries = epss_data.get("data", []) + if not data_entries: + return 0.0, {"max_jump": 0, "spike_detected": False} + + entry = data_entries[0] + ts = list(entry.get("time-series", [])) + if len(ts) < 2: + return 0.0, {"max_jump": 0, "spike_detected": False} + + # Sort oldest-first + points = sorted(ts, key=lambda x: x.get("date", "")) + max_jump = 0.0 + max_jump_date: str | None = None + for i in range(1, len(points)): + prev = float(points[i - 1].get("epss", 0)) + curr = float(points[i].get("epss", 0)) + jump = curr - prev + if jump > max_jump: + max_jump = jump + max_jump_date = points[i].get("date") + + spike_detected = max_jump >= _SPIKE_THRESHOLD + if not spike_detected or not max_jump_date: + return 0.0, {"max_jump": round(max_jump, 4), "spike_detected": False} + + # Exponential decay based on how recent the spike was + try: + spike_dt = datetime.strptime(max_jump_date, "%Y-%m-%d").replace(tzinfo=timezone.utc) + now = datetime.now(timezone.utc) + days_ago = max((now - spike_dt).days, 0) + except (ValueError, TypeError): + days_ago = 30 # fallback + + decay = math.exp(-0.693 * days_ago / _SPIKE_RECENCY_HALFLIFE_DAYS) # ln(2) ≈ 0.693 + # Scale: a spike of 0.10 at day 0 → 1.0; smaller spikes scale linearly + magnitude_factor = min(max_jump / 0.10, 1.0) + score = decay * magnitude_factor + + return min(score, 1.0), { + "max_jump": round(max_jump, 4), + "spike_date": max_jump_date, + "days_ago": days_ago, + "spike_detected": True, + } + + +def score_kev(cve_id: str, kev_set: set[str]) -> tuple[float, bool]: + """Return 1.0 if CVE is in KEV, 0.0 otherwise.""" + in_kev = cve_id.upper() in kev_set + return (1.0 if in_kev else 0.0), in_kev + + +def score_patch_availability(nvd_data: dict) -> tuple[float, bool]: + """ + Check NVD references for patch tags. Patch available → lower urgency (inverted). + Returns (score_0_1, has_patch). Higher score = MORE urgent (no patch). + """ + vulns = nvd_data.get("vulnerabilities", []) + if not vulns: + return 0.5, False # unknown → middle ground + + cve_item = vulns[0].get("cve", {}) + references = cve_item.get("references", []) + has_patch = any("Patch" in ref.get("tags", []) for ref in references) + + # No patch = higher urgency (1.0), patch available = lower urgency (0.0) + return (0.0 if has_patch else 1.0), has_patch + + +def score_age(nvd_data: dict) -> tuple[float, int | None]: + """ + Score based on CVE age. Newer CVEs are more urgent (less time for defenders). + Returns (score_0_1, age_days). + Uses exponential decay: 0 days → 1.0, ~90 days → 0.5, ~365 days → ~0.06. + """ + vulns = nvd_data.get("vulnerabilities", []) + if not vulns: + return 0.5, None # unknown → middle ground + + cve_item = vulns[0].get("cve", {}) + published = cve_item.get("published") + if not published: + return 0.5, None + + try: + # NVD format: "2024-03-29T13:15:00.000" + pub_dt = datetime.fromisoformat(published.replace("Z", "+00:00")) + if pub_dt.tzinfo is None: + pub_dt = pub_dt.replace(tzinfo=timezone.utc) + now = datetime.now(timezone.utc) + age_days = max((now - pub_dt).days, 0) + except (ValueError, TypeError): + return 0.5, None + + # Exponential decay with 90-day half-life + decay = math.exp(-0.693 * age_days / 90) + return decay, age_days + + +# --------------------------------------------------------------------------- +# Composite scorer +# --------------------------------------------------------------------------- + + +def compute_temporal_priority( + cve_id: str, + nvd_data: dict | None = None, + epss_data: dict | None = None, + kev_set: set[str] | None = None, +) -> dict[str, Any]: + """ + Compute the temporal priority score for a CVE. + + Parameters + ---------- + cve_id : str + CVE identifier. + nvd_data : dict, optional + Pre-fetched NVD API response. If None, will be fetched. + epss_data : dict, optional + Pre-fetched EPSS API response. If None, will be fetched. + kev_set : set, optional + Pre-fetched KEV CVE ID set. If None, will be fetched. + + Returns + ------- + dict with keys: score, label, signals, weights, cve_id + """ + cve_id = cve_id.upper().strip() + + # Fetch data if not provided + if nvd_data is None: + nvd_data = _fetch_nvd(cve_id) + if epss_data is None: + epss_data = _fetch_epss(cve_id) + if kev_set is None: + kev_set = _fetch_kev() + + # Compute individual signals + cvss_score, cvss_raw = score_cvss(nvd_data) + epss_score, epss_raw = score_epss_current(epss_data) + spike_score, spike_details = score_epss_spike(epss_data) + kev_score, in_kev = score_kev(cve_id, kev_set) + patch_score, has_patch = score_patch_availability(nvd_data) + age_score, age_days = score_age(nvd_data) + + # Weighted composite + raw_composite = ( + _W_CVSS * cvss_score + + _W_EPSS * epss_score + + _W_SPIKE * spike_score + + _W_KEV * kev_score + + _W_PATCH * patch_score + + _W_AGE * age_score + ) + + # Scale to 0–100 + final_score = round(min(max(raw_composite * 100, 0), 100), 1) + + # Urgency label + if final_score >= 80: + label = "CRITICAL" + elif final_score >= 60: + label = "HIGH" + elif final_score >= 40: + label = "MEDIUM" + elif final_score >= 20: + label = "LOW" + else: + label = "INFORMATIONAL" + + return { + "cve_id": cve_id, + "score": final_score, + "label": label, + "signals": { + "cvss": { + "normalised": round(cvss_score, 4), + "raw_score": cvss_raw, + "weight": _W_CVSS, + }, + "epss_current": { + "normalised": round(epss_score, 4), + "raw_epss": epss_raw, + "weight": _W_EPSS, + }, + "epss_spike": { + "normalised": round(spike_score, 4), + "weight": _W_SPIKE, + **spike_details, + }, + "cisa_kev": { + "normalised": round(kev_score, 4), + "in_kev": in_kev, + "weight": _W_KEV, + }, + "patch_availability": { + "normalised": round(patch_score, 4), + "has_patch": has_patch, + "weight": _W_PATCH, + "note": "higher = no patch (more urgent)", + }, + "age": { + "normalised": round(age_score, 4), + "age_days": age_days, + "weight": _W_AGE, + "note": "higher = newer CVE (more urgent)", + }, + }, + "interpretation": ( + f"{cve_id} has a temporal priority of {final_score}/100 ({label}). " + + (f"CVSS {cvss_raw}/10. " if cvss_raw else "") + + (f"EPSS {epss_raw:.4f}. " if epss_raw else "") + + ("In CISA KEV (actively exploited). " if in_kev else "") + + ("Patch available. " if has_patch else "No patch found. ") + + (f"Published {age_days} days ago." if age_days is not None else "") + ), + } + + +# --------------------------------------------------------------------------- +# Strands tool entry point +# --------------------------------------------------------------------------- + + +def temporal_priority(tool: ToolUse, **kwargs: Any) -> ToolResult: + """Strands SDK tool entry point.""" + 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(): + result: ToolResult = { + "toolUseId": tool_use_id, + "status": "error", + "content": [{"text": "Invalid CVE ID. Must be a non-empty string."}], + } + log_tool_output_size("temporal_priority", result) + return result + + try: + data = compute_temporal_priority(cve_id) + except Exception as exc: + result = { + "toolUseId": tool_use_id, + "status": "error", + "content": [{"text": f"Failed to compute temporal priority: {exc}"}], + } + log_tool_output_size("temporal_priority", result) + return result + + import json + + result = { + "toolUseId": tool_use_id, + "status": "success", + "content": [{"text": json.dumps(data, indent=2)}], + } + log_tool_output_size("temporal_priority", result) + return result diff --git a/tests/test_temporal_priority.py b/tests/test_temporal_priority.py new file mode 100644 index 0000000..1a3e3b4 --- /dev/null +++ b/tests/test_temporal_priority.py @@ -0,0 +1,811 @@ +""" +Comprehensive test suite for temporal_priority tool. + +Tests cover: +- TOOL_SPEC contract (Strands SDK interface) +- Input validation +- HTTP helper retry/back-off +- Individual signal scoring functions +- Composite scoring +- Tool entry point +- CLI subcommand +- Edge cases +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from manus_agent.tools.temporal_priority import ( + _SPIKE_THRESHOLD, + TOOL_SPEC, + _get_with_retry, + compute_temporal_priority, + score_age, + score_cvss, + score_epss_current, + score_epss_spike, + score_kev, + score_patch_availability, + temporal_priority, +) + +# --------------------------------------------------------------------------- +# TOOL_SPEC contract tests +# --------------------------------------------------------------------------- + + +class TestToolSpec: + """Verify the TOOL_SPEC satisfies Strands SDK requirements.""" + + def test_has_name(self): + assert TOOL_SPEC["name"] == "temporal_priority" + + 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_type(self): + prop = TOOL_SPEC["inputSchema"]["json"]["properties"]["cve_id"] + assert prop["type"] == "string" + assert "description" in prop + + +# --------------------------------------------------------------------------- +# Input validation tests +# --------------------------------------------------------------------------- + + +class TestInputValidation: + """Verify the tool entry point validates input.""" + + def test_empty_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-1", "input": {"cve_id": ""}} + result = temporal_priority(tool_use) + assert result["status"] == "error" + assert "Invalid CVE ID" in result["content"][0]["text"] + + def test_none_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-2", "input": {"cve_id": None}} + result = temporal_priority(tool_use) + assert result["status"] == "error" + + def test_whitespace_only_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-3", "input": {"cve_id": " "}} + result = temporal_priority(tool_use) + assert result["status"] == "error" + + def test_missing_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-4", "input": {}} + result = temporal_priority(tool_use) + assert result["status"] == "error" + + def test_non_string_cve_id_returns_error(self): + tool_use = {"toolUseId": "test-5", "input": {"cve_id": 12345}} + result = temporal_priority(tool_use) + assert result["status"] == "error" + + +# --------------------------------------------------------------------------- +# HTTP helper tests +# --------------------------------------------------------------------------- + + +class TestGetWithRetry: + """Test the _get_with_retry HTTP helper.""" + + @patch("manus_agent.tools.temporal_priority.requests.get") + def test_success_first_try(self, mock_get): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"result": "ok"} + mock_resp.raise_for_status = MagicMock() + mock_get.return_value = mock_resp + + result = _get_with_retry("https://example.com/api") + assert result == {"result": "ok"} + assert mock_get.call_count == 1 + + @patch("manus_agent.tools.temporal_priority.time.sleep") + @patch("manus_agent.tools.temporal_priority.requests.get") + def test_retry_on_429(self, mock_get, mock_sleep): + fail_resp = MagicMock() + fail_resp.status_code = 429 + + ok_resp = MagicMock() + ok_resp.status_code = 200 + ok_resp.json.return_value = {"data": "yes"} + ok_resp.raise_for_status = MagicMock() + + mock_get.side_effect = [fail_resp, ok_resp] + result = _get_with_retry("https://example.com/api") + assert result == {"data": "yes"} + assert mock_get.call_count == 2 + mock_sleep.assert_called_once() + + @patch("manus_agent.tools.temporal_priority.time.sleep") + @patch("manus_agent.tools.temporal_priority.requests.get") + def test_retry_on_503(self, mock_get, mock_sleep): + fail_resp = MagicMock() + fail_resp.status_code = 503 + + ok_resp = MagicMock() + ok_resp.status_code = 200 + ok_resp.json.return_value = {"ok": True} + ok_resp.raise_for_status = MagicMock() + + mock_get.side_effect = [fail_resp, ok_resp] + result = _get_with_retry("https://example.com/api") + assert result == {"ok": True} + + @patch("manus_agent.tools.temporal_priority._MAX_RETRIES", 2) + @patch("manus_agent.tools.temporal_priority.time.sleep") + @patch("manus_agent.tools.temporal_priority.requests.get") + def test_exhausted_retries_raises(self, mock_get, mock_sleep): + import requests as req + + mock_get.side_effect = req.exceptions.ConnectionError("timeout") + with pytest.raises(req.exceptions.ConnectionError): + _get_with_retry("https://example.com/api") + assert mock_get.call_count == 2 + + @patch("manus_agent.tools.temporal_priority._MAX_RETRIES", 3) + @patch("manus_agent.tools.temporal_priority.time.sleep") + @patch("manus_agent.tools.temporal_priority.requests.get") + def test_non_retryable_status_exhausts_retries(self, mock_get, mock_sleep): + """Non-retryable HTTP errors caught by the except block still retry.""" + import requests as req + + mock_resp = MagicMock() + mock_resp.status_code = 404 + mock_resp.raise_for_status.side_effect = req.exceptions.HTTPError("404") + mock_get.return_value = mock_resp + + with pytest.raises(req.exceptions.HTTPError): + _get_with_retry("https://example.com/api") + # The HTTPError is caught by the generic except, so all retries are used + assert mock_get.call_count == 3 + + +# --------------------------------------------------------------------------- +# Signal scoring tests — CVSS +# --------------------------------------------------------------------------- + + +class TestScoreCvss: + """Test CVSS signal scoring.""" + + def test_cvss31_extraction(self): + nvd = {"vulnerabilities": [{"cve": {"metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 9.8}}]}}}]} + norm, raw = score_cvss(nvd) + assert raw == 9.8 + assert norm == pytest.approx(0.98, abs=0.001) + + def test_cvss30_fallback(self): + nvd = {"vulnerabilities": [{"cve": {"metrics": {"cvssMetricV30": [{"cvssData": {"baseScore": 7.5}}]}}}]} + norm, raw = score_cvss(nvd) + assert raw == 7.5 + assert norm == pytest.approx(0.75, abs=0.001) + + def test_cvss2_fallback(self): + nvd = {"vulnerabilities": [{"cve": {"metrics": {"cvssMetricV2": [{"cvssData": {"baseScore": 5.0}}]}}}]} + norm, raw = score_cvss(nvd) + assert raw == 5.0 + assert norm == pytest.approx(0.5, abs=0.001) + + def test_empty_vulnerabilities(self): + norm, raw = score_cvss({"vulnerabilities": []}) + assert norm == 0.0 + assert raw is None + + def test_no_metrics(self): + nvd = {"vulnerabilities": [{"cve": {"metrics": {}}}]} + norm, raw = score_cvss(nvd) + assert norm == 0.0 + assert raw is None + + def test_perfect_10(self): + nvd = {"vulnerabilities": [{"cve": {"metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 10.0}}]}}}]} + norm, raw = score_cvss(nvd) + assert norm == 1.0 + assert raw == 10.0 + + +# --------------------------------------------------------------------------- +# Signal scoring tests — EPSS current +# --------------------------------------------------------------------------- + + +class TestScoreEpssCurrent: + """Test EPSS current score extraction.""" + + def test_top_level_epss(self): + data = {"data": [{"epss": "0.95432"}]} + norm, raw = score_epss_current(data) + assert raw == pytest.approx(0.95432, abs=0.0001) + assert norm == pytest.approx(0.95432, abs=0.0001) + + def test_fallback_to_timeseries(self): + data = {"data": [{"time-series": [{"epss": "0.123", "date": "2025-01-01"}]}]} + norm, raw = score_epss_current(data) + assert raw == pytest.approx(0.123, abs=0.001) + + def test_empty_data(self): + norm, raw = score_epss_current({"data": []}) + assert norm == 0.0 + assert raw is None + + def test_no_data_key(self): + norm, raw = score_epss_current({}) + assert norm == 0.0 + assert raw is None + + def test_zero_epss(self): + data = {"data": [{"epss": "0.0"}]} + norm, raw = score_epss_current(data) + assert norm == 0.0 + assert raw == 0.0 + + +# --------------------------------------------------------------------------- +# Signal scoring tests — EPSS spike +# --------------------------------------------------------------------------- + + +class TestScoreEpssSpike: + """Test EPSS spike detection and scoring.""" + + def test_no_data_returns_zero(self): + score, details = score_epss_spike({"data": []}) + assert score == 0.0 + assert details["spike_detected"] is False + + def test_single_point_no_spike(self): + data = {"data": [{"time-series": [{"epss": "0.5", "date": "2025-01-01"}]}]} + score, details = score_epss_spike(data) + assert score == 0.0 + assert details["spike_detected"] is False + + def test_no_significant_jump(self): + data = { + "data": [ + { + "time-series": [ + {"epss": "0.01", "date": "2025-01-01"}, + {"epss": "0.02", "date": "2025-01-02"}, + {"epss": "0.025", "date": "2025-01-03"}, + ] + } + ] + } + score, details = score_epss_spike(data) + assert score == 0.0 + assert details["spike_detected"] is False + assert details["max_jump"] < _SPIKE_THRESHOLD + + @patch("manus_agent.tools.temporal_priority.datetime") + def test_recent_spike_high_score(self, mock_dt): + # Simulate a spike that happened "today" + mock_dt.now.return_value = datetime(2025, 7, 10, tzinfo=timezone.utc) + mock_dt.strptime = datetime.strptime + + data = { + "data": [ + { + "time-series": [ + {"epss": "0.01", "date": "2025-07-08"}, + {"epss": "0.02", "date": "2025-07-09"}, + {"epss": "0.12", "date": "2025-07-10"}, # +0.10 spike + ] + } + ] + } + score, details = score_epss_spike(data) + assert details["spike_detected"] is True + assert details["max_jump"] == pytest.approx(0.10, abs=0.001) + # Recent spike → high score (close to 1.0) + assert score > 0.8 + + @patch("manus_agent.tools.temporal_priority.datetime") + def test_old_spike_decayed(self, mock_dt): + # Simulate a spike from 60 days ago + mock_dt.now.return_value = datetime(2025, 9, 8, tzinfo=timezone.utc) + mock_dt.strptime = datetime.strptime + + data = { + "data": [ + { + "time-series": [ + {"epss": "0.01", "date": "2025-07-09"}, + {"epss": "0.12", "date": "2025-07-10"}, # +0.11 spike, 60 days ago + ] + } + ] + } + score, details = score_epss_spike(data) + assert details["spike_detected"] is True + # 60 days with 14-day half-life → very decayed + assert score < 0.1 + + +# --------------------------------------------------------------------------- +# Signal scoring tests — KEV +# --------------------------------------------------------------------------- + + +class TestScoreKev: + """Test CISA KEV membership scoring.""" + + def test_in_kev(self): + score, in_kev = score_kev("CVE-2024-3094", {"CVE-2024-3094", "CVE-2021-44228"}) + assert score == 1.0 + assert in_kev is True + + def test_not_in_kev(self): + score, in_kev = score_kev("CVE-2099-9999", {"CVE-2024-3094"}) + assert score == 0.0 + assert in_kev is False + + def test_case_insensitive(self): + score, in_kev = score_kev("cve-2024-3094", {"CVE-2024-3094"}) + assert score == 1.0 + assert in_kev is True + + def test_empty_kev_set(self): + score, in_kev = score_kev("CVE-2024-3094", set()) + assert score == 0.0 + assert in_kev is False + + +# --------------------------------------------------------------------------- +# Signal scoring tests — Patch availability +# --------------------------------------------------------------------------- + + +class TestScorePatchAvailability: + """Test patch availability scoring.""" + + def test_patch_available(self): + nvd = { + "vulnerabilities": [ + { + "cve": { + "references": [ + {"url": "https://example.com/fix", "tags": ["Patch"]}, + ] + } + } + ] + } + score, has_patch = score_patch_availability(nvd) + assert score == 0.0 # patch present → lower urgency + assert has_patch is True + + def test_no_patch(self): + nvd = { + "vulnerabilities": [ + { + "cve": { + "references": [ + {"url": "https://example.com/advisory", "tags": ["Third Party Advisory"]}, + ] + } + } + ] + } + score, has_patch = score_patch_availability(nvd) + assert score == 1.0 # no patch → higher urgency + assert has_patch is False + + def test_empty_references(self): + nvd = {"vulnerabilities": [{"cve": {"references": []}}]} + score, has_patch = score_patch_availability(nvd) + assert score == 1.0 + assert has_patch is False + + def test_empty_vulnerabilities_returns_middle(self): + score, has_patch = score_patch_availability({"vulnerabilities": []}) + assert score == 0.5 + assert has_patch is False + + def test_multiple_refs_one_patch(self): + nvd = { + "vulnerabilities": [ + { + "cve": { + "references": [ + {"url": "https://example.com/cve", "tags": ["Vendor Advisory"]}, + {"url": "https://github.com/fix", "tags": ["Patch", "Vendor Advisory"]}, + ] + } + } + ] + } + score, has_patch = score_patch_availability(nvd) + assert score == 0.0 + assert has_patch is True + + +# --------------------------------------------------------------------------- +# Signal scoring tests — Age +# --------------------------------------------------------------------------- + + +class TestScoreAge: + """Test CVE age scoring.""" + + @patch("manus_agent.tools.temporal_priority.datetime") + def test_brand_new_cve(self, mock_dt): + mock_dt.now.return_value = datetime(2025, 7, 10, 12, 0, tzinfo=timezone.utc) + mock_dt.fromisoformat = datetime.fromisoformat + + nvd = {"vulnerabilities": [{"cve": {"published": "2025-07-10T10:00:00.000"}}]} + score, age_days = score_age(nvd) + assert age_days == 0 + assert score == pytest.approx(1.0, abs=0.01) + + @patch("manus_agent.tools.temporal_priority.datetime") + def test_90_day_old_cve(self, mock_dt): + mock_dt.now.return_value = datetime(2025, 10, 8, 12, 0, tzinfo=timezone.utc) + mock_dt.fromisoformat = datetime.fromisoformat + + nvd = {"vulnerabilities": [{"cve": {"published": "2025-07-10T12:00:00.000"}}]} + score, age_days = score_age(nvd) + assert age_days == pytest.approx(90, abs=2) + # 90-day half-life → ~0.5 + assert score == pytest.approx(0.5, abs=0.05) + + @patch("manus_agent.tools.temporal_priority.datetime") + def test_year_old_cve(self, mock_dt): + mock_dt.now.return_value = datetime(2026, 7, 10, 12, 0, tzinfo=timezone.utc) + mock_dt.fromisoformat = datetime.fromisoformat + + nvd = {"vulnerabilities": [{"cve": {"published": "2025-07-10T12:00:00.000"}}]} + score, age_days = score_age(nvd) + assert age_days == pytest.approx(365, abs=2) + # Very old → very low + assert score < 0.1 + + def test_no_published_date(self): + nvd = {"vulnerabilities": [{"cve": {}}]} + score, age_days = score_age(nvd) + assert score == 0.5 + assert age_days is None + + def test_empty_vulnerabilities(self): + score, age_days = score_age({"vulnerabilities": []}) + assert score == 0.5 + assert age_days is None + + +# --------------------------------------------------------------------------- +# Composite scoring tests +# --------------------------------------------------------------------------- + + +class TestComputeTemporalPriority: + """Test the composite temporal priority scorer.""" + + def test_critical_cve_all_signals_high(self): + """A CVE with high CVSS, high EPSS, in KEV, no patch, new → critical.""" + nvd = { + "vulnerabilities": [ + { + "cve": { + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 10.0}}]}, + "references": [], + "published": datetime.now(timezone.utc).isoformat(), + } + } + ] + } + epss = {"data": [{"epss": "0.97", "time-series": []}]} + kev_set = {"CVE-2024-9999"} + + result = compute_temporal_priority("CVE-2024-9999", nvd_data=nvd, epss_data=epss, kev_set=kev_set) + assert result["score"] >= 80 + assert result["label"] == "CRITICAL" + assert result["cve_id"] == "CVE-2024-9999" + + def test_low_risk_cve(self): + """A CVE with low CVSS, low EPSS, not in KEV, patch available, old → low.""" + nvd = { + "vulnerabilities": [ + { + "cve": { + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 2.0}}]}, + "references": [{"url": "https://fix.com", "tags": ["Patch"]}], + "published": "2020-01-01T00:00:00.000", + } + } + ] + } + epss = {"data": [{"epss": "0.001", "time-series": []}]} + kev_set = set() + + result = compute_temporal_priority("CVE-2020-0001", nvd_data=nvd, epss_data=epss, kev_set=kev_set) + assert result["score"] < 30 + assert result["label"] in ("LOW", "INFORMATIONAL") + + def test_medium_risk_cve(self): + """A CVE with moderate signals → medium.""" + nvd = { + "vulnerabilities": [ + { + "cve": { + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 7.0}}]}, + "references": [], + "published": "2025-03-01T00:00:00.000", + } + } + ] + } + epss = {"data": [{"epss": "0.3", "time-series": []}]} + kev_set = set() + + result = compute_temporal_priority("CVE-2025-1234", nvd_data=nvd, epss_data=epss, kev_set=kev_set) + assert 30 <= result["score"] <= 70 + + def test_score_bounded_0_100(self): + """Score should always be 0–100.""" + nvd = {"vulnerabilities": []} + epss = {"data": []} + kev_set = set() + + result = compute_temporal_priority("CVE-0000-0000", nvd_data=nvd, epss_data=epss, kev_set=kev_set) + assert 0 <= result["score"] <= 100 + + def test_kev_membership_significant_boost(self): + """Being in KEV should significantly boost the score.""" + nvd = { + "vulnerabilities": [ + { + "cve": { + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 5.0}}]}, + "references": [], + "published": "2024-06-01T00:00:00.000", + } + } + ] + } + epss = {"data": [{"epss": "0.1", "time-series": []}]} + + result_no_kev = compute_temporal_priority("CVE-2024-5555", nvd_data=nvd, epss_data=epss, kev_set=set()) + result_with_kev = compute_temporal_priority( + "CVE-2024-5555", nvd_data=nvd, epss_data=epss, kev_set={"CVE-2024-5555"} + ) + + assert result_with_kev["score"] > result_no_kev["score"] + # KEV weight is 0.20 → at least 15-point boost + assert result_with_kev["score"] - result_no_kev["score"] >= 15 + + def test_result_has_required_keys(self): + nvd = {"vulnerabilities": []} + epss = {"data": []} + result = compute_temporal_priority("CVE-2024-0001", nvd_data=nvd, epss_data=epss, kev_set=set()) + + assert "cve_id" in result + assert "score" in result + assert "label" in result + assert "signals" in result + assert "interpretation" in result + + def test_signals_have_weights(self): + nvd = {"vulnerabilities": []} + epss = {"data": []} + result = compute_temporal_priority("CVE-2024-0001", nvd_data=nvd, epss_data=epss, kev_set=set()) + + signals = result["signals"] + for key in ("cvss", "epss_current", "epss_spike", "cisa_kev", "patch_availability", "age"): + assert key in signals + assert "weight" in signals[key] + assert "normalised" in signals[key] + + def test_cve_id_uppercased(self): + nvd = {"vulnerabilities": []} + epss = {"data": []} + result = compute_temporal_priority("cve-2024-0001", nvd_data=nvd, epss_data=epss, kev_set=set()) + assert result["cve_id"] == "CVE-2024-0001" + + +# --------------------------------------------------------------------------- +# Tool entry point tests +# --------------------------------------------------------------------------- + + +class TestToolEntryPoint: + """Test the Strands tool entry point function.""" + + @patch("manus_agent.tools.temporal_priority.compute_temporal_priority") + def test_success(self, mock_compute): + mock_compute.return_value = { + "cve_id": "CVE-2024-3094", + "score": 85.5, + "label": "CRITICAL", + "signals": {}, + "interpretation": "test", + } + tool_use = {"toolUseId": "t1", "input": {"cve_id": "CVE-2024-3094"}} + result = temporal_priority(tool_use) + assert result["status"] == "success" + assert result["toolUseId"] == "t1" + content = json.loads(result["content"][0]["text"]) + assert content["score"] == 85.5 + + @patch("manus_agent.tools.temporal_priority.compute_temporal_priority") + def test_exception_returns_error(self, mock_compute): + mock_compute.side_effect = RuntimeError("API down") + tool_use = {"toolUseId": "t2", "input": {"cve_id": "CVE-2024-3094"}} + result = temporal_priority(tool_use) + assert result["status"] == "error" + assert "API down" in result["content"][0]["text"] + + @patch("manus_agent.tools.temporal_priority.compute_temporal_priority") + def test_tool_use_id_preserved(self, mock_compute): + mock_compute.return_value = {"cve_id": "X", "score": 50, "label": "MEDIUM", "signals": {}, "interpretation": ""} + tool_use = {"toolUseId": "unique-id-123", "input": {"cve_id": "CVE-2024-1111"}} + result = temporal_priority(tool_use) + assert result["toolUseId"] == "unique-id-123" + + @patch("manus_agent.tools.temporal_priority.compute_temporal_priority") + def test_cve_id_trimmed(self, mock_compute): + mock_compute.return_value = {"cve_id": "X", "score": 50, "label": "MEDIUM", "signals": {}, "interpretation": ""} + tool_use = {"toolUseId": "t3", "input": {"cve_id": " CVE-2024-1111 "}} + result = temporal_priority(tool_use) + assert result["status"] == "success" + + +# --------------------------------------------------------------------------- +# CLI subcommand tests +# --------------------------------------------------------------------------- + + +class TestCliSubcommand: + """Test the temporal-priority CLI subcommand.""" + + @patch("manus_agent.tools.temporal_priority.compute_temporal_priority") + def test_text_output(self, mock_compute, capsys): + from manus_agent.cli import _run_temporal_priority + + mock_compute.return_value = { + "cve_id": "CVE-2024-3094", + "score": 72.3, + "label": "HIGH", + "signals": { + "cvss": {"normalised": 0.98, "raw_score": 9.8, "weight": 0.25}, + "epss_current": {"normalised": 0.5, "raw_epss": 0.5, "weight": 0.25}, + "epss_spike": {"normalised": 0.0, "spike_detected": False, "max_jump": 0, "weight": 0.15}, + "cisa_kev": {"normalised": 1.0, "in_kev": True, "weight": 0.20}, + "patch_availability": {"normalised": 0.0, "has_patch": True, "weight": 0.05}, + "age": {"normalised": 0.8, "age_days": 30, "weight": 0.10}, + }, + "interpretation": "CVE-2024-3094 has a temporal priority of 72.3/100 (HIGH).", + } + + rc = _run_temporal_priority(["CVE-2024-3094"]) + assert rc == 0 + out = capsys.readouterr().out + assert "72.3/100" in out + assert "HIGH" in out + assert "CVE-2024-3094" in out + + @patch("manus_agent.tools.temporal_priority.compute_temporal_priority") + def test_json_output(self, mock_compute, capsys): + from manus_agent.cli import _run_temporal_priority + + mock_compute.return_value = { + "cve_id": "CVE-2024-3094", + "score": 72.3, + "label": "HIGH", + "signals": {}, + "interpretation": "test", + } + + rc = _run_temporal_priority(["CVE-2024-3094", "--output", "json"]) + assert rc == 0 + out = capsys.readouterr().out + data = json.loads(out) + assert data["score"] == 72.3 + + @patch("manus_agent.tools.temporal_priority.compute_temporal_priority") + def test_exception_returns_nonzero(self, mock_compute, capsys): + from manus_agent.cli import _run_temporal_priority + + mock_compute.side_effect = RuntimeError("Network error") + rc = _run_temporal_priority(["CVE-2024-3094"]) + assert rc == 1 + err = capsys.readouterr().err + assert "Network error" in err + + def test_help_flag(self, capsys): + from manus_agent.cli import _run_temporal_priority + + with pytest.raises(SystemExit) as exc_info: + _run_temporal_priority(["--help"]) + assert exc_info.value.code == 0 + + def test_no_args_errors(self): + from manus_agent.cli import _run_temporal_priority + + with pytest.raises(SystemExit) as exc_info: + _run_temporal_priority([]) + assert exc_info.value.code != 0 + + def test_subcommand_registered(self): + from manus_agent.cli import _SUBCOMMANDS + + assert "temporal-priority" in _SUBCOMMANDS + + +# --------------------------------------------------------------------------- +# Edge case tests +# --------------------------------------------------------------------------- + + +class TestEdgeCases: + """Edge cases and boundary conditions.""" + + def test_all_signals_zero(self): + """When all data is missing, score should be bounded and not crash.""" + result = compute_temporal_priority("CVE-0000-0000", nvd_data={}, epss_data={}, kev_set=set()) + assert 0 <= result["score"] <= 100 + assert result["label"] in ("INFORMATIONAL", "LOW", "MEDIUM", "HIGH", "CRITICAL") + + def test_nvd_missing_cve_key(self): + """NVD response with malformed structure.""" + nvd = {"vulnerabilities": [{"cve": {}}]} + result = compute_temporal_priority("CVE-2024-0001", nvd_data=nvd, epss_data={"data": []}, kev_set=set()) + assert 0 <= result["score"] <= 100 + + def test_epss_data_with_empty_timeseries(self): + data = {"data": [{"epss": "0.5", "time-series": []}]} + score, details = score_epss_spike(data) + assert score == 0.0 + assert details["spike_detected"] is False + + def test_score_labels_thresholds(self): + """Verify label thresholds match documented behaviour.""" + nvd = {"vulnerabilities": []} + epss = {"data": []} + + # Force various score ranges by manipulating weights + with patch("manus_agent.tools.temporal_priority._W_CVSS", 0): + with patch("manus_agent.tools.temporal_priority._W_EPSS", 0): + with patch("manus_agent.tools.temporal_priority._W_SPIKE", 0): + with patch("manus_agent.tools.temporal_priority._W_KEV", 0): + with patch("manus_agent.tools.temporal_priority._W_PATCH", 0): + with patch("manus_agent.tools.temporal_priority._W_AGE", 0): + result = compute_temporal_priority( + "CVE-0000-0000", nvd_data=nvd, epss_data=epss, kev_set=set() + ) + assert result["score"] == 0.0 + assert result["label"] == "INFORMATIONAL" + + @patch("manus_agent.tools.temporal_priority._fetch_nvd") + @patch("manus_agent.tools.temporal_priority._fetch_epss") + @patch("manus_agent.tools.temporal_priority._fetch_kev") + def test_fetch_failures_graceful(self, mock_kev, mock_epss, mock_nvd): + """If all fetches fail, the function still returns a valid result.""" + mock_nvd.return_value = {} + mock_epss.return_value = {} + mock_kev.return_value = set() + + result = compute_temporal_priority("CVE-2024-9999") + assert 0 <= result["score"] <= 100 + assert "cve_id" in result + + def test_interpretation_contains_cve_id(self): + nvd = {"vulnerabilities": []} + epss = {"data": []} + result = compute_temporal_priority("CVE-2024-1234", nvd_data=nvd, epss_data=epss, kev_set=set()) + assert "CVE-2024-1234" in result["interpretation"]