From 457bb900fb7e261b05b3c154b9ddc5da3c40e98b Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Wed, 19 Aug 2026 07:01:13 +0800 Subject: [PATCH] fix(aibox): round-5 audit batch K - max_turns clamp, task input validation, XSS hardening, atomic writes, broadcast timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 第五轮审计(J1~J16)全量修复 + 子代理 max_turns 钳制: - 子代理 max_turns 钳制(0/None 回退默认 10,负数钳制 1)+ 14 项回归测试 - J1 任务系统投毒:add_memo 入库校验(时间/间隔/类型)、调度器预检与触发事务 防御、展示路径兜底、register 移入 try、add_user_task 补 repeat_end_time 透传 - J2/J13 前端 XSS:三页面 marked 输出经 DOMPurify 净化、entry1 系统消息/任务 分隔条改 textContent、CDN 脚本补 SRI - J3 FTS 删除换序(先 fts 后内容表,索引不再残留) - J5 原子写推广:新增 common/atomic.py(mkstemp+fsync+os.replace),config/prompt/ history 全部接入;损坏配置备份后回退 - J6 单例初始化加锁(PromptManager/ConfigManager 并发首例不双重加载) - J7 broadcast 并发发送 + wait_for 超时回收僵死连接 - J4/J8/J9~J12/J14~J16:验证异常脱敏、交互输入容错、MemoDatabase WAL+busy_timeout、 memo 批量删除、/new 竞态、skills 引号转义、voice_client 清理、死代码移除、 skill_manager 同名键/help 截断 --- aibox/SKILLS_FORMAT.md | 11 + aibox/aibox.py | 122 +++-- aibox/common/atomic.py | 32 ++ aibox/common/config.py | 43 +- aibox/entry.html | 21 +- aibox/entry1.html | 37 +- aibox/frontend.html | 19 +- aibox/model_config.py | 11 +- aibox/skills/nmapskills.py | 5 +- aibox/skills/screenshot.py | 8 +- aibox/subsystems/conversation_history.py | 9 +- aibox/subsystems/db.py | 131 ++++-- aibox/subsystems/executors.py | 2 + aibox/subsystems/prompt_manager.py | 21 +- aibox/subsystems/skill_manager.py | 27 +- aibox/tests/test_h_fixes.py | 9 +- aibox/tests/test_i_fixes.py | 12 +- aibox/tests/test_k_fixes.py | 540 +++++++++++++++++++++++ aibox/tests/test_model_config.py | 4 +- aibox/tests/test_subagent_max_turns.py | 178 ++++++++ aibox/tools/db_tools.py | 13 +- aibox/voice_client.py | 18 +- 22 files changed, 1147 insertions(+), 126 deletions(-) create mode 100644 aibox/common/atomic.py create mode 100644 aibox/tests/test_k_fixes.py create mode 100644 aibox/tests/test_subagent_max_turns.py diff --git a/aibox/SKILLS_FORMAT.md b/aibox/SKILLS_FORMAT.md index 3eb1732..9625406 100644 --- a/aibox/SKILLS_FORMAT.md +++ b/aibox/SKILLS_FORMAT.md @@ -227,3 +227,14 @@ except ImportError: ``` **`--help` 要完备**。包含参数说明和使用示例两部分。AI 通过阅读 `--help` 输出学会如何调用技能,信息越完整,AI 调用越准确。 + +--- + +## 安全边界(J12,审计 2026-08-19) + +**技能即可信代码**。可执行技能由 AI(可被网页内容/用户消息诱导)以子进程方式运行,参数为不可信输入: + +- **参数必须经 shell 引号转义**:凡技能内部把参数拼进 shell 命令/脚本串(subprocess shell、PowerShell、python-nmap 等内部拼串库),一律先 `shlex.quote`(POSIX)或对应语言转义(PowerShell 单引号翻倍) +- **禁止把凭据嵌入命令行参数**:命令行参数对同机进程可见(`ps`);密码/令牌应经环境变量或配置文件传递 +- **ssh/sftp 类技能**:`AutoAddPolicy` 信任任意主机密钥存在 MITM 风险,仅限可信网络使用;生产环境改用 `known_hosts` 校验 +- 内置技能示例(`skills/`)已按此规范处理;新增技能请遵守同一条例 diff --git a/aibox/aibox.py b/aibox/aibox.py index 98be13d..30f2742 100644 --- a/aibox/aibox.py +++ b/aibox/aibox.py @@ -178,13 +178,20 @@ class TaskScheduler: for memo in pending: try: # 广播任务开始执行 - if self.ws_handler: + if self.ws_handler and self.broadcast_enabled: await self.ws_handler.broadcast_task_status(memo, "executing", i18n.get('task_executing', memo['title'], memo['id'])) # 【关键修复】执行前最后一次验证 + # J1(审计 2026-08-19):格式异常按到期执行(Q5 语义), + # 不再抛 ValueError 致任务既不触发也不完成、每秒重试热循环 if memo.get('reminder_time'): - reminder_time = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) - if reminder_time > now: + try: + reminder_time = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) + except (ValueError, TypeError) as e: + self.logger.warning( + f"任务 #{memo['id']} 提醒时间格式异常,按到期执行: {e}") + reminder_time = None + if reminder_time is not None and reminder_time > now: self.logger.warning(f"任务 #{memo['id']} 的提醒时间 ({memo['reminder_time']}) 大于当前时间 ({now_str}),跳过执行") continue @@ -196,7 +203,7 @@ class TaskScheduler: result = await self.execute_task_as_conversation(memo, memo_db) # 广播任务完成 - if self.ws_handler: + if self.ws_handler and self.broadcast_enabled: await self.ws_handler.broadcast_task_status(memo, "complete", result) executed_tasks.append({ @@ -211,7 +218,7 @@ class TaskScheduler: self.logger.error(i18n.get('task_failed', memo['title'], str(e))) # 广播任务失败 - if self.ws_handler: + if self.ws_handler and self.broadcast_enabled: await self.ws_handler.broadcast_task_status(memo, "failed", str(e)) executed_tasks.append({ @@ -285,17 +292,11 @@ class TaskScheduler: # 获取当前事件循环 loop = asyncio.get_running_loop() - # 创建一个队列来收集完整的响应 - full_response_queue = queue.Queue() - - # 定义输出回调函数 - 直接广播,同时收集 + # 定义输出回调函数 - 直接广播 def output_callback(msg_type, content): - """AI输出的回调函数 - 直接广播,同时收集""" - # 收集完整响应 - if msg_type == "complete": - full_response_queue.put(content) - - # 直接广播给所有客户端 - 必须带上 task_id 和 execution_id + """AI输出的回调函数 - 直接广播""" + # J15(审计 2026-08-19):full_response_queue 收集后从未被消费, + # 属死代码,已移除;广播路径保留 if self.ws_handler and self.broadcast_enabled: # 使用 run_coroutine_threadsafe 在正确的循环中运行异步函数 asyncio.run_coroutine_threadsafe( @@ -420,7 +421,7 @@ class TaskScheduler: executed_tasks = [] for memo in immediate: try: - if self.ws_handler: + if self.ws_handler and self.broadcast_enabled: await self.ws_handler.broadcast_task_status(memo, "executing", i18n.get('immediate_memo_executing', memo['title'])) self.logger.info(f"标记即时任务 #{memo['id']} 为已触发") @@ -428,7 +429,7 @@ class TaskScheduler: result = await self.execute_task_as_conversation(memo, memo_db) - if self.ws_handler: + if self.ws_handler and self.broadcast_enabled: await self.ws_handler.broadcast_task_status(memo, "complete", result) executed_tasks.append({ @@ -442,7 +443,7 @@ class TaskScheduler: except Exception as e: self.logger.error(i18n.get('task_failed', memo['title'], str(e))) - if self.ws_handler: + if self.ws_handler and self.broadcast_enabled: await self.ws_handler.broadcast_task_status(memo, "failed", str(e)) executed_tasks.append({ @@ -703,6 +704,9 @@ def run_full_subagent(role, task, config, logger=None, model=None, _watch_done = threading.Event() # N5:取消观察者线程退出信号(finally 置位) try: max_tokens = max(1, min(int(max_tokens or 2048), constants.DEFAULT_MAX_TOKENS)) + # max_turns 钳制:0/负数此前致首轮即拦截(只剩收尾报告、无任何执行); + # 0/空回退默认值的惯例对齐 max_tokens,下限 1 保证至少执行一个任务轮 + max_turns = max(1, int(max_turns or 10)) temp = temperature if temperature is not None else constants.DEFAULT_TEMPERATURE # 创建子代理:完整 DeepSeekChat 实例,role 作为 system_prompt @@ -2099,6 +2103,8 @@ class WebSocketHandler: # 排队等锁,多连接并发 /new 可耗尽默认池(阻塞新消息处理) self._cmd_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="ws-cmd") + # J7(审计 2026-08-19):broadcast 单客户端发送超时(半开 TCP 防卡死) + self.broadcast_send_timeout = 5.0 def shutdown(self): """I 批:释放命令专用线程池(服务退出时调用)""" @@ -2269,13 +2275,30 @@ class WebSocketHandler: self.logger.error(f"发送消息失败: {e}") async def broadcast(self, message): - """广播消息到所有客户端""" - for websocket in self.connected_clients.copy(): + """广播消息到所有客户端。 + + J7(审计 2026-08-19):此前串行 await 且 send_message 吞掉全部异常, + broadcast 的 except→unregister 为死代码——ping 禁用(ping_interval=None) + 下半开 TCP 无法探测,backpressure 下 websocket.send 可无限期阻塞, + 卡死广播与调度器循环。现改为每客户端并发发送 + wait_for 超时, + 超时即 unregister 回收(discard 幂等)。 + """ + clients = list(self.connected_clients) + if not clients: + return + + async def _send_one(websocket): try: - await self.send_message(websocket, message) + await asyncio.wait_for( + self.send_message(websocket, message), + timeout=self.broadcast_send_timeout) except Exception: - # N13:不再裸 except——吞掉 CancelledError 会破坏任务取消语义 + # 超时/发送异常:回收该连接(取消事件被 wait_for 传播, + # CancelledError 为 BaseException 不在此列,不影响任务取消语义) await self.unregister(websocket) + + await asyncio.gather(*(_send_one(ws) for ws in clients), + return_exceptions=True) async def broadcast_task_status(self, task: Dict, status: str, message: str = None): """广播任务状态""" @@ -2379,16 +2402,26 @@ class WebSocketHandler: else: reminder_time = (datetime.now() + timedelta(seconds=delay)).strftime(constants.DATETIME_FORMAT) - memo_id = self.ai_manager.memo_db.add_user_task( - title=title, - content=content, - reminder_time=reminder_time, - repeat_type=repeat_type, - repeat_interval=repeat_interval, - repeat_interval_unit=repeat_interval_unit, - repeat_end_time=repeat_end_time, - is_immediate=is_immediate - ) + # J1(审计 2026-08-19):add_memo 入库校验抛 ValueError(畸形时间/ + # 重复间隔等),此处转为友好错误消息,不让 handle_message 顶层 + # 兜底以笼统的"处理失败"收场 + try: + memo_id = self.ai_manager.memo_db.add_user_task( + title=title, + content=content, + reminder_time=reminder_time, + repeat_type=repeat_type, + repeat_interval=repeat_interval, + repeat_interval_unit=repeat_interval_unit, + repeat_end_time=repeat_end_time, + is_immediate=is_immediate + ) + except ValueError as e: + await self.send_message(websocket, { + "type": "error", + "content": f"任务添加失败: {e}" + }) + return immediate_msg = i18n.get('immediate_memo_added') if is_immediate else "" repeat_msg = "" @@ -2437,6 +2470,12 @@ class WebSocketHandler: "type": "error", "content": i18n.get('memo_complete_failed', memo_id) }) + else: + # J15(审计 2026-08-19):未知 action 此前静默吞掉,客户端无任何反馈 + await self.send_message(websocket, { + "type": "error", + "content": f"未知的任务操作: {action}" + }) async def handle_command(self, websocket, data): """处理命令消息""" @@ -2467,6 +2506,14 @@ class WebSocketHandler: await asyncio.sleep(0.1) if not ai.is_busy(): break + # J11(审计 2026-08-19):取消后旧回合可能仍在收尾(回发内容), + # 且新消息的 think 线程可能与 reset 无序抢 _msg_lock(RLock 无 + # 公平性)——等待该连接处理标志清零后再排队 reset,避免新回合 + # 在旧上下文上执行后被 reset 抹掉(消息"消失") + for _ in range(20): + if not session or not session.get("is_processing"): + break + await asyncio.sleep(0.1) # 重置对话状态,重新加载系统提示词 # N3:reset_conversation 内部同步获取 _msg_lock(繁忙对话持锁 # 可达数十秒),直接调用会冻结整个事件循环(所有连接的心跳/ @@ -2716,7 +2763,7 @@ class WebSocketHandler: # 定义输出回调函数 - 实时发送 def output_callback(msg_type, content): """AI输出的回调函数 - 实时发送每个chunk""" - self.logger.info(f"输出回调被调用: {msg_type}, 内容长度: {len(content)}") + self.logger.debug(f"输出回调被调用: {msg_type}, 内容长度: {len(content)}") # 创建发送消息的协程 async def send_msg(): @@ -2938,8 +2985,11 @@ class EnhancedAIManager: async def handle_connection(self, websocket, path): """处理WebSocket连接 - 消息接收与处理解耦,确保中断信号能被即时处理""" self.logger.info(f"新连接: {websocket.remote_address}, path: {path}") - await self.ws_handler.register(websocket) try: + # J1(审计 2026-08-19):register 移入 try——此前在 try 外,握手期 + # 异常(如任务展示路径解析失败)逃逸时 finally 不执行,会话登记 + # 永久泄漏;unregister 为 discard 幂等,提前失败也会被正常清理 + await self.ws_handler.register(websocket) async for message in websocket: # 【关键修复】用create_task异步处理消息,不await阻塞接收循环 # 这样当AI在后台生成响应时,interrupt消息仍能被立即接收和处理 @@ -3307,9 +3357,9 @@ def handle_once_mode(manager, args) -> bool: if state["exit_code"] == 0 and not state["interrupted"] and not state["timed_out"]: if response is None or (isinstance(response, str) and not response.strip()): state["exit_code"] = 2 - except KeyboardInterrupt: - state["interrupted"] = True except Exception as e: + # J15(审计 2026-08-19):except KeyboardInterrupt 为死代码——SIGINT + # 已由自定义 handler 接管(_sigint_handler),KeyboardInterrupt 永不抛出 print(f"{i18n.get('error_prefix')}{e}", file=sys.stderr) state["exit_code"] = 2 finally: diff --git a/aibox/common/atomic.py b/aibox/common/atomic.py new file mode 100644 index 0000000..9d790b8 --- /dev/null +++ b/aibox/common/atomic.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +""" +common.atomic - 原子写文件工具(J5,审计 2026-08-19) + +同目录 mkstemp 临时文件 + fsync + os.replace:进程中断/断电不留半截文件; +mkstemp 随机文件名避免多进程并发写同路径互踩(conversation_history 旧实现 +的固定 .tmp 名在并发场景会互相覆盖)。仅依赖标准库,供 common/subsystems +各写路径复用。 +""" + +import os +import tempfile + + +def atomic_write_text(path: str, content: str) -> None: + """原子写文本文件。失败时清理临时文件并抛出原异常。""" + path = os.path.expanduser(path) + directory = os.path.dirname(path) or "." + os.makedirs(directory, exist_ok=True) + fd, tmp_path = tempfile.mkstemp(prefix=".aibox_tmp_", dir=directory) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_path, path) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + pass + raise diff --git a/aibox/common/config.py b/aibox/common/config.py index d293eea..5f88672 100644 --- a/aibox/common/config.py +++ b/aibox/common/config.py @@ -9,8 +9,12 @@ common.config - aibox 配置管理器 import os import json import copy +import shutil +import threading +from datetime import datetime from typing import Dict, Optional +from .atomic import atomic_write_text from .i18n import i18n class ConfigManager: """配置管理器 - 单例模式,负责加载和保存配置""" @@ -18,15 +22,20 @@ class ConfigManager: _instance = None _config = None _config_path = None + _instance_lock = threading.Lock() def __new__(cls): - if cls._instance is None: - cls._instance = super().__new__(cls) + # J6(审计 2026-08-19):此前单例创建与初始化均无锁——多线程首例 + # 并发可双实例、双重加载配置 + with cls._instance_lock: + if cls._instance is None: + cls._instance = super().__new__(cls) return cls._instance def __init__(self): - if self._config is None: - self._load_config() + with self._instance_lock: + if self._config is None: + self._load_config() def _get_default_config_path(self) -> str: env_config = os.environ.get("AI_CONFIG_PATH") @@ -183,6 +192,10 @@ class ConfigManager: print(i18n.get('config_loaded', config_path)) except Exception as e: print(f"⚠️ {i18n.get('error_prefix')}{e},使用默认配置") + # J5(审计 2026-08-19):损坏文件先备份再继续——此前直接改用 + # 默认配置,任一后续保存会用"默认+本次改动"整体覆盖原文件, + # proxy/api/context 等全部手工配置永久丢失 + self._backup_corrupted_config(config_path) self._config = self._get_default_config() else: print(i18n.get('config_not_found', config_path)) @@ -190,6 +203,15 @@ class ConfigManager: self._save_default_config(config_path) self._migrate_legacy_proxy_password() + def _backup_corrupted_config(self, config_path: str): + """J5:损坏配置先复制备份(带时间戳),防止后续保存整体覆盖丢失""" + try: + backup = f"{config_path}.corrupt.{datetime.now().strftime('%Y%m%d_%H%M%S')}" + shutil.copy2(config_path, backup) + print(f"⚠️ 已备份损坏的配置文件到 {backup}") + except Exception as e: + print(f"⚠️ 备份损坏配置文件失败: {e}") + def _merge_configs(self, default: Dict, user: Dict) -> Dict: result = default.copy() for key, value in user.items(): @@ -201,9 +223,9 @@ class ConfigManager: def _save_default_config(self, config_path: str): try: - os.makedirs(os.path.dirname(config_path), exist_ok=True) - with open(config_path, 'w', encoding='utf-8') as f: - json.dump(self._get_default_config(), f, ensure_ascii=False, indent=2) + atomic_write_text( + config_path, + json.dumps(self._get_default_config(), ensure_ascii=False, indent=2)) print(i18n.get('config_created', config_path)) except: pass @@ -296,9 +318,10 @@ class ConfigManager: 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) + # J5:原子写(mkstemp + fsync + os.replace),崩溃不留半截 JSON + atomic_write_text( + config_path, + json.dumps(safe_config, ensure_ascii=False, indent=2)) print(f"✅ 配置已保存到 {config_path}") return True except Exception as e: diff --git a/aibox/entry.html b/aibox/entry.html index f5aa68f..43863f6 100644 --- a/aibox/entry.html +++ b/aibox/entry.html @@ -6,10 +6,12 @@ AI智能助手 - Socket.IO客户端 - - + + + + - +