shuiyu1123
78fea7ccdd
- 重构 core/ 底层 (7文件): app/runner/engine/parser/history/safety/display - FC原生调用为主，prompt-JSON保底 - 移除所有工具返回截断 (ai_search/git_tools/cmd_executor/browser_auto/network_tools/sub_agent/tool_search) - 删除无用文件 (旧服务器/WebUI/彩蛋/启动器等21个文件) - 清理 constants.py 死代码 - 版本号更新为 7.0.0
138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
"""安全检查"""
|
|
import json, re, threading
|
|
from .types import ToolCall
|
|
|
|
MODE_UNRESTRICTED = 0
|
|
MODE_NORMAL = 1
|
|
MODE_MANUAL = 2
|
|
MODE_PLAN = 3
|
|
|
|
WHITELIST = {
|
|
"code_editor": ["read_range", "find ", "regex ", "symbols", "imports", "stats",
|
|
"structure", "context", "todo", "diff "],
|
|
"git_tools": ["status", "log", "diff", "branch", "remote", "show", "blame"],
|
|
"file_manager": ["list", "read", "info", "search"],
|
|
"ai_search": ["url ", "search ", "web "],
|
|
"network_tools": ["ping", "status", "headers"],
|
|
"code_search": ["find", "regex", "symbols"],
|
|
"task_manager": ["list"],
|
|
"sub_agent": ["list", "status"],
|
|
"auto_engineer": ["metrics", "security", "complexity", "suggest"],
|
|
}
|
|
|
|
|
|
class Safety:
|
|
def __init__(self, cli=None, engine_provider=None):
|
|
self._cli = cli
|
|
self._engine = engine_provider or (lambda: None)
|
|
self._lock = threading.Lock()
|
|
self._cache = {}
|
|
try:
|
|
from xcli_core.config import get_system_config
|
|
self.mode = get_system_config("safety_mode", MODE_NORMAL)
|
|
except Exception:
|
|
self.mode = MODE_NORMAL
|
|
|
|
def check(self, tool_call):
|
|
"""返回 (allowed, reason)"""
|
|
# 白名单
|
|
patterns = WHITELIST.get(tool_call.name, [])
|
|
args_lower = tool_call.args.strip().lower()
|
|
for p in patterns:
|
|
if args_lower.startswith(p.lower()):
|
|
return True, "白名单放行"
|
|
|
|
# 无限制/计划模式
|
|
if self.mode in (MODE_UNRESTRICTED, MODE_PLAN):
|
|
return True, "模式自动放行"
|
|
|
|
# 人工模式
|
|
if self.mode == MODE_MANUAL:
|
|
return self._confirm(tool_call)
|
|
|
|
# 普通模式：AI分析
|
|
return self._ai_analyze(tool_call)
|
|
|
|
def _ai_analyze(self, tool_call):
|
|
engine = self._engine()
|
|
if not engine:
|
|
return self._confirm(tool_call)
|
|
|
|
key = f"{tool_call.name}:{tool_call.args[:100]}"
|
|
with self._lock:
|
|
if key in self._cache:
|
|
cached = self._cache[key]
|
|
if cached.get("safe"):
|
|
return True, "AI分析通过(缓存)"
|
|
return self._confirm(tool_call)
|
|
|
|
prompt = f"""分析工具调用的安全性。只返回JSON：
|
|
{{"safe": true/false, "level": "低/中/高", "reason": "原因"}}
|
|
|
|
工具: {tool_call.name}
|
|
参数: {tool_call.args[:200]}
|
|
|
|
safe=true: 读取/查看/无害操作
|
|
safe=false: 删除/执行命令/修改配置"""
|
|
|
|
try:
|
|
resp = engine.generate_response(prompt, system_prompt="只返回JSON")
|
|
result = self._parse(resp)
|
|
if result:
|
|
with self._lock:
|
|
self._cache[key] = result
|
|
if result.get("safe") is True:
|
|
return True, "AI分析通过"
|
|
return self._confirm(tool_call)
|
|
except Exception:
|
|
pass
|
|
return self._confirm(tool_call)
|
|
|
|
def _parse(self, resp):
|
|
if not resp:
|
|
return None
|
|
resp = resp.strip()
|
|
# 去thinking
|
|
if "</thinking>" in resp:
|
|
resp = resp.split("</thinking>")[-1].strip()
|
|
# 去markdown
|
|
if resp.startswith("```"):
|
|
resp = resp.split("\n", 1)[-1] if "\n" in resp else resp[3:]
|
|
if resp.endswith("```"):
|
|
resp = resp[:-3]
|
|
try:
|
|
return json.loads(resp.strip())
|
|
except json.JSONDecodeError:
|
|
m = re.search(r'\{.*?"safe".*?\}', resp, re.DOTALL)
|
|
if m:
|
|
try:
|
|
return json.loads(m.group())
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return None
|
|
|
|
def _confirm(self, tool_call):
|
|
try:
|
|
from colorama import Fore, Style
|
|
except ImportError:
|
|
Fore = Style = type('D', (), {'RED': '', 'CYAN': '', 'YELLOW': '', 'RESET_ALL': ''})()
|
|
print(f" {Fore.RED}工具: {tool_call.name}{Style.RESET_ALL}")
|
|
print(f" {Fore.CYAN}参数: {tool_call.args[:100]}{Style.RESET_ALL}")
|
|
try:
|
|
choice = input(f" {Fore.YELLOW}是否执行? (y/n): {Style.RESET_ALL}").strip().lower()
|
|
allowed = choice in ('y', 'yes', '是')
|
|
return allowed, "用户确认" if allowed else "用户拒绝"
|
|
except (EOFError, KeyboardInterrupt):
|
|
return True, "非交互模式自动放行"
|
|
|
|
def get_mode(self):
|
|
return self.mode
|
|
|
|
def set_mode(self, mode):
|
|
self.mode = mode
|
|
try:
|
|
from xcli_core.config import set_system_config
|
|
set_system_config("safety_mode", mode)
|
|
except Exception:
|
|
pass
|