from __future__ import annotations from app.core.contracts import ToolResult, UserTask from app.tools.base import BaseTool from app.tools.sandbox import ToolSandbox class ShellExecTool(BaseTool): name = "shell_exec" def __init__(self, sandbox: ToolSandbox) -> None: self._sandbox = sandbox def execute(self, task: UserTask, args: dict[str, object]) -> ToolResult: command = str(args.get("command", "")).strip() if not command: return ToolResult(tool=self.name, ok=False, error="Missing command", metadata={"exit_code": -1}) cwd = args.get("cwd") stdin_secret = args.get("stdin_secret") password = args.get("password") if password: command = f'echo "{password}" | sudo -S {command}' completed = self._sandbox.run_shell( command=command, cwd=str(cwd) if cwd else None, stdin_data=str(stdin_secret) if stdin_secret is not None else None, ) output = completed.stdout if completed.returncode == 0 else completed.stderr or completed.stdout error_output = completed.stderr or completed.stdout is_sudo_error = ( completed.returncode != 0 and ("permission denied" in error_output.lower() or "incorrect password" in error_output.lower() or "sudo: password incorrect" in error_output.lower() or "wrong password" in error_output.lower()) ) return ToolResult( tool=self.name, ok=completed.returncode == 0, output=output, error=None if completed.returncode == 0 else f"Command failed with exit code {completed.returncode}", metadata={"exit_code": completed.returncode, "needs_sudo": is_sudo_error}, )