# -*- coding: utf-8 -*- """The bash tool in agentscope.""" import os from typing import AsyncGenerator, Any, List import re from ._bash_parser import BashCommandParser from .._base import ToolBase, ToolMiddlewareBase from .._constants import ( DEFAULT_DANGEROUS_FILES, DEFAULT_DANGEROUS_DIRECTORIES, ) from ...permission import ( PermissionContext, PermissionDecision, PermissionBehavior, PermissionMode, PermissionRule, ) from ...message import TextBlock, ToolResultState from .._response import ToolChunk from ._backend import BackendBase class Bash(ToolBase): """The bash tool.""" name: str = "Bash" """The tool name presented to the agent.""" description: str = """Executes a bash command and returns its output. The working directory persists between commands, but shell state does not. The shell environment is initialized from the user's profile (bash or zsh). IMPORTANT: Avoid using this tool to run `find`, `grep`, `cat`, `head`, `tail`, `sed`, `awk`, or `echo` commands, unless explicitly instructed or after you have verified that a dedicated tool cannot accomplish your task. Instead, use the appropriate dedicated tool as this will provide a much better experience for the user: - File search: Use Glob (NOT find or ls) - Content search: Use Grep (NOT grep or rg) - Read files: Use Read (NOT cat/head/tail) - Edit files: Use Edit (NOT sed/awk) - Write files: Use Write (NOT echo >/cat < None: """Initialize the bash tool. Args: dangerous_files (`list[str]`, optional): Sensitive files that require explicit user confirmation, even in BYPASS mode. Matched by basename (case-insensitive). Defaults to `DEFAULT_DANGEROUS_FILES`. Pass a custom list to fully replace the defaults, or `[]` to disable the filename check. dangerous_directories (`list[str]`, optional): Sensitive directories that require explicit user confirmation. Matched when any path segment equals an entry (case-insensitive). Defaults to `DEFAULT_DANGEROUS_DIRECTORIES`. Pass a custom list to fully replace the defaults, or `[]` to disable the directory check. cwd (`str | os.PathLike[str] | None`, optional): The working directory used when executing bash commands. middlewares (`List[ToolMiddlewareBase] | None`, optional): Tool middlewares wrapping the tool execution. backend (`BackendBase | None`, optional): The sandbox backend to use for shell execution. When ``None``, a :class:`LocalBackend` is created. """ from ._backend import LocalBackend super().__init__(middlewares=middlewares) self._bash_parser = BashCommandParser() self.dangerous_files = list(dangerous_files) self.dangerous_directories = list(dangerous_directories) self._cwd = os.fspath(cwd) if cwd is not None else None self._backend = backend or LocalBackend() async def check_read_only( self, tool_input: dict[str, Any], ) -> bool: """Decide whether this specific bash invocation is read-only. Inspects the command and returns ``True`` for known-safe read-only commands (e.g. ``ls``, ``cat``, ``grep``, ``git status``). The static :attr:`is_read_only` class attribute is ``False`` because Bash can execute arbitrary commands; this method overrides that with a per-invocation answer. """ command = tool_input.get("command", "") if not command: return self.is_read_only return self._bash_parser.is_read_only_command(command) async def check_permissions( self, tool_input: dict[str, Any], context: PermissionContext, ) -> PermissionDecision: """Check permissions for bash command execution. This method implements Bash-specific permission checks: 0. Injection risk check (bypass-immune safety ASK if command contains dynamic expansion like ``$(...)`` or ``<(...)``) 1. Read-only command check — auto-ALLOW in **every mode** (including DEFAULT) for known-safe read-only commands (``ls``, ``pwd``, ``git status``, ``cat``, etc.). This is the static counterpart to :meth:`check_read_only`. 2. Dangerous command pattern check (bypass-immune safety ASK) 3. Sed in-place constraint check (bypass-immune safety ASK) 4. Dangerous path check for config files (bypass-immune safety ASK) 5. Dangerous removal path check for system dirs (bypass-immune safety ASK) 6. ACCEPT_EDITS auto-allow for ``mkdir``/``touch``/``rm``/ ``rmdir``/``mv``/``cp``/``sed`` — only when **every** target path resolves inside a working directory 7. PASSTHROUGH (engine continues with rule matching) "Bypass-immune" decisions set :attr:`PermissionDecision.bypass_immune` so they cannot be silenced by allow rules in DEFAULT mode. In BYPASS mode all bypass-immune ASKs are intentionally skipped — see :attr:`PermissionMode.BYPASS`. Args: tool_input (`dict[str, Any]`): The tool input containing "command" key context (`PermissionContext`): The permission context with mode and rules Returns: `PermissionDecision`: ALLOW for safe operations, ASK for dangerous operations, PASSTHROUGH to let Engine continue with rule matching """ command = tool_input.get("command", "") if not command: return PermissionDecision( behavior=PermissionBehavior.PASSTHROUGH, message="Empty command", ) # 0. Injection check: detect dynamic shell structures that cannot be # statically analyzed (command substitution, process substitution, # control flow, etc.). Must run before read-only check so that # `$(rm -rf /)` inside an otherwise-safe command is caught. injection_reason = self._bash_parser.check_injection_risk(command) if injection_reason: return PermissionDecision( behavior=PermissionBehavior.ASK, message=f"Permission required: {injection_reason}", decision_reason="Safety check: command contains dynamic " "expansion that cannot be statically analyzed", bypass_immune=True, ) # 1. Check if command is read-only (auto-allow) if self._bash_parser.is_read_only_command(command): return PermissionDecision( behavior=PermissionBehavior.ALLOW, message="Permission granted for read-only command", decision_reason="Read-only command is allowed", ) # 2. Check for dangerous commands (safety check, bypass-immune) dangerous_pattern = self._bash_parser.check_dangerous_command(command) if dangerous_pattern: return PermissionDecision( behavior=PermissionBehavior.ASK, message=f"Permission required: Command contains dangerous " f"pattern: {dangerous_pattern}", decision_reason="Safety check: dangerous command pattern " "detected", bypass_immune=True, ) # 3. Check for sed constraints (safety check, bypass-immune) sed_error = self._bash_parser.check_sed_constraints( command, self.dangerous_files, ) if sed_error: return PermissionDecision( behavior=PermissionBehavior.ASK, message=f"Permission required: {sed_error}", decision_reason="Safety check: sed in-place modification " "of dangerous file", bypass_immune=True, ) # 4. Check for dangerous paths in sensitive config files/dirs # (safety check, bypass-immune) dangerous_paths = self._extract_dangerous_paths_from_bash(command) if dangerous_paths: paths_str = ", ".join(dangerous_paths) return PermissionDecision( behavior=PermissionBehavior.ASK, message=f"Permission required: Bash command operates on " f"sensitive paths: {paths_str}", decision_reason="Safety check: dangerous file or " "directory in bash command", bypass_immune=True, ) # 5. Check for dangerous removal paths: rm/rmdir targeting system # critical directories like /, /usr, /etc, ~ (bypass-immune). # Checked separately from step 4 because these paths are not in the # dangerous_files/directories lists — they are system-level paths # that should never be removed regardless of user configuration. removal_path = await self._check_dangerous_removal_path(command) if removal_path: return PermissionDecision( behavior=PermissionBehavior.ASK, message=f"Dangerous removal operation detected: " f"'{removal_path}'\n\nThis command would remove a critical " f"system directory. This requires explicit approval and " f"cannot be auto-allowed by permission rules.", decision_reason="Safety check: dangerous removal of " "critical system path", bypass_immune=True, ) # 6. ACCEPT_EDITS auto-allow for filesystem commands whose targets # all live inside a working directory. Mirrors Write/Edit's strict # working-directory check — we never auto-allow a bash command that # would touch a path outside the configured working set (e.g. # ``cp /etc/hosts /tmp/x`` must not pass even though ``cp`` is in # the auto-allow list). if context.mode == PermissionMode.ACCEPT_EDITS: filesystem_commands = { "mkdir", "touch", "rm", "rmdir", "mv", "cp", "sed", } base_command = ( command.strip().split()[0] if command.strip() else "" ) if base_command in filesystem_commands: # Collect every target path: file arguments AND output # redirections. ``extract_file_paths`` includes both. target_paths = [ path for _cmd, path in self._bash_parser.extract_file_paths( command, ) ] # Conservative: only auto-allow when we extracted at least # one target AND every target resolves inside a working # directory. An empty list means the parser found nothing # actionable (or the command has no args) — in that case # we fall through to PASSTHROUGH rather than blindly # allowing. if target_paths and all( self._path_in_allowed_working_path(path, context) for path in target_paths ): return PermissionDecision( behavior=PermissionBehavior.ALLOW, message=f"Permission granted for '{base_command}' " f"command (accept edits mode - filesystem command, " f"all targets in working directory)", decision_reason=( f"Filesystem command '{base_command}' is " f"auto-allowed in accept edits mode because " f"all target paths are within a working " f"directory" ), ) # 7. Passthrough to let Engine continue with rule matching return PermissionDecision( behavior=PermissionBehavior.PASSTHROUGH, message=f"Execute bash command: {command}", ) async def match_rule( self, rule_content: str | None, tool_input: dict[str, Any], ) -> bool: r"""Match Bash command using regex-based wildcard matching. Implements wildcard matching with escape sequences: - Supports \* for literal asterisk and \\ for literal backslash - Special optimization: "git *" matches both "git" and "git add" - Prefix pattern (e.g., "git:*"): matches commands starting with "git " - Wildcard pattern: converts to regex with proper escape handling - Substring pattern: exact substring matching - If rule_content is None, matches all invocations (tool-name-level rule) Args: rule_content: The command pattern to match, or None to match all tool_input: Must contain a "command" key with the command string Returns: True if pattern matches the command """ # None = tool-name-level rule, matches everything if rule_content is None: return True command = tool_input.get("command", "") # Check if pattern is a prefix pattern (ends with :*) if rule_content.endswith(":*"): prefix = rule_content[:-2].strip() return command.startswith(prefix + " ") or command == prefix # Check if pattern has unescaped wildcards def has_wildcards(pattern: str) -> bool: """Check if pattern contains unescaped * wildcards.""" i = 0 while i < len(pattern): if pattern[i] == "\\": i += 2 # Skip escaped character elif pattern[i] == "*": return True else: i += 1 return False if not has_wildcards(rule_content): # No wildcards, but may have escape sequences # Convert escape sequences for matching pattern = rule_content pattern = pattern.replace("\\\\", "\x00BACKSLASH\x00") pattern = pattern.replace("\\*", "*") pattern = pattern.replace("\x00BACKSLASH\x00", "\\") # Use substring matching with converted pattern return pattern in command # Convert wildcard pattern to regex with escape handling # Use placeholders for escaped sequences ESCAPED_STAR = "\x00ESCAPED_STAR\x00" ESCAPED_BACKSLASH = "\x00ESCAPED_BACKSLASH\x00" pattern = rule_content # Replace \\ with placeholder pattern = pattern.replace("\\\\", ESCAPED_BACKSLASH) # Replace \* with placeholder pattern = pattern.replace("\\*", ESCAPED_STAR) # Manually escape regex special characters (except *) # Don't use re.escape() as it escapes spaces too special_chars = r".^$+?{}[]|()" for char in special_chars: pattern = pattern.replace(char, "\\" + char) # Convert * to regex .* (match any characters) pattern = pattern.replace("*", ".*") # Restore escaped sequences pattern = pattern.replace(ESCAPED_STAR, r"\*") pattern = pattern.replace(ESCAPED_BACKSLASH, r"\\") # Special optimization: "git *" should match both "git" and "git add" # Pattern: if ends with .*, make it optional if pattern.endswith(".*"): base_pattern = pattern[:-2] # Remove .* # Try exact match first (handles trailing space) base_pattern = base_pattern.rstrip() if re.fullmatch(base_pattern, command): return True # Full regex match try: return bool(re.fullmatch(pattern, command)) except re.error: # Invalid regex, fall back to substring matching return rule_content.replace("*", "") in command async def generate_suggestions( self, tool_input: dict[str, Any], ) -> List["PermissionRule"]: """Generate suggested permission rules for Bash commands. Generates prefix rules based on command + subcommand (two words). For example, "git commit -m 'xxx'" generates "git commit:*". Args: tool_input (`dict[str, Any]`): The tool input data containing "command" key Returns: `List[PermissionRule]`: List of suggested permission rules based on command prefixes """ command = tool_input.get("command", "") if not command: return [] # Use bash parser to extract command prefixes prefixes = self._bash_parser.extract_command_prefixes( command, max_prefixes=5, ) if not prefixes: # Cannot extract any prefix, return empty return [] # Generate rules for each prefix rules = [] for prefix in prefixes: rules.append( PermissionRule( tool_name="Bash", rule_content=f"{prefix}:*", behavior=PermissionBehavior.ALLOW, source="suggested", ), ) return rules def _extract_dangerous_paths_from_bash( self, command: str, ) -> list[str]: """Extract dangerous paths from a bash command using tree-sitter. Checks for dangerous paths in: - File-manipulating commands (rm, mv, cp, chmod, chown, sed, touch) - Output redirections (>, >>) Args: command (`str`): The bash command string Returns: `list[str]`: List of dangerous paths found in the command """ dangerous_paths = [] # Use tree-sitter to extract file paths file_paths = self._bash_parser.extract_file_paths(command) for _cmd_name, path in file_paths: if self._is_dangerous_path(path): dangerous_paths.append(path) return dangerous_paths async def _check_dangerous_removal_path(self, command: str) -> str | None: """Check if a rm/rmdir command targets a critical system path. Detects commands like `rm -rf /`, `rm -rf /usr`, `rmdir ~` that would destroy critical system directories. Unlike _is_dangerous_path (which checks against a configurable list of sensitive config files), this checks against a fixed set of system-level paths that must never be removed regardless of user configuration. Dangerous paths are: - Root directory (/) - Home directory (~) - Wildcard alone (*) or as dir/* (removes everything) - Direct children of root (/usr, /etc, /tmp, /var, etc.) Args: command (`str`): The bash command string Returns: `str | None`: The dangerous path if found, None otherwise """ tokens = command.strip().split() if not tokens: return None # Find rm or rmdir subcommands (handle compound commands) try: tree = self._bash_parser.parser.parse(bytes(command, "utf8")) subcommands = self._bash_parser.split_compound_command( tree.root_node, command, ) except Exception: subcommands = [command] # Check each subcommand for rm/rmdir for subcmd in subcommands: subcmd_tokens = subcmd.strip().split() if not subcmd_tokens: continue base = subcmd_tokens[0] if base not in ("rm", "rmdir"): continue # Collect non-flag arguments as potential paths i = 1 while i < len(subcmd_tokens): tok = subcmd_tokens[i] # Skip flags if tok.startswith("-"): i += 1 continue path = tok.strip("'\"") if await self._is_dangerous_removal_path(path): return path i += 1 return None async def _is_dangerous_removal_path(self, path: str) -> bool: """Check if a path is a critical system directory that must not be removed. All path resolution is performed via the backend so that the check operates on the **backend environment's** ``$HOME`` / ``cwd`` / path semantics, not the host process's. Args: path (`str`): The path to check (may be relative, absolute, or contain ~) Returns: `bool`: True if removing this path would be catastrophic """ # Bare wildcard if path in ("*", "./*", "/"): return True # Ends with /* — removes everything in a directory if path.endswith("/*") or path.endswith("\\*"): return True # Expand tilde and resolve to an absolute path inside the # backend environment. Don't resolve symlinks — ``/tmp`` is a # symlink on macOS but is still a root-child and should be # flagged. expanded = await self._backend.expanduser(path) backend_cwd = await self._backend.getcwd() abs_path = self._backend.abspath(expanded, cwd=backend_cwd) # Home directory home = await self._backend.expanduser("~") if abs_path == home: return True # Root itself: ``dirname(root) == root`` on both POSIX # (``"/"``) and Windows (``"C:\\"``), so this check is # path-flavor agnostic. parent = self._backend.dirname(abs_path) if abs_path == parent: return True # Direct children of root (e.g. ``/usr``, ``/etc``, ``/tmp``): # the *parent* of these is the root, where # ``dirname(parent) == parent``. if self._backend.dirname(parent) == parent: return True return False async def call( # type: ignore[override] # pylint: disable=unused-argument self, command: str, description: str = "", timeout: int = 120000, ) -> AsyncGenerator[ToolChunk, None]: """Execute the bash and return the output. Args: command: The bash command to execute. description: Optional description of what the command does. timeout: Timeout in milliseconds (default: 120000, max: 600000). Yields: ToolChunk: The tool execution result with stdout/stderr content. """ # Clamp timeout to max 600000ms and convert to seconds timeout_ms = min(timeout, 600000) timeout_sec = timeout_ms / 1000.0 try: # ``command`` is a full shell command line (it may contain # pipes, redirects, ``&&``, …), so wrap it in a shell — the # backend primitive runs the argv directly without one. Pick # the platform's native shell so the Windows experience that # ``main`` had (commands interpreted by ``cmd.exe``) is # preserved; POSIX hosts use ``/bin/sh``. if os.name == "nt": shell_command = ["cmd", "/c", command] else: shell_command = ["/bin/sh", "-c", command] result = await self._backend.exec_shell( shell_command, cwd=self._cwd, timeout=timeout_sec, ) # Decode and normalize line endings stdout = result.stdout.decode( "utf-8", errors="replace", ).replace("\r\n", "\n") stderr = result.stderr.decode( "utf-8", errors="replace", ).replace("\r\n", "\n") # Check for timeout (backend returns exit_code=-1, # stderr=b"timed out") if result.exit_code == -1 and result.stderr == b"timed out": error_msg = ( f"Command timed out after {timeout_ms}ms: {command}" ) yield ToolChunk( content=[TextBlock(text=error_msg)], state=ToolResultState.ERROR, is_last=True, ) return # Combine output output = stdout if stderr: if output: output += "\n" output += stderr # Truncate if exceeds 30000 characters if len(output) > 30000: output = output[:30000] + "\n... (output truncated)" # Check exit code if not result.ok(): # Command failed error_result = f"Command failed: {command}\n" if stdout: error_result += f"\nStdout:\n{stdout}" if stderr: error_result += f"\nStderr:\n{stderr}" # Truncate error message if needed if len(error_result) > 30000: error_result = ( error_result[:30000] + "\n... (output truncated)" ) yield ToolChunk( content=[TextBlock(text=error_result)], state=ToolResultState.ERROR, is_last=True, ) else: # Command succeeded - note: ToolChunk uses "running" state # which will be converted to "finished" in ToolResponse yield ToolChunk( content=[TextBlock(text=output)], state=ToolResultState.RUNNING, is_last=True, ) except Exception as e: # Other errors error_msg = f"Command failed: {command}\nError: {str(e)}" yield ToolChunk( content=[TextBlock(text=error_msg)], state=ToolResultState.ERROR, is_last=True, )