From 92a251d98e6a9b49675063fe7c7ee3dc738e0dad Mon Sep 17 00:00:00 2001 From: manusjs Date: Sat, 4 Jul 2026 00:04:53 +0000 Subject: [PATCH] test(tools): comprehensive test suite for search_exploit_db, search_packetstorm, query_threat_intelligence_feeds, obtain_cves --- tests/test_exploit_search_tools.py | 956 +++++++++++++++++++++++++++++ 1 file changed, 956 insertions(+) create mode 100644 tests/test_exploit_search_tools.py diff --git a/tests/test_exploit_search_tools.py b/tests/test_exploit_search_tools.py new file mode 100644 index 0000000..1555ed6 --- /dev/null +++ b/tests/test_exploit_search_tools.py @@ -0,0 +1,956 @@ +""" +Comprehensive test suite for exploit-search and threat-intelligence tools: + - search_exploit_db + - search_packetstorm + - query_threat_intelligence_feeds + - obtain_cves (helpers + tool entry point) + +All HTTP calls are 100% mocked — no real network requests. +""" + +from __future__ import annotations + +import importlib +from types import ModuleType +from unittest.mock import MagicMock, patch + +import pytest +import requests + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_tool_use(tool_use_id: str, input_dict: dict) -> dict: + return {"toolUseId": tool_use_id, "input": input_dict} + + +def _make_response( + text: str = "", + status_code: int = 200, + json_data: dict | None = None, + links: dict | None = None, +) -> MagicMock: + """Build a mock requests.Response.""" + mock = MagicMock() + mock.status_code = status_code + mock.text = text + mock.links = links or {} + if json_data is not None: + mock.json.return_value = json_data + mock.raise_for_status = MagicMock() + if status_code >= 400: + mock.raise_for_status.side_effect = requests.exceptions.HTTPError(f"{status_code} Error", response=mock) + return mock + + +# --------------------------------------------------------------------------- +# search_exploit_db +# --------------------------------------------------------------------------- + + +class TestSearchExploitDb: + """Tests for manus_agent.tools.search_exploit_db""" + + @pytest.fixture(autouse=True) + def _import(self): + self.mod: ModuleType = importlib.import_module("manus_agent.tools.search_exploit_db") + + # --- TOOL_SPEC contract --- + + def test_tool_spec_present(self): + assert hasattr(self.mod, "TOOL_SPEC") + + def test_tool_spec_name(self): + assert self.mod.TOOL_SPEC["name"] == "search_exploit_db" + + def test_tool_spec_description_nonempty(self): + desc = self.mod.TOOL_SPEC.get("description", "") + assert isinstance(desc, str) and len(desc) > 10 + + def test_tool_spec_input_schema_has_query(self): + props = self.mod.TOOL_SPEC["inputSchema"]["json"]["properties"] + assert "query" in props + + def test_tool_spec_required_contains_query(self): + required = self.mod.TOOL_SPEC["inputSchema"]["json"]["required"] + assert "query" in required + + # --- Input validation --- + + def test_empty_string_query_returns_error(self): + tool = _make_tool_use("t1", {"query": ""}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + assert "Invalid query" in result["content"][0]["text"] + + def test_whitespace_only_query_returns_error(self): + tool = _make_tool_use("t2", {"query": " "}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + + def test_none_query_returns_error(self): + tool = _make_tool_use("t3", {"query": None}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + + def test_integer_query_returns_error(self): + tool = _make_tool_use("t4", {"query": 12345}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + + def test_missing_query_key_returns_error(self): + tool = _make_tool_use("t5", {}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + + # --- Successful parse with results --- + + def _html_with_entries(self, n: int = 2) -> str: + """Minimal Exploit-DB HTML that the parser can handle.""" + rows = [] + for i in range(n): + rows.append( + f'' + f'Exploit Title {i}' + f'webapps' + f"" + ) + return "".join(rows) + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_returns_success_with_parsed_results(self, mock_get): + mock_get.return_value = _make_response(text=self._html_with_entries(2)) + tool = _make_tool_use("t6", {"query": "CVE-2024-1234"}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "success" + payload = result["content"][0]["json"] + assert "exploits" in payload + assert "summary" in payload + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_url_contains_quoted_query(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("t7", {"query": "CVE-2024-1234"}) + self.mod.search_exploit_db(tool) + called_url = mock_get.call_args[0][0] + assert "CVE" in called_url or "cve" in called_url.lower() + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_empty_html_returns_no_exploits_found(self, mock_get): + mock_get.return_value = _make_response(text="nothing here") + tool = _make_tool_use("t8", {"query": "CVE-2024-0001"}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "success" + payload = result["content"][0]["json"] + assert payload["exploits"] == [] + assert "No exploits found" in payload["summary"] + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_limits_to_at_most_5_results(self, mock_get): + mock_get.return_value = _make_response(text=self._html_with_entries(10)) + tool = _make_tool_use("t9", {"query": "log4j"}) + result = self.mod.search_exploit_db(tool) + if result["status"] == "success": + exploits = result["content"][0]["json"]["exploits"] + assert len(exploits) <= 5 + + # --- Error handling --- + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_connection_error_returns_error_status(self, mock_get): + mock_get.side_effect = requests.exceptions.ConnectionError("no route") + tool = _make_tool_use("t10", {"query": "CVE-2024-9999"}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + assert "Exploit-DB" in result["content"][0]["text"] or "failed" in result["content"][0]["text"].lower() + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_timeout_error_returns_error_status(self, mock_get): + mock_get.side_effect = requests.exceptions.Timeout("timed out") + tool = _make_tool_use("t11", {"query": "CVE-2024-9999"}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_http_error_returns_error_status(self, mock_get): + mock_get.return_value = _make_response(status_code=503) + tool = _make_tool_use("t12", {"query": "CVE-2024-9999"}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_generic_exception_returns_error_status(self, mock_get): + mock_get.side_effect = RuntimeError("unexpected boom") + tool = _make_tool_use("t13", {"query": "CVE-2024-9999"}) + result = self.mod.search_exploit_db(tool) + assert result["status"] == "error" + + # --- tool_use_id echo --- + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_tool_use_id_echoed_in_success(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("echo-id-1", {"query": "log4shell"}) + result = self.mod.search_exploit_db(tool) + assert result["toolUseId"] == "echo-id-1" + + def test_tool_use_id_echoed_in_validation_error(self): + tool = _make_tool_use("echo-id-2", {"query": ""}) + result = self.mod.search_exploit_db(tool) + assert result["toolUseId"] == "echo-id-2" + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_tool_use_id_echoed_in_request_error(self, mock_get): + mock_get.side_effect = requests.exceptions.RequestException("boom") + tool = _make_tool_use("echo-id-3", {"query": "CVE-2024-0001"}) + result = self.mod.search_exploit_db(tool) + assert result["toolUseId"] == "echo-id-3" + + # --- result structure --- + + @patch("manus_agent.tools.search_exploit_db.requests.get") + def test_success_result_has_content_list(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("t20", {"query": "struts"}) + result = self.mod.search_exploit_db(tool) + assert isinstance(result["content"], list) + assert len(result["content"]) > 0 + + +# --------------------------------------------------------------------------- +# search_packetstorm +# --------------------------------------------------------------------------- + + +class TestSearchPacketstorm: + """Tests for manus_agent.tools.search_packetstorm""" + + @pytest.fixture(autouse=True) + def _import(self): + self.mod: ModuleType = importlib.import_module("manus_agent.tools.search_packetstorm") + + # --- TOOL_SPEC contract --- + + def test_tool_spec_present(self): + assert hasattr(self.mod, "TOOL_SPEC") + + def test_tool_spec_name(self): + assert self.mod.TOOL_SPEC["name"] == "search_packetstorm" + + def test_tool_spec_description_nonempty(self): + desc = self.mod.TOOL_SPEC.get("description", "") + assert isinstance(desc, str) and len(desc) > 10 + + def test_tool_spec_input_schema_has_query(self): + props = self.mod.TOOL_SPEC["inputSchema"]["json"]["properties"] + assert "query" in props + + def test_tool_spec_required_contains_query(self): + required = self.mod.TOOL_SPEC["inputSchema"]["json"]["required"] + assert "query" in required + + # --- Input validation --- + + def test_empty_string_query_returns_error(self): + tool = _make_tool_use("ps1", {"query": ""}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + assert "Invalid query" in result["content"][0]["text"] + + def test_whitespace_only_query_returns_error(self): + tool = _make_tool_use("ps2", {"query": "\t\n"}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + + def test_none_query_returns_error(self): + tool = _make_tool_use("ps3", {"query": None}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + + def test_integer_query_returns_error(self): + tool = _make_tool_use("ps4", {"query": 42}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + + def test_missing_query_key_returns_error(self): + tool = _make_tool_use("ps5", {}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + + # --- Successful parse --- + + def _html_with_entries(self, n: int = 2) -> str: + """Minimal PacketStorm HTML the parser can handle.""" + parts = [""] + for i in range(n): + parts.append(f'
Exploit Entry {i}
') + parts.append("") + return "".join(parts) + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_returns_success_with_parsed_results(self, mock_get): + mock_get.return_value = _make_response(text=self._html_with_entries(2)) + tool = _make_tool_use("ps6", {"query": "CVE-2024-1234"}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "success" + payload = result["content"][0]["json"] + assert "exploits" in payload + assert "summary" in payload + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_url_contains_query_string(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("ps7", {"query": "CVE-2024-5678"}) + self.mod.search_packetstorm(tool) + called_url = mock_get.call_args[0][0] + assert "packetstormsecurity.com" in called_url + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_empty_html_returns_no_exploits_found(self, mock_get): + mock_get.return_value = _make_response(text="nothing") + tool = _make_tool_use("ps8", {"query": "CVE-2023-0001"}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "success" + payload = result["content"][0]["json"] + assert payload["exploits"] == [] + assert "No exploits found" in payload["summary"] + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_limits_to_at_most_5_results(self, mock_get): + mock_get.return_value = _make_response(text=self._html_with_entries(10)) + tool = _make_tool_use("ps9", {"query": "shellshock"}) + result = self.mod.search_packetstorm(tool) + if result["status"] == "success": + exploits = result["content"][0]["json"]["exploits"] + assert len(exploits) <= 5 + + # --- Error handling --- + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_connection_error_returns_error_status(self, mock_get): + mock_get.side_effect = requests.exceptions.ConnectionError("no route") + tool = _make_tool_use("ps10", {"query": "CVE-2024-9999"}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + assert "Packet Storm" in result["content"][0]["text"] or "failed" in result["content"][0]["text"].lower() + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_timeout_returns_error_status(self, mock_get): + mock_get.side_effect = requests.exceptions.Timeout("timeout") + tool = _make_tool_use("ps11", {"query": "CVE-2024-9999"}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_http_error_returns_error_status(self, mock_get): + mock_get.return_value = _make_response(status_code=404) + tool = _make_tool_use("ps12", {"query": "CVE-2024-9999"}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_generic_exception_returns_error_status(self, mock_get): + mock_get.side_effect = ValueError("unexpected") + tool = _make_tool_use("ps13", {"query": "CVE-2024-9999"}) + result = self.mod.search_packetstorm(tool) + assert result["status"] == "error" + + # --- tool_use_id echo --- + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_tool_use_id_echoed_in_success(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("ps-echo-1", {"query": "CVE-2024-1234"}) + result = self.mod.search_packetstorm(tool) + assert result["toolUseId"] == "ps-echo-1" + + def test_tool_use_id_echoed_in_validation_error(self): + tool = _make_tool_use("ps-echo-2", {"query": ""}) + result = self.mod.search_packetstorm(tool) + assert result["toolUseId"] == "ps-echo-2" + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_tool_use_id_echoed_in_request_error(self, mock_get): + mock_get.side_effect = requests.exceptions.RequestException("boom") + tool = _make_tool_use("ps-echo-3", {"query": "CVE-2024-0001"}) + result = self.mod.search_packetstorm(tool) + assert result["toolUseId"] == "ps-echo-3" + + # --- result structure --- + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_success_result_has_content_list(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("ps-struct-1", {"query": "eternalblue"}) + result = self.mod.search_packetstorm(tool) + assert isinstance(result["content"], list) + assert len(result["content"]) > 0 + + @patch("manus_agent.tools.search_packetstorm.requests.get") + def test_exploit_entries_have_title_and_link(self, mock_get): + mock_get.return_value = _make_response(text=self._html_with_entries(1)) + tool = _make_tool_use("ps-struct-2", {"query": "CVE-2021-44228"}) + result = self.mod.search_packetstorm(tool) + if result["status"] == "success": + exploits = result["content"][0]["json"]["exploits"] + for exp in exploits: + assert "title" in exp + assert "link" in exp + + +# --------------------------------------------------------------------------- +# query_threat_intelligence_feeds +# --------------------------------------------------------------------------- + + +class TestQueryThreatIntelligenceFeeds: + """Tests for manus_agent.tools.query_threat_intelligence_feeds""" + + @pytest.fixture(autouse=True) + def _import(self): + self.mod: ModuleType = importlib.import_module("manus_agent.tools.query_threat_intelligence_feeds") + + # --- TOOL_SPEC contract --- + + def test_tool_spec_present(self): + assert hasattr(self.mod, "TOOL_SPEC") + + def test_tool_spec_name(self): + assert self.mod.TOOL_SPEC["name"] == "query_threat_intelligence_feeds" + + def test_tool_spec_description_nonempty(self): + desc = self.mod.TOOL_SPEC.get("description", "") + assert isinstance(desc, str) and len(desc) > 10 + + def test_tool_spec_input_schema_has_cve_id(self): + props = self.mod.TOOL_SPEC["inputSchema"]["json"]["properties"] + assert "cve_id" in props + + def test_tool_spec_required_contains_cve_id(self): + required = self.mod.TOOL_SPEC["inputSchema"]["json"]["required"] + assert "cve_id" in required + + # --- Input validation --- + + def test_empty_cve_id_returns_error(self): + tool = _make_tool_use("ti1", {"cve_id": ""}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "error" + assert "Invalid CVE ID" in result["content"][0]["text"] + + def test_whitespace_cve_id_returns_error(self): + tool = _make_tool_use("ti2", {"cve_id": " "}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "error" + + def test_none_cve_id_returns_error(self): + tool = _make_tool_use("ti3", {"cve_id": None}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "error" + + def test_integer_cve_id_returns_error(self): + tool = _make_tool_use("ti4", {"cve_id": 12345}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "error" + + def test_missing_cve_id_key_returns_error(self): + tool = _make_tool_use("ti5", {}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "error" + + # --- CVE found in feed --- + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_cve_found_in_feed_returns_intelligence(self, mock_get): + rss_body = ( + "" + "Advisory for CVE-2024-3094" + "Critical vulnerability CVE-2024-3094 exploited in the wild." + "" + ) + mock_get.return_value = _make_response(text=rss_body) + tool = _make_tool_use("ti6", {"cve_id": "CVE-2024-3094"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "success" + payload = result["content"][0]["json"] + assert "intelligence" in payload + assert len(payload["intelligence"]) > 0 + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_cve_not_in_feed_returns_empty_intelligence(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("ti7", {"cve_id": "CVE-2024-9999"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "success" + payload = result["content"][0]["json"] + assert payload["intelligence"] == [] + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_case_insensitive_cve_match(self, mock_get): + """The tool should find CVE-2024-1111 even if the feed has lowercase 'cve-2024-1111'.""" + mock_get.return_value = _make_response( + text="cve-2024-1111 is exploited" + ) + tool = _make_tool_use("ti8", {"cve_id": "CVE-2024-1111"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "success" + # Should detect the lowercase match + payload = result["content"][0]["json"] + assert len(payload["intelligence"]) > 0 + + # --- Error handling per feed (should not raise, just continue) --- + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_request_exception_skips_feed_and_returns_empty(self, mock_get): + mock_get.side_effect = requests.exceptions.ConnectionError("no route") + tool = _make_tool_use("ti9", {"cve_id": "CVE-2024-3094"}) + result = self.mod.query_threat_intelligence_feeds(tool) + # Should succeed gracefully even if all feeds fail + assert result["status"] == "success" + payload = result["content"][0]["json"] + assert payload["intelligence"] == [] + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_timeout_skips_feed_gracefully(self, mock_get): + mock_get.side_effect = requests.exceptions.Timeout("timeout") + tool = _make_tool_use("ti10", {"cve_id": "CVE-2024-0001"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "success" + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_http_error_skips_feed_gracefully(self, mock_get): + mock_get.return_value = _make_response(status_code=500) + tool = _make_tool_use("ti11", {"cve_id": "CVE-2024-0001"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "success" + + # --- Summary field --- + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_found_intelligence_summary_mentions_cve(self, mock_get): + mock_get.return_value = _make_response(text="CVE-2024-3094 advisory text here for testing") + tool = _make_tool_use("ti12", {"cve_id": "CVE-2024-3094"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "success" + payload = result["content"][0]["json"] + if len(payload["intelligence"]) > 0: + assert "CVE-2024-3094" in payload["summary"] or "feeds" in payload["summary"].lower() + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_not_found_summary_mentions_no_intelligence(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("ti13", {"cve_id": "CVE-9999-0000"}) + result = self.mod.query_threat_intelligence_feeds(tool) + payload = result["content"][0]["json"] + assert "No" in payload["summary"] or "no" in payload["summary"].lower() + + # --- tool_use_id echo --- + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_tool_use_id_echoed_in_success(self, mock_get): + mock_get.return_value = _make_response(text="") + tool = _make_tool_use("ti-echo-1", {"cve_id": "CVE-2024-0001"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["toolUseId"] == "ti-echo-1" + + def test_tool_use_id_echoed_in_validation_error(self): + tool = _make_tool_use("ti-echo-2", {"cve_id": ""}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["toolUseId"] == "ti-echo-2" + + # --- Intelligence entry structure --- + + @patch("manus_agent.tools.query_threat_intelligence_feeds.requests.get") + def test_intelligence_entry_has_expected_keys(self, mock_get): + mock_get.return_value = _make_response(text="Advisory: CVE-2024-3094 has been added to CISA KEV") + tool = _make_tool_use("ti14", {"cve_id": "CVE-2024-3094"}) + result = self.mod.query_threat_intelligence_feeds(tool) + assert result["status"] == "success" + for entry in result["content"][0]["json"]["intelligence"]: + assert "feed_name" in entry + assert "feed_url" in entry + assert "cve_found" in entry + + +# --------------------------------------------------------------------------- +# obtain_cves — helper functions & tool entry point +# --------------------------------------------------------------------------- + + +class TestObtainCves: + """Tests for manus_agent.tools.obtain_cves""" + + @pytest.fixture(autouse=True) + def _import(self): + self.mod: ModuleType = importlib.import_module("manus_agent.tools.obtain_cves") + + # --- TOOL_SPEC contract --- + + def test_tool_spec_present(self): + assert hasattr(self.mod, "TOOL_SPEC") + + def test_tool_spec_name(self): + assert self.mod.TOOL_SPEC["name"] == "obtain_cves" + + def test_tool_spec_description_nonempty(self): + desc = self.mod.TOOL_SPEC.get("description", "") + assert isinstance(desc, str) and len(desc) > 5 + + def test_tool_spec_required_start_date(self): + required = self.mod.TOOL_SPEC["inputSchema"]["json"]["required"] + assert "start_date" in required + + def test_tool_spec_required_end_date(self): + required = self.mod.TOOL_SPEC["inputSchema"]["json"]["required"] + assert "end_date" in required + + # --- _get_all_cves_from_nvd --- + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_nvd_single_page(self, mock_get): + nvd_data = { + "vulnerabilities": [{"cve": {"id": "CVE-2024-0001"}}], + "totalResults": 1, + } + mock_get.return_value = _make_response(json_data=nvd_data) + result = self.mod._get_all_cves_from_nvd("2024-01-01T00:00:00.000Z", "2024-01-31T23:59:59.000Z") + assert len(result) == 1 + assert result[0]["cve"]["id"] == "CVE-2024-0001" + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_nvd_pagination_two_pages(self, mock_get): + page1 = { + "vulnerabilities": [{"cve": {"id": f"CVE-2024-{i:04d}"}} for i in range(3)], + "totalResults": 5, + } + page2 = { + "vulnerabilities": [{"cve": {"id": f"CVE-2024-{i:04d}"}} for i in range(3, 5)], + "totalResults": 5, + } + mock_get.side_effect = [_make_response(json_data=page1), _make_response(json_data=page2)] + result = self.mod._get_all_cves_from_nvd("2024-01-01T00:00:00.000Z", "2024-01-31T23:59:59.000Z") + assert len(result) == 5 + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_nvd_empty_result(self, mock_get): + mock_get.return_value = _make_response(json_data={"vulnerabilities": [], "totalResults": 0}) + result = self.mod._get_all_cves_from_nvd("2024-01-01T00:00:00.000Z", "2024-01-31T23:59:59.000Z") + assert result == [] + + # --- _get_all_cves_from_github --- + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_github_single_page(self, mock_get): + advisories = [ + { + "cve_id": "CVE-2024-0001", + "summary": "Test advisory", + "published_at": "2024-01-15T00:00:00Z", + "cvss_severities": {"cvss_v3": {"score": 9.8}}, + } + ] + mock_resp = _make_response(json_data=advisories) + mock_resp.links = {} + mock_get.return_value = mock_resp + result = self.mod._get_all_cves_from_github("2024-01-01T00:00:00.000Z", "2024-01-31T23:59:59.000Z") + assert len(result) == 1 + assert result[0]["cve"]["id"] == "CVE-2024-0001" + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_github_skips_advisories_without_cve_id(self, mock_get): + advisories = [ + {"cve_id": None, "summary": "No CVE advisory"}, + {"cve_id": "CVE-2024-0002", "summary": "Has CVE", "published_at": "2024-01-01T00:00:00Z"}, + ] + mock_resp = _make_response(json_data=advisories) + mock_resp.links = {} + mock_get.return_value = mock_resp + result = self.mod._get_all_cves_from_github("2024-01-01T00:00:00.000Z", "2024-01-31T23:59:59.000Z") + assert len(result) == 1 + assert result[0]["cve"]["id"] == "CVE-2024-0002" + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_github_request_exception_returns_empty(self, mock_get): + mock_get.side_effect = requests.exceptions.RequestException("GitHub down") + result = self.mod._get_all_cves_from_github("2024-01-01T00:00:00.000Z", "2024-01-31T23:59:59.000Z") + assert result == [] + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_github_pagination_follows_next_link(self, mock_get): + page1 = [{"cve_id": "CVE-2024-0001", "summary": "a", "published_at": "2024-01-01T00:00:00Z"}] + page2 = [{"cve_id": "CVE-2024-0002", "summary": "b", "published_at": "2024-01-02T00:00:00Z"}] + mock_resp1 = _make_response(json_data=page1) + mock_resp1.links = {"next": {"url": "https://api.github.com/advisories?page=2"}} + mock_resp2 = _make_response(json_data=page2) + mock_resp2.links = {} + mock_get.side_effect = [mock_resp1, mock_resp2] + result = self.mod._get_all_cves_from_github("2024-01-01T00:00:00.000Z", "2024-01-31T23:59:59.000Z") + assert len(result) == 2 + + # --- _filter_cves_by_epss --- + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_epss_filter_passes_high_epss(self, mock_get): + cves = [{"cve": {"id": "CVE-2024-0001"}}] + epss_response = {"data": [{"cve": "CVE-2024-0001", "epss": "0.9", "percentile": "0.99"}]} + mock_get.return_value = _make_response(json_data=epss_response) + result = self.mod._filter_cves_by_epss(cves) + assert len(result) == 1 + assert result[0]["epss_data"]["cve"] == "CVE-2024-0001" + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_epss_filter_blocks_low_epss(self, mock_get): + cves = [{"cve": {"id": "CVE-2024-0002"}}] + epss_response = {"data": [{"cve": "CVE-2024-0002", "epss": "0.001", "percentile": "0.10"}]} + mock_get.return_value = _make_response(json_data=epss_response) + result = self.mod._filter_cves_by_epss(cves) + assert len(result) == 0 + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_epss_filter_passes_high_percentile_even_if_low_epss(self, mock_get): + """percentile > 0.5 should pass even if absolute EPSS score is low.""" + cves = [{"cve": {"id": "CVE-2024-0003"}}] + epss_response = {"data": [{"cve": "CVE-2024-0003", "epss": "0.001", "percentile": "0.75"}]} + mock_get.return_value = _make_response(json_data=epss_response) + result = self.mod._filter_cves_by_epss(cves) + assert len(result) == 1 + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_epss_filter_empty_input_returns_empty(self, mock_get): + result = self.mod._filter_cves_by_epss([]) + assert result == [] + mock_get.assert_not_called() + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_epss_filter_missing_from_epss_response_excluded(self, mock_get): + """CVE not in EPSS response should be excluded.""" + cves = [{"cve": {"id": "CVE-2024-9999"}}] + epss_response = {"data": []} # CVE not in EPSS + mock_get.return_value = _make_response(json_data=epss_response) + result = self.mod._filter_cves_by_epss(cves) + assert len(result) == 0 + + # --- obtain_cves (tool entry point) --- + + @patch("manus_agent.tools.obtain_cves._filter_cves_by_epss") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_github") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_nvd") + def test_obtain_cves_success_no_cves(self, mock_nvd, mock_github, mock_epss): + mock_nvd.return_value = [] + mock_github.return_value = [] + mock_epss.return_value = [] + tool = _make_tool_use( + "oc1", + { + "start_date": "2024-01-01T00:00:00.000Z", + "end_date": "2024-01-31T23:59:59.000Z", + }, + ) + result = self.mod.obtain_cves(tool) + assert result["status"] == "success" + assert "No new high/critical CVEs found" in result["content"][0]["text"] + + @patch("manus_agent.tools.obtain_cves._filter_cves_by_epss") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_github") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_nvd") + def test_obtain_cves_merges_and_deduplicates(self, mock_nvd, mock_github, mock_epss): + cve = {"cve": {"id": "CVE-2024-0001"}} + mock_nvd.return_value = [cve] + mock_github.return_value = [cve] # duplicate + mock_epss.return_value = [cve] + tool = _make_tool_use( + "oc2", + { + "start_date": "2024-01-01T00:00:00.000Z", + "end_date": "2024-01-31T23:59:59.000Z", + }, + ) + result = self.mod.obtain_cves(tool) + assert result["status"] == "success" + # Confirm dedup: total_found should be 1 + json_payload = next((c for c in result["content"] if "json" in c), None) + assert json_payload is not None + assert json_payload["json"]["total_found"] == 1 + + @patch("manus_agent.tools.obtain_cves._filter_cves_by_epss") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_github") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_nvd") + def test_obtain_cves_github_cve_added_when_not_in_nvd(self, mock_nvd, mock_github, mock_epss): + nvd_cve = {"cve": {"id": "CVE-2024-0001"}} + gh_cve = {"cve": {"id": "CVE-2024-0002"}} + mock_nvd.return_value = [nvd_cve] + mock_github.return_value = [gh_cve] + mock_epss.return_value = [nvd_cve, gh_cve] + tool = _make_tool_use( + "oc3", + { + "start_date": "2024-01-01T00:00:00.000Z", + "end_date": "2024-01-31T23:59:59.000Z", + }, + ) + result = self.mod.obtain_cves(tool) + json_payload = next(c for c in result["content"] if "json" in c) + assert json_payload["json"]["total_found"] == 2 + + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_nvd") + def test_obtain_cves_exception_returns_error(self, mock_nvd): + mock_nvd.side_effect = RuntimeError("NVD exploded") + tool = _make_tool_use( + "oc4", + { + "start_date": "2024-01-01T00:00:00.000Z", + "end_date": "2024-01-31T23:59:59.000Z", + }, + ) + result = self.mod.obtain_cves(tool) + assert result["status"] == "error" + assert "error occurred" in result["content"][0]["text"].lower() + + @patch("manus_agent.tools.obtain_cves._filter_cves_by_epss") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_github") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_nvd") + def test_obtain_cves_tool_use_id_echoed(self, mock_nvd, mock_github, mock_epss): + mock_nvd.return_value = [] + mock_github.return_value = [] + mock_epss.return_value = [] + tool = _make_tool_use( + "oc-echo-1", + { + "start_date": "2024-01-01T00:00:00.000Z", + "end_date": "2024-01-31T23:59:59.000Z", + }, + ) + result = self.mod.obtain_cves(tool) + assert result["toolUseId"] == "oc-echo-1" + + @patch("manus_agent.tools.obtain_cves._filter_cves_by_epss") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_github") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_nvd") + def test_obtain_cves_result_includes_totals_in_json(self, mock_nvd, mock_github, mock_epss): + cve = {"cve": {"id": "CVE-2024-0001"}} + mock_nvd.return_value = [cve] + mock_github.return_value = [] + mock_epss.return_value = [cve] + tool = _make_tool_use( + "oc-json-1", + { + "start_date": "2024-01-01T00:00:00.000Z", + "end_date": "2024-01-31T23:59:59.000Z", + }, + ) + result = self.mod.obtain_cves(tool) + json_payload = next(c for c in result["content"] if "json" in c) + assert "total_found" in json_payload["json"] + assert "total_with_high_epss" in json_payload["json"] + assert "cves_with_high_epss" in json_payload["json"] + + @patch("manus_agent.tools.obtain_cves._filter_cves_by_epss") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_github") + @patch("manus_agent.tools.obtain_cves._get_all_cves_from_nvd") + def test_obtain_cves_epss_filter_applied_in_chunks(self, mock_nvd, mock_github, mock_epss): + """Confirm _filter_cves_by_epss is called with at-most-100 item chunks.""" + cves = [{"cve": {"id": f"CVE-2024-{i:04d}"}} for i in range(150)] + mock_nvd.return_value = cves + mock_github.return_value = [] + mock_epss.return_value = cves # all pass + + tool = _make_tool_use( + "oc-chunk-1", + { + "start_date": "2024-01-01T00:00:00.000Z", + "end_date": "2024-01-31T23:59:59.000Z", + }, + ) + self.mod.obtain_cves(tool) + # Should have been called twice (chunks of 100 and 50) + call_count = mock_epss.call_count + assert call_count >= 2 + + # --- _enrich_with_cisa_kev --- + + @patch("manus_agent.tools.obtain_cves.requests.get") + def test_enrich_kev_marks_known_exploited(self, mock_get): + kev_data = {"vulnerabilities": [{"cveID": "CVE-2024-0001"}]} + mock_get.return_value = _make_response(json_data=kev_data) + cves = [{"cve": {"id": "CVE-2024-0001"}}, {"cve": {"id": "CVE-2024-0002"}}] + result = self.mod._enrich_with_cisa_kev(cves) + enriched = {c["cve"]["id"]: c for c in result} + assert enriched["CVE-2024-0001"]["cisa_kev"] is True + assert enriched["CVE-2024-0002"]["cisa_kev"] is False + + +# --------------------------------------------------------------------------- +# Cross-tool: vi_agent wiring +# --------------------------------------------------------------------------- + + +class TestViAgentWiresExploitSearchTools: + """Confirm all three exploit-search tools are referenced in vi_agent.""" + + @pytest.fixture(autouse=True) + def _load_source(self): + vi_agent_path = importlib.import_module("manus_agent.agents.vi_agent").__file__ + with open(vi_agent_path) as fh: + self.source = fh.read() + + def test_search_exploit_db_in_vi_agent(self): + assert "search_exploit_db" in self.source + + def test_search_packetstorm_in_vi_agent(self): + assert "search_packetstorm" in self.source + + def test_query_threat_intelligence_feeds_in_vi_agent(self): + assert "query_threat_intelligence_feeds" in self.source + + def test_search_exploit_db_module_importable(self): + mod = importlib.import_module("manus_agent.tools.search_exploit_db") + assert hasattr(mod, "search_exploit_db") + + def test_search_packetstorm_module_importable(self): + mod = importlib.import_module("manus_agent.tools.search_packetstorm") + assert hasattr(mod, "search_packetstorm") + + def test_query_threat_intelligence_feeds_module_importable(self): + mod = importlib.import_module("manus_agent.tools.query_threat_intelligence_feeds") + assert hasattr(mod, "query_threat_intelligence_feeds") + + +# --------------------------------------------------------------------------- +# Cross-tool: Tool Spec contract (parametrised) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "tool_module,tool_name", + [ + ("manus_agent.tools.search_exploit_db", "search_exploit_db"), + ("manus_agent.tools.search_packetstorm", "search_packetstorm"), + ("manus_agent.tools.query_threat_intelligence_feeds", "query_threat_intelligence_feeds"), + ("manus_agent.tools.obtain_cves", "obtain_cves"), + ], +) +class TestToolSpecContractParametrised: + def test_tool_spec_name_matches(self, tool_module, tool_name): + mod = importlib.import_module(tool_module) + assert mod.TOOL_SPEC["name"] == tool_name + + def test_tool_spec_has_description(self, tool_module, tool_name): + mod = importlib.import_module(tool_module) + assert "description" in mod.TOOL_SPEC + assert len(mod.TOOL_SPEC["description"]) > 5 + + def test_tool_spec_has_input_schema(self, tool_module, tool_name): + mod = importlib.import_module(tool_module) + assert "inputSchema" in mod.TOOL_SPEC + + def test_tool_spec_input_schema_has_required(self, tool_module, tool_name): + mod = importlib.import_module(tool_module) + schema = mod.TOOL_SPEC["inputSchema"]["json"] + assert "required" in schema + assert len(schema["required"]) > 0 + + def test_callable_exists_with_same_name_as_tool(self, tool_module, tool_name): + mod = importlib.import_module(tool_module) + assert callable(getattr(mod, tool_name, None))