"""TerminalTool - 命令行工具 为Agent提供安全的命令行执行能力，支持： - 文件系统操作（ls, cat, head, tail, find, grep） - 文本处理（wc, sort, uniq） - 目录导航（pwd, cd） - 安全限制（白名单命令、路径限制、超时控制） 使用场景： - JIT（即时）文件检索与分析 - 代码仓库探索 - 日志文件分析 - 数据文件预览 安全特性： - 命令白名单（只允许安全的只读命令） - 工作目录限制（沙箱） - 超时控制 - 输出大小限制 - 禁止危险操作（rm, mv, chmod等） """ from typing import Dict, Any, List, Optional import subprocess import os from pathlib import Path import shlex import re from ..base import Tool, ToolParameter class TerminalTool(Tool): """命令行工具 提供安全的命令行执行能力，支持常用的文件系统和文本处理命令。 安全限制： - 只允许白名单中的命令 - 限制在指定工作目录内 - 超时控制（默认30秒） - 输出大小限制（默认10MB） 用法示例： ```python terminal = TerminalTool(workspace="./project") # 列出文件 result = terminal.run({"command": "ls -la"}) # 查看文件内容 result = terminal.run({"command": "cat README.md"}) # 搜索文件 result = terminal.run({"command": "grep -r 'TODO' src/"}) # 查看文件前10行 result = terminal.run({"command": "head -n 10 data.csv"}) ``` """ # 允许的命令白名单 # 这些命令被认为是安全的，主要用于文件查看、文本处理和信息获取 # 不包含可能修改系统或造成安全风险的命令（如rm、mv、chmod等） ALLOWED_COMMANDS = { # 文件列表与信息 'ls', 'dir', 'tree', # 文件内容查看 'cat', 'head', 'tail', 'less', 'more', # 文件搜索 'find', 'grep', 'egrep', 'fgrep', 'rg', # 文本处理 'wc', 'sort', 'uniq', 'cut', 'awk', 'sed', # shell 常见内建（用于管道/小脚本；仍受整体策略约束） 'echo', 'printf', # 目录/文件创建（受路径沙箱约束） 'mkdir', # 目录操作 'pwd', 'cd', # 文件信息 'file', 'stat', 'du', 'df', # 其他 'which', 'whereis', # 版本控制（只读子命令会被进一步限制） 'git', } # 常见 shell 元字符（用于检测"组合命令/写盘/子命令"等风险点；不再一刀切禁止） # 这些元字符在shell中有特殊含义，可能用于组合命令或执行危险操作 # 系统会检测这些字符的存在，并根据安全策略决定是否允许执行 SHELL_META_TOKENS = ["|", "||", "&&", ";", ">", ">>", "<", "$(", "`"] # 需要人类确认的高风险命令（MVP：只做识别；是否放行由上层策略决定） # 这些命令可能对系统造成不可逆的修改，需要用户明确确认才能执行 DANGEROUS_BASE_COMMANDS = {"rm", "chmod"} # Git的高风险子命令，可能造成代码丢失或历史修改 DANGEROUS_GIT_SUBCOMMANDS = {("reset", "--hard"), ("reset", "--hard", "HEAD")} def __init__( self, workspace: str = ".", timeout: int = 30, max_output_size: int = 10 * 1024 * 1024, # 10MB allow_cd: bool = True, confirm_dangerous: bool = False, default_shell_mode: bool = False, ): """初始化TerminalTool实例 Args: workspace: 工作目录路径，所有命令将在此目录或其子目录中执行 timeout: 命令执行超时时间（秒），防止长时间运行的命令 max_output_size: 输出大小限制（字节），防止过大输出消耗资源 allow_cd: 是否允许cd命令，控制目录切换权限 confirm_dangerous: 是否在执行高风险命令时提示用户确认 default_shell_mode: 默认是否启用shell模式（支持管道、重定向等） """ super().__init__( name="terminal", description="命令行工具 - 执行安全的文件系统、文本处理和代码执行命令（ls, cat, grep, head, tail等）" ) # 将工作目录转换为绝对路径并规范化 self.workspace = Path(workspace).resolve() self.timeout = timeout self.max_output_size = max_output_size self.allow_cd = allow_cd self.confirm_dangerous = confirm_dangerous self.default_shell_mode = default_shell_mode # 当前工作目录（相对于workspace） # 初始设置为工作目录根目录，可通过cd命令更改 self.current_dir = self.workspace # 确保工作目录存在，如果不存在则创建 self.workspace.mkdir(parents=True, exist_ok=True) def run(self, parameters: Dict[str, Any]) -> str: """执行工具的主入口方法 根据参数解析命令并执行，包含完整的安全检查流程： 1. 参数验证 - 确保输入参数格式正确且必要参数存在 2. 命令解析和分类 - 将命令字符串解析为可执行的参数列表 3. 安全策略检查 - 多层安全验证，包括白名单检查、危险操作检测等 4. 命令执行和结果处理 - 安全执行命令并格式化返回结果 安全机制说明： - 命令白名单：只执行预定义的安全命令，防止恶意命令执行 - 路径沙箱：所有文件操作限制在工作目录内，防止越权访问 - 危险操作确认：高风险命令需要用户明确确认才能执行 - 超时控制：防止长时间运行的命令消耗系统资源 - 输出限制：防止过大输出导致内存问题 Args: parameters: 包含command、allow_dangerous、shell_mode等参数的字典 - command: 要执行的命令字符串 - allow_dangerous: 是否允许执行危险操作 - shell_mode: 是否启用shell模式（支持管道、重定向等） Returns: str: 命令执行结果或错误信息，包含详细的错误说明 """ # 第一步：参数验证 - 确保输入参数符合预期格式 if not self.validate_parameters(parameters): return "❌ 参数验证失败" # 提取并清理命令参数 command = parameters.get("command", "").strip() allow_dangerous = bool(parameters.get("allow_dangerous", False)) shell_mode = bool(parameters.get("shell_mode", self.default_shell_mode)) # 基础安全检查：拒绝空命令，防止无意义的系统调用 if not command: return "❌ 命令不能为空" # 执行模式选择：根据shell_mode参数决定执行方式 # shell_mode=True: 支持管道、重定向等复杂shell语法，但需要更严格的安全检查 # shell_mode=False: 使用argv模式，更安全但不支持shell特性 if shell_mode: return self._execute_shell(command, allow_dangerous=allow_dangerous) # 第二步：命令解析 - 使用shlex进行安全的命令分割，处理引号和转义 try: parts = shlex.split(command) except ValueError as e: return f"❌ 命令解析失败: {e}" # 解析后再次验证，确保命令不为空 if not parts: return "❌ 命令不能为空" base_command = parts[0] # 第三步：安全策略检查 - 命令白名单验证 # 这是第一道安全防线，确保只能执行预定义的安全命令 if base_command not in self.ALLOWED_COMMANDS: return f"❌ 不允许的命令: {base_command}\n允许的命令: {', '.join(sorted(self.ALLOWED_COMMANDS))}" # 特殊命令处理：git命令需要额外的子命令安全检查 if base_command == "git": return self._handle_git(parts, allow_dangerous) # 第四步：危险操作确认机制 # 当用户明确允许危险操作且启用了确认机制时，进行交互式确认 # 这为高风险操作提供了最后一道人工确认防线 if allow_dangerous and self.confirm_dangerous: ans = input(f"\n⚠️ 高风险命令：{command}\n允许执行？(y/n)\nconfirm> ").strip().lower() if ans not in {"y", "yes"}: return "⛔️ 已取消执行（用户未确认）。" # 特殊命令处理：cd命令需要单独处理以维护工作目录状态 if base_command == 'cd': return self._handle_cd(parts) # 第五步：执行命令 - 通过所有安全检查后执行命令 return self._execute_argv(parts, allow_dangerous=allow_dangerous) def get_parameters(self) -> List[ToolParameter]: """获取工具参数定义""" return [ ToolParameter( name="command", type="string", description=( f"要执行的命令（白名单: {', '.join(sorted(list(self.ALLOWED_COMMANDS)[:10]))}...）\n" "示例: 'ls -la', 'cat file.txt', 'grep pattern *.py', 'head -n 20 data.csv'" ), required=True ), ToolParameter( name="allow_dangerous", type="boolean", description="是否允许高风险命令（默认false；仅在用户明确确认后才可设置为true）", required=False ), ToolParameter( name="shell_mode", type="boolean", description="是否允许 shell 语义（管道/重定向/多段命令等）。默认继承工具配置。", required=False, ), ] def _contains_shell_meta(self, command: str) -> bool: """检查命令中是否包含shell元字符 Args: command: 要检查的命令字符串 Returns: bool: 如果包含元字符返回True，否则返回False """ return any(tok in command for tok in self.SHELL_META_TOKENS) # --- shell parsing helpers (ignore operators inside quotes) --- def _split_shell_segments(self, command: str) -> List[str]: """ 按管道/逻辑与/逻辑或/分号操作符分割shell命令，忽略引号内的操作符。 返回分割后的段列表（不包含操作符）。 这个方法用于分析复杂的shell命令，确保每个段都是安全的。 Args: command: 要分割的shell命令字符串 Returns: List[str]: 分割后的命令段列表 """ ops = ["||", "&&", "|", ";"] segs: List[str] = [] buf: List[str] = [] i = 0 quote: Optional[str] = None while i < len(command): ch = command[i] if ch in {"'", '"'}: if quote is None: quote = ch elif quote == ch: quote = None buf.append(ch) i += 1 continue if ch == "\\": buf.append(ch) if i + 1 < len(command): buf.append(command[i + 1]) i += 2 else: i += 1 continue if quote is None: matched = False for op in ops: if command.startswith(op, i): seg = "".join(buf).strip() if seg: segs.append(seg) buf = [] i += len(op) matched = True break if matched: continue buf.append(ch) i += 1 seg = "".join(buf).strip() if seg: segs.append(seg) return segs def _has_unquoted(self, command: str, token: str) -> bool: """检查token（如>或$()或|）是否出现在引号外 这个方法用于检测可能存在安全风险的shell操作符， 确保它们不是在引号内（引号内是安全的字符串字面量）。 Args: command: 要检查的命令字符串 token: 要查找的token字符串 Returns: bool: 如果token出现在引号外返回True，否则返回False """ q: Optional[str] = None i = 0 while i < len(command): ch = command[i] if ch in {"'", '"'}: if q is None: q = ch elif q == ch: q = None i += 1 continue if ch == "\\": i += 2 continue if q is None and command.startswith(token, i): return True i += 1 return False def _shell_requires_allow_dangerous(self, command: str) -> bool: """检查shell命令是否需要危险操作权限 此方法分析shell命令，判断其是否包含可能对系统造成风险的操作。 这是安全机制的重要组成部分，用于识别需要额外权限确认的命令。 安全检查逻辑： 1. 检查文件写入操作（>、>>重定向），但排除到/dev/null的重定向 2. 检查命令替换（`command`或$(command)） 3. 检查已知的危险基础命令（rm、chmod等） 4. 检查Git的危险子命令（如reset --hard） Args: command: 要检查的shell命令字符串 Returns: bool: 如果命令需要危险操作权限则返回True，否则返回False Note: - 此方法不会阻止命令执行，只是标识需要额外确认的命令 - 实际的权限检查在执行阶段进行 - 采用相对保守的策略，对/dev/null重定向等安全操作给予宽容 """ # 写盘/命令替换通常视为高风险，但常见的只读掩埋（如 2>/dev/null、|| echo）可放宽 # 宽容规则：仅当重定向目标不是 /dev/null 时才视为写盘；简单的 "|| echo ..." 视为安全。 if self._has_unquoted(command, ">") or self._has_unquoted(command, ">>"): # 忽略 /dev/null 重定向（这是安全的丢弃输出操作） if re.search(r">\s*/dev/null", command) or re.search(r">>\s*/dev/null", command): pass else: # 其他重定向操作可能修改文件内容，需要危险权限 return True # 检查命令替换：`command`（反引号）和 $(command) 格式 # 命令替换可能执行任意代码，存在代码注入风险 if self._has_unquoted(command, "$(") or self._has_unquoted(command, "`"): return True # 检查已知的危险基础命令 # 这些命令可能对系统造成不可逆的影响，需要特别关注 if re.search(r"(^|\s)rm(\s|$)", command): return True if re.search(r"(^|\s)chmod(\s|$)", command): return True # 检查Git的危险子命令 # git reset --hard 可能导致代码丢失，属于高风险操作 if re.search(r"git\s+reset\s+--hard", command): return True # 如果没有检测到危险操作，则不需要额外权限 return False def _shell_all_commands_whitelisted(self, command: str) -> bool: """ 静态检查shell命令中的所有段是否都在白名单中（尽力而为的检查） 此方法通过分割shell命令为多个段，然后检查每个段的第一个命令是否在白名单中。 这是对shell命令的安全预检查，用于在未允许危险操作时确保命令的安全性。 安全检查策略： 1. 使用shell元字符分割命令为多个独立段 2. 对每个段进行命令解析，提取基础命令 3. 检查基础命令是否在白名单中 4. 对git命令进行特殊处理，只允许安全的只读子命令 Args: command: 要检查的完整shell命令字符串 Returns: bool: 如果所有命令段都在白名单中则返回True，否则返回False Note: - 这是一个尽力而为的检查，不能保证100%准确 - 对于复杂的shell语法，可能存在误判 - git命令只允许status和diff子命令，其他子命令被视为危险操作 """ # 预处理：将换行符替换为空格，便于统一处理 cmd = command.replace("\n", " ") # 分割命令为多个段，每个段包含一个独立的命令 segments = self._split_shell_segments(cmd) # 对每个命令段进行安全检查 for seg in segments: seg = seg.strip() if not seg: continue # 跳过空段 try: # 使用shlex进行安全的命令分割，处理引号和转义 argv = shlex.split(seg) except Exception: # 如果解析失败，认为不安全 return False if not argv: continue # 跳过空命令段 # 获取基础命令（命令名的第一部分） base = argv[0] # 检查基础命令是否在白名单中 if base not in self.ALLOWED_COMMANDS: return False # 对git命令进行特殊处理 # 只允许安全的只读子命令，其他子命令被视为危险操作 if base == "git": if len(argv) < 2: return False # git命令必须包含子命令 if argv[1] not in {"status", "diff"}: return False # 只允许status和diff子命令 # 所有命令段都通过了安全检查 return True def _execute_shell(self, command: str, allow_dangerous: bool = False) -> str: """ 执行shell命令字符串（支持Claude Code风格的shell特性） 此方法提供安全的shell命令执行能力，支持管道、重定向、命令替换等复杂shell语法。 通过多层安全检查机制确保命令执行的安全性。 安全防护措施： - 如果shell命令包含重定向/命令替换/已知危险操作 -> 需要allow_dangerous权限 - 如果未允许危险操作 -> 要求所有命令段都在白名单中（尽力而为的检查） - confirm_dangerous可以提示用户确认包含shell元字符或需要allow_dangerous的命令 执行流程： 1. 检查命令是否需要危险权限 2. 验证命令白名单（如果未允许危险操作） 3. 用户确认（如果启用confirm_dangerous） 4. 执行命令并处理输出 5. 返回执行结果或错误信息 Args: command: 要执行的完整shell命令字符串 allow_dangerous: 是否允许执行危险操作（默认False） Returns: str: 命令执行结果或错误信息 Note: - 支持管道操作而无需确认（如 'ls | grep .py'） - 只有在可能写入文件/转义/执行危险操作时才需要确认 - 输出会被截断以防止内存问题 """ needs_allow = self._shell_requires_allow_dangerous(command) # 第一层安全检查：危险操作检测 # 如果命令包含危险操作但未允许危险操作，则拒绝执行 # 这是第一道安全防线，防止潜在的恶意操作 if needs_allow and not allow_dangerous: return "❌ 该命令包含写盘/子命令替换/高风险操作，需用户确认后再执行（allow_dangerous=true）" # 第二层安全检查：白名单验证 # 如果未允许危险操作，则检查所有命令段是否都在白名单中 # 这是对命令的进一步安全验证，确保只能执行预定义的安全命令 if not allow_dangerous and not self._shell_all_commands_whitelisted(command): return "❌ shell_mode 下检测到非白名单命令/不允许的 git 子命令。需要用户确认后再执行（allow_dangerous=true）" # Claude Code-like: pipes are allowed without confirmation; only confirm when it may write/escape/execute dangerous ops. if self.confirm_dangerous and (allow_dangerous or needs_allow): ans = input(f"\n⚠️ 即将执行高风险 shell 命令：{command}\n允许执行？(y/n)\nconfirm> ").strip().lower() if ans not in {"y", "yes"}: return "⛔️ 已取消执行（用户未确认）。" try: result = subprocess.run( command, shell=True, cwd=str(self.current_dir), capture_output=True, text=True, timeout=self.timeout, env=os.environ.copy(), ) output = (result.stdout or "") + (result.stderr or "") if len(output.encode("utf-8", errors="ignore")) > self.max_output_size: output = output[: self.max_output_size] + "\n...output truncated...\n" if result.returncode != 0: return f"命令执行失败 (返回码 {result.returncode}):\n{output}" return output.strip() if output.strip() else "(no output)" except subprocess.TimeoutExpired: return f"❌ 命令超时（>{self.timeout}s）" except Exception as e: return f"❌ 命令执行异常: {e}" def _handle_cd(self, parts: List[str]) -> str: """处理cd命令，实现安全的目录切换 此方法专门处理cd命令，确保目录切换在允许的工作空间范围内进行。 支持绝对路径、相对路径和特殊路径（如..、~）的处理。 Args: parts: cd命令的参数列表，parts[1]是目标路径 Returns: str: 执行结果或错误信息 Note: - 只允许切换到工作空间内的目录 - 支持路径规范化，处理..和.等相对路径 - 不允许切换到工作空间之外的目录 """ if not self.allow_cd: return "❌ cd 命令已禁用" if len(parts) < 2: # cd命令没有参数时，返回当前目录信息 return f"当前目录: {self.current_dir}" target_dir = parts[1] # 处理特殊的相对路径 if target_dir == "..": # 切换到父目录 new_dir = self.current_dir.parent elif target_dir == ".": # 当前目录 new_dir = self.current_dir elif target_dir == "~": # 切换到工作目录根目录 new_dir = self.workspace else: # 解析相对或绝对路径 new_dir = (self.current_dir / target_dir).resolve() # 检查新目录是否在工作空间范围内 try: new_dir.relative_to(self.workspace) except ValueError: return f"❌ 不允许访问工作目录外的路径: {new_dir}" # 验证目录存在性 if not new_dir.exists(): return f"❌ 目录不存在: {new_dir}" # 确保目标是目录而非文件 if not new_dir.is_dir(): return f"❌ 不是目录: {new_dir}" # 更新当前工作目录 self.current_dir = new_dir return f"✅ 切换到目录: {self.current_dir}" def _execute_argv(self, argv: List[str], allow_dangerous: bool = False) -> str: """执行参数向量形式的命令（不使用shell解释） 此方法直接执行命令及其参数，不通过shell解释，因此不支持管道、重定向等shell特性。 这种方式更安全，因为避免了shell注入攻击的风险，但功能相对有限。 安全检查流程： 1. 检查高风险命令（rm、chmod等）是否需要用户确认 2. 对允许的高风险命令进行路径沙箱限制 3. 对mkdir命令进行路径沙箱限制 4. 执行命令并处理结果 Args: argv: 命令及其参数的列表，argv[0]是命令名，其余是参数 allow_dangerous: 是否允许执行高风险命令（默认False） Returns: str: 命令执行结果或错误信息 Note: - 不支持shell特性如管道、重定向、变量替换等 - 更安全，避免了shell注入攻击 - 适用于简单的命令执行场景 - 所有路径操作都被限制在工作空间内 """ # 第一层安全检查：高风险命令二次门禁 # 对明确高风险基命令（rm/chmod等）进行额外检查 # 这些命令可能对系统造成不可逆的修改，需要用户明确确认 # 这是安全防护的第一道防线，防止意外的系统修改 if argv and argv[0] in self.DANGEROUS_BASE_COMMANDS and not allow_dangerous: return f"❌ 高风险命令 {argv[0]} 需要人类确认（allow_dangerous=true）" # 第二层安全检查：路径沙箱限制 # 对带路径参数的高风险命令做路径沙箱限制（仅当放行时） # 确保即使允许执行危险命令，也只能在工作空间内操作 # 这防止了恶意用户通过相对路径或符号链接逃逸沙箱 if argv and argv[0] in {"rm", "chmod"} and allow_dangerous: # 保守检查：所有看起来像路径的非标志参数都必须在工作空间内 # 使用current_dir作为基准，确保相对路径也被正确限制 for a in argv[1:]: if a.startswith("-"): continue # 跳过选项参数（如 -r, -f 等） candidate = (self.current_dir / a).resolve() try: # 检查解析后的绝对路径是否在工作空间内 candidate.relative_to(self.workspace) except ValueError: # 路径超出工作空间范围，拒绝执行 return f"❌ 拒绝在工作目录外操作: {a}" # 第三层安全检查：mkdir命令路径沙箱 # mkdir虽然相对安全，但仍需确保不会在工作空间外创建目录 # 这防止了通过mkdir命令在工作空间外创建后门或敏感目录 if argv and argv[0] == "mkdir": for a in argv[1:]: if a.startswith("-"): continue # 跳过选项参数（如 -p, -m 等） candidate = (self.current_dir / a).resolve() try: # 检查要创建的目录是否在工作空间内 candidate.relative_to(self.workspace) except ValueError: # 尝试在工作空间外创建目录，拒绝执行 return f"❌ 不允许在工作目录外创建目录: {a}" try: # 直接执行命令，不通过shell # 使用shell=False确保命令不被shell解释，避免注入攻击 # 使用capture_output=True捕获标准输出和标准错误 # 使用text=True确保输出为文本格式 result = subprocess.run( argv, shell=False, cwd=str(self.current_dir), capture_output=True, text=True, timeout=self.timeout, env=os.environ.copy(), ) # 合并标准输出和标准错误 # 这样用户可以看到完整的执行结果 output = result.stdout if result.stderr: output += f"\n[stderr]\n{result.stderr}" # 检查输出大小，防止过大输出消耗内存 if len(output) > self.max_output_size: output = output[:self.max_output_size] output += f"\n\n⚠️ 输出被截断（超过 {self.max_output_size} 字节）" # 添加返回码信息，帮助用户了解命令执行状态 if result.returncode != 0: output = f"⚠️ 命令返回码: {result.returncode}\n\n{output}" return output if output else "✅ 命令执行成功（无输出）" except subprocess.TimeoutExpired: # 命令执行超时，可能是死循环或处理大量数据 return f"❌ 命令执行超时（超过 {self.timeout} 秒）" except Exception as e: # 捕获其他异常，如权限错误、文件不存在等 return f"❌ 命令执行失败: {e}" def _truncate_output(self, output: str) -> str: """截断过大的输出，防止内存问题 此方法检查输出字符串的长度，如果超过配置的最大输出大小， 则截断输出并添加提示信息，防止过大的输出消耗过多内存。 Args: output: 需要检查的输出字符串 Returns: str: 原始输出或截断后的输出（如果超过大小限制） Note: - 截断时会保留开头的内容，丢弃超出限制的部分 - 会在截断的输出末尾添加提示信息，说明输出已被截断 """ if len(output) > self.max_output_size: return output[: self.max_output_size] + f"\n[输出被截断，超过 {self.max_output_size} 字节限制]" return output def get_current_dir(self) -> str: """获取当前工作目录""" return str(self.current_dir) def reset_dir(self): """重置到工作目录根""" self.current_dir = self.workspace