diff --git a/tests/test_verify_exploit.py b/tests/test_verify_exploit.py new file mode 100644 index 0000000..6f230d4 --- /dev/null +++ b/tests/test_verify_exploit.py @@ -0,0 +1,1206 @@ +"""Comprehensive test suite for the verify_exploit tool function. + +Tests cover the full verify_exploit() orchestration function including: +- Docker preflight failures (DockerConnectionError, generic exceptions) +- Image build failures +- Target container start failures +- Remote mode: wait-for-target timeout, exploit execution (success/failure), + InterpreterNotFoundError, transient Docker errors +- Local mode: container health wait failures, exploit execution (success/failure), + InterpreterNotFoundError, transient Docker errors +- Successful verification (exit_code=0 → "verified") +- Failed verification (non-zero exit_code → "failed") +- Unexpected top-level exceptions +- _truncate_text edge cases (already covered elsewhere but we test integration) +- Input validation (missing dockerfile/exploit_code) + +All tests are fully mocked — no real Docker calls. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from manus_agent.tools.verify_exploit import ( + TOOL_SPEC, + verify_exploit, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def _make_tool_use( + *, + dockerfile_content: str = "FROM ubuntu:20.04\nRUN apt-get update", + exploit_code: str = "import socket; print('pwned')", + exploit_language: str = "python", + cve_id: str = "CVE-2024-1234", + target_info: dict | None = None, + target_port: int = 80, + timeout: int = 300, + exploit_mode: str = "remote", + target_env: dict | None = None, +) -> dict: + """Build a ToolUse dict for verify_exploit.""" + if target_info is None: + target_info = { + "affected_software": "TestApp", + "affected_versions": "1.0-2.0", + "vulnerability_type": "RCE", + } + return { + "toolUseId": "test-tool-use-id", + "input": { + "dockerfile_content": dockerfile_content, + "exploit_code": exploit_code, + "exploit_language": exploit_language, + "cve_id": cve_id, + "target_info": target_info, + "target_port": target_port, + "timeout": timeout, + "exploit_mode": exploit_mode, + "target_env": target_env, + }, + } + + +def _get_json(result: dict) -> dict: + """Extract the JSON payload from a ToolResult.""" + return result["content"][0]["json"] + + +# --------------------------------------------------------------------------- +# TOOL_SPEC sanity +# --------------------------------------------------------------------------- + + +class TestToolSpec: + """Verify the tool specification is well-formed.""" + + def test_tool_spec_name(self): + assert TOOL_SPEC["name"] == "verify_exploit" + + def test_tool_spec_has_description(self): + assert "Proof-of-Concept" in TOOL_SPEC["description"] + + def test_tool_spec_required_fields(self): + required = TOOL_SPEC["inputSchema"]["json"]["required"] + assert "dockerfile_content" in required + assert "exploit_code" in required + assert "exploit_language" in required + assert "cve_id" in required + assert "target_info" in required + + def test_tool_spec_has_exploit_mode(self): + props = TOOL_SPEC["inputSchema"]["json"]["properties"] + assert "exploit_mode" in props + assert props["exploit_mode"]["enum"] == ["remote", "local"] + + def test_tool_spec_has_target_env(self): + props = TOOL_SPEC["inputSchema"]["json"]["properties"] + assert "target_env" in props + + +# --------------------------------------------------------------------------- +# Input validation +# --------------------------------------------------------------------------- + + +class TestInputValidation: + """Test early-exit paths for invalid input.""" + + def test_missing_dockerfile_content(self): + tool_use = _make_tool_use(dockerfile_content="") + result = verify_exploit(tool_use) + assert result["status"] == "error" + assert "required" in result["content"][0]["text"].lower() + + def test_missing_exploit_code(self): + tool_use = _make_tool_use(exploit_code="") + result = verify_exploit(tool_use) + assert result["status"] == "error" + assert "required" in result["content"][0]["text"].lower() + + def test_both_missing(self): + tool_use = _make_tool_use(dockerfile_content="", exploit_code="") + result = verify_exploit(tool_use) + assert result["status"] == "error" + + def test_tool_use_id_preserved(self): + tool_use = _make_tool_use(dockerfile_content="") + result = verify_exploit(tool_use) + assert result["toolUseId"] == "test-tool-use-id" + + +# --------------------------------------------------------------------------- +# Docker preflight failures +# --------------------------------------------------------------------------- + + +class TestDockerPreflight: + """Test Docker daemon preflight check failures.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + def test_docker_connection_error(self, mock_get_client): + """DockerConnectionError yields infra_error with diagnosis.""" + from manus_agent.utils.docker_client import DockerConnectionError + + mock_get_client.side_effect = DockerConnectionError( + message="Cannot connect", + diagnosis="Docker not running", + remediation="Start Docker daemon", + ) + + tool_use = _make_tool_use() + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "infra_error" + assert "Cannot connect" in data["summary"] + assert data["error"]["category"] == "infra" + assert data["error"]["stage"] == "docker_preflight" + assert data["error"]["retryable"] is False + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + def test_generic_docker_exception(self, mock_get_client): + """Generic exception during Docker preflight.""" + mock_get_client.side_effect = RuntimeError("Docker socket gone") + + tool_use = _make_tool_use() + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "infra_error" + assert "Docker socket gone" in data["summary"] + assert data["error"]["stage"] == "docker_preflight" + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + def test_transient_docker_error_is_retryable(self, mock_get_client, mock_is_transient): + """Transient Docker errors are marked retryable.""" + mock_get_client.side_effect = ConnectionError("temporary failure") + mock_is_transient.return_value = True + + tool_use = _make_tool_use() + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["error"]["retryable"] is True + + +# --------------------------------------------------------------------------- +# Build failures +# --------------------------------------------------------------------------- + + +class TestBuildFailures: + """Test image build failures.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_build_target_exception(self, mock_sandbox, mock_get_client): + """build_target raising Exception yields build_error.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.side_effect = RuntimeError("Dockerfile syntax error") + sandbox.build_log = "Step 1/3: FROM invalid\nERROR: invalid reference" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + tool_use = _make_tool_use() + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "build_error" + assert "Dockerfile syntax error" in data["summary"] + assert "invalid reference" in data["build_log"] + sandbox.cleanup.assert_called_once() + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_build_log_truncation(self, mock_sandbox, mock_get_client): + """Very long build logs are truncated.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.side_effect = RuntimeError("build failed") + # Create a very long build log + sandbox.build_log = "line\n" * 3000 + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + tool_use = _make_tool_use() + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "build_error" + assert "[truncated" in data["build_log"] + + +# --------------------------------------------------------------------------- +# Target start failures +# --------------------------------------------------------------------------- + + +class TestTargetStartFailures: + """Test target container start failures.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_start_target_exception(self, mock_sandbox, mock_get_client): + """start_target raising Exception yields target_error.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:abc123" + sandbox.start_target.side_effect = RuntimeError("Port already in use") + sandbox.build_log = "Build OK" + sandbox.get_target_logs.return_value = "some log" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 1 + + tool_use = _make_tool_use() + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "target_error" + assert "Port already in use" in data["summary"] + assert data["target_exit_code"] == 1 + sandbox.cleanup.assert_called_once() + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_start_target_passes_environment(self, mock_sandbox, mock_get_client): + """target_env is forwarded to start_target.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:abc123" + sandbox.start_target.side_effect = RuntimeError("fail") + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + env = {"DB_HOST": "localhost", "DB_PASS": "secret"} + tool_use = _make_tool_use(target_env=env) + verify_exploit(tool_use) + + sandbox.start_target.assert_called_once_with("sha256:abc123", environment=env) + + +# --------------------------------------------------------------------------- +# Remote mode: wait for target +# --------------------------------------------------------------------------- + + +class TestRemoteWaitForTarget: + """Test remote-mode wait_for_target timeout.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_target_not_ready(self, mock_sandbox, mock_get_client): + """wait_for_target returning False yields target_error.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = False + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "target crashed" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 137 + + tool_use = _make_tool_use(exploit_mode="remote", target_port=8080) + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "target_error" + assert "8080" in data["summary"] + assert data["target_exit_code"] == 137 + sandbox.wait_for_target.assert_called_once_with(port=8080, timeout=60) + + +# --------------------------------------------------------------------------- +# Remote mode: exploit execution +# --------------------------------------------------------------------------- + + +class TestRemoteExploitExecution: + """Test remote-mode exploit execution paths.""" + + def _setup_sandbox_to_exploit(self, mock_sandbox, mock_get_client): + """Helper to get past preflight, build, start, and wait.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.build_log = "OK" + sandbox.get_target_logs.return_value = "target logs here" + sandbox.get_docker_ps_all.return_value = "CONTAINER ID..." + sandbox.get_target_exit_code.return_value = 0 + return sandbox + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_exploit_verified_exit_0(self, mock_sandbox, mock_get_client): + """Exit code 0 means verified.""" + sandbox = self._setup_sandbox_to_exploit(mock_sandbox, mock_get_client) + sandbox.run_exploit.return_value = { + "stdout": "pwned!", + "stderr": "", + "exit_code": 0, + } + + tool_use = _make_tool_use(exploit_mode="remote") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "verified" + assert "successfully" in data["summary"] + assert "CVE-2024-1234" in data["summary"] + assert "TestApp" in data["summary"] + assert "RCE" in data["summary"] + assert data["exploit_output"]["stdout"] == "pwned!" + assert data["execution_time_seconds"] >= 0 + sandbox.cleanup.assert_called_once() + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_exploit_failed_nonzero_exit(self, mock_sandbox, mock_get_client): + """Non-zero exit code means failed.""" + sandbox = self._setup_sandbox_to_exploit(mock_sandbox, mock_get_client) + sandbox.run_exploit.return_value = { + "stdout": "", + "stderr": "connection refused", + "exit_code": 1, + } + + tool_use = _make_tool_use(exploit_mode="remote") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "failed" + assert "exited with code 1" in data["summary"] + assert data["exploit_output"]["stderr"] == "connection refused" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_exploit_interpreter_not_found(self, mock_sandbox, mock_get_client): + """InterpreterNotFoundError yields infra_error.""" + from manus_agent.sandbox.exploit_sandbox import InterpreterNotFoundError + + sandbox = self._setup_sandbox_to_exploit(mock_sandbox, mock_get_client) + sandbox.run_exploit.side_effect = InterpreterNotFoundError( + container_role="exploit", + language="python", + candidates=["python3", "python"], + ) + + tool_use = _make_tool_use(exploit_mode="remote") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "infra_error" + assert "python" in data["summary"].lower() + assert data["error"]["category"] == "infra" + assert data["error"]["stage"] == "interpreter_preflight" + assert data["error"]["retryable"] is False + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_exploit_transient_error(self, mock_sandbox, mock_get_client, mock_is_transient): + """Transient Docker error during exploit run yields infra_error.""" + sandbox = self._setup_sandbox_to_exploit(mock_sandbox, mock_get_client) + sandbox.run_exploit.side_effect = ConnectionError("Docker socket reset") + mock_is_transient.return_value = True + + tool_use = _make_tool_use(exploit_mode="remote") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "infra_error" + assert data["error"]["retryable"] is True + assert data["error"]["stage"] == "run_exploit" + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_exploit_non_transient_error(self, mock_sandbox, mock_get_client, mock_is_transient): + """Non-transient error during exploit run yields target_error.""" + sandbox = self._setup_sandbox_to_exploit(mock_sandbox, mock_get_client) + sandbox.run_exploit.side_effect = ValueError("Bad exploit format") + mock_is_transient.return_value = False + + tool_use = _make_tool_use(exploit_mode="remote") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "target_error" + assert data["error"]["retryable"] is False + assert data["error"]["stage"] == "run_exploit" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_exploit_env_includes_target_port(self, mock_sandbox, mock_get_client): + """TARGET_PORT env var is passed to run_exploit.""" + sandbox = self._setup_sandbox_to_exploit(mock_sandbox, mock_get_client) + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + + tool_use = _make_tool_use(exploit_mode="remote", target_port=9090) + verify_exploit(tool_use) + + sandbox.run_exploit.assert_called_once_with( + code="import socket; print('pwned')", + language="python", + env={"TARGET_PORT": "9090"}, + ) + + +# --------------------------------------------------------------------------- +# Local mode: health wait failures +# --------------------------------------------------------------------------- + + +class TestLocalModeHealthWait: + """Test local-mode container health wait failures.""" + + @patch("manus_agent.tools.verify_exploit.wait_for_container_healthy") + @patch("manus_agent.tools.verify_exploit.wait_for_container_running") + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_wait_running_fails_non_transient( + self, mock_sandbox, mock_get_client, mock_is_transient, mock_wait_running, mock_wait_healthy + ): + """Non-transient failure in wait_for_container_running yields target_error.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + mock_is_transient.return_value = False + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.target_container = MagicMock() + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "container exited" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 1 + + mock_wait_running.side_effect = RuntimeError("Container exited immediately") + + tool_use = _make_tool_use(exploit_mode="local") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "target_error" + assert "Container exited immediately" in data["summary"] + assert data["error"]["stage"] == "wait_target_local" + assert data["error"]["retryable"] is False + + @patch("manus_agent.tools.verify_exploit.wait_for_container_healthy") + @patch("manus_agent.tools.verify_exploit.wait_for_container_running") + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_wait_healthy_fails_transient( + self, mock_sandbox, mock_get_client, mock_is_transient, mock_wait_running, mock_wait_healthy + ): + """Transient failure in wait_for_container_healthy yields infra_error.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + mock_is_transient.return_value = True + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.target_container = MagicMock() + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + mock_wait_running.return_value = None + mock_wait_healthy.side_effect = ConnectionError("Docker socket reset") + + tool_use = _make_tool_use(exploit_mode="local") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "infra_error" + assert data["error"]["retryable"] is True + assert data["error"]["stage"] == "wait_target_local" + + @patch("manus_agent.tools.verify_exploit.wait_for_container_healthy") + @patch("manus_agent.tools.verify_exploit.wait_for_container_running") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_skips_health_wait_when_no_container( + self, mock_sandbox, mock_get_client, mock_wait_running, mock_wait_healthy + ): + """When target_container is None, skip health wait and go to exploit.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.target_container = None + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + sandbox.run_local_exploit.return_value = {"stdout": "ok", "stderr": "", "exit_code": 0} + + tool_use = _make_tool_use(exploit_mode="local") + result = verify_exploit(tool_use) + data = _get_json(result) + + # Should reach exploit execution without calling wait functions + mock_wait_running.assert_not_called() + mock_wait_healthy.assert_not_called() + assert data["verification_status"] == "verified" + + +# --------------------------------------------------------------------------- +# Local mode: exploit execution +# --------------------------------------------------------------------------- + + +class TestLocalExploitExecution: + """Test local-mode exploit execution paths.""" + + def _setup_local_sandbox(self, mock_sandbox, mock_get_client): + """Helper to get past preflight, build, start, and health checks.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.target_container = None # Skip health wait + sandbox.build_log = "build ok" + sandbox.get_target_logs.return_value = "target logs" + sandbox.get_docker_ps_all.return_value = "containers" + sandbox.get_target_exit_code.return_value = 0 + return sandbox + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_local_exploit_verified(self, mock_sandbox, mock_get_client): + """Local exploit with exit_code=0 means verified.""" + sandbox = self._setup_local_sandbox(mock_sandbox, mock_get_client) + sandbox.run_local_exploit.return_value = { + "stdout": "root shell obtained", + "stderr": "", + "exit_code": 0, + } + + tool_use = _make_tool_use(exploit_mode="local") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "verified" + assert "successfully" in data["summary"] + assert data["exploit_output"]["stdout"] == "root shell obtained" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_local_exploit_failed(self, mock_sandbox, mock_get_client): + """Local exploit with non-zero exit code means failed.""" + sandbox = self._setup_local_sandbox(mock_sandbox, mock_get_client) + sandbox.run_local_exploit.return_value = { + "stdout": "", + "stderr": "permission denied", + "exit_code": 126, + } + + tool_use = _make_tool_use(exploit_mode="local") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "failed" + assert "exited with code 126" in data["summary"] + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_local_interpreter_not_found(self, mock_sandbox, mock_get_client): + """InterpreterNotFoundError in local mode yields target_error.""" + from manus_agent.sandbox.exploit_sandbox import InterpreterNotFoundError + + sandbox = self._setup_local_sandbox(mock_sandbox, mock_get_client) + sandbox.run_local_exploit.side_effect = InterpreterNotFoundError( + container_role="target", + language="ruby", + candidates=["ruby", "ruby3.0"], + ) + + tool_use = _make_tool_use(exploit_mode="local", exploit_language="ruby") + result = verify_exploit(tool_use) + data = _get_json(result) + + # In local mode, InterpreterNotFoundError is target_error (not infra_error) + assert data["verification_status"] == "target_error" + assert "ruby" in data["summary"].lower() + assert data["error"]["category"] == "target" + assert data["error"]["stage"] == "interpreter_preflight" + assert data["error"]["retryable"] is False + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_local_transient_error(self, mock_sandbox, mock_get_client, mock_is_transient): + """Transient error during local exploit run yields infra_error.""" + sandbox = self._setup_local_sandbox(mock_sandbox, mock_get_client) + sandbox.run_local_exploit.side_effect = OSError("Docker API timeout") + mock_is_transient.return_value = True + + tool_use = _make_tool_use(exploit_mode="local") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "infra_error" + assert data["error"]["retryable"] is True + assert data["error"]["stage"] == "run_local_exploit" + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_local_non_transient_error(self, mock_sandbox, mock_get_client, mock_is_transient): + """Non-transient error during local exploit run yields target_error.""" + sandbox = self._setup_local_sandbox(mock_sandbox, mock_get_client) + sandbox.run_local_exploit.side_effect = ValueError("Bad format") + mock_is_transient.return_value = False + + tool_use = _make_tool_use(exploit_mode="local") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "target_error" + assert data["error"]["retryable"] is False + assert data["error"]["stage"] == "run_local_exploit" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_local_exploit_called_correctly(self, mock_sandbox, mock_get_client): + """run_local_exploit receives correct code and language.""" + sandbox = self._setup_local_sandbox(mock_sandbox, mock_get_client) + sandbox.run_local_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + + tool_use = _make_tool_use( + exploit_mode="local", + exploit_code="#!/bin/bash\nwhoami", + exploit_language="bash", + ) + verify_exploit(tool_use) + + sandbox.run_local_exploit.assert_called_once_with( + code="#!/bin/bash\nwhoami", + language="bash", + ) + + +# --------------------------------------------------------------------------- +# Unexpected top-level exceptions +# --------------------------------------------------------------------------- + + +class TestUnexpectedExceptions: + """Test the outer try/except catching unexpected errors.""" + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_unexpected_exception_non_transient(self, mock_sandbox, mock_get_client, mock_is_transient): + """Unexpected exception yields target_error when not transient.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + mock_is_transient.return_value = False + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "ok", "stderr": "", "exit_code": 0} + sandbox.build_log = "ok" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + # get_target_logs is called once after exploit success to assemble the result. + # Make it raise — this triggers the outer except handler which calls + # get_target_logs again (in _result via the except block). Use a counter + # to raise only on the first call, then return empty on subsequent calls. + call_count = {"n": 0} + original_error = MemoryError("out of memory") + + def _get_target_logs_side_effect(): + call_count["n"] += 1 + if call_count["n"] == 1: + raise original_error + return "" + + sandbox.get_target_logs.side_effect = _get_target_logs_side_effect + + tool_use = _make_tool_use(exploit_mode="remote") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "target_error" + assert "Unexpected error" in data["summary"] + assert data["error"]["stage"] == "unexpected" + sandbox.cleanup.assert_called() + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_unexpected_exception_transient(self, mock_sandbox, mock_get_client, mock_is_transient): + """Unexpected transient exception yields infra_error.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + mock_is_transient.return_value = True + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + sandbox.build_log = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + call_count = {"n": 0} + + def _get_target_logs_side_effect(): + call_count["n"] += 1 + if call_count["n"] == 1: + raise ConnectionResetError("socket reset") + return "" + + sandbox.get_target_logs.side_effect = _get_target_logs_side_effect + + tool_use = _make_tool_use(exploit_mode="remote") + result = verify_exploit(tool_use) + data = _get_json(result) + + assert data["verification_status"] == "infra_error" + assert data["error"]["retryable"] is True + + +# --------------------------------------------------------------------------- +# Cleanup guarantees +# --------------------------------------------------------------------------- + + +class TestCleanup: + """Verify that sandbox.cleanup() is always called.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_cleanup_on_success(self, mock_sandbox, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + verify_exploit(_make_tool_use()) + sandbox.cleanup.assert_called_once() + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_cleanup_on_build_error(self, mock_sandbox, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.side_effect = RuntimeError("fail") + sandbox.build_log = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + verify_exploit(_make_tool_use()) + sandbox.cleanup.assert_called_once() + + @patch("manus_agent.tools.verify_exploit.is_transient_docker_error") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_cleanup_on_unexpected_error(self, mock_sandbox, mock_get_client, mock_is_transient): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + mock_is_transient.return_value = False + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + sandbox.build_log = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + call_count = {"n": 0} + + def _get_target_logs_side_effect(): + call_count["n"] += 1 + if call_count["n"] == 1: + raise RuntimeError("crash") + return "" + + sandbox.get_target_logs.side_effect = _get_target_logs_side_effect + + verify_exploit(_make_tool_use()) + sandbox.cleanup.assert_called_once() + + +# --------------------------------------------------------------------------- +# Default values and edge cases +# --------------------------------------------------------------------------- + + +class TestDefaults: + """Test default values from input.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_default_timeout_300(self, mock_sandbox, mock_get_client): + """Default timeout is 300 seconds.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.side_effect = RuntimeError("stop") + sandbox.build_log = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + tool_use = { + "toolUseId": "tid", + "input": { + "dockerfile_content": "FROM x", + "exploit_code": "x", + "exploit_language": "python", + "cve_id": "CVE-2024-1234", + "target_info": {"affected_software": "x", "affected_versions": "1", "vulnerability_type": "RCE"}, + }, + } + verify_exploit(tool_use) + mock_sandbox.assert_called_once_with(timeout=300) + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_custom_timeout(self, mock_sandbox, mock_get_client): + """Custom timeout is forwarded to ExploitSandbox.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.side_effect = RuntimeError("stop") + sandbox.build_log = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + tool_use = _make_tool_use(timeout=600) + verify_exploit(tool_use) + mock_sandbox.assert_called_once_with(timeout=600) + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_default_target_port_80(self, mock_sandbox, mock_get_client): + """Default target port is 80.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = False + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + tool_use = { + "toolUseId": "tid", + "input": { + "dockerfile_content": "FROM x", + "exploit_code": "x", + "exploit_language": "bash", + "cve_id": "CVE-2024-1234", + "target_info": {"affected_software": "x", "affected_versions": "1", "vulnerability_type": "RCE"}, + }, + } + verify_exploit(tool_use) + sandbox.wait_for_target.assert_called_once_with(port=80, timeout=60) + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_default_exploit_mode_remote(self, mock_sandbox, mock_get_client): + """Default exploit mode is remote.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + tool_use = { + "toolUseId": "tid", + "input": { + "dockerfile_content": "FROM x", + "exploit_code": "echo test", + "exploit_language": "bash", + "cve_id": "CVE-2024-1234", + "target_info": {"affected_software": "x", "affected_versions": "1", "vulnerability_type": "RCE"}, + }, + } + verify_exploit(tool_use) + # Should call run_exploit (remote), not run_local_exploit + sandbox.run_exploit.assert_called_once() + sandbox.run_local_exploit.assert_not_called() + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_empty_target_env_default(self, mock_sandbox, mock_get_client): + """None target_env becomes empty dict.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.side_effect = RuntimeError("fail") + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + tool_use = _make_tool_use(target_env=None) + verify_exploit(tool_use) + sandbox.start_target.assert_called_once_with("sha256:img", environment={}) + + +# --------------------------------------------------------------------------- +# Result structure validation +# --------------------------------------------------------------------------- + + +class TestResultStructure: + """Verify result payloads have the expected structure.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_success_result_structure(self, mock_sandbox, mock_get_client): + """Verified result has all expected keys.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "ok", "stderr": "", "exit_code": 0} + sandbox.build_log = "build log" + sandbox.get_target_logs.return_value = "target log" + sandbox.get_docker_ps_all.return_value = "ps output" + sandbox.get_target_exit_code.return_value = 0 + + result = verify_exploit(_make_tool_use()) + assert result["toolUseId"] == "test-tool-use-id" + assert result["status"] == "success" + + data = _get_json(result) + expected_keys = { + "verification_status", + "summary", + "exploit_output", + "target_logs", + "build_log", + "docker_ps", + "target_exit_code", + "execution_time_seconds", + } + assert set(data.keys()) == expected_keys + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_error_result_structure(self, mock_sandbox, mock_get_client): + """Error result has all expected keys including error object.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.side_effect = RuntimeError("build failed") + sandbox.build_log = "log" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + result = verify_exploit(_make_tool_use()) + data = _get_json(result) + + expected_keys = { + "verification_status", + "summary", + "exploit_output", + "target_logs", + "build_log", + "docker_ps", + "target_exit_code", + "execution_time_seconds", + "error", + } + assert set(data.keys()) == expected_keys + assert data["exploit_output"] == {"stdout": "", "stderr": "", "exit_code": -1} + + +# --------------------------------------------------------------------------- +# log_tool_output_size integration +# --------------------------------------------------------------------------- + + +class TestLogging: + """Verify log_tool_output_size is called.""" + + @patch("manus_agent.tools.verify_exploit.log_tool_output_size") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_log_called_on_success(self, mock_sandbox, mock_get_client, mock_log): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + verify_exploit(_make_tool_use()) + mock_log.assert_called_once() + assert mock_log.call_args[0][0] == "verify_exploit" + + @patch("manus_agent.tools.verify_exploit.log_tool_output_size") + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_log_called_on_error(self, mock_sandbox, mock_get_client, mock_log): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.side_effect = RuntimeError("fail") + sandbox.build_log = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = None + + verify_exploit(_make_tool_use()) + mock_log.assert_called_once() + + +# --------------------------------------------------------------------------- +# CVE ID and target_info in summary +# --------------------------------------------------------------------------- + + +class TestSummaryContent: + """Verify summary text includes relevant CVE/target info.""" + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_verified_summary_has_cve_and_software(self, mock_sandbox, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + tool_use = _make_tool_use( + cve_id="CVE-2023-9999", + target_info={ + "affected_software": "Apache Struts", + "affected_versions": "2.0-2.5", + "vulnerability_type": "RCE", + }, + ) + result = verify_exploit(tool_use) + data = _get_json(result) + + assert "CVE-2023-9999" in data["summary"] + assert "Apache Struts" in data["summary"] + assert "RCE" in data["summary"] + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_failed_summary_has_exit_code(self, mock_sandbox, mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 42} + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + result = verify_exploit(_make_tool_use()) + data = _get_json(result) + + assert "exited with code 42" in data["summary"] + + @patch("manus_agent.tools.verify_exploit.get_docker_client") + @patch("manus_agent.tools.verify_exploit.ExploitSandbox") + def test_missing_target_info_fields_use_unknown(self, mock_sandbox, mock_get_client): + """When target_info fields are missing, summary uses 'unknown'.""" + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + sandbox = mock_sandbox.return_value + sandbox.build_target.return_value = "sha256:img" + sandbox.start_target.return_value = None + sandbox.wait_for_target.return_value = True + sandbox.run_exploit.return_value = {"stdout": "", "stderr": "", "exit_code": 0} + sandbox.build_log = "" + sandbox.get_target_logs.return_value = "" + sandbox.get_docker_ps_all.return_value = "" + sandbox.get_target_exit_code.return_value = 0 + + tool_use = _make_tool_use(target_info={}) + result = verify_exploit(tool_use) + data = _get_json(result) + + assert "unknown" in data["summary"]