From f9089af8f9a5fc1e75578393cdc9b8fd5791bb60 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 13 Apr 2026 16:23:24 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=9D=83=E9=99=90=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E7=AE=A1=E9=81=93=E5=91=BD=E4=BB=A4=E5=B9=B6?= =?UTF-8?q?=E4=BC=98=E5=8C=96prompt=E6=B3=A8=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/main_terminal_parts/context.py | 61 ++++++++++ core/main_terminal_parts/tools_execution.py | 120 ++++++++++++++++++- prompts/permission_mode.txt | 10 +- utils/context_manager.py | 125 -------------------- 4 files changed, 183 insertions(+), 133 deletions(-) diff --git a/core/main_terminal_parts/context.py b/core/main_terminal_parts/context.py index 7020e71..fa496e7 100644 --- a/core/main_terminal_parts/context.py +++ b/core/main_terminal_parts/context.py @@ -87,6 +87,62 @@ logger = setup_logger(__name__) DISABLE_LENGTH_CHECK = True class MainTerminalContextMixin: + # 权限模式说明（代码中存储的简要版本） + _PERMISSION_MODE_DESC = { + "unrestricted": "当前处于无限制模式，所有工具均可直接使用，无需额外批准。", + "readonly": "当前处于只读模式，仅能执行不会修改工作区的读取类操作。", + "approval": "当前处于批准模式，修改工作区的操作需要用户批准后方可执行。", + } + + _PERMISSION_MODE_LABEL = { + "unrestricted": "无限制", + "readonly": "只读", + "approval": "批准", + } + + def _build_permission_mode_message(self) -> Optional[str]: + """根据当前权限模式构建权限说明消息（从模板读取并替换占位符）""" + template = self.load_prompt("permission_mode") + if not template: + return None + + mode = self.get_permission_mode() + + # 根据模式生成详细规则 + if mode == "unrestricted": + detailed_rules = ( + "- 所有命令和工具均可直接使用，无需用户批准\n" + "- run_command：支持任意命令，包括管道、重定向、子shell等\n" + "- 文件操作：write_file、edit_file、create_file、delete_file、rename_file、create_folder 均可直接使用" + ) + elif mode == "readonly": + detailed_rules = ( + "- run_command：仅允许纯只读命令（ls, cat, grep, find, head, tail, wc, stat, file, sort, uniq 等）\n" + "- 允许只读管道：如 `find . -name '*.py' | head -20`、`cat file | sort | uniq`\n" + "- 禁止：子shell（$() ``）、逻辑运算符（&& || ;）、重定向（> <）\n" + "- 禁止危险管道：管道指向 shell（bash/sh）、tee、xargs、chmod 等写入或执行操作\n" + "- 文件操作：仅允许 read_file、view_image、view_video 等读取类工具\n" + "- 禁止：write_file、edit_file、create_file、delete_file、rename_file、create_folder 等修改类工具" + ) + elif mode == "approval": + detailed_rules = ( + "- 免批准直通：纯只读命令（ls, cat, grep, find 等）及只读管道可直接执行\n" + "- 免批准直通：如 `find . -name '*.py' | head -20`、`grep 'error' log | wc -l`\n" + "- 需要用户批准：非只读命令（带重定向、子shell、逻辑运算符）\n" + "- 需要用户批准：危险管道（指向 bash、tee、xargs、chmod 等）\n" + "- 需要用户批准：terminal_input、run_python、write_file、edit_file、create_file、delete_file、rename_file、create_folder、save_webpage\n" + "- 系统会在执行前弹出批准请求，请等待用户确认后再继续" + ) + else: + detailed_rules = "" + + return template.format( + permission_mode=mode, + permission_mode_label=self._PERMISSION_MODE_LABEL.get(mode, mode), + mode_description=self._PERMISSION_MODE_DESC.get(mode, ""), + detailed_rules=detailed_rules, + ) + def build_context(self) -> Dict: """构建主终端上下文""" # 读取记忆 @@ -167,6 +223,11 @@ class MainTerminalContextMixin: if workspace_system: messages.append({"role": "system", "content": workspace_system}) + # 注入权限模式说明（根据当前模式动态生成） + permission_mode_message = self._build_permission_mode_message() + if permission_mode_message: + messages.append({"role": "system", "content": permission_mode_message}) + if self.tool_category_states.get("sub_agent", True): sub_agent_prompt = self.load_prompt("sub_agent_guidelines").strip() if sub_agent_prompt: diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index 5ca4376..984bad9 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -152,23 +152,89 @@ class MainTerminalToolsExecutionMixin: self.context_manager.add_conversation("system", message, metadata=metadata) print(f"{OUTPUT_FORMATS['info']} {message}") + # 扩展的只读命令白名单（包含常用管道命令） + _READONLY_ALLOWED_EXECUTABLES = { + "grep", "find", "ls", "pwd", "tree", "cat", "head", "tail", + "less", "rg", "wc", "du", "stat", "file", "sed", "awk", "git", + "sort", "uniq", "cut", "tr", "echo", "printf", "date", + "basename", "dirname", "which", "strings", "ps", "who", "id", + } + + # 管道中绝对禁止的目标命令（宁可错杀） + _DANGEROUS_PIPE_TARGETS = { + "sh", "bash", "zsh", "fish", "dash", "ksh", "csh", "tcsh", + "rm", "mv", "cp", "dd", "rmdir", "unlink", "mkdir", "touch", + "chmod", "chown", "chgrp", "chattr", "setfacl", + "tee", "sponge", "xargs", "parallel", + "exec", "source", ".", "eval", + } + def _is_readonly_run_command_allowed(self, command: Any) -> bool: cmd = str(command or "").strip() if not cmd: return False lowered = cmd.lower() - forbidden = list(READONLY_RUN_COMMAND_BLOCKED_TOKENS) - if any(token in cmd for token in forbidden): + + # 1. 绝对禁止的token（子shell、逻辑运算符、重定向） + # 管道符 | 单独处理，不在此禁止 + absolute_forbidden = ("&&", "||", ";", ">", "<", "$(", "`", "&",) + if any(token in cmd for token in absolute_forbidden): return False - parts = cmd.split() + + # 2. 如果包含管道，使用管道专用检查 + if "|" in cmd: + return self._is_readonly_pipeline_allowed(cmd) + + # 3. 普通单命令检查 + return self._is_readonly_single_command_allowed(cmd) + + def _is_readonly_pipeline_allowed(self, command: str) -> bool: + """检查管道链中每个命令是否都是只读的""" + # 分割管道段（保留空字符串检查） + segments = [seg.strip() for seg in command.split("|") if seg.strip()] + if not segments: + return False + + for segment in segments: + # 段内禁止子shell（双重检查） + if "$(" in segment or "`" in segment: + return False + + # 提取命令名 + parts = segment.split() + if not parts: + return False + executable = parts[0].lower() + + # 检查段内是否有输出重定向（管道内也不允许） + if ">" in segment: + return False + + # 检查是否是安全的只读命令 + if not self._is_safe_readonly_executable(executable, segment): + return False + + # 检查是否是危险的管道目标 + if self._is_dangerous_pipe_target(segment): + return False + + return True + + def _is_readonly_single_command_allowed(self, command: str) -> bool: + """检查单个命令是否是只读的""" + lowered = command.lower() + parts = command.split() if not parts: return False executable = parts[0].lower() + + # 使用基础白名单 basic_allowed = set(READONLY_RUN_COMMAND_ALLOWED or ()) if executable in basic_allowed and executable not in {"git", "sed", "awk"}: return True + # 特殊命令检查 if executable == "git": if len(parts) < 2: return False @@ -179,11 +245,57 @@ class MainTerminalToolsExecutionMixin: return len(parts) >= 2 and parts[1] == "-n" if executable == "awk": - # 只读模式下允许基础文本处理，但拒绝 awk 的命令执行能力 return "system(" not in lowered return False + def _is_safe_readonly_executable(self, executable: str, segment: str) -> bool: + """检查单个命令是否在扩展白名单中""" + if executable not in self._READONLY_ALLOWED_EXECUTABLES: + return False + + # 特殊命令的安全检查 + lowered = segment.lower() + + if executable == "sed": + return "-n" in segment + + if executable == "awk": + return "system(" not in lowered + + if executable == "git": + parts = segment.split() + if len(parts) < 2: + return False + git_sub = parts[1].lower() + return git_sub in set(READONLY_RUN_COMMAND_ALLOWED_GIT_SUBCOMMANDS or ()) + + # 检查其他命令是否有危险的参数组合 + # 例如：sort -o（输出到文件）是危险的 + if executable == "sort": + return "-o" not in segment and "--output" not in lowered + + return True + + def _is_dangerous_pipe_target(self, segment: str) -> bool: + """检查管道目标是否会执行写入/修改/危险操作""" + lowered = segment.lower() + parts = lowered.split() + if not parts: + return False + cmd = parts[0] + + # 检查是否在危险命令列表中 + if cmd in self._DANGEROUS_PIPE_TARGETS: + return True + + # 检查是否有危险的参数（如 -exec, -delete 等） + dangerous_args = ("-exec", "-delete", "-ok", "-execdir", "-okdir") + if any(arg in lowered for arg in dangerous_args): + return True + + return False + def evaluate_tool_permission(self, tool_name: str, arguments: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: mode = "unrestricted" try: diff --git a/prompts/permission_mode.txt b/prompts/permission_mode.txt index 2d28e20..6f65d55 100644 --- a/prompts/permission_mode.txt +++ b/prompts/permission_mode.txt @@ -1,4 +1,6 @@ -## 权限模式约束（始终生效） -- 当前权限模式：{permission_mode_label}（{permission_mode}） -- 执行规则：{permission_rules} -- 注意：工具始终可见；若超出权限，调用会返回拒绝或等待批准结果。 +## 权限模式：{permission_mode_label}（{permission_mode}） + +{mode_description} + +### 当前规则 +{detailed_rules} diff --git a/utils/context_manager.py b/utils/context_manager.py index fda86ed..a63dc09 100644 --- a/utils/context_manager.py +++ b/utils/context_manager.py @@ -1972,129 +1972,4 @@ class ContextManager: return content - def build_messages(self, context: Dict, user_input: str) -> List[Dict]: - """构建消息列表（添加终端内容注入）""" - # 加载系统提示（Qwen3.5 使用专用提示） - model_key = getattr(self.main_terminal, "model_key", "kimi") if hasattr(self, "main_terminal") else "kimi" - prompt_name = "main_system_qwenvl" if (model_supports_image(model_key) or model_supports_video(model_key)) else "main_system" - system_prompt = self.load_prompt(prompt_name) - # 格式化系统提示 - container_path = self.container_mount_path or "/workspace" - container_cpus = self.container_cpu_limit - container_memory = self.container_memory_limit - project_storage = self.project_storage_limit - prompt_replacements = get_model_prompt_replacements(model_key) - now = datetime.now() - current_time_text = f"{now.year}年{now.month}月{now.day}日 {now.hour}点（24小时制）" - system_prompt = system_prompt.format( - project_path=container_path, - container_path=container_path, - container_cpus=container_cpus, - container_memory=container_memory, - project_storage=project_storage, - file_tree=context["project_info"]["file_tree"], - memory=context["memory"], - current_time=current_time_text, - model_description=prompt_replacements.get("model_description", "") - ) - - messages = [ - {"role": "system", "content": system_prompt} - ] - - try: - from modules.personalization_manager import load_personalization_config - from modules.skills_manager import ( - get_skills_catalog, - build_skills_list, - merge_enabled_skills, - build_skills_prompt, - ) - personalization_config = getattr(self, "custom_personalization_config", None) or load_personalization_config(self.data_dir) - skills_catalog = get_skills_catalog() - enabled_skills = merge_enabled_skills( - personalization_config.get("enabled_skills") if isinstance(personalization_config, dict) else None, - skills_catalog, - personalization_config.get("skills_catalog_snapshot") if isinstance(personalization_config, dict) else None, - ) - skills_template = self.load_prompt("skills_system").strip() - skills_list = build_skills_list(skills_catalog, enabled_skills) - skills_prompt = build_skills_prompt(skills_template, skills_list) - if skills_prompt: - messages.append({"role": "system", "content": skills_prompt}) - except Exception as exc: - print(f"[Skills] 系统提示生成失败: {exc}") - - workspace_system = self._build_workspace_system_message(context) - if workspace_system: - messages.append({"role": "system", "content": workspace_system}) - - # 浅压缩替换开关：仅当个人空间开启自动浅层压缩时，才将已打标 tool 结果替换为占位符 - shallow_replace_enabled = False - try: - personalization_config = ( - getattr(self, "custom_personalization_config", None) - or load_personalization_config(self.data_dir) - ) - shallow_replace_enabled = bool(personalization_config.get("auto_shallow_compress_enabled", False)) - except Exception: - shallow_replace_enabled = False - - replaced_tool_count = 0 - # 添加对话历史 - for conv in context["conversation"]: - if conv["role"] == "assistant": - message = { - "role": conv["role"], - "content": conv["content"] - } - reasoning = conv.get("reasoning_content") - if reasoning: - message["reasoning_content"] = reasoning - if "tool_calls" in conv and conv["tool_calls"]: - message["tool_calls"] = conv["tool_calls"] - messages.append(message) - elif conv["role"] == "tool": - conv_meta = conv.get("metadata") or {} - if shallow_replace_enabled and conv_meta.get("auto_shallow_compacted"): - message = { - "role": "tool", - "content": AUTO_SHALLOW_PLACEHOLDER, - "tool_call_id": conv.get("tool_call_id", ""), - "name": conv.get("name", "") - } - messages.append(message) - replaced_tool_count += 1 - continue - images = conv.get("images") or (conv.get("metadata") or {}).get("images") or [] - videos = conv.get("videos") or (conv.get("metadata") or {}).get("videos") or [] - content_value = conv.get("content") - if isinstance(content_value, list): - content_payload = content_value - elif images or videos: - content_payload = self._build_content_with_images(content_value, images, videos) - else: - content_payload = content_value - message = { - "role": "tool", - "content": content_payload, - "tool_call_id": conv.get("tool_call_id", ""), - "name": conv.get("name", "") - } - messages.append(message) - else: - images = conv.get("images") or (conv.get("metadata") or {}).get("images") or [] - videos = conv.get("videos") or (conv.get("metadata") or {}).get("videos") or [] - content_payload = self._build_content_with_images(conv["content"], images, videos) if (images or videos) else conv["content"] - messages.append({ - "role": conv["role"], - "content": content_payload - }) - if shallow_replace_enabled: - print(f"[ContextCompression] build_messages 替换tool占位符: {replaced_tool_count} 条") - - # 添加终端内容（如果有的话） - # 这里需要从参数传入或获取 - - return messages