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))