diff --git a/app/integrated_app/error_surface.py b/app/integrated_app/error_surface.py new file mode 100644 index 00000000..99656d33 --- /dev/null +++ b/app/integrated_app/error_surface.py @@ -0,0 +1,83 @@ +"""对外错误面:把异常文本压成不含服务端路径、长度受限的提示。 + +Security [D6]: + 路由层把 ``str(exc)`` 原文写进响应体会泄露绝对路径、SQL 片段与堆栈细节。 + ``routes/model.py`` 已有这套脱敏逻辑(S-R6),但它是私有的,其它路由各自 + ``str(exc)``,等于同一类漏洞修了一处、漏了七处。本模块把它提为唯一出口, + 新增路由请调用 :func:`safe_error_message` 而不是 ``str(exc)``。 + +注意:``html.escape`` 只防 XSS,不防信息泄露,不能替代本模块。 +""" + +from __future__ import annotations + +import asyncio +import re + +from .exceptions import ( + EngineSwitchError, + InsufficientVRAMError, + ModelLoadError, + TTSError, +) + +#: 匹配 Windows 绝对路径(``C:\\...``)与 Unix 多级路径(``a/b/c``) +SENSITIVE_PATH_PATTERN = re.compile(r"[A-Za-z]:\\[^\s\"'<>|*?]+|/(?:[^\s\"'<>|*?]+/)+[^\s\"'<>|*?]*") + +#: 对外消息的最大字符数 +ERROR_MESSAGE_MAX_LENGTH = 200 + + +def redact_paths(message: str, max_length: int = ERROR_MESSAGE_MAX_LENGTH) -> str: + """把文本里的文件路径替换为 ``[PATH]`` 并截断长度。 + + 先脱敏再截断:截断可能把路径切成半截,反而留下更难识别的残片。 + + Args: + message: 原始文本(可能来自异常消息或日志行)。 + max_length: 返回消息的最大字符数。 + + Returns: + 脱敏后的文本。 + """ + if not message: + return "" + msg = SENSITIVE_PATH_PATTERN.sub("[PATH]", message) + if len(msg) > max_length: + msg = msg[:max_length] + "..." + return msg + + +def safe_error_message(exc: BaseException | None, max_length: int = ERROR_MESSAGE_MAX_LENGTH) -> str: + """对异常消息脱敏,得到可安全返回给客户端的字符串。 + + Args: + exc: 异常对象;``None`` 表示未知错误。 + max_length: 返回消息的最大字符数。 + + Returns: + 脱敏后的错误消息。 + """ + if exc is None: + return "未知错误" + + # 以下四条领域异常历史上直接返回未脱敏的 str(exc),而这几个类恰恰最常把 + # 文件路径写进消息(模型加载/切换/显存报错都带 model/ 下的路径)。 + if isinstance(exc, InsufficientVRAMError): + return f"显存不足:{redact_paths(str(exc), max_length)}" + if isinstance(exc, EngineSwitchError): + return f"引擎切换失败:{redact_paths(str(exc), max_length)}" + if isinstance(exc, ModelLoadError): + return f"模型加载失败:{redact_paths(str(exc), max_length)}" + if isinstance(exc, TTSError): + return redact_paths(str(exc), max_length) + if isinstance(exc, FileNotFoundError): + return "文件不存在或已被删除" + if isinstance(exc, (asyncio.TimeoutError, TimeoutError)): + return "操作超时,请稍后重试" + if isinstance(exc, PermissionError): + return "权限不足,无法访问所需资源" + if isinstance(exc, OSError): + return f"系统错误:{redact_paths(str(exc), max_length)}" + + return redact_paths(str(exc), max_length) diff --git a/app/integrated_app/generation.py b/app/integrated_app/generation.py index c71b280a..d92c0004 100644 --- a/app/integrated_app/generation.py +++ b/app/integrated_app/generation.py @@ -36,6 +36,7 @@ from .config import SAVE_DIR from .exceptions import AudioProcessingError, ValidationError +from .path_guard import resolve_bare_in_dir from .watermark import WATERMARK_SOURCE_ID logger = logging.getLogger("tts_multimodel") @@ -794,6 +795,12 @@ def preprocess_and_save_temp( 提示用户支持的输入类型。 AudioProcessingError: 当音频读取、重采样或写入磁盘失败时抛出。 """ + # 文件名由调用方给出(表单值或音色名派生):钉死为 SAVE_DIR 内的单段裸名。 + # 少了这一步,os.replace 会把临时文件搬到 SAVE_DIR 之外,等于任意位置写文件。 + out_path = resolve_bare_in_dir(SAVE_DIR, filename) + if out_path is None: + raise ValidationError(f"非法的临时文件名: {filename!r}(只允许 SAVE_DIR 内的裸文件名)") + tmp_p: str | None = None try: # 形态 1:本地文件路径 (str) @@ -848,8 +855,6 @@ def preprocess_and_save_temp( except Exception as exc: raise AudioProcessingError(f"参考音频重采样失败: {exc}") from exc - out_path = os.path.join(SAVE_DIR, filename) - # 写入策略:先落到 SAVE_DIR 内的临时文件,再 os.replace 原子替换 # 保证并发写入时不会读到半截文件 dir_name = os.path.dirname(out_path) diff --git a/app/integrated_app/path_guard.py b/app/integrated_app/path_guard.py new file mode 100644 index 00000000..cec37a63 --- /dev/null +++ b/app/integrated_app/path_guard.py @@ -0,0 +1,95 @@ +"""把外部传入的名字解析为受控目录内的裸文件路径。 + +Security [D6] / py/path-injection: + 仓库里此前有两种"看似安全"的写法,各自都有洞: + + 1. ``os.path.basename(user_input)`` —— 挡得住 ``../`` 遍历,挡不住同目录内的 + 命名混淆(覆盖 ``metadata.json``、写出隐藏文件),而且一旦忘记调用就没有任何防线; + 2. ``realpath(p).startswith(realpath(base))`` 且少了 ``+ os.sep`` —— + ``base`` 为 ``personas`` 时,兄弟目录 ``personas_evil/x.wav`` 同样以 + ``personas`` 开头,判定直接通过。这类前缀比对是假包含。 + + 本模块把"裸文件名 + 真实路径严格前缀"两件事一次做对,供音色/输出目录共用, + 避免每处再写一遍 basename。 +""" + +from __future__ import annotations + +import os +import re + +#: 单个文件名的长度上限(各主流文件系统的单段上限均为 255) +MAX_FILENAME_LENGTH = 255 + +#: 平台分隔符集合:POSIX 下 ``os.sep`` 是 ``/``、Windows 下是 ``\\``, +#: 但两种平台都可能收到另一种分隔符,所以四个都要显式排除。 +_SEPARATORS: frozenset[str] = frozenset(s for s in (os.sep, os.altsep, "/", "\\") if s) + +_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]") + + +def is_bare_filename(value: str) -> bool: + """判断 ``value`` 是否是单段裸文件名(不含目录成分、非隐藏、非 ``.``/``..``)。 + + Args: + value: 待判定的名字(来自请求体、表单或配置文件)。 + + Returns: + 是裸文件名返回 True。 + """ + if not value or len(value) > MAX_FILENAME_LENGTH: + return False + if _CONTROL_CHARS.search(value): + return False + if any(sep in value for sep in _SEPARATORS): + return False + # 以点开头的既包括 "." 与 "..",也包括 `.metadata.json` 这类隐藏文件。 + return not value.startswith(".") + + +def ensure_within_dir(base_dir: str, candidate: str) -> str | None: + """校验 ``candidate`` 解析后确实落在 ``base_dir`` 之内。 + + 接受任意(相对或绝对)路径,返回规范化真实路径或 ``None``。 + 判定用 ``base + os.sep`` 前缀,因此兄弟目录同前缀的情况会被拒绝。 + + Args: + base_dir: 允许的根目录。 + candidate: 待校验路径(可以是绝对路径)。 + + Returns: + 命中时返回真实绝对路径,越界或解析失败返回 ``None``。 + """ + if not candidate: + return None + try: + root = os.path.realpath(base_dir) + real = os.path.realpath(candidate) + except OSError: + return None + if real == root or not real.startswith(root + os.sep): + return None + return real + + +def resolve_bare_in_dir( + base_dir: str, + name: str, + *, + allowed_extensions: frozenset[str] | set[str] | None = None, +) -> str | None: + """把裸文件名解析为 ``base_dir`` 内的真实路径;非法或越界返回 ``None``。 + + Args: + base_dir: 目标目录。 + name: 外部传入的文件名(含扩展名)。 + allowed_extensions: 扩展名白名单(小写含点),``None`` 表示不校验扩展名。 + + Returns: + 合法时返回真实绝对路径,否则 ``None``。 + """ + if not is_bare_filename(name): + return None + if allowed_extensions is not None and os.path.splitext(name)[1].lower() not in allowed_extensions: + return None + return ensure_within_dir(base_dir, os.path.join(base_dir, name)) diff --git a/app/integrated_app/persona_manager.py b/app/integrated_app/persona_manager.py index 7cf5ce55..68a1174a 100644 --- a/app/integrated_app/persona_manager.py +++ b/app/integrated_app/persona_manager.py @@ -43,10 +43,12 @@ _PERSONA_NAME_RE, PERSONA_DIR, ) +from .error_surface import safe_error_message from .exceptions import EngineNotLoadedError from .generation import preprocess_and_save_temp from .model_manager import _model_lock, _persona_embedding_cache from .model_registry import registry +from .path_guard import resolve_bare_in_dir from .persona_metadata import ( PersonaMetadata, load_persona_metadata, @@ -174,10 +176,12 @@ def fn_save_persona( tmp_p: str | None = None try: - wav_path = os.path.join(PERSONA_DIR, f"{name}.wav") - txt_path = os.path.join(PERSONA_DIR, f"{name}.txt") - wav_real = os.path.realpath(wav_path) - if not wav_real.startswith(os.path.realpath(PERSONA_DIR)): + # 裸名 + 真实路径严格前缀(含 os.sep),一次判定 wav/txt 两个写入点。 + # 此前的 startswith(realpath(PERSONA_DIR)) 少了 os.sep,兄弟目录 + # personas_evil/x.wav 会以 personas 前缀通过判定,属假包含。 + wav_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.wav") + txt_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.txt") + if not wav_path or not txt_path: return "❌ 非法路径", False existing = os.path.exists(wav_path) or os.path.exists(txt_path) @@ -248,7 +252,7 @@ def fn_save_persona( with contextlib.suppress(Exception): os.unlink(tmp_p) logger.exception(f"[音色固化] 音色 [{name}] 固化失败") - return f"❌ 固化失败: {str(e)}", False + return f"❌ 固化失败: {safe_error_message(e)}", False def get_persona_list(search_keyword: str = "") -> list[str]: @@ -433,11 +437,14 @@ def load_persona_embedding(name: str) -> Any | None: EngineNotLoadedError: 走到在线计算分支但 VoxCPM2 模型尚未加载时抛出, ``engine`` 属性固定为 ``"voxcpm2"``。 """ - # 与 _save/_delete 一致的第二段防线(realpath 前缀比对)放在函数入口: - # 读路径的调用方各自用 os.path.basename 兜底,收敛到这里后不再依赖调用方自觉, - # 也避免越出 PERSONA_DIR 的 name 命中内存缓存。 - _persona_root = os.path.realpath(PERSONA_DIR) - if not os.path.realpath(os.path.join(_persona_root, f"{name}.wav")).startswith(_persona_root): + # 与 _save/_delete 一致的第二段防线放在函数入口:读路径的调用方各自兜底不可靠, + # 收敛到这里后不再依赖调用方自觉,也避免越出 PERSONA_DIR 的 name 命中内存缓存。 + # 三个后缀分别判定(wav/txt/pt 都是独立落盘点),且带 os.sep 的严格前缀, + # 因此 personas 与 personas_evil 这类同前缀兄弟目录不再被误判为"在目录内"。 + wav_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.wav") + txt_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.txt") + pt_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.pt") + if not wav_path or not txt_path or not pt_path: logger.warning(f"[嵌入加载] 音色名越出 PERSONA_DIR,拒绝加载: {name!r}") return None @@ -445,10 +452,6 @@ def load_persona_embedding(name: str) -> Any | None: if cached is not None: return cached - wav_path = os.path.join(PERSONA_DIR, f"{name}.wav") - txt_path = os.path.join(PERSONA_DIR, f"{name}.txt") - pt_path = os.path.join(PERSONA_DIR, f"{name}.pt") - wav_exists = os.path.exists(wav_path) txt_exists = os.path.exists(txt_path) pt_exists = os.path.exists(pt_path) @@ -595,13 +598,12 @@ def delete_persona(name: str) -> tuple[bool, str]: if not valid: return False, err_msg - wav_path = os.path.join(PERSONA_DIR, f"{name}.wav") - txt_path = os.path.join(PERSONA_DIR, f"{name}.txt") - pt_path = os.path.join(PERSONA_DIR, f"{name}.pt") - meta_path = os.path.join(PERSONA_DIR, f"{name}.metadata.json") - - real_wav = os.path.realpath(wav_path) - if not real_wav.startswith(os.path.realpath(PERSONA_DIR)): + # 四个待删路径逐个判定,判定标准与写入/加载侧共用 path_guard(含 os.sep 前缀)。 + wav_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.wav") + txt_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.txt") + pt_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.pt") + meta_path = resolve_bare_in_dir(PERSONA_DIR, f"{name}.metadata.json") + if wav_path is None or txt_path is None or pt_path is None or meta_path is None: return False, "非法路径" deleted_any = False @@ -613,7 +615,8 @@ def delete_persona(name: str) -> tuple[bool, str]: os.remove(path) deleted_any = True except OSError as e: - errors.append(f"删除 {os.path.basename(path)} 失败: {e}") + # 只回传文件名 + 脱敏后的原因:OSError 原文带绝对路径。 + errors.append(f"删除 {os.path.basename(path)} 失败: {safe_error_message(e)}") if name in _persona_embedding_cache: with contextlib.suppress(Exception): diff --git a/app/integrated_app/persona_metadata.py b/app/integrated_app/persona_metadata.py index b1e75d67..b26c2075 100644 --- a/app/integrated_app/persona_metadata.py +++ b/app/integrated_app/persona_metadata.py @@ -34,6 +34,8 @@ from datetime import datetime from typing import Any +from .path_guard import resolve_bare_in_dir + logger = logging.getLogger("tts_multimodel") @@ -356,7 +358,12 @@ def load_persona_metadata(persona_dir: str, persona_name: str) -> PersonaMetadat PersonaMetadata: 加载到的元数据实例,永远不会返回 None。 """ legacy_shared = os.path.join(persona_dir, "metadata.json") - meta_path = os.path.join(persona_dir, f"{persona_name}.metadata.json") + # 按名拼接的两个路径都是读点:音色名可能来自配置或导入清单,不假定已校验。 + meta_path = resolve_bare_in_dir(persona_dir, f"{persona_name}.metadata.json") + txt_path = resolve_bare_in_dir(persona_dir, f"{persona_name}.txt") + if meta_path is None or txt_path is None: + logger.warning(f"音色名越出目录或不是裸文件名,跳过按名读取元数据: {persona_name!r}") + return PersonaMetadata(name=persona_name) if os.path.exists(meta_path): try: @@ -366,7 +373,6 @@ def load_persona_metadata(persona_dir: str, persona_name: str) -> PersonaMetadat except Exception as e: logger.warning(f"加载 {persona_name} 的 {persona_name}.metadata.json 失败: {e}") - txt_path = os.path.join(persona_dir, f"{persona_name}.txt") if os.path.exists(txt_path): try: with open(txt_path, encoding="utf-8") as f: @@ -399,10 +405,14 @@ def save_persona_metadata(persona_dir: str, persona_name: str, meta: PersonaMeta persona_name: 音色名称,用于生成 .metadata.json / .txt 文件名。 meta: 待保存的元数据实例。 """ - meta_path = os.path.join(persona_dir, f"{persona_name}.metadata.json") + # 写点比读点更需要硬失败:越界的名字会把元数据写到目录外,静默返回等于丢失。 + meta_path = resolve_bare_in_dir(persona_dir, f"{persona_name}.metadata.json") + txt_path = resolve_bare_in_dir(persona_dir, f"{persona_name}.txt") + if meta_path is None or txt_path is None: + raise ValueError(f"非法音色名,拒绝写入元数据: {persona_name!r}") + with open(meta_path, "w", encoding="utf-8") as f: json.dump(meta.to_dict(), f, ensure_ascii=False, indent=2) - txt_path = os.path.join(persona_dir, f"{persona_name}.txt") with open(txt_path, "w", encoding="utf-8") as f: f.write(meta.to_legacy_text()) diff --git a/app/integrated_app/routes/generate/utils.py b/app/integrated_app/routes/generate/utils.py index 22b47c9d..0048b51c 100644 --- a/app/integrated_app/routes/generate/utils.py +++ b/app/integrated_app/routes/generate/utils.py @@ -40,6 +40,7 @@ from ...audio_processing import enhance_audio from ...config import MAX_UPLOAD_SIZE_BYTES, SAVE_DIR, get_config +from ...error_surface import redact_paths from ...exceptions import ( EngineSwitchError, GenerationCancelledError, @@ -51,6 +52,7 @@ from ...history_db import get_history_db from ...model_manager import _time_estimator from ...monitor import get_health_monitor +from ...path_guard import ensure_within_dir, is_bare_filename from ...security.audit import log_audit from ..system import increment_generation, log_operation @@ -74,6 +76,8 @@ # E6-1 SECURITY/ROBUSTNESS: 单次生成硬超时 (秒) — 防止超长文本耗尽信号量池 # 默认 600s (10 分钟),可按硬件调优。生成超时后释放信号量,返回友好错误。 _GENERATION_HARD_TIMEOUT_S: float = float(os.environ.get("TTS_GENERATION_TIMEOUT_S", "600.0")) +# 对外错误消息长度上限:比 error_surface 的默认 200 宽松,留给中文校验提示可读性。 +_CLIENT_MESSAGE_MAX_LENGTH: int = 500 # --------------------------------------------------------------------------- # 生成结果缓存(BACKEND_DESIGN_ASSESSMENT §长期能力建设 T9):相同请求短 TTL 命中, @@ -765,11 +769,11 @@ def _safe_error_msg(exc: BaseException) -> str: 用户可读的错误描述(中文,不含文件路径/堆栈等技术细节)。 """ if isinstance(exc, InsufficientVRAMError): - return f"显存不足:{str(exc)}" + return f"显存不足:{redact_paths(str(exc))}" if isinstance(exc, EngineSwitchError): - return f"引擎切换失败:{str(exc)}" + return f"引擎切换失败:{redact_paths(str(exc))}" if isinstance(exc, TTSError): - return str(exc) + return redact_paths(str(exc)) exc_str: str = str(exc) lowered: str = exc_str.lower() @@ -782,9 +786,9 @@ def _safe_error_msg(exc: BaseException) -> str: if isinstance(exc, RuntimeError): if "cuda" in lowered or "vram" in lowered or "out of memory" in lowered: return "显存不足,请尝试缩短文本、关闭其他GPU程序,或在设置中切换到CPU模式" - return f"运行时错误:{exc_str[:200]}" + return f"运行时错误:{redact_paths(exc_str)}" if isinstance(exc, ValueError): - return f"参数错误:{exc_str[:200]}" + return f"参数错误:{redact_paths(exc_str)}" if isinstance(exc, FileNotFoundError): return "参考音频文件不存在或已被删除,请重新上传" if isinstance(exc, TimeoutError): @@ -1050,6 +1054,10 @@ def _error_html( """ from ...i18n import get_lang, t + # 对外错误面:调用方常把 str(exc) 原样传进来,绝对路径会随响应外泄。 + # 在唯一的渲染出口统一脱敏,比要求每个调用方都记得脱敏可靠。 + error_message = redact_paths(error_message, _CLIENT_MESSAGE_MAX_LENGTH) + lang: str = get_lang(request) # 合并自定义头与内置 toast 头(自定义头优先,避免覆盖 HX-Trigger) merged_headers: dict[str, str] = { @@ -1224,7 +1232,7 @@ async def resolve_persona_ref( Args: request: FastAPI 请求(用于渲染错误 HTML)。 - persona_name: Persona 音色名称(basename)。 + persona_name: Persona 音色名称(必须是裸文件名,不含目录成分)。 Returns: 成功: (wav_path, None) @@ -1235,29 +1243,33 @@ async def resolve_persona_ref( from ...persona_manager import PERSONA_DIR, load_persona_embedding - safe_name: str = os.path.basename(persona_name) + # 音色名必须是单段裸名:此前只靠 os.path.basename 削掉分隔符,既挡不住 + # 同目录内的命名混淆,也让非法输入被静默改写成另一个名字去查。 + if not is_bare_filename(persona_name): + return None, _error_html(request, f"音色名格式不合法: {persona_name}") + + safe_name: str = persona_name persona_data: Any | None = load_persona_embedding(safe_name) - if persona_data is not None: - # 处理不同返回格式(兼容 .pt 缓存嵌入和在线计算) - # 情况 1: 在线计算分支 -> 返回二元组 (wav_path, ref_text) - # 情况 2: .pt 缓存分支 -> 直接返回嵌入对象(此时音频文件必然存在) - wav_path: str | None = None - if isinstance(persona_data, tuple) and len(persona_data) == 2: - wav_path, ref_text = persona_data - elif isinstance(persona_data, (str, os.PathLike)) and os.path.isfile(str(persona_data)): - wav_path = str(persona_data) - else: - # 缓存嵌入对象(张量或其他嵌入数据),wav 文件必然存在 - # (.pt 缓存只在 wav 存在后才会写入) - candidate = os.path.join(PERSONA_DIR, f"{safe_name}.wav") - wav_path = candidate + if persona_data is None: + return None, _error_html(request, f"音色不存在: {safe_name}") - if wav_path and os.path.isfile(wav_path): - return wav_path, None - else: - return None, _error_html(request, f"音色文件不存在: {safe_name}") + # 处理不同返回格式(兼容 .pt 缓存嵌入和在线计算) + # 情况 1: 在线计算分支 -> 返回二元组 (wav_path, ref_text) + # 情况 2: .pt 缓存分支 -> 直接返回嵌入对象(此时音频文件必然存在) + # 情况 3: 其它对象 -> 按音色名回到 PERSONA_DIR 内定位 + wav_path: str | None = None + if isinstance(persona_data, tuple) and len(persona_data) == 2: + wav_path = str(persona_data[0]) + elif isinstance(persona_data, (str, os.PathLike)): + wav_path = str(persona_data) else: - return None, _error_html(request, f"音色不存在: {safe_name}") + wav_path = os.path.join(PERSONA_DIR, f"{safe_name}.wav") + + # 返回值来自引擎侧,不假定它仍在 PERSONA_DIR 内:用同一把包含性尺子再过一次。 + contained = ensure_within_dir(PERSONA_DIR, wav_path or "") + if contained and os.path.isfile(contained): + return contained, None + return None, _error_html(request, f"音色文件不存在: {safe_name}") def pre_validate( diff --git a/app/integrated_app/routes/generate/voxcpm2/design.py b/app/integrated_app/routes/generate/voxcpm2/design.py index 410ff807..db82cffb 100644 --- a/app/integrated_app/routes/generate/voxcpm2/design.py +++ b/app/integrated_app/routes/generate/voxcpm2/design.py @@ -56,6 +56,7 @@ from ....config import get_engine_text_limit from ....model_registry import registry +from ....path_guard import ensure_within_dir, is_bare_filename from ..utils import ( _check_engine_ready, _error_html, @@ -175,28 +176,30 @@ async def generate_voxcpm_design( # ------------------------------------------------------------------ actual_ref_path: str | None = None if persona_name: - from ....persona_manager import load_persona_embedding + from ....persona_manager import PERSONA_DIR, load_persona_embedding - safe_name: str = os.path.basename(persona_name) + # 音色名必须是单段裸名,非法输入不再静默 basename 改写后继续查。 + if not is_bare_filename(persona_name): + return _error_html(request, f"音色名格式不合法: {persona_name}") + + safe_name: str = persona_name persona_data = load_persona_embedding(safe_name) if persona_data is not None: # 兼容不同 .pt 缓存格式: # 新格式:二元组 (wav_path, ref_text) — 在线计算/新缓存分支 # 旧格式:字典 {'items': [嵌入数据]} — 嵌入缓存分支 - wav_path = None + wav_path: str | None = None if isinstance(persona_data, tuple) and len(persona_data) == 2: - wav_path, _ = persona_data - elif isinstance(persona_data, (str, os.PathLike)) and os.path.isfile(str(persona_data)): + wav_path = str(persona_data[0]) + elif isinstance(persona_data, (str, os.PathLike)): wav_path = str(persona_data) else: # 嵌入缓存对象(张量等),wav 文件必然存在 - from ....persona_manager import PERSONA_DIR as _PD - - candidate = os.path.join(_PD, f"{safe_name}.wav") - wav_path = candidate if os.path.isfile(candidate) else None + wav_path = os.path.join(PERSONA_DIR, f"{safe_name}.wav") - if wav_path and os.path.isfile(wav_path): - actual_ref_path = wav_path + contained = ensure_within_dir(PERSONA_DIR, wav_path or "") + if contained and os.path.isfile(contained): + actual_ref_path = contained logger.info(f"[VoxCPM声音设计] 已加载音色 '{safe_name}' 的参考音频") else: return _error_html(request, f"音色文件不存在: {safe_name}") diff --git a/app/integrated_app/routes/generate/voxcpm2/script.py b/app/integrated_app/routes/generate/voxcpm2/script.py index 3d86788f..50dc2fad 100644 --- a/app/integrated_app/routes/generate/voxcpm2/script.py +++ b/app/integrated_app/routes/generate/voxcpm2/script.py @@ -65,6 +65,7 @@ from ....config import MAX_TEXT_LENGTH from ....model_registry import registry +from ....path_guard import ensure_within_dir, is_bare_filename from ..utils import ( _check_engine_ready, _error_html, @@ -171,29 +172,32 @@ async def generate_voxcpm_script( # ------------------------------------------------------------------ persona_map_with_wav: dict[str, str] = {} if persona_names.strip(): - from ....persona_manager import load_persona_embedding + from ....persona_manager import PERSONA_DIR, load_persona_embedding persona_name_list = [n.strip() for n in persona_names.split(",") if n.strip()] for pname in persona_name_list: - safe_name: str = os.path.basename(pname) + # 单个非法角色名只降级跳过(下方局部降级策略),不阻断整份剧本。 + if not is_bare_filename(pname): + logger.warning(f"[VoxCPM剧本工坊] 音色名格式不合法,已跳过: {pname!r}") + continue + safe_name: str = pname persona_data = load_persona_embedding(safe_name) if persona_data is not None: # 兼容不同 .pt 缓存格式: # 新格式:二元组 (wav_path, ref_text);旧格式:dict {'items': [嵌入数据]} - wav_path = None + wav_path: str | None = None if isinstance(persona_data, tuple) and len(persona_data) == 2: - wav_path, _ = persona_data - elif isinstance(persona_data, (str, os.PathLike)) and os.path.isfile(str(persona_data)): + wav_path = str(persona_data[0]) + elif isinstance(persona_data, (str, os.PathLike)): wav_path = str(persona_data) else: # 嵌入缓存对象(张量等),wav 文件必然存在 - from ....persona_manager import PERSONA_DIR as _PD + candidate = os.path.join(PERSONA_DIR, f"{safe_name}.wav") + wav_path = candidate - candidate = os.path.join(_PD, f"{safe_name}.wav") - wav_path = candidate if os.path.isfile(candidate) else None - - if wav_path and os.path.isfile(wav_path): - persona_map_with_wav[safe_name] = wav_path + contained = ensure_within_dir(PERSONA_DIR, wav_path or "") + if contained and os.path.isfile(contained): + persona_map_with_wav[safe_name] = contained logger.info(f"[VoxCPM剧本工坊] 已加载音色 '{safe_name}' 的参考音频") else: # Why:单行失败局部降级策略。 diff --git a/app/integrated_app/routes/model.py b/app/integrated_app/routes/model.py index d2e4c4aa..97f77424 100644 --- a/app/integrated_app/routes/model.py +++ b/app/integrated_app/routes/model.py @@ -38,7 +38,6 @@ import json import logging import os -import re from typing import Any from fastapi import APIRouter, Form, HTTPException, Request @@ -50,6 +49,10 @@ fn_voxcpm_set_lora_enabled, fn_voxcpm_unload_lora, ) +from ..error_surface import ( + redact_paths, + safe_error_message, +) from ..exceptions import ( EngineSwitchError, GenerationError, @@ -80,56 +83,6 @@ logger = logging.getLogger("tts_multimodel.model_routes") -# S-R6: 错误消息脱敏 — 匹配 Windows/Unix 文件路径 -_SENSITIVE_PATH_PATTERN = re.compile(r"[A-Za-z]:\\[^\s\"'<>|*?]+|/(?:[^\s\"'<>|*?]+/)+[^\s\"'<>|*?]*") -_ERROR_MESSAGE_MAX_LENGTH = 200 - - -def _safe_error_message(exc: Exception, max_length: int = _ERROR_MESSAGE_MAX_LENGTH) -> str: - """对错误消息进行脱敏,避免向客户端泄露敏感信息。 - - Security [D6]: - 错误消息可能包含文件路径、SQL 语句、堆栈细节等敏感信息。 - - Args: - exc: 异常对象。 - max_length: 返回消息的最大字符数。 - - Returns: - 脱敏后的错误消息字符串。 - """ - if exc is None: - return "未知错误" - - def _redact(msg: str) -> str: - # 先脱敏再截断:截断可能把路径切成半截,反而留下更难识别的残片。 - return _SENSITIVE_PATH_PATTERN.sub("[PATH]", msg) - - # 以下四条分支历史上直接返回未脱敏的 str(exc),而这几个类恰恰是最常把 - # 文件路径写进消息的(模型加载/切换/显存报错都带 model/ 下的绝对或相对路径)。 - # S-R6 的脱敏只覆盖了 OSError 与兜底分支,等于漏了最容易泄的一类。 - if isinstance(exc, InsufficientVRAMError): - return f"显存不足:{_redact(str(exc))[:max_length]}" - if isinstance(exc, EngineSwitchError): - return f"引擎切换失败:{_redact(str(exc))[:max_length]}" - if isinstance(exc, ModelLoadError): - return f"模型加载失败:{_redact(str(exc))[:max_length]}" - if isinstance(exc, TTSError): - return _redact(str(exc))[:max_length] - if isinstance(exc, FileNotFoundError): - return "文件不存在或已被删除" - if isinstance(exc, (asyncio.TimeoutError, TimeoutError)): - return "操作超时,请稍后重试" - if isinstance(exc, PermissionError): - return "权限不足,无法访问所需资源" - if isinstance(exc, OSError): - return f"系统错误:{_redact(str(exc))[:max_length]}" - - msg = _redact(str(exc)) - if len(msg) > max_length: - msg = msg[:max_length] + "..." - return msg - def _get_vram_used_mb() -> int: """获取当前进程已占用的 GPU 显存(MB)。失败时返回 0。 @@ -284,8 +237,9 @@ def _run_load() -> list: _notify_load("显存不足,正在清理后重试...") free_gpu_memory() continue - _notify_load(last_msg, status="failed", error=last_msg) - safe_msg = _SENSITIVE_PATH_PATTERN.sub("[PATH]", last_msg) + # 进度状态与响应体都是客户端可见面,脱敏要一次做在两者之前。 + safe_msg = redact_paths(last_msg) + _notify_load(safe_msg, status="failed", error=safe_msg) return JSONResponse({"status": "error", "message": safe_msg, "engine": engine}) _notify_load(last_msg, status="completed") log_operation("model", f"{engine} 加载完成") @@ -308,7 +262,7 @@ def _run_load() -> list: continue if isinstance(exc, ImportError): _notify_load("加载失败:模型文件缺失", status="failed", error=str(exc)) - raise ModelLoadError(f"模型文件缺失: {_safe_error_message(exc)}") from exc + raise ModelLoadError(f"模型文件缺失: {safe_error_message(exc)}") from exc if isinstance(exc, RuntimeError) and is_oom_error(exc): _notify_load("显存不足,加载失败", status="failed", error=str(exc)) raise InsufficientVRAMError(str(exc)) from exc @@ -316,9 +270,9 @@ def _run_load() -> list: _notify_load(f"加载失败:{exc}", status="failed", error=str(exc)) raise _notify_load(f"加载异常:{exc}", status="failed", error=str(exc)) - raise GenerationError(f"模型加载异常: {_safe_error_message(exc)}") from exc + raise GenerationError(f"模型加载异常: {safe_error_message(exc)}") from exc - safe_error = _safe_error_message(last_error) if last_error else "unknown error" + safe_error = safe_error_message(last_error) if last_error else "unknown error" logger.error("模型加载在 %d 次重试后失败: %s", MAX_RETRIES, last_error, exc_info=True) _notify_load(f"加载失败(重试{MAX_RETRIES}次后)", status="failed", error=safe_error) return JSONResponse( @@ -328,7 +282,7 @@ def _run_load() -> list: raise except Exception as exc: # noqa: BLE001 logger.error("模型加载失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.post("/unload", summary="卸载模型", description="从 GPU 卸载当前模型,释放显存") @@ -356,7 +310,7 @@ async def unload_model_endpoint(request: Request) -> Response: raise except Exception as exc: # noqa: BLE001 logger.error("模型卸载失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.post("/preload", summary="预加载模型", description="后台触发预加载模型到 GPU") @@ -387,7 +341,7 @@ async def preload_model_endpoint(request: Request) -> Response: raise except Exception as exc: # noqa: BLE001 logger.error("模型预加载失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.get("/preload/status", summary="预加载状态", description="查询模型预加载进度") @@ -402,7 +356,7 @@ async def preload_status_endpoint() -> Response: return JSONResponse({"status": "ok", "preload": status}) except Exception as exc: # noqa: BLE001 logger.error("预加载状态查询失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.post("/switch", summary="切换引擎", description="切换当前激活的 TTS 引擎") @@ -470,7 +424,7 @@ def _run_switch() -> str: rolled_back_engine = registry.current_engine if registry.current_engine else prev_engine rollback_msg = f"已自动回滚到 {rolled_back_engine} 引擎" if rolled_back_engine else "" - safe_err = _safe_error_message(exc) + safe_err = safe_error_message(exc) request.app.state.engine_switch_state = { "active": True, "step": f"切换失败 - {rollback_msg}", @@ -550,7 +504,7 @@ async def lora_load_endpoint(request: Request) -> Response: raise except Exception as exc: # noqa: BLE001 logger.error("LoRA 加载失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.post("/lora/unload", summary="卸载 LoRA", description="卸载当前 LoRA 权重") @@ -577,7 +531,7 @@ async def lora_unload_endpoint(request: Request) -> Response: raise except Exception as exc: # noqa: BLE001 logger.error("LoRA 卸载失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.post("/lora/toggle", summary="切换 LoRA", description="启用或禁用 LoRA 权重") @@ -607,7 +561,7 @@ async def lora_toggle_endpoint(request: Request) -> Response: raise except Exception as exc: # noqa: BLE001 logger.error("LoRA 切换失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.get("/lora/state", summary="LoRA 状态", description="获取当前 LoRA 启用/加载状态") @@ -629,7 +583,7 @@ async def lora_state_endpoint() -> Response: return JSONResponse({"status": "ok", "state": state}) except Exception as exc: # noqa: BLE001 logger.error("LoRA 状态查询失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.get("/lora/list", summary="LoRA 列表", description="列出可用的 LoRA 检查点") @@ -663,7 +617,7 @@ async def lora_list_endpoint() -> Response: return JSONResponse({"status": "ok", "checkpoints": checkpoints}) except Exception as exc: # noqa: BLE001 logger.error("LoRA 列表查询失败: %s", exc, exc_info=True) - return JSONResponse({"status": "error", "message": _safe_error_message(exc)}) + return JSONResponse({"status": "error", "message": safe_error_message(exc)}) @router.get("/download_hints", summary="模型下载提示", description="返回缺失模型的下载命令与链接") diff --git a/app/integrated_app/routes/persona.py b/app/integrated_app/routes/persona.py index 6ee22d2d..384e7a48 100644 --- a/app/integrated_app/routes/persona.py +++ b/app/integrated_app/routes/persona.py @@ -29,8 +29,10 @@ from fastapi.responses import HTMLResponse, JSONResponse from ..config import SAVE_DIR +from ..error_surface import safe_error_message from ..exceptions import PersonaNotFoundError from ..exceptions import ValidationError as TTSValidationError +from ..path_guard import resolve_bare_in_dir from ..persona_manager import ( delete_persona, fn_save_persona, @@ -97,15 +99,11 @@ def _resolve_generated_audio(value: str) -> str | None: Returns: 命中且合法时返回绝对路径,否则 None。 """ - if not value or os.path.basename(value) != value or value in {".", ".."}: + # 判定逻辑收敛到 path_guard:裸文件名 + 扩展名白名单 + 带 os.sep 的真实路径前缀。 + contained = resolve_bare_in_dir(SAVE_DIR, value, allowed_extensions=ALLOWED_AUDIO_EXTENSIONS) + if contained is None: return None - if os.path.splitext(value)[1].lower() not in ALLOWED_AUDIO_EXTENSIONS: - return None - save_root = os.path.realpath(SAVE_DIR) - candidate = os.path.realpath(os.path.join(SAVE_DIR, value)) - if candidate != save_root and not candidate.startswith(save_root + os.sep): - return None - return candidate if os.path.isfile(candidate) else None + return contained if os.path.isfile(contained) else None @router.post( @@ -322,7 +320,7 @@ async def persona_delete(name: str) -> JSONResponse: except OSError as fs_err: logger.error(f"删除 Persona 底层文件失败 name={name}: {fs_err}") return JSONResponse( - {"status": "error", "message": f"删除文件失败: {fs_err}"}, + {"status": "error", "message": f"删除文件失败: {safe_error_message(fs_err)}"}, status_code=400, ) diff --git a/app/integrated_app/routes/system/settings.py b/app/integrated_app/routes/system/settings.py index 37c56f98..ce396371 100644 --- a/app/integrated_app/routes/system/settings.py +++ b/app/integrated_app/routes/system/settings.py @@ -34,6 +34,7 @@ router = APIRouter(prefix="/api/system", tags=["system"]) +from ...error_surface import safe_error_message # noqa: E402 from .gpu import _get_gpu_device, _get_gpu_utilization # noqa: E402 from .logs import log_operation # noqa: E402 @@ -636,7 +637,7 @@ async def update_settings(request: Request) -> dict[str, Any]: logger.error(f"[settings/PUT] 写入 config.yaml 失败: {exc}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=str(exc), + detail=safe_error_message(exc), ) from exc _apply_patch_to_runtime(payload) @@ -699,7 +700,7 @@ def reset_settings() -> dict[str, Any]: except RuntimeError as exc: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=str(exc), + detail=safe_error_message(exc), ) from exc with contextlib.suppress(Exception): @@ -722,7 +723,7 @@ def get_advanced_params() -> dict[str, Any]: return {"status": "ok", "params": params.to_dict() if hasattr(params, "to_dict") else dict(params)} except Exception as exc: # noqa: BLE001 logger.warning(f"获取高级参数失败: {exc}") - return {"status": "error", "message": str(exc), "params": {}} + return {"status": "error", "message": safe_error_message(exc), "params": {}} @router.post("/advanced_params", summary="保存高级生成参数", description="更新并持久化 VoxCPM2 高级参数") @@ -766,7 +767,7 @@ async def save_advanced_params(request: Request) -> dict[str, Any]: raise except Exception as exc: # noqa: BLE001 logger.error(f"保存高级参数失败: {exc}", exc_info=True) - return {"status": "error", "message": str(exc)} + return {"status": "error", "message": safe_error_message(exc)} # --------------------------------------------------------------------------- @@ -799,7 +800,7 @@ async def save_general_settings(request: Request) -> dict[str, Any]: return {"status": "ok", "settings": existing} except (OSError, RuntimeError, TypeError, ValueError) as exc: logger.error(f"保存通用设置失败: {exc}", exc_info=True) - return {"status": "error", "message": str(exc)} + return {"status": "error", "message": safe_error_message(exc)} # --------------------------------------------------------------------------- @@ -816,7 +817,7 @@ async def get_generation_defaults() -> dict[str, Any]: return {"status": "ok", "params": params} except Exception as exc: # noqa: BLE001 logger.warning(f"获取默认生成参数失败: {exc}") - return {"status": "error", "message": str(exc), "params": dict(_DEFAULT_GENERATION_DEFAULTS)} + return {"status": "error", "message": safe_error_message(exc), "params": dict(_DEFAULT_GENERATION_DEFAULTS)} @router.post("/generation_defaults", summary="保存默认生成参数", description="持久化默认生成参数") @@ -849,4 +850,4 @@ async def save_generation_defaults(request: Request) -> dict[str, Any]: return {"status": "ok", "params": existing} except (OSError, RuntimeError, TypeError, ValueError) as exc: logger.error(f"保存默认生成参数失败: {exc}", exc_info=True) - return {"status": "error", "message": str(exc)} + return {"status": "error", "message": safe_error_message(exc)} diff --git a/app/integrated_app/routes/training.py b/app/integrated_app/routes/training.py index 2edbdc0f..1bdb2464 100644 --- a/app/integrated_app/routes/training.py +++ b/app/integrated_app/routes/training.py @@ -42,6 +42,8 @@ from fastapi import APIRouter, Request from fastapi.responses import JSONResponse +from ..error_surface import safe_error_message + try: import yaml except ImportError: @@ -409,8 +411,10 @@ async def start_training(request: Request) -> JSONResponse: train_script: str = os.path.join(project_root, "scripts", "train_voxcpm_finetune.py") if not os.path.isfile(train_script): + # 绝对路径只进日志:响应里带上它等于向客户端披露服务端目录结构。 + logger.error(f"训练脚本不存在: {train_script}") return JSONResponse( - {"status": "error", "message": f"Training script not found: {train_script}"}, + {"status": "error", "message": "Training script not found: scripts/train_voxcpm_finetune.py"}, status_code=400, ) @@ -535,7 +539,7 @@ async def start_training(request: Request) -> JSONResponse: logger.error(f"训练启动失败: {spawn_err}") _training_process = None return JSONResponse( - {"status": "error", "message": f"训练启动失败,请检查配置和日志: {spawn_err}"}, + {"status": "error", "message": f"训练启动失败,请检查配置和日志: {safe_error_message(spawn_err)}"}, status_code=500, ) diff --git a/docs/SECURITY_CODEQL_TRIAGE.md b/docs/SECURITY_CODEQL_TRIAGE.md index 5bb5b214..52e9a57d 100644 --- a/docs/SECURITY_CODEQL_TRIAGE.md +++ b/docs/SECURITY_CODEQL_TRIAGE.md @@ -298,8 +298,173 @@ basename、`tau` 等直接插进 HTML 文本与 4 处属性,`basename()` 不 - **#97(锁集自相矛盾)已按「退」路解决**:main 锁内为 `antlr4 4.9.3`、`mpmath 1.3.0`、 `tokenizers 0.21.0`、`transformers 4.52.1`,且 `pyproject.toml`/`requirements.txt` 同步把 声明收窄成 `transformers>=4.52.1,<4.53`、`tokenizers>=0.21.0,<0.22` —— 声明与锁一致了。 - **仍开放的代价**:注释里写明 4.52.x 带 16 条 transformers 公告(按 api.osv.dev 实测), - 但没看到 pip-audit 对这 16 条的命中/豁免说明,属于"已知未结"。 + **仍开放的代价**:4.52.x 带 16 条 transformers 公告(按 api.osv.dev 实测)。 + ⚠️ 本条原写法「没看到 pip-audit 的命中/豁免说明,属于已知未结」是错的,已在 §8.5 更正: + 豁免与逐条判定在 `security.yml` 与 `docs/SECURITY_DEPENDABOT_TRIAGE.md` §1/§1a/§2/§4, + 并由 `tests/test_dependency_consistency.py` 的 D3/D4 钉住。 - **#99(前端 XSS)**:其中 3 处真问题(文件名进 `innerHTML` 的 `onclick` 属性,只做了 JS 字符串转义)在告警表里已消失(`js/incomplete-sanitization` 整族为 0),说明已被修; 具体修法本轮未复核。 + +--- + +## 8. 第二阶段(2026-09-24):54 条逐条读完后落地 + +### 8.0 先更正开工基线:open 不是 30 条,是 54 条 + +交下来的基线写的是「open 30(`py/stack-trace-exposure` 18 + `py/path-injection` 12)、dismissed 30 / fixed 10」。 +按 §0 的命令实测 `origin/main@2433c10`: + +| 口径 | 交下来的 | 实测 | 差额出处 | +|---|---|---|---| +| open | 30 | **54** | §7.2 判过但**没执行**的 24 条:19 条「已缓解」+ 5 条 `settings.py` 真修 | +| `py/path-injection` open | 12 | **36** | 同上(24 条差额全落在这一族) | +| `py/stack-trace-exposure` open | 18 | 18 | 一致 | +| dismissed | 30 | **48** | 09-20→09-23 三批累计;`js/xss-through-dom` 10 + `overly-large-range` 7 也在里面 | +| fixed | 10 | 10 | 一致 | + +差额不是数据漂移:54 条 open 的 `created_at` 与 `updated_at` 全部等于 `2026-09-10T03:38:29Z`, +说明**没有任何一条是被重扫新开出来的**;而 09-23 16:24 那次扫描(run 35888530990,`2433c10`,success) +确实生效过——它把 `py/reflective-xss` #107/#108 翻成 fixed(PR #153 的真修)。所以「open 54」是当前有效的判定。 + +### 8.1 读的过程中挖到一个真缺陷:假包含 + +§7.2 把 `persona_manager` 那 18 条记成「已缓解」,依据是「白名单 + realpath 前缀比对」。 +逐字读守卫后:**前缀比对少了 `+ os.sep`**。 + +```python +# 旧(persona_manager.py:180 / :440 / :604 三处同一写法) +if not os.path.realpath(p).startswith(os.path.realpath(PERSONA_DIR)): +``` + +`PERSONA_DIR` 为 `.../personas` 时,`.../personas_evil/trap.wav` 同样以 `personas` 开头 → 判定放行。 +可利用路径:`load_persona_embedding` 的入口守卫**没有**白名单(只有这条前缀比对), +所以传 `name="../personas_evil/trap"` 就能把读越界变成实际发生 —— `wav_exists` 用的是 +`os.path.join(PERSONA_DIR, name)`,那个 `../` 会真的解析到兄弟目录。 +`fn_save_persona`/`delete_persona` 因为白名单在前(`_PERSONA_NAME_RE` 不含任何分隔符), +走不到这一步,但那三行守卫本身是假的。 + +现在由 `path_guard.ensure_within_dir` 统一做 `base + os.sep` 前缀判定, +反例与正例都钉在测试里(`test_old_prefix_check_is_the_bug` 断言旧写法确实放行, +`test_sibling_dir_prefix_is_rejected` 断言新写法拒绝)。 + +### 8.2 逐条台账(54 条,全部读完落点函数原文后判定) + +| 告警 | 规则 | 文件:行 | 落点函数 | 判定 | 依据组 | +|---|---|---|---|---|---| +| #10 | `path-injection` | `generation.py:857` | preprocess_and_save_temp | **真修** | A | +| #9 | `path-injection` | `generation.py:857` | preprocess_and_save_temp | **真修** | A | +| #11 | `path-injection` | `generation.py:863` | preprocess_and_save_temp | **真修** | A | +| #22 | `path-injection` | `persona_manager.py:183` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | B | +| #21 | `path-injection` | `persona_manager.py:183` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | B | +| #24 | `path-injection` | `persona_manager.py:202` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | B | +| #23 | `path-injection` | `persona_manager.py:202` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | B | +| #25 | `path-injection` | `persona_manager.py:205` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | C | +| #26 | `path-injection` | `persona_manager.py:207` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | C | +| #31 | `path-injection` | `persona_manager.py:222` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | B | +| #27 | `path-injection` | `persona_manager.py:238` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | C | +| #28 | `path-injection` | `persona_manager.py:240` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | C | +| #29 | `path-injection` | `persona_manager.py:247` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | C | +| #30 | `path-injection` | `persona_manager.py:249` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | C | +| #32 | `path-injection` | `persona_manager.py:452` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | D | +| #33 | `path-injection` | `persona_manager.py:453` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | D | +| #34 | `path-injection` | `persona_manager.py:454` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | D | +| #35 | `path-injection` | `persona_manager.py:458` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | D | +| #36 | `path-injection` | `persona_manager.py:491` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | D | +| #37 | `path-injection` | `persona_manager.py:611` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | E | +| #38 | `path-injection` | `persona_manager.py:613` | fn_save_persona / load_persona_embedding / delete_persona | **真修** | E | +| #15 | `path-injection` | `persona_metadata.py:361` | load/save_persona_metadata | **真修** | F | +| #16 | `path-injection` | `persona_metadata.py:363` | load/save_persona_metadata | **真修** | F | +| #17 | `path-injection` | `persona_metadata.py:370` | load/save_persona_metadata | **真修** | F | +| #18 | `path-injection` | `persona_metadata.py:372` | load/save_persona_metadata | **真修** | F | +| #19 | `path-injection` | `persona_metadata.py:403` | load/save_persona_metadata | **真修** | F | +| #20 | `path-injection` | `persona_metadata.py:407` | load/save_persona_metadata | **真修** | F | +| #51 | `path-injection` | `routes/generate/utils.py:1247` | resolve_persona_ref / _error_html | **真修** | G | +| #52 | `path-injection` | `routes/generate/utils.py:1255` | resolve_persona_ref / _error_html | **真修** | G | +| #6 | `path-injection` | `routes/generate/voxcpm2/design.py:189` | generate_voxcpm_design | **真修** | H | +| #7 | `path-injection` | `routes/generate/voxcpm2/design.py:196` | generate_voxcpm_design | **真修** | H | +| #8 | `path-injection` | `routes/generate/voxcpm2/design.py:198` | generate_voxcpm_design | **真修** | H | +| #39 | `path-injection` | `routes/generate/voxcpm2/script.py:186` | generate_voxcpm_script | **真修** | I | +| #40 | `path-injection` | `routes/generate/voxcpm2/script.py:193` | generate_voxcpm_script | **真修** | I | +| #41 | `path-injection` | `routes/generate/voxcpm2/script.py:195` | generate_voxcpm_script | **真修** | I | +| #14 | `path-injection` | `routes/persona.py:108` | _resolve_generated_audio / persona_delete | **已缓解** | — | +| #90 | `stack-trace-exposure` | `routes/generate/utils.py:829` | resolve_persona_ref / _error_html | **已缓解** | — | +| #91 | `stack-trace-exposure` | `routes/generate/utils.py:1096` | resolve_persona_ref / _error_html | **真修** | N | +| #83 | `stack-trace-exposure` | `routes/generate/voxcpm2/streaming.py:526` | stream_generate SSE | **真修** | N | +| #74 | `stack-trace-exposure` | `routes/persona.py:84` | _resolve_generated_audio / persona_delete | **真修** | L | +| #75 | `stack-trace-exposure` | `routes/persona.py:325` | _resolve_generated_audio / persona_delete | **真修** | M | +| #76 | `stack-trace-exposure` | `routes/persona.py:330` | _resolve_generated_audio / persona_delete | **真修** | L | +| #77 | `stack-trace-exposure` | `routes/persona.py:331` | _resolve_generated_audio / persona_delete | **真修** | L | +| #78 | `stack-trace-exposure` | `routes/system/settings.py:725` | advanced_params / general_settings / generation_defaults | **真修** | K | +| #79 | `stack-trace-exposure` | `routes/system/settings.py:769` | advanced_params / general_settings / generation_defaults | **真修** | K | +| #80 | `stack-trace-exposure` | `routes/system/settings.py:802` | advanced_params / general_settings / generation_defaults | **真修** | K | +| #81 | `stack-trace-exposure` | `routes/system/settings.py:819` | advanced_params / general_settings / generation_defaults | **真修** | K | +| #82 | `stack-trace-exposure` | `routes/system/settings.py:852` | advanced_params / general_settings / generation_defaults | **真修** | K | +| #84 | `stack-trace-exposure` | `routes/training.py:373` | start_training / get_training_log | **误报** | — | +| #85 | `stack-trace-exposure` | `routes/training.py:421` | start_training / get_training_log | **误报** | P | +| #86 | `stack-trace-exposure` | `routes/training.py:426` | start_training / get_training_log | **误报** | P | +| #87 | `stack-trace-exposure` | `routes/training.py:431` | start_training / get_training_log | **误报** | P | +| #88 | `stack-trace-exposure` | `routes/training.py:538` | start_training / get_training_log | **真修** | J | +| #89 | `stack-trace-exposure` | `routes/training.py:601` | start_training / get_training_log | **记录不修** | — | + +合计:**真修 47|误报 4|已缓解 2|记录不修 1**。 + +### 8.3 真修怎么修的(两处统一,不在各写一遍) + +1. **`app/integrated_app/path_guard.py`(新增)** —— `is_bare_filename`(单段裸名:无分隔符、 + 无控制字符、非隐藏、非 `.`/`..`、长度上限)、`ensure_within_dir`(真实路径 + `os.sep` 严格前缀)、 + `resolve_bare_in_dir`(两者合用,可选扩展名白名单)。 + 接入点:`generation.preprocess_and_save_temp`、`persona_manager` 的固化/嵌入加载/删除、 + `persona_metadata` 读写、`routes/generate/utils.resolve_persona_ref`、 + `voxcpm2/design`、`voxcpm2/script`、`routes/persona._resolve_generated_audio`。 + `routes/training._validate_path` 本来就带 `os.sep`,保持不变,作为对照。 +2. **`app/integrated_app/error_surface.py`(新增)** —— 把 `routes/model.py` 私有的 + `_safe_error_message` 提为共用的 `safe_error_message` / `redact_paths`(行为不变, + 四条领域异常分支仍先脱敏再截断)。接入点:`settings.py` 7 处 `str(exc)` + (CodeQL 标了 5 处,同形的 `HTTPException(detail=…)` 639/702 一并改)、 + `routes/persona.py:325`、`training.py:538` 与 `:413`(后者不是异常,但把 + `train_script` 绝对路径原样写进了响应)、`persona_manager` 的两条用户可见消息、 + `routes/generate/utils._safe_error_msg` 六条分支,以及 `_error_html` 的渲染入口 + (一处管住 toast 头 / 模板 / 内联降级三个出口,SSE 的 #83 与 #91 都在这条链上)。 + 顺带修掉 `model.py` 里脱敏**晚了一行**的问题:`_notify_load(last_msg, …)` 把未脱敏的 + 原文写进进度状态,客户端轮询拿到的仍是原文。 + +新增测试:`tests/test_path_guard.py`、`tests/test_error_surface_leaks.py`; +扩写 `tests/test_persona_embedding_load.py`(兄弟目录那一条)、 +`tests/test_error_message_redaction.py`(改指向共用模块)。 +本地:ruff check/format 全绿、`pytest -m "not gpu and not cuda and not integration and not benchmark and not e2e"` +**2114 passed / 12 skipped**、mypy 棘轮 **103→103** 不变。 + +### 8.4 dismiss 的先后次序(不给没落地的证据交差) + +- **现在就按 `false positive` 单独 dismiss 的**:#84、#85、#86、#87(`training.py` + 的 JSON 解析回显与 `_validate_path` 只回显 `user_path`);#90(`_partial_success_html` + 三个插值无异常文本流入);#14(`_resolve_generated_audio` 原本就是正确实现,本 PR 只是提为共用)。 +- **合并后按 `mitigated` 单独 dismiss 的**:47 条真修点位,理由绑定**合并 SHA + 具体守卫行 + 测试名**。 + 这一族清不掉数字是预期内的:CodeQL 不认自定义净化器,`model.py` 那 11 条在 #96 合并、 + 测试也钉住之后仍是 open(§7.1 已记录),所以只能靠绑定证据的逐条交差。 +- **按 `won't fix` 交差并记录口径的**:#89 `get_training_log` —— 端点契约就是回传训练子进程 + stdout(路径与 traceback 是排障需要的东西),本机单用户应用无跨信任边界。**不为清数字而 dismiss。** + +### 8.5 `transformers 4.52.x` 的 16 条公告:账本在哪(更正 §7.3) + +§7.3 里「没看到 pip-audit 对这 16 条的命中/豁免说明,属于已知未结」**是我没翻 `security.yml` 的错话**。 +实际状态: + +- `.github/workflows/security.yml` 的 pip-audit 步骤带 16 个 `--ignore-vuln`, + 号取自 run 35575129704 job 106255145288 的真实表格; +- `docs/SECURITY_DEPENDABOT_TRIAGE.md` §1 表 A1–A16 逐条给可达性判定,§1a 给两道扫描器 + 口径差异(Trivy 因 `CRITICAL,HIGH` + `ignore-unfixed` 只见 3 条,pip-audit 无过滤所以 16 条), + §2 给「为什么不升」(4.57.6 实测让 IndexTTS 2.0/2.5 的 `infer_v2` / `infer_v2_5` 直接 ImportError), + §4 给复点条件,复审截止 **2026-12-31**; +- 同步关系由 `tests/test_dependency_consistency.py` 的 D3/D4 钉住: + 工作流里的号集合 == 分诊表里的号集合,且豁免的号必须真的绑在被扫的版本上。 + 本地实测 13 passed。 + +本轮不新增、不放宽任何豁免:16 个号是 `api.osv.dev` 对 4.52.1 返回的全部公告(24 条记录按 CVE +去重后 16 个,与 §1a 记的 4.52.4 集合一致),新出现的公告仍会让这一步变红。 + +缺的那一件是**跟踪载体**:§4 的复点条件、D3/D4 的机器门禁都在,但 `security.yml` 的注释与本节 +都没有指向一个 open issue,Dependabot 又只会为 A1–A8(GHSA 有记录的那批)开单, +A9–A16 那 8 条 PYSEC-only 的永远不会自己冒出来。要不要为这 16 条开一个显式跟踪 issue +(标题挂 `transformers 4.52.x` + 截止 2026-12-31),留在第二阶段的四态清单里等一句话。 diff --git a/tests/test_error_message_redaction.py b/tests/test_error_message_redaction.py index 8b2203a8..443afaa5 100644 --- a/tests/test_error_message_redaction.py +++ b/tests/test_error_message_redaction.py @@ -1,7 +1,7 @@ -"""`routes/model.py::_safe_error_message` 的脱敏覆盖面。 +"""`error_surface.safe_error_message` 的脱敏覆盖面。 回归动机(2026-09-20 复核 py/stack-trace-exposure 时发现):S-R6 的 -`_SENSITIVE_PATH_PATTERN` 脱敏只用在 `OSError` 与兜底分支上,而 +`SENSITIVE_PATH_PATTERN` 脱敏只用在 `OSError` 与兜底分支上,而 `InsufficientVRAMError` / `EngineSwitchError` / `ModelLoadError` / `TTSError` 四条分支直接返回未脱敏的 `str(exc)` —— 这几个类恰恰是最常把 `model/` 下的 真实路径写进消息的。这里锁住「每一条分支都过脱敏」,防止以后新增分支时又漏。 @@ -15,13 +15,13 @@ _PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(_PROJECT_ROOT / "app")) +from integrated_app.error_surface import safe_error_message # noqa: E402 from integrated_app.exceptions import ( # noqa: E402 EngineSwitchError, InsufficientVRAMError, ModelLoadError, TTSError, ) -from integrated_app.routes.model import _safe_error_message # noqa: E402 POSIX_PATH = "/app/model/VoxCPM2/config.json" WIN_PATH = "C:\\Users\\dev\\model\\IndexTTS-2.5\\config.json" @@ -43,32 +43,32 @@ class TestDomainErrorsAreRedacted: ], ) def test_posix_path_never_escapes(self, exc_cls, label): - out = _safe_error_message(exc_cls(f"读取 {POSIX_PATH} 失败")) + out = safe_error_message(exc_cls(f"读取 {POSIX_PATH} 失败")) assert out.startswith(label) or not label _assert_redacted(out, POSIX_PATH) @pytest.mark.parametrize("exc_cls", [InsufficientVRAMError, EngineSwitchError, ModelLoadError, TTSError]) def test_windows_path_never_escapes(self, exc_cls): - out = _safe_error_message(exc_cls(f"加载失败:{WIN_PATH} 缺失")) + out = safe_error_message(exc_cls(f"加载失败:{WIN_PATH} 缺失")) _assert_redacted(out, WIN_PATH) class TestOtherBranchesUnchanged: def test_file_not_found_is_generic_message(self): - assert _safe_error_message(FileNotFoundError(POSIX_PATH)) == "文件不存在或已被删除" + assert safe_error_message(FileNotFoundError(POSIX_PATH)) == "文件不存在或已被删除" def test_permission_error_is_generic_message(self): - assert "权限不足" in _safe_error_message(PermissionError(POSIX_PATH)) + assert "权限不足" in safe_error_message(PermissionError(POSIX_PATH)) def test_oserror_still_redacts(self): - _assert_redacted(_safe_error_message(OSError(f"IO 失败 {POSIX_PATH}")), POSIX_PATH) + _assert_redacted(safe_error_message(OSError(f"IO 失败 {POSIX_PATH}")), POSIX_PATH) def test_unknown_exception_still_redacts(self): - _assert_redacted(_safe_error_message(RuntimeError(f"boom {POSIX_PATH}")), POSIX_PATH) + _assert_redacted(safe_error_message(RuntimeError(f"boom {POSIX_PATH}")), POSIX_PATH) def test_none_returns_unknown(self): - assert _safe_error_message(None) == "未知错误" + assert safe_error_message(None) == "未知错误" def test_long_message_truncated(self): - out = _safe_error_message(RuntimeError("x" * 900)) + out = safe_error_message(RuntimeError("x" * 900)) assert len(out) <= 203 and out.endswith("...") diff --git a/tests/test_error_surface_leaks.py b/tests/test_error_surface_leaks.py new file mode 100644 index 00000000..9c0acbde --- /dev/null +++ b/tests/test_error_surface_leaks.py @@ -0,0 +1,157 @@ +"""对外错误面不得带出服务端路径(py/stack-trace-exposure 的整改回归)。 + +覆盖 2026-09-24 分诊确认的两类真实外泄: + +1. ``routes/system/settings.py`` —— 7 处把 ``str(exc)`` 原样放进响应 + (CodeQL 标了 725/769/802/819/852 五处,另有 639/702 两处 + ``HTTPException(detail=str(exc))`` 同类); +2. ``persona_manager`` —— 固化失败与逐文件删除失败把 ``OSError`` 原文 + 拼进用户可见消息(Windows 下该原文含绝对路径)。 + +统一走 ``error_surface.safe_error_message`` 后,这里断言的是**行为**: +异常消息里的路径在响应中不再出现,同时可读提示仍在。 +""" + +import sys +from pathlib import Path +from typing import Any + +import pytest + +_PROJECT_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(_PROJECT_ROOT / "app")) + +from integrated_app.exceptions import ( # noqa: E402 + EngineSwitchError, + InsufficientVRAMError, + ModelLoadError, + TTSError, +) + +WIN_PATH = r"C:\Users\dev\TTS_MultiModel\config\general_settings.json" +POSIX_PATH = "/opt/tts/config/general_settings.json" + + +def _assert_no_path(out: str, secret: str) -> None: + assert secret not in out, out + assert "[PATH]" in out or "不存在" in out or "权限" in out, out + + +class TestSettingsErrorSurface: + @pytest.fixture + def settings_mod(self): + from integrated_app.routes.system import settings + + return settings + + async def test_get_generation_defaults_hides_path(self, settings_mod, monkeypatch): + async def _boom(*_a: Any, **_k: Any) -> None: + raise OSError(f"[Errno 13] Permission denied: '{POSIX_PATH}'") + + monkeypatch.setattr(settings_mod, "_load_json_file", _boom) + out = await settings_mod.get_generation_defaults() + + assert out["status"] == "error" + _assert_no_path(out["message"], POSIX_PATH) + + async def test_save_general_settings_hides_path(self, settings_mod, monkeypatch): + async def _boom(*_a: Any, **_k: Any) -> None: + raise OSError(f"[Errno 28] No space left on device: '{WIN_PATH}'") + + class _Req: + async def json(self) -> dict[str, Any]: + return {"language": "zh", "theme": "dark"} + + monkeypatch.setattr(settings_mod, "_save_json_file", _boom) + out = await settings_mod.save_general_settings(_Req()) # type: ignore[arg-type] + + assert out["status"] == "error" + _assert_no_path(out["message"], WIN_PATH) + + def test_reset_settings_http_detail_hides_path(self, settings_mod, monkeypatch): + from fastapi import HTTPException + + def _boom(*_a: Any, **_k: Any) -> None: + raise RuntimeError(f"写入 config.yaml 失败: {WIN_PATH}") + + monkeypatch.setattr(settings_mod, "_save_yaml_raw", _boom) + with pytest.raises(HTTPException) as ei: + settings_mod.reset_settings() + + _assert_no_path(str(ei.value.detail), WIN_PATH) + + +class TestPersonaManagerMessageRedaction: + @pytest.fixture + def persona_env(self, tmp_path, monkeypatch): + from integrated_app import persona_manager as pm + + monkeypatch.setattr(pm, "PERSONA_DIR", str(tmp_path)) + return tmp_path, pm + + def test_delete_failure_message_has_no_path(self, persona_env, monkeypatch): + root, pm = persona_env + (root / "alice.wav").write_bytes(b"RIFF....WAVEfmt ") + + def _locked(path: str) -> None: + raise OSError(32, f"另一个程序正在使用此文件: '{path}'") + + monkeypatch.setattr(pm.os, "remove", _locked) + ok, msg = pm.delete_persona("alice") + + assert ok is False + assert "alice.wav" in msg, "文件名要留下,否则用户不知道该关哪个程序" + _assert_no_path(msg, str(root)) + + def test_save_failure_message_has_no_path(self, persona_env, monkeypatch): + root, pm = persona_env + + def _boom(*_a: Any, **_k: Any) -> None: + raise RuntimeError(f"ffmpeg 退出码 1,输出目录 {POSIX_PATH}") + + monkeypatch.setattr(pm, "preprocess_and_save_temp", _boom) + msg, needs_confirm = pm.fn_save_persona("alice", b"RIFF", "参考文本") + + assert needs_confirm is False + assert "❌" in msg + _assert_no_path(msg, POSIX_PATH) + + +class TestGenerateSideErrorSurface: + """生成侧的 ``_safe_error_msg`` 与 SSE / 错误片段共用同一把尺子。""" + + @pytest.mark.parametrize( + "exc", + [ + TTSError(f"读取 {POSIX_PATH} 失败"), + ModelLoadError(f"权重缺失 {WIN_PATH}"), + InsufficientVRAMError(f"需要 6.2GB,{POSIX_PATH} 所在盘空闲不足"), + EngineSwitchError(f"切换到 voxcpm2 失败:{WIN_PATH}"), + RuntimeError(f"CUDA 初始化失败 at {POSIX_PATH}"), + ValueError(f"参数非法 {WIN_PATH}"), + ], + ) + def test_safe_error_msg_redacts(self, exc: BaseException) -> None: + from integrated_app.routes.generate.utils import _safe_error_msg + + out = _safe_error_msg(exc) + + assert POSIX_PATH not in out + assert WIN_PATH not in out + + def test_error_html_redacts_at_render(self): + """``_error_html`` 是错误文本的统一出口,脱敏挂在渲染入口而非各调用方。""" + from integrated_app.routes.generate import utils as gu + + class _Req: + """只够走通 ``get_lang``;没有 ``app`` 属性,因此渲染落到内联降级分支。""" + + query_params: dict[str, str] = {} + cookies: dict[str, str] = {} + + resp = gu._error_html(_Req(), f"生成失败: 无法读取 {POSIX_PATH}") + body = resp.body.decode("utf-8") + + assert POSIX_PATH not in body + assert "[PATH]" in body + assert POSIX_PATH not in resp.headers.get("HX-Trigger", "") diff --git a/tests/test_path_guard.py b/tests/test_path_guard.py new file mode 100644 index 00000000..600c70bc --- /dev/null +++ b/tests/test_path_guard.py @@ -0,0 +1,168 @@ +"""``path_guard`` 的判定边界,以及三个调用侧对恶意名字的拦截。 + +回归动机(2026-09-24 CodeQL py/path-injection 逐条复核):仓库里此前的两种写法 +各有洞 —— ``os.path.basename`` 只削分隔符,``startswith(realpath(dir))`` 少了 +``os.sep``。后者是假包含:``PERSONA_DIR`` 为 ``personas`` 时,兄弟目录 +``personas_evil`` 里的文件同样以 ``personas`` 开头,旧守卫直接放行。 +本文件把这条边界钉住(``test_old_prefix_check_is_the_bug`` 是反面对照)。 +""" + +import sys +from pathlib import Path + +import pytest + +_PROJECT_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(_PROJECT_ROOT / "app")) + +from integrated_app.path_guard import ( # noqa: E402 + MAX_FILENAME_LENGTH, + ensure_within_dir, + is_bare_filename, + resolve_bare_in_dir, +) + +EVIL_NAMES = [ + "../outside.wav", + "..\\outside.wav", + "sub/inner.wav", + "C:\\Windows\\win.ini", + "/etc/passwd", + "./same.wav", + "", + ".", + "..", + ".hidden.wav", + "nul\x00.wav", + "line\nbreak.wav", + "x" * (MAX_FILENAME_LENGTH + 1), +] + + +class TestIsBareFilename: + def test_legit_names_pass(self): + for name in ("a.wav", "音色_1.mp3", "temp_ref.wav", "metadata.json", "x" * MAX_FILENAME_LENGTH): + assert is_bare_filename(name), name + + @pytest.mark.parametrize("name", EVIL_NAMES) + def test_hostile_names_rejected(self, name): + assert not is_bare_filename(name) + + +class TestEnsureWithinDir: + def test_inside_returns_realpath(self, tmp_path): + root = tmp_path / "personas" + root.mkdir() + target = root / "a.wav" + target.write_bytes(b"RIFF") + + assert ensure_within_dir(str(root), str(root / ".." / "personas" / "a.wav")) == str(target.resolve()) + + def test_outside_returns_none(self, tmp_path): + root = tmp_path / "personas" + root.mkdir() + + assert ensure_within_dir(str(root), str(tmp_path / "secret.wav")) is None + assert ensure_within_dir(str(root), str(root)) is None + assert ensure_within_dir(str(root), "") is None + + def test_old_prefix_check_is_the_bug(self, tmp_path): + """少了 ``os.sep`` 的前缀比对会把兄弟目录判成"目录内"。""" + root = tmp_path / "personas" + root.mkdir() + sibling = tmp_path / "personas_evil" + sibling.mkdir() + target = sibling / "trap.wav" + target.write_bytes(b"RIFF") + + # 反面对照:旧写法确实放行 —— 这就是本文件要拦的那一类。 + assert str(target.resolve()).startswith(str(root.resolve())) + # 新写法拒绝。 + assert ensure_within_dir(str(root), str(target)) is None + + +class TestResolveBareInDir: + def test_bare_name_inside(self, tmp_path): + root = tmp_path / "outputs" + root.mkdir() + (root / "temp_ref.wav").write_bytes(b"RIFF") + + assert resolve_bare_in_dir(str(root), "temp_ref.wav") == str((root / "temp_ref.wav").resolve()) + + def test_extension_whitelist(self, tmp_path): + root = tmp_path / "outputs" + root.mkdir() + + assert resolve_bare_in_dir(str(root), "a.wav", allowed_extensions=frozenset({".wav"})) is not None + assert resolve_bare_in_dir(str(root), "a.exe", allowed_extensions=frozenset({".wav"})) is None + + @pytest.mark.parametrize("name", EVIL_NAMES) + def test_hostile_names_rejected(self, tmp_path, name): + root = tmp_path / "personas" + root.mkdir() + + assert resolve_bare_in_dir(str(root), name) is None + + def test_missing_file_still_resolves(self, tmp_path): + """本模块只做路径判定,不做存在性检查——写路径需要能解析尚未创建的文件。""" + root = tmp_path / "personas" + root.mkdir() + + assert resolve_bare_in_dir(str(root), "brand_new.wav") == str((root / "brand_new.wav").resolve()) + + +class TestGenerationTempFilenameGuard: + def test_hostile_filename_raises_before_io(self): + from integrated_app.exceptions import ValidationError + from integrated_app.generation import preprocess_and_save_temp + + with pytest.raises(ValidationError, match="非法的临时文件名"): + preprocess_and_save_temp(None, "../../outside.wav") + + def test_default_filename_is_accepted(self, tmp_path, monkeypatch): + """默认 ``temp_ref.wav`` 必须仍然通过名字校验(守卫只拦非法名)。""" + from integrated_app import generation + + monkeypatch.setattr(generation, "SAVE_DIR", str(tmp_path)) + with pytest.raises(Exception, match="不支持的音频输入类型"): + generation.preprocess_and_save_temp(None) + + +class TestPersonaManagerContainment: + """三个按名拼路径的入口都要用同一把尺子。""" + + @pytest.fixture + def persona_env(self, tmp_path, monkeypatch): + from integrated_app import persona_manager as pm + + monkeypatch.setattr(pm, "PERSONA_DIR", str(tmp_path)) + monkeypatch.setattr(pm.registry, "is_voxcpm_ready", lambda: False) + return tmp_path, pm + + @pytest.mark.parametrize( + "name", + ["../personas_evil/trap", "..\\personas_evil\\trap", "sub/dir", "C:\\Windows\\win.ini", "/etc/passwd"], + ) + def test_load_rejects_non_bare_name(self, persona_env, name): + root, pm = persona_env + sibling = root.parent / "personas_evil" + sibling.mkdir(exist_ok=True) + (sibling / "trap.wav").write_bytes(b"RIFF....WAVEfmt ") + + assert pm.load_persona_embedding(name) is None + + def test_delete_rejects_non_bare_name(self, persona_env): + root, pm = persona_env + + ok, msg = pm.delete_persona("../outside") + assert ok is False + assert "格式不合法" in msg or "非法路径" in msg + + def test_save_rejects_non_bare_name(self, persona_env): + root, pm = persona_env + + msg, needs_confirm = pm.fn_save_persona("../evil", b"RIFF", "ref") + assert needs_confirm is False + assert "❌" in msg + assert list(root.iterdir()) == [], "非法名不得在 PERSONA_DIR 内留下半截文件" + assert not (root.parent / "evil.wav").exists(), "非法名不得写到 PERSONA_DIR 之外" diff --git a/tests/test_persona_embedding_load.py b/tests/test_persona_embedding_load.py index 58d6f89f..89e924ab 100644 --- a/tests/test_persona_embedding_load.py +++ b/tests/test_persona_embedding_load.py @@ -146,6 +146,20 @@ def test_absolute_name_is_rejected(self, persona_env): assert pm.load_persona_embedding(str(outside)) is None + def test_sibling_dir_prefix_is_rejected(self, persona_env): + """同前缀兄弟目录 ``personas_evil`` 不算「PERSONA_DIR 内」。 + + 入口守卫此前写的是 ``realpath(p).startswith(realpath(PERSONA_DIR))``,少了 + ``os.sep``,于是 ``../personas_evil/trap`` 解析出的路径以 ``personas`` 开头 + 并通过判定 —— 这条就是那个洞的直接回归。 + """ + root, pm = persona_env + sibling = root.parent / (root.name + "_evil") + sibling.mkdir() + (sibling / "trap.wav").write_bytes(b"RIFF....WAVEfmt ") + + assert pm.load_persona_embedding(f"../{sibling.name}/trap") is None + def test_in_dir_name_still_loads(self, persona_env): """守卫不能顺手挡掉正常音色(回归断言,与上面两条互为对照)。""" root, pm = persona_env