Support sudo password flow for shell commands

duck_core/tools/gateway.py

@@ -4,7 +4,6 @@ from duck_core.tools.base import Tool, ToolResult
 from duck_core.tools.file_read import FileReadTool
 from duck_core.tools.file_write import FileWriteTool
 from duck_core.tools.list_dir import ListDirTool
-from duck_core.tools.os_update_check import OsUpdateCheckTool
 from duck_core.tools.search_files import SearchFilesTool
 from duck_core.tools.shell_exec_safe import ShellExecSafeTool
 
@@ -21,12 +20,13 @@ class ToolGateway:
                 FileWriteTool(workspace),
                 ListDirTool(workspace),
                 SearchFilesTool(workspace),
-                OsUpdateCheckTool(workspace),
                 ShellExecSafeTool(workspace),
             ]
         )
 
-    async def run_action(self, action: dict[str, Any], approved: bool = False) -> ToolResult:
+    async def run_action(
+        self, action: dict[str, Any], approved: bool = False, password: str | None = None
+    ) -> ToolResult:
         tool_name = str(action.get("tool", ""))
         tool = self.tools.get(tool_name)
         if tool is None:
@@ -36,4 +36,6 @@ class ToolGateway:
             return ToolResult(ok=False, error="Tool args must be an object")
         if approved:
             args = {**args, "_approved": True}
+        if password is not None:
+            args = {**args, "_password": password}
         return await tool.run(args)

duck_core/tools/os_update_check.py

@@ -1,58 +0,0 @@
-import shutil
-import subprocess
-from typing import Any
-
-from duck_core.tools.base import ToolResult
-
-
-class OsUpdateCheckTool:
-    name = "os_update_check"
-    risk_level = "low"
-
-    def __init__(self, workspace: str, timeout_seconds: int = 60, max_lines: int = 300):
-        self.workspace = workspace
-        self.timeout_seconds = timeout_seconds
-        self.max_lines = max_lines
-
-    async def run(self, args: dict[str, Any]) -> ToolResult:
-        if shutil.which("apt"):
-            return self._run_apt()
-        return ToolResult(
-            ok=False,
-            error="No supported package manager found for update checks.",
-            metadata={"supported_package_managers": ["apt"]},
-        )
-
-    def _run_apt(self) -> ToolResult:
-        try:
-            completed = subprocess.run(
-                ["apt", "list", "--upgradable"],
-                cwd=self.workspace,
-                text=True,
-                capture_output=True,
-                timeout=self.timeout_seconds,
-                check=False,
-            )
-        except subprocess.SubprocessError as exc:
-            return ToolResult(ok=False, error=str(exc), metadata={"package_manager": "apt"})
-
-        lines = completed.stdout.splitlines()
-        package_lines = [line for line in lines if "/" in line and "upgradable from:" in line]
-        output_lines = lines[: self.max_lines]
-        truncated = len(lines) > self.max_lines
-        if truncated:
-            output_lines.append(f"... truncated after {self.max_lines} lines")
-
-        return ToolResult(
-            ok=completed.returncode == 0,
-            output="\n".join(output_lines),
-            error=completed.stderr or None,
-            metadata={
-                "package_manager": "apt",
-                "command": "apt list --upgradable",
-                "returncode": completed.returncode,
-                "upgradable_count": len(package_lines),
-                "refreshed_cache": False,
-                "truncated": truncated,
-            },
-        )

duck_core/tools/shell_exec_safe.py

@@ -13,6 +13,8 @@ ALLOWLIST = {
     "tail",
     "grep",
     "find",
+    "apt list",
+    "apt-cache policy",
     "pytest",
     "python -m pytest",
     "python3 -m pytest",
@@ -23,7 +25,6 @@ ALLOWLIST = {
 
 BLOCKLIST = {
     "rm",
-    "sudo",
     "su",
     "dd",
     "mkfs",
@@ -58,17 +59,27 @@ class ShellExecSafeTool:
     async def run(self, args: dict[str, Any]) -> ToolResult:
         command = str(args.get("command", "")).strip()
         approved = bool(args.get("_approved"))
+        password = args.get("_password")
         allowed, reason, blocked = self._is_allowed(command, approved=approved)
         if not allowed:
             metadata = {"blocked": True} if blocked else {"requires_approval": True}
             return ToolResult(ok=False, error=reason, metadata=metadata)
+        if self._is_sudo_command(command) and not password:
+            return ToolResult(
+                ok=False,
+                error="Sudo password is required to run this command.",
+                metadata={"requires_password": True},
+            )
+        run_command = self._sudo_stdin_command(command) if self._is_sudo_command(command) else command
+        input_text = f"{password}\n" if self._is_sudo_command(command) else None
         try:
             completed = subprocess.run(
-                command,
+                run_command,
                 cwd=self.workspace,
                 shell=True,
                 text=True,
                 capture_output=True,
+                input=input_text,
                 timeout=self.timeout_seconds,
                 check=False,
             )
@@ -93,6 +104,8 @@ class ShellExecSafeTool:
                 return False, f"Command is blocked: {blocked}", True
         if approved:
             return True, None, False
+        if self._is_sudo_command(command):
+            return False, "Sudo command requires approval.", False
         prefix1 = parts[0] if parts else ""
         prefix2 = " ".join(parts[:2])
         prefix3 = " ".join(parts[:3])
@@ -107,3 +120,14 @@ class ShellExecSafeTool:
         if " " in lowered_blocked or "|" in lowered_blocked:
             return lowered_command.startswith(lowered_blocked) or lowered_blocked in lowered_command
         return bool(parts) and parts[0].lower() == lowered_blocked
+
+    def _is_sudo_command(self, command: str) -> bool:
+        try:
+            parts = shlex.split(command)
+        except ValueError:
+            return False
+        return bool(parts) and parts[0] == "sudo"
+
+    def _sudo_stdin_command(self, command: str) -> str:
+        parts = shlex.split(command)
+        return shlex.join(["sudo", "-S", "-p", "", *parts[1:]])