fix(aibox): batch H audit fixes - subagent cancel propagation, scheduler timeout, exec process-group reaping

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
This commit is contained in:
qing committed 2026-08-18 02:33:17 +00:00
1 parent d1f780e381
commit d3acccd146
14 files changed
+643 -79

No files matched your search

+55 -12
View File
@@ -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)
+20 -15
View File
@@ -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:
"""获取代理配置的详细信息(用于显示,不回传任何密码值)"""
+16 -4
View File
@@ -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}")
+2 -1
View File
@@ -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:
+15 -5
View File
@@ -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:
+69 -18
View File
@@ -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
+14 -14
View File
@@ -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:
+408
View File
@@ -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
+16
View File
@@ -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)
+3 -1
View File
@@ -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}"
+5 -2
View File
@@ -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你现在可以询问用户。"
)
+5 -2
View File
@@ -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)
+7 -2
View File
@@ -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()
}
+8 -3
View File
@@ -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)。