Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 83 additions & 0 deletions app/integrated_app/error_surface.py
Original file line number Diff line number Diff line change
@@ -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)
9 changes: 7 additions & 2 deletions app/integrated_app/generation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
95 changes: 95 additions & 0 deletions app/integrated_app/path_guard.py
Original file line number Diff line number Diff line change
@@ -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))
47 changes: 25 additions & 22 deletions app/integrated_app/persona_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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]:
Expand Down Expand Up @@ -433,22 +437,21 @@ 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

cached = _persona_embedding_cache.get(name)
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)
Expand Down Expand Up @@ -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
Expand All @@ -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):
Expand Down
18 changes: 14 additions & 4 deletions app/integrated_app/persona_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")


Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand Down Expand Up @@ -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())
Loading
Loading