| 1 | """Shell execution tool.""" |
| 2 | |
| 3 | import asyncio |
| 4 | import os |
| 5 | import re |
| 6 | import shutil |
| 7 | import sys |
| 8 | from pathlib import Path |
| 9 | from typing import Any |
| 10 | |
| 11 | from loguru import logger |
| 12 | |
| 13 | from nanobot.agent.tools.base import Tool, tool_parameters |
| 14 | from nanobot.agent.tools.sandbox import wrap_command |
| 15 | from nanobot.agent.tools.schema import IntegerSchema, StringSchema, tool_parameters_schema |
| 16 | from nanobot.config.paths import get_media_dir |
| 17 | |
| 18 | _IS_WINDOWS = sys.platform == "win32" |
| 19 | |
| 20 | |
| 21 | @tool_parameters( |
| 22 | tool_parameters_schema( |
| 23 | command=StringSchema("The shell command to execute"), |
| 24 | working_dir=StringSchema("Optional working directory for the command"), |
| 25 | timeout=IntegerSchema( |
| 26 | 60, |
| 27 | description=( |
| 28 | "Timeout in seconds. Increase for long-running commands " |
| 29 | "like compilation or installation (default 60, max 600)." |
| 30 | ), |
| 31 | minimum=1, |
| 32 | maximum=600, |
| 33 | ), |
| 34 | required=["command"], |
| 35 | ) |
| 36 | ) |
| 37 | class ExecTool(Tool): |
| 38 | """Tool to execute shell commands.""" |
| 39 | |
| 40 | def __init__( |
| 41 | self, |
| 42 | timeout: int = 60, |
| 43 | working_dir: str | None = None, |
| 44 | deny_patterns: list[str] | None = None, |
| 45 | allow_patterns: list[str] | None = None, |
| 46 | restrict_to_workspace: bool = False, |
| 47 | sandbox: str = "", |
| 48 | path_append: str = "", |
| 49 | allowed_env_keys: list[str] | None = None, |
| 50 | ): |
| 51 | self.timeout = timeout |
| 52 | self.working_dir = working_dir |
| 53 | self.sandbox = sandbox |
| 54 | self.deny_patterns = deny_patterns or [ |
| 55 | r"\brm\s+-[rf]{1,2}\b", # rm -r, rm -rf, rm -fr |
| 56 | r"\bdel\s+/[fq]\b", # del /f, del /q |
| 57 | r"\brmdir\s+/s\b", # rmdir /s |
| 58 | r"(?:^|[;&|]\s*)format\b", # format (as standalone command only) |
| 59 | r"\b(mkfs|diskpart)\b", # disk operations |
| 60 | r"\bdd\s+if=", # dd |
| 61 | r">\s*/dev/sd", # write to disk |
| 62 | r"\b(shutdown|reboot|poweroff)\b", # system power |
| 63 | r":\(\)\s*\{.*\};\s*:", # fork bomb |
| 64 | # Block writes to nanobot internal state files (#2989). |
| 65 | # history.jsonl / .dream_cursor are managed by append_history(); |
| 66 | # direct writes corrupt the cursor format and crash /dream. |
| 67 | r">>?\s*\S*(?:history\.jsonl|\.dream_cursor)", # > / >> redirect |
| 68 | r"\btee\b[^|;&<>]*(?:history\.jsonl|\.dream_cursor)", # tee / tee -a |
| 69 | r"\b(?:cp|mv)\b(?:\s+[^\s|;&<>]+)+\s+\S*(?:history\.jsonl|\.dream_cursor)", # cp/mv target |
| 70 | r"\bdd\b[^|;&<>]*\bof=\S*(?:history\.jsonl|\.dream_cursor)", # dd of= |
| 71 | r"\bsed\s+-i[^|;&<>]*(?:history\.jsonl|\.dream_cursor)", # sed -i |
| 72 | ] |
| 73 | self.allow_patterns = allow_patterns or [] |
| 74 | self.restrict_to_workspace = restrict_to_workspace |
| 75 | self.path_append = path_append |
| 76 | self.allowed_env_keys = allowed_env_keys or [] |
| 77 | |
| 78 | @property |
| 79 | def name(self) -> str: |
| 80 | return "exec" |
| 81 | |
| 82 | _MAX_TIMEOUT = 600 |
| 83 | _MAX_OUTPUT = 10_000 |
| 84 | |
| 85 | @property |
| 86 | def description(self) -> str: |
| 87 | return ( |
| 88 | "Execute a shell command and return its output. " |
| 89 | "Prefer read_file/write_file/edit_file over cat/echo/sed, " |
| 90 | "and grep/glob over shell find/grep. " |
| 91 | "Use -y or --yes flags to avoid interactive prompts. " |
| 92 | "Output is truncated at 10 000 chars; timeout defaults to 60s." |
| 93 | ) |
| 94 | |
| 95 | @property |
| 96 | def exclusive(self) -> bool: |
| 97 | return True |
| 98 | |
| 99 | async def execute( |
| 100 | self, command: str, working_dir: str | None = None, |
| 101 | timeout: int | None = None, **kwargs: Any, |
| 102 | ) -> str: |
| 103 | cwd = working_dir or self.working_dir or os.getcwd() |
| 104 | |
| 105 | # Prevent an LLM-supplied working_dir from escaping the configured |
| 106 | # workspace when restrict_to_workspace is enabled (#2826). Without |
| 107 | # this, a caller can pass working_dir="/etc" and then all absolute |
| 108 | # paths under /etc would pass the _guard_command check that anchors |
| 109 | # on cwd. |
| 110 | if self.restrict_to_workspace and self.working_dir: |
| 111 | try: |
| 112 | requested = Path(cwd).expanduser().resolve() |
| 113 | workspace_root = Path(self.working_dir).expanduser().resolve() |
| 114 | except Exception: |
| 115 | return "Error: working_dir could not be resolved" |
| 116 | if requested != workspace_root and workspace_root not in requested.parents: |
| 117 | return "Error: working_dir is outside the configured workspace" |
| 118 | |
| 119 | guard_error = self._guard_command(command, cwd) |
| 120 | if guard_error: |
| 121 | return guard_error |
| 122 | |
| 123 | if self.sandbox: |
| 124 | if _IS_WINDOWS: |
| 125 | logger.warning( |
| 126 | "Sandbox '{}' is not supported on Windows; running unsandboxed", |
| 127 | self.sandbox, |
| 128 | ) |
| 129 | else: |
| 130 | workspace = self.working_dir or cwd |
| 131 | command = wrap_command(self.sandbox, command, workspace, cwd) |
| 132 | cwd = str(Path(workspace).resolve()) |
| 133 | |
| 134 | effective_timeout = min(timeout or self.timeout, self._MAX_TIMEOUT) |
| 135 | env = self._build_env() |
| 136 | |
| 137 | if self.path_append: |
| 138 | if _IS_WINDOWS: |
| 139 | env["PATH"] = env.get("PATH", "") + ";" + self.path_append |
| 140 | else: |
| 141 | command = f'export PATH="$PATH:{self.path_append}"; {command}' |
| 142 | |
| 143 | try: |
| 144 | process = await self._spawn(command, cwd, env) |
| 145 | |
| 146 | try: |
| 147 | stdout, stderr = await asyncio.wait_for( |
| 148 | process.communicate(), |
| 149 | timeout=effective_timeout, |
| 150 | ) |
| 151 | except asyncio.TimeoutError: |
| 152 | await self._kill_process(process) |
| 153 | return f"Error: Command timed out after {effective_timeout} seconds" |
| 154 | except asyncio.CancelledError: |
| 155 | await self._kill_process(process) |
| 156 | raise |
| 157 | |
| 158 | output_parts = [] |
| 159 | |
| 160 | if stdout: |
| 161 | output_parts.append(stdout.decode("utf-8", errors="replace")) |
| 162 | |
| 163 | if stderr: |
| 164 | stderr_text = stderr.decode("utf-8", errors="replace") |
| 165 | if stderr_text.strip(): |
| 166 | output_parts.append(f"STDERR:\n{stderr_text}") |
| 167 | |
| 168 | output_parts.append(f"\nExit code: {process.returncode}") |
| 169 | |
| 170 | result = "\n".join(output_parts) if output_parts else "(no output)" |
| 171 | |
| 172 | max_len = self._MAX_OUTPUT |
| 173 | if len(result) > max_len: |
| 174 | half = max_len // 2 |
| 175 | result = ( |
| 176 | result[:half] |
| 177 | + f"\n\n... ({len(result) - max_len:,} chars truncated) ...\n\n" |
| 178 | + result[-half:] |
| 179 | ) |
| 180 | |
| 181 | return result |
| 182 | |
| 183 | except Exception as e: |
| 184 | return f"Error executing command: {str(e)}" |
| 185 | |
| 186 | @staticmethod |
| 187 | async def _spawn( |
| 188 | command: str, cwd: str, env: dict[str, str], |
| 189 | ) -> asyncio.subprocess.Process: |
| 190 | """Launch *command* in a platform-appropriate shell.""" |
| 191 | if _IS_WINDOWS: |
| 192 | comspec = env.get("COMSPEC", os.environ.get("COMSPEC", "cmd.exe")) |
| 193 | return await asyncio.create_subprocess_exec( |
| 194 | comspec, "/c", command, |
| 195 | stdout=asyncio.subprocess.PIPE, |
| 196 | stderr=asyncio.subprocess.PIPE, |
| 197 | cwd=cwd, |
| 198 | env=env, |
| 199 | ) |
| 200 | bash = shutil.which("bash") or "/bin/bash" |
| 201 | return await asyncio.create_subprocess_exec( |
| 202 | bash, "-l", "-c", command, |
| 203 | stdout=asyncio.subprocess.PIPE, |
| 204 | stderr=asyncio.subprocess.PIPE, |
| 205 | cwd=cwd, |
| 206 | env=env, |
| 207 | ) |
| 208 | |
| 209 | @staticmethod |
| 210 | async def _kill_process(process: asyncio.subprocess.Process) -> None: |
| 211 | """Kill a subprocess and reap it to prevent zombies.""" |
| 212 | process.kill() |
| 213 | try: |
| 214 | await asyncio.wait_for(process.wait(), timeout=5.0) |
| 215 | except asyncio.TimeoutError: |
| 216 | pass |
| 217 | finally: |
| 218 | if not _IS_WINDOWS: |
| 219 | try: |
| 220 | os.waitpid(process.pid, os.WNOHANG) |
| 221 | except (ProcessLookupError, ChildProcessError) as e: |
| 222 | logger.debug("Process already reaped or not found: {}", e) |
| 223 | |
| 224 | def _build_env(self) -> dict[str, str]: |
| 225 | """Build a minimal environment for subprocess execution. |
| 226 | |
| 227 | On Unix, only HOME/LANG/TERM are passed; ``bash -l`` sources the |
| 228 | user's profile which sets PATH and other essentials. |
| 229 | |
| 230 | On Windows, ``cmd.exe`` has no login-profile mechanism, so a curated |
| 231 | set of system variables (including PATH) is forwarded. API keys and |
| 232 | other secrets are still excluded. |
| 233 | """ |
| 234 | if _IS_WINDOWS: |
| 235 | sr = os.environ.get("SYSTEMROOT", r"C:\Windows") |
| 236 | env = { |
| 237 | "SYSTEMROOT": sr, |
| 238 | "COMSPEC": os.environ.get("COMSPEC", f"{sr}\\system32\\cmd.exe"), |
| 239 | "USERPROFILE": os.environ.get("USERPROFILE", ""), |
| 240 | "HOMEDRIVE": os.environ.get("HOMEDRIVE", "C:"), |
| 241 | "HOMEPATH": os.environ.get("HOMEPATH", "\\"), |
| 242 | "TEMP": os.environ.get("TEMP", f"{sr}\\Temp"), |
| 243 | "TMP": os.environ.get("TMP", f"{sr}\\Temp"), |
| 244 | "PATHEXT": os.environ.get("PATHEXT", ".COM;.EXE;.BAT;.CMD"), |
| 245 | "PATH": os.environ.get("PATH", f"{sr}\\system32;{sr}"), |
| 246 | "APPDATA": os.environ.get("APPDATA", ""), |
| 247 | "LOCALAPPDATA": os.environ.get("LOCALAPPDATA", ""), |
| 248 | "ProgramData": os.environ.get("ProgramData", ""), |
| 249 | "ProgramFiles": os.environ.get("ProgramFiles", ""), |
| 250 | "ProgramFiles(x86)": os.environ.get("ProgramFiles(x86)", ""), |
| 251 | "ProgramW6432": os.environ.get("ProgramW6432", ""), |
| 252 | } |
| 253 | for key in self.allowed_env_keys: |
| 254 | val = os.environ.get(key) |
| 255 | if val is not None: |
| 256 | env[key] = val |
| 257 | return env |
| 258 | home = os.environ.get("HOME", "/tmp") |
| 259 | env = { |
| 260 | "HOME": home, |
| 261 | "LANG": os.environ.get("LANG", "C.UTF-8"), |
| 262 | "TERM": os.environ.get("TERM", "dumb"), |
| 263 | } |
| 264 | for key in self.allowed_env_keys: |
| 265 | val = os.environ.get(key) |
| 266 | if val is not None: |
| 267 | env[key] = val |
| 268 | return env |
| 269 | |
| 270 | def _guard_command(self, command: str, cwd: str) -> str | None: |
| 271 | """Best-effort safety guard for potentially destructive commands.""" |
| 272 | cmd = command.strip() |
| 273 | lower = cmd.lower() |
| 274 | |
| 275 | for pattern in self.deny_patterns: |
| 276 | if re.search(pattern, lower): |
| 277 | return "Error: Command blocked by safety guard (dangerous pattern detected)" |
| 278 | |
| 279 | if self.allow_patterns: |
| 280 | if not any(re.search(p, lower) for p in self.allow_patterns): |
| 281 | return "Error: Command blocked by safety guard (not in allowlist)" |
| 282 | |
| 283 | from nanobot.security.network import contains_internal_url |
| 284 | if contains_internal_url(cmd): |
| 285 | return "Error: Command blocked by safety guard (internal/private URL detected)" |
| 286 | |
| 287 | if self.restrict_to_workspace: |
| 288 | if "..\\" in cmd or "../" in cmd: |
| 289 | return "Error: Command blocked by safety guard (path traversal detected)" |
| 290 | |
| 291 | cwd_path = Path(cwd).resolve() |
| 292 | |
| 293 | for raw in self._extract_absolute_paths(cmd): |
| 294 | try: |
| 295 | expanded = os.path.expandvars(raw.strip()) |
| 296 | p = Path(expanded).expanduser().resolve() |
| 297 | except Exception: |
| 298 | continue |
| 299 | |
| 300 | media_path = get_media_dir().resolve() |
| 301 | if (p.is_absolute() |
| 302 | and cwd_path not in p.parents |
| 303 | and p != cwd_path |
| 304 | and media_path not in p.parents |
| 305 | and p != media_path |
| 306 | ): |
| 307 | return "Error: Command blocked by safety guard (path outside working dir)" |
| 308 | |
| 309 | return None |
| 310 | |
| 311 | @staticmethod |
| 312 | def _extract_absolute_paths(command: str) -> list[str]: |
| 313 | # Windows: match drive-root paths like `C:\` as well as `C:\path\to\file` |
| 314 | # NOTE: `*` is required so `C:\` (nothing after the slash) is still extracted. |
| 315 | win_paths = re.findall(r"[A-Za-z]:\\[^\s\"'|><;]*", command) |
| 316 | posix_paths = re.findall(r"(?:^|[\s|>'\"])(/[^\s\"'>;|<]+)", command) # POSIX: /absolute only |
| 317 | home_paths = re.findall(r"(?:^|[\s|>'\"])(~[^\s\"'>;|<]*)", command) # POSIX/Windows home shortcut: ~ |
| 318 | return win_paths + posix_paths + home_paths |
| 319 |