From d3acccd146ae369722e4646bfa2241db5ed8182b Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Tue, 18 Aug 2026 02:33:17 +0000 Subject: [PATCH] fix(aibox): batch H audit fixes - subagent cancel propagation, scheduler timeout, exec process-group reaping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fourth audit round (see docs/reports/aibox_audit_20260817.md §11): - H1: shared parent cancel event no longer cleared on subagent first think (N5 regression); parent-already-cancelled skips subagent start - M1: /reload_prompts moved off event loop (run_in_executor, N3 sibling) - M2: skill --help scan prewarmed before serving connections - M3: scheduler task-level timeout (scheduler.task_timeout, default 600s) - M4: get_recent_memories(None) capped instead of full-table load - M5: memo delete now physically removes rows (was complete-only) - M6: PromptManager.reload builds new list then atomically swaps - L1-L12: read_file arg coercion, proxy save failure reported, SSE utf-8 tolerance, persona realpath check, truncation char boundary, compressor div0, skill path quoting, ask_user required, POSIX process-group kill (pgid captured at Popen), web scheme whitelist, registry redact substring match, Bearer redaction in failover logs --- aibox/aibox.py | 67 +++- aibox/common/config.py | 35 +- aibox/llm_client.py | 20 +- aibox/subsystems/conversation_compressor.py | 3 +- aibox/subsystems/db.py | 20 +- aibox/subsystems/executors.py | 87 ++++- aibox/subsystems/prompt_manager.py | 28 +- aibox/tests/test_h_fixes.py | 408 ++++++++++++++++++++ aibox/tools/basic_tools.py | 16 + aibox/tools/db_tools.py | 4 +- aibox/tools/misc_tools.py | 7 +- aibox/tools/proxy_tools.py | 7 +- aibox/tools/registry.py | 9 +- aibox/tools/skill_tools.py | 11 +- 14 files changed, 643 insertions(+), 79 deletions(-) create mode 100644 aibox/tests/test_h_fixes.py diff --git a/aibox/aibox.py b/aibox/aibox.py index 45e095a..07ce313 100644 --- a/aibox/aibox.py +++ b/aibox/aibox.py @@ -328,14 +328,35 @@ class TaskScheduler: ai_thread.daemon = True ai_thread.start() - # 等待 AI 完成，同时处理可能的取消 + # M3（审计 2026-08-18）：任务级超时兜底。此前 await ai_future + # 无任何超时/取消路径——AI think 卡死（LLM 挂起、子进程不可杀） + # 时该任务永不返回，调度器 run 循环顺序 await 被阻塞，后续全部 + # 定时/即时任务不再执行且任务已 mark_as_triggered 不会重试。 + # 到点 call_later 触发 ai.cancel()（置取消事件 + 关在途 HTTP 流）， + # worker 线程自然收尾，调度循环继续。 + task_timeout = float(ConfigManager().get("scheduler.task_timeout", 600)) + timed_out = threading.Event() + + def _timeout_cancel(): + timed_out.set() + try: + ai.cancel() + except Exception: + pass + + timeout_handle = loop.call_later(task_timeout, _timeout_cancel) try: response = await ai_future result = response or "AI无响应" except Exception as e: - self.logger.error(f"AI对话失败: {e}") - result = f"任务执行但AI对话失败: {str(e)}" + if timed_out.is_set(): + self.logger.error(f"任务执行超时（{task_timeout}秒），AI 已取消") + result = f"任务执行超时（{task_timeout}秒），AI 已取消" + else: + self.logger.error(f"AI对话失败: {e}") + result = f"任务执行但AI对话失败: {str(e)}" finally: + timeout_handle.cancel() # 确保线程结束 ai_thread.join(timeout=1.0) @@ -534,10 +555,11 @@ class ExpressEmotionTool(Tool): if not name: return "" path = os.path.join(constants.SKILLS_DIR, name + ".md") - # N12：persona 为 LLM 可控输入，做路径边界校验——normcase 统一大小写与 - # 分隔符后要求解析结果仍在 SKILLS_DIR 内，防止 ../../ 越界读取并注入 + # N12 + L4（审计 2026-08-18）：persona 为 LLM 可控输入。abspath 前缀 + # 检查拦截 ../ 越界，但符号链接可绕过（skills 目录内软链指向任意文件 + # 即被读取注入）；realpath 解析软链后再做前缀校验。 skills_root = os.path.normcase(os.path.abspath(constants.SKILLS_DIR)) - resolved = os.path.normcase(os.path.abspath(path)) + resolved = os.path.normcase(os.path.realpath(path)) if not resolved.startswith(skills_root + os.sep): self.logger.warning(f"人格技能名非法（越界路径）: {name}") return "" @@ -673,6 +695,14 @@ def run_full_subagent(role, task, config, logger=None, model=None, # 深度中断（sub_chat.cancel() 强关在途 HTTP 流并 kill 其子进程）。 if cancel_event is not None: sub_chat._cancel_event = cancel_event + # H1（审计 2026-08-18）：共享父事件后禁止子代理首轮 clear—— + # _think_and_respond_impl 首轮无条件 clear() 会吞掉父级已置位的 + # 取消信号（用户中断 / --once 超时恰落在子代理生成窗口时被清掉， + # 观察者线程随之失效，子代理跑完全程）。置位后立即检查一次， + # 父级已取消则不再启动子代理。 + sub_chat._shared_cancel_event = True + if cancel_event.is_set(): + return "❌ 子代理已因父级取消而终止" def _watch_parent_cancel(): while not _watch_done.wait(timeout=0.2): @@ -1456,8 +1486,9 @@ class DeepSeekChat: def _think_and_respond_impl(self, input_text: str, output_callback: Callable = None, is_continuation: bool = False) -> Optional[str]: """think_and_respond 的实际实现，在 _msg_lock 保护下执行。""" - # 首轮开始时清除取消标志 - if not is_continuation: + # 首轮开始时清除取消标志（H1：共享父事件的子代理不清除——清除会 + # 吞掉父级已置位的取消信号，使中断/--once 超时失效） + if not is_continuation and not getattr(self, "_shared_cancel_event", False): self._cancel_event.clear() # 立即检查是否已被取消 @@ -2492,10 +2523,17 @@ class WebSocketHandler: elif command == "/reload_prompts": session = self._get_session(websocket) ai = session.get("ai") if session else None - if ai: - new_count = ai.reload_prompts() - else: - new_count = self.ai_manager.prompt_manager.reload() + # M1（审计 2026-08-18）：reload_prompts 内部同步获取 _msg_lock + # （繁忙对话持锁可达数十秒），直接调用会冻结整个事件循环—— + # N3 只修了 /new，这里同类遗漏。移入线程池执行。 + loop = asyncio.get_running_loop() + + def _do_reload_prompts(): + if ai: + return ai.reload_prompts() + return self.ai_manager.prompt_manager.reload() + + new_count = await loop.run_in_executor(None, _do_reload_prompts) await self.send_message(websocket, { "type": "command_result", "content": i18n.get('prompts_reloaded', new_count) @@ -2883,12 +2921,17 @@ class EnhancedAIManager: port = constants.WEBSOCKET_PORT self.ws_handler = WebSocketHandler(self) + # M2（审计 2026-08-18）：技能扫描对每个可执行技能串行跑 --help + # 子进程（各 ≤10s），懒加载于首条消息/首个任务时在事件循环线程 + # 执行会冻结整个服务。启动前预热，把扫描成本移到连接建立之前。 + self.ws_handler._get_shared_skill_manager() # 如果启用了任务调度器，创建并启动 if constants.SCHEDULER_ENABLED: self.scheduler = TaskScheduler(asyncio.get_running_loop(), self.logger) self.scheduler.set_websocket_handler(self.ws_handler) self.scheduler.set_ai_manager(self) # 设置AI管理器引用 + self.scheduler._get_shared_skill_manager() async def handler(websocket): await self.handle_connection(websocket, None) diff --git a/aibox/common/config.py b/aibox/common/config.py index bc68887..d293eea 100644 --- a/aibox/common/config.py +++ b/aibox/common/config.py @@ -278,27 +278,32 @@ class ConfigManager: "password_env": password_env } if save: - self._save_config() + return self._save_config() return True - def _save_config(self): - """将当前配置保存到配置文件（覆盖）。 + def _save_config(self) -> bool: + """将当前配置保存到配置文件（覆盖）。返回是否成功落盘。 落盘前对代理密码做纵深防御脱敏：无论调用路径如何，明文密码一律不写入文件。 + L2（审计 2026-08-18）：此前异常仅 print，set_proxy 无条件报"已保存"， + 磁盘只读等场景下用户以为已持久化，重启后配置丢失。 """ config_path = self._config_path - if config_path: - try: - safe_config = copy.deepcopy(self._config) - socks = safe_config.get("proxy", {}).get("socks5") - if socks and socks.get("password"): - socks["password"] = None - os.makedirs(os.path.dirname(config_path), exist_ok=True) - with open(config_path, 'w', encoding='utf-8') as f: - json.dump(safe_config, f, ensure_ascii=False, indent=2) - print(f"✅ 配置已保存到 {config_path}") - except Exception as e: - print(f"⚠️ 保存配置失败: {e}") + if not config_path: + return False + try: + safe_config = copy.deepcopy(self._config) + socks = safe_config.get("proxy", {}).get("socks5") + if socks and socks.get("password"): + socks["password"] = None + os.makedirs(os.path.dirname(config_path), exist_ok=True) + with open(config_path, 'w', encoding='utf-8') as f: + json.dump(safe_config, f, ensure_ascii=False, indent=2) + print(f"✅ 配置已保存到 {config_path}") + return True + except Exception as e: + print(f"⚠️ 保存配置失败: {e}") + return False def get_proxy_info(self) -> Dict: """获取代理配置的详细信息（用于显示，不回传任何密码值）""" diff --git a/aibox/llm_client.py b/aibox/llm_client.py index 026aed3..c095279 100644 --- a/aibox/llm_client.py +++ b/aibox/llm_client.py @@ -22,9 +22,21 @@ from typing import Dict, Iterator, List, Optional import json import os +import re import time import requests +# L12（审计 2026-08-18）：瞬时错误体可能回显网关/WAF 错误页中的 +# Authorization 头，抹掉 Bearer 令牌后再进日志，防密钥落盘 +_AUTH_HEADER_RE = re.compile(r"Bearer\s+[A-Za-z0-9._~+/=-]+") + + +def _sanitize_log_msg(msg: str) -> str: + """日志脱敏：替换错误消息中的 Bearer 令牌。""" + if not msg: + return msg + return _AUTH_HEADER_RE.sub("Bearer ***", msg) + class LLMClient(ABC): """大模型客户端抽象基类。子类需实现流式/非流式两种调用，并归一化为统一事件格式。""" @@ -64,7 +76,7 @@ class LLMClient(ABC): def _log(self, level: str, msg: str): if self.logger is not None and hasattr(self.logger, level): try: - getattr(self.logger, level)(msg) + getattr(self.logger, level)(_sanitize_log_msg(msg)) except Exception: pass @@ -219,7 +231,7 @@ class DeepSeekClient(LLMClient): for line in response.iter_lines(): if not line: continue - line_text = line.decode("utf-8") + line_text = line.decode("utf-8", errors="replace") # L3：非 UTF-8 行不中断流 if not line_text.startswith("data: "): continue data = line_text[6:] @@ -352,7 +364,7 @@ class FailoverClient: def _log(self, level: str, msg: str): if self.logger is not None and hasattr(self.logger, level): try: - getattr(self.logger, level)(msg) + getattr(self.logger, level)(_sanitize_log_msg(msg)) except Exception: pass @@ -382,7 +394,7 @@ class FailoverClient: self._stats["failovers"] += 1 try: if self.on_failover is not None: - self.on_failover(from_name, to_name, reason) + self.on_failover(from_name, to_name, _sanitize_log_msg(reason)) except Exception as e: self._log("warning", f"on_failover 回调异常（忽略）: {e}") diff --git a/aibox/subsystems/conversation_compressor.py b/aibox/subsystems/conversation_compressor.py index a5ebc20..c0b1a59 100644 --- a/aibox/subsystems/conversation_compressor.py +++ b/aibox/subsystems/conversation_compressor.py @@ -514,7 +514,8 @@ class ConversationCompressor: final_tokens = self.estimate_tokens([m for m in final_messages if m.get("role") != "system"]) print(f"\n✅ {i18n.get('conversation_compressed', original_count, final_count)}") - print(f" Token: {original_tokens} -> {final_tokens} (压缩率: {final_tokens/original_tokens:.1%})") + # L6：压缩块全空时 original_tokens 可能为 0，除零防护 + print(f" Token: {original_tokens} -> {final_tokens} (压缩率: {final_tokens / max(original_tokens, 1):.1%})") # 检查是否压缩到阈值以下 if final_count >= threshold: diff --git a/aibox/subsystems/db.py b/aibox/subsystems/db.py index 75f3671..f50366f 100644 --- a/aibox/subsystems/db.py +++ b/aibox/subsystems/db.py @@ -232,14 +232,15 @@ class MemoryDatabase: return results def get_recent_memories(self, limit: int = 10) -> List[Dict]: - """获取最近的记忆。limit=None 时不限制返回条数（供分页使用）。""" + """获取最近的记忆。limit=None 时使用默认上限（M4：此前 None 无 LIMIT + 全表加载，记忆库增长后单次调用 fetchall 全量入内存；与 search_memories + 的 None→默认上限语义对齐）。""" + if limit is None: + limit = constants.MEMORY_MAX_SEARCH_RESULTS with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() - if limit is None: - cursor.execute('SELECT * FROM memories ORDER BY end_time DESC') - else: - cursor.execute('SELECT * FROM memories ORDER BY end_time DESC LIMIT ?', (limit,)) + cursor.execute('SELECT * FROM memories ORDER BY end_time DESC LIMIT ?', (limit,)) results = [] for row in cursor.fetchall(): @@ -672,6 +673,15 @@ class MemoDatabase: return True return False + def delete_memo(self, memo_id: int) -> bool: + """物理删除任务（M5：delete 语义此前被 complete_memo 顶替——只标记 + 完成不删行，重复任务还会被推进而非移除，stats 计数失真）""" + with sqlite3.connect(self.db_path) as conn: + cursor = conn.cursor() + cursor.execute("DELETE FROM memos WHERE id = ?", (memo_id,)) + conn.commit() + return cursor.rowcount > 0 + def get_memo_stats(self) -> Dict: """获取任务统计信息""" with sqlite3.connect(self.db_path) as conn: diff --git a/aibox/subsystems/executors.py b/aibox/subsystems/executors.py index ead795d..a7889b3 100644 --- a/aibox/subsystems/executors.py +++ b/aibox/subsystems/executors.py @@ -13,6 +13,7 @@ subsystems.executors - 命令执行器 import os import queue +import signal import subprocess import threading import time @@ -71,17 +72,44 @@ class SubprocessCommandExecutor(CommandExecutor): def __init__(self, logger=None): super().__init__(logger) self._current_process = None + self._current_pgid = None # L9：Popen 后立即捕获，shell 死亡后 getpgid 会失败 self._process_lock = threading.Lock() + def _terminate_tree(self, proc, force: bool = False): + """终止整个进程组（L9：POSIX 下 shell=True 只杀 shell 时，孙进程 + （`(x &)` 等）仍持管道写端——reader 阻塞无 EOF、fd/线程悬挂至孙 + 进程退出，写满 64KB 管道缓冲后孙进程也卡死。start_new_session 使 + shell 成为会话首进程，按组信号可一次覆盖整棵进程树）。 + 组 ID 在 Popen 后立即保存：shell 死亡后 os.getpgid(pid) 会抛 + ProcessLookupError，届时无法再从进程反查组。 + Windows 无进程组概念，退回单进程 terminate/kill。""" + try: + if os.name == "posix": + sig = signal.SIGKILL if force else signal.SIGTERM + with self._process_lock: + pgid = self._current_pgid + if pgid is None: + pgid = os.getpgid(proc.pid) + os.killpg(pgid, sig) + elif force: + proc.kill() + else: + proc.terminate() + except Exception: + try: + proc.kill() + except Exception: + pass + def cancel(self): - """取消当前正在运行的命令（kill子进程）。线程安全。""" + """取消当前正在运行的命令（kill子进程组）。线程安全。""" with self._process_lock: proc = self._current_process if proc is not None: try: if proc.poll() is None: - proc.kill() - self.logger.info("已终止正在运行的子进程") + self._terminate_tree(proc, force=True) + self.logger.info("已终止正在运行的子进程组") except Exception: pass @@ -117,14 +145,14 @@ class SubprocessCommandExecutor(CommandExecutor): while True: # 检查超时（readline 阻塞不再影响本判断） if timeout > 0 and time.time() - start_time > timeout: - process.terminate() + self._terminate_tree(process) timeout_occurred = True break # 检查输出大小 if total_bytes >= self.MAX_OUTPUT_BYTES: truncated = True - process.terminate() + self._terminate_tree(process) break try: @@ -140,6 +168,12 @@ class SubprocessCommandExecutor(CommandExecutor): if shell_dead_since is None: shell_dead_since = time.time() elif time.time() - shell_dead_since > 1.0: + # L9：shell 已死但孙进程仍持管道写端（`(sleep 300 &) ; + # echo done` 形态）。只靠 daemon reader 兜底会悬挂线程 + # 与管道 fd 至孙进程退出，写满 64KB 缓冲后孙进程自身 + # 也卡死。此处强杀进程组收尾（nohup 重定向输出的 + # 真正脱离场景不持管道，不会走到这里）。 + self._terminate_tree(process, force=True) break continue @@ -147,10 +181,12 @@ class SubprocessCommandExecutor(CommandExecutor): if total_bytes + line_bytes > self.MAX_OUTPUT_BYTES and not truncated: allowed = self.MAX_OUTPUT_BYTES - total_bytes if allowed > 0: - output_chunks.append(line.encode('utf-8', errors='replace')[:allowed].decode('utf-8', errors='replace')) + # L5：decode 用 errors='ignore'——按字节截断可能落在多字节 + # 字符中间，errors='replace' 会在行尾留下半个字符（�） + output_chunks.append(line.encode('utf-8', errors='replace')[:allowed].decode('utf-8', errors='ignore')) total_bytes = self.MAX_OUTPUT_BYTES truncated = True - process.terminate() + self._terminate_tree(process) break output_chunks.append(line) @@ -209,25 +245,30 @@ class SubprocessCommandExecutor(CommandExecutor): bufsize=1, universal_newlines=True, env=exec_env, - cwd=os.path.expanduser(work_dir) + cwd=os.path.expanduser(work_dir), + # L9：POSIX 下 shell 成为会话首进程，超时/取消可按进程组 + # 一次终止整棵进程树（含 & 后台孙进程），消除管道悬挂 + start_new_session=(os.name == "posix") ) # 保存进程引用以便cancel()可以kill它 with self._process_lock: self._current_process = process + if os.name == "posix": + try: + self._current_pgid = os.getpgid(process.pid) + except Exception: + self._current_pgid = None start_time = time.time() output_str, timeout_occurred, truncated = self._read_with_limit(process, timeout, start_time) # 确保进程终止 if process.poll() is None: - if timeout_occurred or truncated: - process.kill() - else: - process.terminate() + self._terminate_tree(process, force=timeout_occurred or truncated) try: process.wait(timeout=2) except: - process.kill() + self._terminate_tree(process, force=True) return_code = process.returncode if process.returncode is not None else -1 @@ -255,10 +296,11 @@ class SubprocessCommandExecutor(CommandExecutor): return f"❌ 执行错误: {str(e)}" finally: if process and process.poll() is None: - process.kill() + self._terminate_tree(process, force=True) # 清除当前进程引用 with self._process_lock: self._current_process = None + self._current_pgid = None def execute_streaming(self, command: str, output_queue: queue.Queue, timeout: int = 30, work_dir: str = None, env: Dict = None, **kwargs): @@ -296,12 +338,20 @@ class SubprocessCommandExecutor(CommandExecutor): bufsize=1, universal_newlines=True, env=exec_env, - cwd=os.path.expanduser(work_dir) + cwd=os.path.expanduser(work_dir), + # L9：POSIX 下 shell 成为会话首进程，超时/取消可按进程组 + # 一次终止整棵进程树（含 & 后台孙进程），消除管道悬挂 + start_new_session=(os.name == "posix") ) self.process = process # 保存进程引用以便cancel()可以kill它 with self._process_lock: self._current_process = process + if os.name == "posix": + try: + self._current_pgid = os.getpgid(process.pid) + except Exception: + self._current_pgid = None start_time = time.time() # H2 修复：复用线程化读取（reader 线程 + 超时轮询），行经 output_queue 流式转发 @@ -315,7 +365,7 @@ class SubprocessCommandExecutor(CommandExecutor): # 如果超时或截断，确保进程被终止 if (timeout_occurred or truncated) and process.poll() is None: - process.kill() + self._terminate_tree(process, force=True) try: process.wait(timeout=2) except: @@ -352,11 +402,12 @@ class SubprocessCommandExecutor(CommandExecutor): output_queue.put(("complete", f"❌ 执行错误: {str(e)}")) finally: if process and process.poll() is None: - process.terminate() + self._terminate_tree(process) try: process.wait(timeout=2) except: - process.kill() + self._terminate_tree(process, force=True) self.process = None with self._process_lock: self._current_process = None + self._current_pgid = None diff --git a/aibox/subsystems/prompt_manager.py b/aibox/subsystems/prompt_manager.py index 6816e98..3de4b0c 100644 --- a/aibox/subsystems/prompt_manager.py +++ b/aibox/subsystems/prompt_manager.py @@ -60,10 +60,16 @@ class PromptManager: prompts_dir = config.get("system.prompts_dir", "~/.aibox/prompts") return os.path.expanduser(prompts_dir) - def _load_prompts(self): + def _load_prompts(self, target: Optional[List[Dict]] = None): + """加载提示词到 target 列表（缺省 self._prompts）。 + + M6：reload 传入新列表本地构建，完成后由 reload 原子替换—— + 此前先清空 self._prompts 再逐文件重读，并发读者（其他连接构建 + system 消息）会读到空/半截提示词列表。 + """ + prompts = target if target is not None else self._prompts self._prompts_dir = self._get_prompts_dir() os.makedirs(self._prompts_dir, exist_ok=True) - self._prompts = [] # 确保核心文件存在 self._ensure_identity_file() @@ -86,17 +92,14 @@ class PromptManager: with open(filepath, 'r', encoding='utf-8') as f: content = f.read().strip() if content: - self._prompts.append({ + prompts.append({ 'file': filename, 'content': content }) except Exception as e: print(f"⚠️ 读取提示词文件 {filename} 失败: {e}") - essential_files = [self.IDENTITY_FILE, self.STATUS_FILE, self.SOUL_FILE] - other_files = [f for f in prompt_files if f not in essential_files] if prompt_files else [] - - if not self._prompts: + if not prompts: self._create_default_prompts() def _get_template_dir(self) -> str: @@ -240,10 +243,6 @@ class PromptManager: return """我是DeepSeek，一个智能AI助手。 请创建 99_SOUL.md 文件来定义我的完整身份、角色、性格和能力边界。""" - def get_skill_prompts(self) -> List[Dict]: - """获取所有基础技能提示词（以1_开头的文件）""" - return [p for p in self._prompts if p['file'].startswith(self.SKILL_PREFIX)] - def get_soul_prompt(self) -> Optional[Dict]: """获取灵魂提示词（99_SOUL.md）""" for p in self._prompts: @@ -296,9 +295,10 @@ class PromptManager: return [p for p in self._prompts if self._skill_sort_key(p['file']) is not None] def reload(self): - """重新加载所有提示词""" - self._prompts = [] - self._load_prompts() + """重新加载所有提示词（M6：构建新列表后原子替换，并发读者始终读到完整列表）""" + new_prompts: List[Dict] = [] + self._load_prompts(new_prompts) + self._prompts = new_prompts return len(self._prompts) def get_prompt_count(self) -> int: diff --git a/aibox/tests/test_h_fixes.py b/aibox/tests/test_h_fixes.py new file mode 100644 index 0000000..6e92883 --- /dev/null +++ b/aibox/tests/test_h_fixes.py @@ -0,0 +1,408 @@ +"""aibox 第四轮审计 Batch H 修复回归测试（2026-08-18，审计报告 §十一） + +覆盖： + - H1：子代理共享父取消事件后首轮 think 不再清除父事件（N5 回归）+ 父已取消时跳过启动 + - M3：调度器任务级超时兜底（AI think 卡死不再停摆调度循环） + - M4：get_recent_memories(limit=None) 使用默认上限（不再全表加载） + - M5：memo delete 物理删除（此前只标记完成） + - M6：PromptManager.reload 原子替换（并发读者不再读到空/半截列表） + - L1：read_file 的 sed 参数强转 int（防引号注入） + - L2：set_proxy 落盘失败如实报告（不再静默报"已保存"） + - L3：SSE 非 UTF-8 行不中断流（errors=replace） + - L4：persona 路径校验解析符号链接（软链越界拒绝） + - L5：1MB 截断不再产生半个字符（�） + - L6：compressor 空内容压缩不再除零 + - L7：InstallSkillTool 指引命令路径引用（防拆词/注入） + - L8：ask_user 缺参返回明确错误（不再 TypeError） + - L10：web 工具仅允许 http/https（file:// 等拒绝） + - L11：registry 日志打码改包含匹配（apiKey/access_token 等变体） + - L12：FailoverClient 日志 Bearer 令牌脱敏 + +M1（/reload_prompts 移线程池）/M2（技能扫描启动前预热）为 async WebSocket / +启动路径，依赖事件循环与真实连接对象，由既有 WebSocket 测试面与启动逻辑回归。 +""" + +import asyncio +import os +import queue +import sys +import threading +import types +from unittest import mock + +import pytest + +# ---- 环境桩：websockets 未安装时保证 import aibox.py 不失败 ---- +if "websockets" not in sys.modules: + sys.modules["websockets"] = types.ModuleType("websockets") + +import aibox # noqa: E402 +from aibox import DeepSeekChat, TaskScheduler, run_full_subagent # noqa: E402 + +from llm_client import DeepSeekClient, FailoverClient, _sanitize_log_msg # noqa: E402 +from subsystems.db import MemoryDatabase, MemoDatabase # noqa: E402 +from subsystems.prompt_manager import PromptManager # noqa: E402 +from subsystems.conversation_compressor import ConversationCompressor # noqa: E402 +from common import constants # noqa: E402 + +from test_conversation_compressor import FakeLLMClient # noqa: E402 +from test_once_mode import _make_ai # noqa: E402 + + +def _fake_sub_chat(): + fake = mock.MagicMock() + fake.think_and_respond = mock.MagicMock(return_value="任务完成报告") + fake.stats = {} + fake.messages = [] + fake._cancel_event = threading.Event() + return fake + + +# ==================== H1：子代理取消事件清除回归 ==================== + +class TestH1SubagentCancelClear: + def test_shared_event_not_cleared_on_first_think(self): + """共享父事件后首轮 think 不清除事件——父级已置位的取消信号不被吞掉""" + ai = _make_ai() + parent_event = threading.Event() + parent_event.set() + ai._cancel_event = parent_event + ai._shared_cancel_event = True + with pytest.raises(InterruptedError): + ai._think_and_respond_impl("hi") + assert parent_event.is_set() + + def test_own_event_still_cleared_on_first_think(self): + """未共享（普通会话）时首轮仍清除自身事件——原语义不回归""" + import contextlib + ai = _make_ai() + ai._cancel_event.set() + with contextlib.suppress(Exception): + ai._think_and_respond_impl("hi") + assert not ai._cancel_event.is_set() + + def test_subagent_not_started_when_parent_already_cancelled(self): + """父事件已置位 → 直接返回取消提示，不启动子代理""" + fake = _fake_sub_chat() + parent_event = threading.Event() + parent_event.set() + with mock.patch("aibox.DeepSeekChat", return_value=fake), \ + mock.patch("aibox.create_llm_client"), \ + mock.patch("aibox.SubprocessCommandExecutor"): + result = run_full_subagent("r", "t", None, cancel_event=parent_event) + assert "取消" in result + fake.think_and_respond.assert_not_called() + + +# ==================== M3：调度器任务超时 ==================== + +class _BlockingAI: + """think 卡死直到被 cancel 的伪 AI（模拟 LLM 挂起）""" + + def __init__(self): + self.cancelled = threading.Event() + self.cancel_called = 0 + + def cancel(self): + self.cancel_called += 1 + self.cancelled.set() + + def think_and_respond(self, input_text, output_callback=None, + is_continuation=False): + self.cancelled.wait(timeout=15) + raise InterruptedError("用户中断") + + +class _FakeMemoDB: + def get_memo(self, memo_id): + return None + + +class TestM3SchedulerTimeout: + def test_stuck_task_times_out_and_scheduler_continues(self): + from aibox import ConfigManager + loop = asyncio.new_event_loop() + try: + scheduler = TaskScheduler(loop=loop, logger=None) + fake_ai = _BlockingAI() + memo = {"id": 7, "title": "卡死任务", "content": "内容", "trigger_count": 0} + orig_get = ConfigManager.get + + def fake_get(self, key, default=None): + # 只把任务超时压到 0.3s，其余配置读取走原实现（防止 + # DATETIME_FORMAT 等属性读到错误类型） + if key == "scheduler.task_timeout": + return 0.3 + return orig_get(self, key, default) + + with mock.patch.object(ConfigManager, "get", fake_get), \ + mock.patch.object(scheduler, "_create_task_ai", return_value=fake_ai): + result = loop.run_until_complete( + scheduler.execute_task_as_conversation(memo, _FakeMemoDB())) + assert "超时" in result + assert fake_ai.cancel_called >= 1 + finally: + loop.close() + + +# ==================== M4/M5：记忆上限与任务删除 ==================== + +class TestM4M5Db: + def test_recent_memories_none_capped(self, tmp_path, monkeypatch): + """limit=None 不再全表加载，使用默认上限（对齐 search_memories）""" + from test_conversation_compressor import patch_const + db = MemoryDatabase(db_path=str(tmp_path / "m.db")) + for i in range(7): + db.add_memory(topic=f"t{i}", summary=f"s{i}", content="c", + start_time="2026-08-18 00:00:00", + end_time="2026-08-18 00:00:01", importance=1) + patch_const(monkeypatch, "MEMORY_MAX_SEARCH_RESULTS", 5) + res = db.get_recent_memories(limit=None) + assert len(res) == 5 + + def test_delete_memo_physically_removes(self, tmp_path): + """delete_memo 真实删除行（此前 complete_memo 只标记完成）""" + db = MemoDatabase(db_path=str(tmp_path / "m.db")) + mid = db.add_memo(title="待删任务", content="内容") + assert db.get_memo(mid) is not None + assert db.delete_memo(mid) is True + assert db.get_memo(mid) is None + assert db.delete_memo(mid) is False # 已不存在 + + def test_memo_tool_delete_uses_physical_delete(self, tmp_path): + """MemoTool delete 走 delete_memo：行消失而非仅标记完成""" + from tools import MemoTool + db = MemoDatabase(db_path=str(tmp_path / "m.db")) + mid = db.add_memo(title="重复任务", content="内容", repeat_type="daily") + tool = MemoTool(db, None) + result = tool.execute(action="delete", memo_id=mid, confirm=True) + assert "已删除" in result + assert db.get_memo(mid) is None + stats = db.get_memo_stats() + assert stats["total"] == 0 + + +# ==================== M6：PromptManager 原子 reload ==================== + +class TestM6PromptReloadAtomic: + def test_reload_builds_new_list_then_swaps(self): + """reload 在本地新列表构建完成后才原子替换——读者始终看到完整列表""" + pm = PromptManager() + old_list = pm._prompts + seen = [] + + def fake_load(target): + seen.append(target) + target.append({"file": "X_test.md", "content": "y"}) + + with mock.patch.object(pm, "_load_prompts", side_effect=fake_load): + n = pm.reload() + assert len(seen) == 1 + assert seen[0] is not old_list # 构建用的是新列表 + assert pm._prompts is seen[0] # 完成后才替换 + assert n == 1 + + +# ==================== L1：read_file 参数强转 ==================== + +class TestL1ReadFileCoercion: + def test_non_integer_lines_rejected(self, tmp_path): + from tools import ReadFileTool + f = tmp_path / "f.txt" + f.write_text("line1\nline2\n", encoding="utf-8") + tool = ReadFileTool() + r = tool.execute(filepath=str(f), start_line="1'; rm -rf /; echo '", + end_line=5) + assert "参数错误" in r + + def test_valid_integers_still_work(self, tmp_path): + from tools import ReadFileTool + f = tmp_path / "f.txt" + f.write_text("line1\nline2\n", encoding="utf-8") + tool = ReadFileTool() + r = tool.execute(filepath=str(f), start_line=1, end_line=2) + assert "sed -n" in r and "参数错误" not in r + + +# ==================== L2：set_proxy 落盘失败如实报告 ==================== + +class TestL2ProxySaveFailure: + def test_set_proxy_returns_false_on_save_failure(self, tmp_path): + from common import ConfigManager + cm = ConfigManager() + cm._config_path = str(tmp_path / "config.json") + with mock.patch("builtins.open", side_effect=PermissionError("disk ro")): + ok = cm.set_proxy("127.0.0.1", 1080, save=True) + assert ok is False + + def test_proxy_tool_reports_save_failure(self, tmp_path): + from tools import SetProxyTool + tool = SetProxyTool(None) + with mock.patch("builtins.open", side_effect=PermissionError("disk ro")): + r = tool.execute(host="127.0.0.1", port=1080, save=True, + skip_check=True, enabled=True) + assert "保存失败" in r + + +# ==================== L3：SSE 非 UTF-8 行不中断流 ==================== + +class TestL3SseDecode: + def test_non_utf8_line_does_not_break_stream(self): + c = DeepSeekClient(api_key="k", model="m", base_url="https://x") + resp = mock.MagicMock() + resp.status_code = 200 + resp.iter_lines.return_value = [ + b"data: \xff\xfe", + b'data: {"choices":[{"delta":{"content":"ok"}}]}', + b"data: [DONE]", + ] + with mock.patch("llm_client.requests.post", return_value=resp): + events = list(c.stream_chat([{"role": "user", "content": "hi"}])) + assert any(e.get("type") == "content" and e.get("text") == "ok" + for e in events) + + +# ==================== L4：persona 软链越界 ==================== + +class TestL4PersonaSymlink: + def test_symlink_outside_skills_dir_rejected(self, tmp_path, monkeypatch): + from test_conversation_compressor import patch_const + from aibox import ExpressEmotionTool + skills_dir = tmp_path / "skills" + skills_dir.mkdir() + outside = tmp_path / "outside.md" + outside.write_text("秘密内容", encoding="utf-8") + (skills_dir / "evil.md").symlink_to(outside) + tool = ExpressEmotionTool() + patch_const(monkeypatch, "SKILLS_DIR", str(skills_dir)) + content = tool._load_persona("evil", "") + assert content == "" + + def test_normal_persona_still_loads(self, tmp_path, monkeypatch): + from test_conversation_compressor import patch_const + from aibox import ExpressEmotionTool + skills_dir = tmp_path / "skills" + skills_dir.mkdir() + (skills_dir / "happy.md").write_text("快乐人格", encoding="utf-8") + tool = ExpressEmotionTool() + patch_const(monkeypatch, "SKILLS_DIR", str(skills_dir)) + content = tool._load_persona("happy", "") + assert content == "快乐人格" + + +# ==================== L5：截断无半个字符 ==================== + +class TestL5NoPartialChar: + def test_truncated_output_has_no_replacement_char(self): + from subsystems.executors import SubprocessCommandExecutor + ex = SubprocessCommandExecutor() + r = ex.execute("python3 -c \"print('中' * 600000)\"", timeout=60) + assert "截断" in r + assert "\ufffd" not in r + + @pytest.mark.skipif(os.name != "posix", reason="Linux 进程组语义") + def test_grandchild_holding_pipe_is_reaped(self): + """`(sleep N &) ; echo done` 形态：shell 退出后孙进程持管道写端， + 宽限窗口到期强杀进程组——不悬挂 reader、孙进程被清理""" + import subprocess as sp + from subsystems.executors import SubprocessCommandExecutor + marker = f"sleep {12340}" + ex = SubprocessCommandExecutor() + r = ex.execute(f"({marker} &) ; echo done", timeout=8) + assert "done" in r + leftover = sp.run(["pgrep", "-f", marker], + capture_output=True, text=True).stdout.strip() + assert leftover == "" + + +# ==================== L6：compressor 除零 ==================== + +class TestL6CompressorDivZero: + def test_empty_content_compress_no_div0(self, monkeypatch): + comp = ConversationCompressor(llm_client=FakeLLMClient(), + logger=None) + from test_conversation_compressor import patch_const + patch_const(monkeypatch, "COMPRESS_MESSAGE_THRESHOLD", 10) + messages = [{"role": "system", "content": "SYS"}] + [ + {"role": "user" if i % 2 == 0 else "assistant", "content": ""} + for i in range(15) + ] + result = comp.compress_conversation(messages) + assert isinstance(result, list) + + +# ==================== L7：InstallSkillTool 路径引用 ==================== + +class TestL7SkillPathQuoting: + def test_guide_commands_quote_paths(self): + from tools import InstallSkillTool + tool = InstallSkillTool() + r = tool.execute(skill_path="/tmp/a b; rm -rf /") + assert "ls -l '/tmp/a b; rm -rf /'" in r + assert "cp '/tmp/a b; rm -rf /' " in r + + +# ==================== L8：ask_user 缺参 ==================== + +class TestL8AskUser: + def test_missing_question_returns_error(self): + from tools import AskUserTool + tool = AskUserTool() + assert "需要提供问题" in tool.execute() + + def test_question_passed_through(self): + from tools import AskUserTool + tool = AskUserTool() + assert "测试问题" in tool.execute(question="测试问题") + + +# ==================== L10：web 工具 scheme 白名单 ==================== + +class TestL10WebScheme: + def test_file_url_rejected(self): + from tools import SimpleWebCrawlerTool + tool = SimpleWebCrawlerTool() + r = tool.execute(action="get", url="file:///etc/shadow") + assert "仅支持 http/https" in r + + def test_http_url_accepted(self): + from tools import SimpleWebCrawlerTool + tool = SimpleWebCrawlerTool() + r = tool.execute(action="get", url="https://example.com") + assert "curl" in r and "仅支持" not in r + + +# ==================== L11：registry 打码包含匹配 ==================== + +class TestL11SanitizeSubstring: + def test_variant_keys_redacted(self): + from tools.registry import _sanitize_kwargs + out = _sanitize_kwargs({ + "apiKey": "x", "access_token": "y", "passwd": "z", + "url": "https://ok", + }) + assert out == {"apiKey": "***", "access_token": "***", + "passwd": "***", "url": "https://ok"} + + def test_normal_keys_untouched(self): + from tools.registry import _sanitize_kwargs + out = _sanitize_kwargs({"filepath": "/tmp/a", "question": "hi"}) + assert out == {"filepath": "/tmp/a", "question": "hi"} + + +# ==================== L12：failover 日志 Bearer 脱敏 ==================== + +class TestL12LogSanitize: + def test_sanitize_msg_redacts_bearer(self): + msg = "瞬时故障: HTTP 401: Bearer sk-abc123xyz" + out = _sanitize_log_msg(msg) + assert "sk-abc123xyz" not in out + assert "Bearer ***" in out + + def test_failover_log_redacts_bearer(self): + logger = mock.MagicMock() + c = FailoverClient([], logger=logger) + c._log("warning", "瞬时故障（Bearer sk-secret456）") + logged = logger.warning.call_args[0][0] + assert "sk-secret456" not in logged + assert "Bearer ***" in logged diff --git a/aibox/tools/basic_tools.py b/aibox/tools/basic_tools.py index e656d2d..f06dad2 100644 --- a/aibox/tools/basic_tools.py +++ b/aibox/tools/basic_tools.py @@ -96,6 +96,16 @@ class ReadFileTool(FileToolBase): # 限制max_lines上限，防止请求超大范围 max_lines = min(max_lines, 500) + # L1（审计 2026-08-18）：start/end_line 为 LLM 可控参数，此前未强转 + # 直拼 sed 单引号模板——LLM 传含引号的字符串可闭合引号注入命令。 + # 强转 int（非法输入在此拦截）并夹取合理范围。 + try: + start_line = int(start_line) + end_line = int(end_line) + except (TypeError, ValueError): + return "❌ 参数错误: start_line/end_line 必须为整数" + start_line = max(1, start_line) + end_line = max(0, min(end_line, 10_000_000)) quoted = shlex.quote(abs_path) # 构造等价的命令行读取命令 @@ -320,6 +330,12 @@ class SimpleWebCrawlerTool(Tool): if action not in ("get", "text", "title", "links", "search"): return f"❌ 未知操作: {action}" + # L10（审计 2026-08-18）：url 为 LLM 可控输入（可被网页内容诱导）， + # 此前不限 scheme——file:// 可经生成的 curl 指引读取本地文件（SSRF）。 + # 白名单 http/https 后其余协议直接拒绝。 + if not url or not str(url).strip().lower().startswith(("http://", "https://")): + return "❌ 仅支持 http/https 协议地址（已拦截 file:// 等非网络协议）" + quoted_url = shlex.quote(url) timeout = constants.WEB_TIMEOUT user_agent = shlex.quote(constants.WEB_USER_AGENT) diff --git a/aibox/tools/db_tools.py b/aibox/tools/db_tools.py index af530db..5fc71e2 100644 --- a/aibox/tools/db_tools.py +++ b/aibox/tools/db_tools.py @@ -752,7 +752,9 @@ class MemoTool(Tool): memo_id = kwargs.get("memo_id") if memo_id: - if self.memo_db.complete_memo(memo_id, force_complete=True): + # M5：真实物理删除（此前调 complete_memo 只标记完成，行不删除， + # 重复任务被推进而非移除，stats 计数失真） + if self.memo_db.delete_memo(memo_id): return self.format_result(True, f"任务 #{memo_id} 已删除") else: return f"❌ 无法删除任务 #{memo_id}" diff --git a/aibox/tools/misc_tools.py b/aibox/tools/misc_tools.py index 0cfebc4..18cf603 100644 --- a/aibox/tools/misc_tools.py +++ b/aibox/tools/misc_tools.py @@ -39,10 +39,13 @@ class AskUserTool(Tool): ) } }, - "required": [] + # L8：此前 required 为空，LLM 不传 question 时 str + None 抛 TypeError + "required": ["question"] } - def execute(self, question: str, **kwargs) -> str: + def execute(self, question: str = None, **kwargs) -> str: + if not question: + return "❌ 需要提供问题内容（question 参数）" return ( "你遇到了想搞清楚的事情: " + question + "\n你现在可以询问用户。" ) diff --git a/aibox/tools/proxy_tools.py b/aibox/tools/proxy_tools.py index 350c8f8..52b6ec0 100644 --- a/aibox/tools/proxy_tools.py +++ b/aibox/tools/proxy_tools.py @@ -146,7 +146,7 @@ class SetProxyTool(Tool): check_detail = detail config = ConfigManager() - config.set_proxy(host, port, username, password, password_env, enabled, save) + saved = config.set_proxy(host, port, username, password, password_env, enabled, save) proxy_info = config.get_proxy_info() if proxy_info.get("password_env"): pw_source = "环境变量 " + proxy_info["password_env"] @@ -159,10 +159,13 @@ class SetProxyTool(Tool): "服务器": f"{proxy_info['host']}:{proxy_info['port']}", "认证": "有" if proxy_info.get("username") else "无", "密码来源": pw_source, - "持久化": "已保存" if save else "仅内存" + "持久化": "已保存" if saved else "保存失败（仅本次会话有效，重启后丢失）" } if check_detail: data["连通性"] = check_detail + # L2：落盘失败（磁盘只读等）如实报告，不再静默报"已保存" + if save and not saved: + return self.format_result(False, "代理已配置但保存失败（仅本次会话有效，重启后丢失）", data) return self.format_result(True, "代理设置成功", data) diff --git a/aibox/tools/registry.py b/aibox/tools/registry.py index 7cc9c33..631c250 100644 --- a/aibox/tools/registry.py +++ b/aibox/tools/registry.py @@ -16,13 +16,18 @@ from common import Logger, constants, i18n from .base import Tool # 工具参数名命中这些关键词时，debug 日志里打码（防密码/密钥进日志文件） -_SENSITIVE_PARAM_KEYS = {"password", "api_key", "token", "secret", "authorization"} +# L11（审计 2026-08-18）：精确匹配漏掉 apiKey/access_token/passwd 等变体， +# 改包含匹配——日志场景下过度打码可接受，漏打码才是问题。 +_SENSITIVE_PARAM_KEYS = ( + "password", "passwd", "api_key", "apikey", "token", "secret", + "authorization", "credential", +) def _sanitize_kwargs(kwargs: Dict) -> Dict: """复制一份参数并把敏感键的值替换为 ***（仅用于日志输出）""" return { - k: ("***" if str(k).lower() in _SENSITIVE_PARAM_KEYS else v) + k: ("***" if any(s in str(k).lower() for s in _SENSITIVE_PARAM_KEYS) else v) for k, v in kwargs.items() } diff --git a/aibox/tools/skill_tools.py b/aibox/tools/skill_tools.py index 39d9407..3f337c9 100644 --- a/aibox/tools/skill_tools.py +++ b/aibox/tools/skill_tools.py @@ -16,6 +16,7 @@ tools.skill_tools - 技能相关工具 import os import queue import re +import shlex from typing import Dict, Optional from common import constants, i18n, Logger @@ -253,10 +254,14 @@ class InstallSkillTool(Tool): skills_dir = constants.SKILLS_DIR # 如果给了具体路径，在说明里带入，否则用占位符 - src = skill_path.strip() if skill_path else "/path/to/your_skill.md" + raw_src = skill_path.strip() if skill_path else "" # 去掉扩展名作为技能名称，再统一补 .md 作为目标文件名 - base = os.path.splitext(os.path.basename(src))[0] if src else "your_skill" + base = os.path.splitext(os.path.basename(raw_src))[0] if raw_src else "your_skill" dst = os.path.join(skills_dir, base + ".md") + # L7（审计 2026-08-18）：src 为 LLM 可控输入，含空格/元字符时 + # 未经引用的命令会被 shell 拆词/注入（指引由子代理经 run_command 执行） + src = shlex.quote(raw_src) if raw_src else "/path/to/your_skill.md" + dst_q = shlex.quote(dst) guide = f"""📦 扩展技能安装指引 {"=" * 55} @@ -282,7 +287,7 @@ class InstallSkillTool(Tool): 调用 create_subagent 派生子代理执行复制命令（目标文件名统一补 .md 扩展名）： - cp {src} {dst} + cp {src} {dst_q} ⚠️ 注意：目标文件名必须带 .md 扩展名， 系统以 .md 文件作为技能加载，且技能名称取文件名（不含 .md）。