feat: 权限模式支持管道命令并优化prompt注入
This commit is contained in:
parent
3219df388b
commit
f9089af8f9
@ -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:
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -1,4 +1,6 @@
|
||||
## 权限模式约束（始终生效）
|
||||
- 当前权限模式：{permission_mode_label}（{permission_mode}）
|
||||
- 执行规则：{permission_rules}
|
||||
- 注意：工具始终可见；若超出权限，调用会返回拒绝或等待批准结果。
|
||||
## 权限模式：{permission_mode_label}（{permission_mode}）
|
||||
|
||||
{mode_description}
|
||||
|
||||
### 当前规则
|
||||
{detailed_rules}
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user