diff --git a/src/praisonai-agents/tests/managed/test_managed_factory.py b/src/praisonai-agents/tests/managed/test_managed_factory.py index 782a348bd..76bcbd6c6 100644 --- a/src/praisonai-agents/tests/managed/test_managed_factory.py +++ b/src/praisonai-agents/tests/managed/test_managed_factory.py @@ -53,7 +53,7 @@ def test_defaults(self): cfg = LocalManagedConfig() assert cfg.name == "Agent" assert cfg.model == "gpt-4o" - assert cfg.sandbox_type == "subprocess" + assert cfg.host_packages_ok is False assert cfg.max_turns == 25 assert "execute_command" in cfg.tools @@ -471,3 +471,293 @@ def test_provision_execute_shutdown_local(self): asyncio.run(agent.shutdown_compute()) assert agent._compute_instance_id is None + + +# ====================================================================== # +# Security tests - package installation and sandbox behavior +# ====================================================================== # + +class TestManagedSandboxSafety: + def test_install_packages_without_compute_raises(self): + """Test that package installation without compute provider raises ManagedSandboxRequired.""" + import pytest + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + from praisonai.integrations.managed_agents import ManagedSandboxRequired + + cfg = LocalManagedConfig(packages={"pip": ["requests"]}) + agent = LocalManagedAgent(config=cfg) + + with pytest.raises(ManagedSandboxRequired, match="Package installation requires compute provider"): + agent._install_packages() + + def test_install_packages_with_host_packages_ok_works(self): + """Test that package installation with explicit opt-out works.""" + from unittest.mock import patch, MagicMock + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + cfg = LocalManagedConfig(packages={"pip": ["requests"]}, host_packages_ok=True) + agent = LocalManagedAgent(config=cfg) + + with patch('subprocess.run') as mock_run: + mock_run.return_value = MagicMock() + agent._install_packages() + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + assert "pip" in args + assert "install" in args + assert "requests" in args + + def test_install_packages_with_compute_runs_in_sandbox(self): + """Test that packages install in compute sandbox when compute provider attached.""" + import asyncio + from unittest.mock import patch, MagicMock, AsyncMock + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + mock_compute = MagicMock() + mock_compute.execute = AsyncMock(return_value={"exit_code": 0, "stdout": "success", "stderr": ""}) + + cfg = LocalManagedConfig(packages={"pip": ["requests"]}) + agent = LocalManagedAgent(config=cfg, compute=mock_compute) + agent._compute_instance_id = "test_instance" + + with patch('subprocess.run') as mock_subprocess: + agent._install_packages() + # subprocess.run should NOT be called when compute is attached + mock_subprocess.assert_not_called() + # compute.execute should be called instead + mock_compute.execute.assert_called_once() + + def test_no_packages_skips_installation(self): + """Test that no packages specified skips installation entirely.""" + from unittest.mock import patch + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + cfg = LocalManagedConfig() + agent = LocalManagedAgent(config=cfg) + + with patch('subprocess.run') as mock_run: + agent._install_packages() + mock_run.assert_not_called() + + def test_empty_pip_packages_skips_installation(self): + """Test that empty pip packages list skips installation.""" + from unittest.mock import patch + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + cfg = LocalManagedConfig(packages={"pip": []}) + agent = LocalManagedAgent(config=cfg) + + with patch('subprocess.run') as mock_run: + agent._install_packages() + mock_run.assert_not_called() + + def test_exception_message_includes_remediation(self): + """Test that ManagedSandboxRequired exception includes actionable remediation.""" + import pytest + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + from praisonai.integrations.managed_agents import ManagedSandboxRequired + + cfg = LocalManagedConfig(packages={"pip": ["dangerous-package"]}) + agent = LocalManagedAgent(config=cfg) + + with pytest.raises(ManagedSandboxRequired) as exc_info: + agent._install_packages() + + error_msg = str(exc_info.value) + assert "dangerous-package" in error_msg + assert "compute='docker'" in error_msg + assert "host_packages_ok=True" in error_msg + + def test_managed_sandbox_required_exception_creation(self): + """Test ManagedSandboxRequired exception can be created and has correct default message.""" + from praisonai.integrations.managed_agents import ManagedSandboxRequired + + exc = ManagedSandboxRequired() + assert "Package installation requires compute provider for security" in str(exc) + + custom_exc = ManagedSandboxRequired("Custom message") + assert str(custom_exc) == "Custom message" + + +class TestComputeToolBridge: + """Test compute-bridged tool execution routing.""" + + def test_tools_use_compute_bridge_when_compute_attached(self): + """Test that shell tools use compute bridge when compute provider attached.""" + from unittest.mock import MagicMock + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + mock_compute = MagicMock() + cfg = LocalManagedConfig(tools=["execute_command", "read_file", "write_file", "list_files"]) + agent = LocalManagedAgent(config=cfg, compute=mock_compute) + + tools = agent._resolve_tools() + + # Should have 4 compute-bridged tools + shell_tools = [t for t in tools if hasattr(t, '__name__') and + t.__name__ in {"execute_command", "read_file", "write_file", "list_files"}] + assert len(shell_tools) == 4 + + def test_tools_use_host_when_no_compute(self): + """Test that tools use host versions when no compute provider.""" + from unittest.mock import patch, MagicMock + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + cfg = LocalManagedConfig(tools=["execute_command"]) + agent = LocalManagedAgent(config=cfg) + + with patch('praisonaiagents.tools.execute_command') as mock_tool: + mock_tool.__name__ = "execute_command" + tools = agent._resolve_tools() + # Should use host tool, not compute bridge + assert mock_tool in tools + + def test_compute_execute_command_bridge(self): + """Test compute-bridged execute_command works correctly.""" + import asyncio + from unittest.mock import MagicMock, AsyncMock + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + mock_compute = MagicMock() + mock_compute.execute = AsyncMock(return_value={ + "stdout": "hello world", + "stderr": "", + "exit_code": 0 + }) + + agent = LocalManagedAgent(compute=mock_compute) + agent._compute_instance_id = "test_instance" + + execute_command = agent._create_compute_execute_command() + result = execute_command("echo hello world") + + assert result == "hello world" + mock_compute.execute.assert_called_once_with("test_instance", "echo hello world", timeout=300) + + def test_compute_execute_command_with_stderr(self): + """Test compute-bridged execute_command handles stderr correctly.""" + import asyncio + from unittest.mock import MagicMock, AsyncMock + from praisonai.integrations.managed_local import LocalManagedAgent + + mock_compute = MagicMock() + mock_compute.execute = AsyncMock(return_value={ + "stdout": "output", + "stderr": "warning", + "exit_code": 1 + }) + + agent = LocalManagedAgent(compute=mock_compute) + agent._compute_instance_id = "test_instance" + + execute_command = agent._create_compute_execute_command() + result = execute_command("failing_command") + + assert "output" in result + assert "STDERR: warning" in result + assert "Exit code: 1" in result + + def test_compute_read_file_bridge(self): + """Test compute-bridged read_file works correctly.""" + from unittest.mock import MagicMock, AsyncMock + from praisonai.integrations.managed_local import LocalManagedAgent + + mock_compute = MagicMock() + mock_compute.execute = AsyncMock(return_value={ + "stdout": "file contents", + "stderr": "", + "exit_code": 0 + }) + + agent = LocalManagedAgent(compute=mock_compute) + agent._compute_instance_id = "test_instance" + + read_file = agent._create_compute_read_file() + result = read_file("/path/to/file.txt") + + assert result == "file contents" + mock_compute.execute.assert_called_once_with("test_instance", "cat /path/to/file.txt", timeout=60) + + def test_compute_write_file_bridge(self): + """Test compute-bridged write_file works correctly.""" + from unittest.mock import MagicMock, AsyncMock + from praisonai.integrations.managed_local import LocalManagedAgent + + mock_compute = MagicMock() + mock_compute.execute = AsyncMock(return_value={ + "stdout": "", + "stderr": "", + "exit_code": 0 + }) + + agent = LocalManagedAgent(compute=mock_compute) + agent._compute_instance_id = "test_instance" + + write_file = agent._create_compute_write_file() + result = write_file("/path/to/file.txt", "file content") + + assert "File written successfully" in result + # Check that the command was properly escaped + mock_compute.execute.assert_called_once() + call_args = mock_compute.execute.call_args + assert "echo" in call_args[0][1] + assert "/path/to/file.txt" in call_args[0][1] + + def test_compute_list_files_bridge(self): + """Test compute-bridged list_files works correctly.""" + from unittest.mock import MagicMock, AsyncMock + from praisonai.integrations.managed_local import LocalManagedAgent + + mock_compute = MagicMock() + mock_compute.execute = AsyncMock(return_value={ + "stdout": "file1.txt\nfile2.txt\n", + "stderr": "", + "exit_code": 0 + }) + + agent = LocalManagedAgent(compute=mock_compute) + agent._compute_instance_id = "test_instance" + + list_files = agent._create_compute_list_files() + result = list_files("/some/dir") + + assert "file1.txt" in result + assert "file2.txt" in result + mock_compute.execute.assert_called_once_with("test_instance", "ls -la /some/dir", timeout=60) + + def test_compute_tools_require_provisioned_instance(self): + """Test that compute tools raise error when no instance is provisioned.""" + import pytest + from praisonai.integrations.managed_local import LocalManagedAgent + + mock_compute = MagicMock() + agent = LocalManagedAgent(compute=mock_compute) + # Don't set _compute_instance_id + + execute_command = agent._create_compute_execute_command() + with pytest.raises(RuntimeError, match="No compute provider provisioned"): + execute_command("echo test") + + def test_auto_provision_compute_in_ensure_agent(self): + """Test that _ensure_agent auto-provisions compute when needed.""" + import asyncio + from unittest.mock import patch, MagicMock, AsyncMock + from praisonai.integrations.managed_local import LocalManagedAgent, LocalManagedConfig + + mock_compute = MagicMock() + mock_info = MagicMock() + mock_info.instance_id = "auto_provisioned_instance" + + cfg = LocalManagedConfig() + agent = LocalManagedAgent(config=cfg, compute=mock_compute) + + with patch.object(agent, 'provision_compute', new_callable=AsyncMock) as mock_provision: + mock_provision.return_value = mock_info + with patch('praisonaiagents.Agent') as mock_agent_class: + mock_agent_class.return_value = MagicMock() + + inner_agent = agent._ensure_agent() + + # Should have auto-provisioned compute + mock_provision.assert_called_once() + assert agent._compute_instance_id == "auto_provisioned_instance" diff --git a/src/praisonai/praisonai/integrations/managed_agents.py b/src/praisonai/praisonai/integrations/managed_agents.py index 761384abc..7bdb0a191 100644 --- a/src/praisonai/praisonai/integrations/managed_agents.py +++ b/src/praisonai/praisonai/integrations/managed_agents.py @@ -36,6 +36,22 @@ logger = logging.getLogger(__name__) +class ManagedSandboxRequired(Exception): + """Raised when package installation or tool execution requires a compute provider for security. + + This exception is raised when: + - Packages are specified without a compute provider attached + - host_packages_ok=False (the default for security) + + To resolve: + 1. Attach a compute provider: LocalManagedAgent(compute="docker") + 2. Or explicitly opt-out: LocalManagedConfig(host_packages_ok=True) + """ + + def __init__(self, message: str = "Package installation requires compute provider for security"): + super().__init__(message) + + # --------------------------------------------------------------------------- # ManagedConfig — Anthropic-specific configuration dataclass # Lives in the Wrapper (not Core SDK) because its fields map directly to diff --git a/src/praisonai/praisonai/integrations/managed_local.py b/src/praisonai/praisonai/integrations/managed_local.py index dc27489fc..51a9f405a 100644 --- a/src/praisonai/praisonai/integrations/managed_local.py +++ b/src/praisonai/praisonai/integrations/managed_local.py @@ -78,11 +78,13 @@ class LocalManagedConfig: metadata: Dict[str, Any] = field(default_factory=dict) # ── Environment fields ── - sandbox_type: str = "subprocess" working_dir: str = "" env: Dict[str, str] = field(default_factory=dict) packages: Optional[Dict[str, List[str]]] = None networking: Dict[str, Any] = field(default_factory=lambda: {"type": "unrestricted"}) + + # ── Security fields ── + host_packages_ok: bool = False # Safety: require explicit opt-out for host pip installs # ── Session fields ── session_title: str = "PraisonAI local session" @@ -284,6 +286,128 @@ def _resolve_model(self) -> str: return model + # ------------------------------------------------------------------ + # Compute-based tool execution bridge + # ------------------------------------------------------------------ + def _create_compute_execute_command(self) -> Callable: + """Create a compute-bridged execute_command tool.""" + def execute_command(command: str, timeout: int = 300) -> str: + """Execute a shell command in the compute sandbox.""" + if self._compute is None or self._compute_instance_id is None: + raise RuntimeError("No compute provider provisioned for command execution") + + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + result = loop.run_until_complete(self._compute.execute( + self._compute_instance_id, command, timeout=timeout + )) + stdout = result.get("stdout", "") + stderr = result.get("stderr", "") + exit_code = result.get("exit_code", 0) + + output = "" + if stdout: + output += stdout + if stderr: + if output: + output += "\n" + output += f"STDERR: {stderr}" + if exit_code != 0: + output += f"\nExit code: {exit_code}" + + return output or "Command completed successfully" + finally: + loop.close() + + execute_command.__name__ = "execute_command" + execute_command.__doc__ = "Execute a shell command in the compute sandbox." + return execute_command + + def _create_compute_read_file(self) -> Callable: + """Create a compute-bridged read_file tool.""" + def read_file(file_path: str) -> str: + """Read a file from the compute sandbox.""" + if self._compute is None or self._compute_instance_id is None: + raise RuntimeError("No compute provider provisioned for file operations") + + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + # Use cat command to read file content + import shlex + result = loop.run_until_complete(self._compute.execute( + self._compute_instance_id, f"cat -- {shlex.quote(file_path)}", timeout=60 + )) + if result.get("exit_code", 0) != 0: + stderr = result.get("stderr", "") + return f"Error reading file: {stderr}" + return result.get("stdout", "") + finally: + loop.close() + + read_file.__name__ = "read_file" + read_file.__doc__ = "Read a file from the compute sandbox." + return read_file + + def _create_compute_write_file(self) -> Callable: + """Create a compute-bridged write_file tool.""" + def write_file(file_path: str, content: str) -> str: + """Write content to a file in the compute sandbox.""" + if self._compute is None or self._compute_instance_id is None: + raise RuntimeError("No compute provider provisioned for file operations") + + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + # Use printf for safe file writing (preserves newlines) + import shlex + escaped_content = shlex.quote(content) + escaped_path = shlex.quote(file_path) + result = loop.run_until_complete(self._compute.execute( + self._compute_instance_id, f"printf '%s' {escaped_content} > {escaped_path}", timeout=60 + )) + if result.get("exit_code", 0) != 0: + stderr = result.get("stderr", "") + return f"Error writing file: {stderr}" + return f"File written successfully to {file_path}" + finally: + loop.close() + + write_file.__name__ = "write_file" + write_file.__doc__ = "Write content to a file in the compute sandbox." + return write_file + + def _create_compute_list_files(self) -> Callable: + """Create a compute-bridged list_files tool.""" + def list_files(directory: str = ".") -> str: + """List files in a directory within the compute sandbox.""" + if self._compute is None or self._compute_instance_id is None: + raise RuntimeError("No compute provider provisioned for file operations") + + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + # Use ls -la command to list files + import shlex + result = loop.run_until_complete(self._compute.execute( + self._compute_instance_id, f"ls -la -- {shlex.quote(directory)}", timeout=60 + )) + if result.get("exit_code", 0) != 0: + stderr = result.get("stderr", "") + return f"Error listing directory: {stderr}" + return result.get("stdout", "") + finally: + loop.close() + + list_files.__name__ = "list_files" + list_files.__doc__ = "List files in a directory within the compute sandbox." + return list_files + # ------------------------------------------------------------------ # Tool resolution # ------------------------------------------------------------------ @@ -297,9 +421,25 @@ def _resolve_tools(self) -> List: for name in tool_names: resolved_names.append(TOOL_ALIAS_MAP.get(name, name)) - # Import tools lazily + # Import tools lazily, using compute-bridged versions when available tools = [] + compute_tools = {"execute_command", "read_file", "write_file", "list_files"} + for name in resolved_names: + # Use compute-bridged versions for shell/file tools when compute attached + if self._compute is not None and name in compute_tools: + if name == "execute_command": + tools.append(self._create_compute_execute_command()) + elif name == "read_file": + tools.append(self._create_compute_read_file()) + elif name == "write_file": + tools.append(self._create_compute_write_file()) + elif name == "list_files": + tools.append(self._create_compute_list_files()) + logger.debug("[local_managed] using compute-bridged tool: %s", name) + continue + + # Use host tools for non-shell operations or when no compute attached try: from praisonaiagents import tools as tool_module func = getattr(tool_module, name, None) @@ -451,27 +591,104 @@ def _restore_state(self) -> None: self.provider = saved_cfg["provider"] def _install_packages(self) -> None: - """Install packages specified in config before agent starts.""" + """Install packages specified in config before agent starts. + + Security: When compute provider is attached, packages install in sandbox. + When no compute provider, requires explicit host_packages_ok=True. + """ packages = self._cfg.get("packages") if not packages: return pip_pkgs = packages.get("pip", []) if isinstance(packages, dict) else [] - if pip_pkgs: - cmd = [sys.executable, "-m", "pip", "install", "-q"] + pip_pkgs - logger.info("[local_managed] installing pip packages: %s", pip_pkgs) + if not pip_pkgs: + return + + # If compute provider attached, install in sandbox + if self._compute is not None: + logger.info("[local_managed] installing pip packages in sandbox: %s", pip_pkgs) + # Auto-provision compute if needed + if self._compute_instance_id is None: + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + self._compute_instance_id = loop.run_until_complete(self._provision_compute_if_needed()) + finally: + loop.close() + + # Install packages in compute sandbox + import shlex + cmd = "pip install -q " + " ".join(shlex.quote(pkg) for pkg in pip_pkgs) try: - subprocess.run(cmd, check=True, capture_output=True, timeout=120) - except subprocess.CalledProcessError as e: - logger.warning("[local_managed] pip install failed: %s", e.stderr) - except subprocess.TimeoutExpired: - logger.warning("[local_managed] pip install timed out") + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + result = loop.run_until_complete(self._compute.execute( + self._compute_instance_id, cmd, timeout=120 + )) + if result.get("exit_code", 0) != 0: + stderr = result.get("stderr", "") + raise RuntimeError( + f"pip install failed in compute sandbox: {stderr}. " + f"Packages: {pip_pkgs}" + ) + finally: + loop.close() + except Exception as e: + if "RuntimeError" in str(type(e)): + raise # Re-raise our structured error + raise RuntimeError( + f"pip install failed in compute sandbox: {e}. " + f"Packages: {pip_pkgs}" + ) + return + + # No compute provider - check if host install is explicitly allowed + host_packages_ok = self._cfg.get("host_packages_ok", False) + if not host_packages_ok: + from .managed_agents import ManagedSandboxRequired + raise ManagedSandboxRequired( + f"Package installation requires compute provider for security. " + f"Packages requested: {pip_pkgs}. " + f"To fix: 1) Add compute='docker' (recommended) or 2) Set host_packages_ok=True (unsafe)." + ) + + # Host install with explicit opt-out + logger.warning("[local_managed] installing pip packages on HOST (unsafe): %s", pip_pkgs) + cmd = [sys.executable, "-m", "pip", "install", "-q"] + pip_pkgs + try: + subprocess.run(cmd, check=True, capture_output=True, timeout=120) + except subprocess.CalledProcessError as e: + logger.warning("[local_managed] pip install failed: %s", e.stderr) + except subprocess.TimeoutExpired: + logger.warning("[local_managed] pip install timed out") + + async def _provision_compute_if_needed(self) -> str: + """Auto-provision compute if needed and return instance ID.""" + if self._compute is None: + raise RuntimeError("No compute provider attached") + if self._compute_instance_id is not None: + return self._compute_instance_id + info = await self.provision_compute() + return info.instance_id def _ensure_agent(self) -> Any: """Create or return the inner PraisonAI Agent.""" if self._inner_agent is not None: return self._inner_agent + # Auto-provision compute if needed for tool execution + if self._compute is not None and self._compute_instance_id is None: + import asyncio + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + self._compute_instance_id = loop.run_until_complete(self._provision_compute_if_needed()) + finally: + loop.close() + self._install_packages() from praisonaiagents import Agent