fix: TestSubtitleFilter补work_dir参数适配新安全校验

P1-1: 路径校验startswith绕过
- multi_track_mixer: 改用resolved_path.relative_to()替代原始字符串startswith
- concat_engine: 同上
- subtitle_render_engine: 同上
- 原因: /tmp/work/../../opt/secret.aac 原始字符串以work_dir开头，
  旧代码误判为在目录内从而跳过白名单校验

P1-2: build_subtitle_filter work_dir默认值绕过
- work_dir=None时不再用ass_path父目录自证清白
- 不传/None/空字符串一律抛PathSecurityError
- 调用方必须显式传入可信的work_dir

新增4个单测验证绕过漏洞已修复，24个测试全绿

apps/worker/video_processing/concat_engine.py

@@ -24,12 +24,17 @@ from pathlib import Path
 from typing import Any
 
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, probe_video_info, run_ffmpeg
+from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
 
 logger = logging.getLogger(__name__)
 
 
 # ── 常量 ──────────────────────────────────────────────────────────────────────
 
+MAX_CONCAT_SEGMENTS = 50  # 最大拼接段数（安全上限，防止OOM）
+
+ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"}
+
 # concat demuxer 要求一致的参数列表
 CONCAT_DEMUXER_REQUIRED_PARAMS = [
     "codec_name",  # 视频编码
@@ -144,6 +149,50 @@ class ConcatConfig:
         return len([s for s in self.segments if s.video_path])
 
 
+# ── 路径安全校验 ────────────────────────────────────────────────────────────
+
+
+def _validate_video_path(video_path: str, work_dir: Path) -> None:
+    """校验视频文件路径安全性.
+
+    规则：
+    - local:// schema → 必须在 work_dir 内
+    - 相对路径 → 必须在 work_dir 内
+    - 绝对路径 → 必须在允许目录白名单内
+    - 扩展名必须是视频格式
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not video_path or not isinstance(video_path, str):
+        raise PathSecurityError("视频路径不能为空")
+
+    # 本地路径（local:// 或相对路径 / 绝对路径）
+    if video_path.startswith("local://") or not video_path.startswith(("http://", "https://", "oss://")):
+        is_abs = video_path.startswith("/") and not video_path.startswith("local://")
+        resolved_path = safe_resolve_path(
+            video_path,
+            work_dir,
+            allow_outside=is_abs,
+            allowed_extensions=ALLOWED_VIDEO_EXTENSIONS,
+        )
+        # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过）
+        if is_abs:
+            resolved_work_dir = work_dir.resolve()
+            try:
+                resolved_path.relative_to(resolved_work_dir)
+            except ValueError:
+                if not is_in_allowed_dirs(resolved_path):
+                    raise PathSecurityError(f"视频路径不在允许目录内: {video_path[:80]}")
+    # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责）
+    # 但检查扩展名
+    else:
+        path_part = video_path.split("?")[0].split("#")[0]
+        ext = Path(path_part).suffix.lower()
+        if ext and ext not in ALLOWED_VIDEO_EXTENSIONS:
+            raise PathSecurityError(f"不允许的视频文件类型: {ext}")
+
+
 # ── 视频拼接引擎 ──────────────────────────────────────────────────────────────
 
 
@@ -179,6 +228,27 @@ class ConcatEngine:
         if not valid_segments:
             raise ValueError("No valid video segments to concat")
 
+        # ── 安全校验：段数上限 ──
+        if len(valid_segments) > MAX_CONCAT_SEGMENTS:
+            raise ValueError(f"Too many concat segments: {len(valid_segments)} > {MAX_CONCAT_SEGMENTS}")
+
+        # ── 安全校验：所有视频路径白名单校验 ──
+        safe_segments = []
+        for seg in valid_segments:
+            try:
+                _validate_video_path(seg.video_path, self.work_dir)
+                safe_segments.append(seg)
+            except PathSecurityError as e:
+                logger.warning("[concat] skip segment: path security check failed: %s", e)
+
+        if len(safe_segments) != len(valid_segments):
+            valid_segments = safe_segments
+            config.segments = safe_segments
+            logger.info("[concat] %d segments passed security check", len(safe_segments))
+
+        if not valid_segments:
+            raise ValueError("No valid video segments after security check")
+
         if len(valid_segments) == 1:
             # 只有一段，直接复制
             import shutil

apps/worker/video_processing/multi_track_mixer.py

@@ -21,6 +21,7 @@ from pathlib import Path
 from typing import TYPE_CHECKING
 
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
+from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
 
 if TYPE_CHECKING:
     from video_processing.render_audio import RenderContext
@@ -36,6 +37,8 @@ TRACK_TYPE_VOICEOVER = "voiceover"  # 配音（TTS/人声）
 TRACK_TYPE_SFX = "sfx"  # 音效
 TRACK_TYPE_AMBIENT = "ambient"  # 环境音
 
+MAX_AUDIO_TRACKS = 8  # 最大混音轨道数（安全上限，防止资源耗尽）
+
 # 各轨道默认音量（相对主音频）
 DEFAULT_VOLUMES = {
     TRACK_TYPE_MAIN: 1.0,
@@ -153,6 +156,56 @@ class MultiTrackMixConfig:
         return len([t for t in self.tracks if t.enabled and t.audio_path]) > 0
 
 
+# ── 路径安全校验 ────────────────────────────────────────────────────────────
+
+
+ALLOWED_AUDIO_EXTENSIONS = {".mp3", ".wav", ".aac", ".ogg", ".flac", ".m4a", ".wma"}
+
+
+def _validate_audio_path(audio_path: str, work_dir: Path) -> None:
+    """校验音频文件路径安全性.
+
+    规则：
+    - local:// schema → 必须在 work_dir 内
+    - 相对路径 → 必须在 work_dir 内
+    - 绝对路径 → 必须在允许目录白名单内
+    - 扩展名必须是音频格式
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not audio_path or not isinstance(audio_path, str):
+        raise PathSecurityError("音频路径不能为空")
+
+    # 本地路径（local:// 或相对路径）
+    if audio_path.startswith("local://") or not audio_path.startswith(("http://", "https://", "oss://")):
+        is_abs = audio_path.startswith("/") and not audio_path.startswith("local://")
+        resolved_path = safe_resolve_path(
+            audio_path,
+            work_dir,
+            allow_outside=is_abs,
+            allowed_extensions=ALLOWED_AUDIO_EXTENSIONS,
+        )
+        # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过）
+        if is_abs:
+            resolved_work_dir = work_dir.resolve()
+            try:
+                resolved_path.relative_to(resolved_work_dir)
+            except ValueError:
+                if not is_in_allowed_dirs(resolved_path):
+                    raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}")
+    # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责）
+    # 但检查扩展名
+    else:
+        # URL路径，检查扩展名白名单（取 ? 之前的部分）
+        path_part = audio_path.split("?")[0].split("#")[0]
+        from pathlib import Path as _P
+
+        ext = _P(path_part).suffix.lower()
+        if ext and ext not in ALLOWED_AUDIO_EXTENSIONS:
+            raise PathSecurityError(f"不允许的音频文件类型: {ext}")
+
+
 # ── 单轨道预处理 ────────────────────────────────────────────────────────────
 
 
@@ -289,6 +342,39 @@ def mix_multi_track(
     if target_duration <= 0:
         target_duration = 5.0
 
+    # ── 安全校验：轨道数量上限 ──
+    enabled_tracks = [t for t in config.tracks if t.enabled and t.audio_path]
+    if len(enabled_tracks) > MAX_AUDIO_TRACKS:
+        logger.warning(
+            "[multi-track] too many tracks: %d > %d, truncating to max",
+            len(enabled_tracks),
+            MAX_AUDIO_TRACKS,
+        )
+        enabled_tracks = enabled_tracks[:MAX_AUDIO_TRACKS]
+        # 更新 config.tracks 为截断后的列表
+        config.tracks = enabled_tracks
+
+    # ── 安全校验：所有音频路径白名单校验 ──
+    # 主音频路径
+    try:
+        _validate_audio_path(str(main_audio_path), ctx.work_dir)
+    except PathSecurityError as e:
+        logger.error("[multi-track] main audio path security check failed: %s", e)
+        raise
+
+    # 各轨道音频路径
+    valid_tracks = []
+    for track in enabled_tracks:
+        try:
+            _validate_audio_path(track.audio_path, ctx.work_dir)
+            valid_tracks.append(track)
+        except PathSecurityError as e:
+            logger.warning("[multi-track] skip track %s: path security check failed: %s", track.track_id, e)
+
+    if len(valid_tracks) != len(enabled_tracks):
+        config.tracks = valid_tracks
+        logger.info("[multi-track] %d tracks passed security check", len(valid_tracks))
+
     # 收集所有有效轨道（已预处理好的）
     prepared_tracks: list[Path] = []
 

apps/worker/video_processing/path_security.py

@@ -0,0 +1,303 @@
+"""路径安全校验工具 — 路径遍历防护.
+
+统一的文件路径安全校验方案，覆盖所有渲染管线中的路径处理场景：
+- 本地素材路径校验
+- local:// 路径 schema 校验
+- 工作目录内路径安全约束
+- 防止路径遍历攻击 (../)
+
+防护要点：
+1. 所有用户可控路径必须在允许的目录内
+2. 解析符号链接后的真实路径仍需在允许目录内
+3. 禁止空路径、相对路径遍历、绝对路径逃逸
+4. 路径字符限制与规范化
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# 最大路径长度
+MAX_PATH_LENGTH = 4096
+
+# 允许的文件扩展名（渲染相关）
+ALLOWED_MEDIA_EXTENSIONS = {
+    ".mp4",
+    ".mov",
+    ".avi",
+    ".mkv",
+    ".webm",
+    ".flv",
+    ".wmv",  # 视频
+    ".mp3",
+    ".wav",
+    ".aac",
+    ".ogg",
+    ".flac",
+    ".m4a",
+    ".wma",  # 音频
+    ".jpg",
+    ".jpeg",
+    ".png",
+    ".gif",
+    ".bmp",
+    ".webp",
+    ".tiff",  # 图片
+    ".srt",
+    ".ass",
+    ".vtt",
+    ".sub",  # 字幕
+    ".txt",
+    ".json",  # 文本/配置
+}
+
+# local:// schema 前缀
+LOCAL_SCHEMA_PREFIX = "local://"
+
+
+class PathSecurityError(ValueError):
+    """路径安全校验失败."""
+
+    pass
+
+
+def safe_resolve_path(
+    input_path: str | Path,
+    base_dir: str | Path,
+    *,
+    allow_outside: bool = False,
+    allowed_extensions: set[str] | None = None,
+) -> Path:
+    """安全解析路径，确保最终路径在 base_dir 内.
+
+    Args:
+        input_path: 输入路径（相对或绝对）
+        base_dir: 基路径目录，解析后的路径必须在此目录内
+        allow_outside: 是否允许路径在 base_dir 外（默认禁止）
+        allowed_extensions: 允许的文件扩展名集合（None 表示不限制）
+
+    Returns:
+        解析后的绝对路径 Path 对象
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if input_path is None:
+        raise PathSecurityError("路径不能为空")
+
+    path_str = str(input_path).strip()
+    if not path_str:
+        raise PathSecurityError("路径不能为空")
+
+    if len(path_str) > MAX_PATH_LENGTH:
+        raise PathSecurityError(f"路径过长 ({len(path_str)} > {MAX_PATH_LENGTH})")
+
+    # 空字节检测（必须在 Path() 之前）
+    if "\x00" in path_str:
+        raise PathSecurityError("路径包含空字节")
+
+    # 处理 local:// schema
+    if path_str.startswith(LOCAL_SCHEMA_PREFIX):
+        path_str = path_str[len(LOCAL_SCHEMA_PREFIX) :]
+        # local:// 后必须是相对路径（相对于 base_dir），不能是绝对路径
+        if os.path.isabs(path_str):
+            raise PathSecurityError("local:// 路径不能是绝对路径")
+
+    # 规范化 base_dir
+    base_dir = Path(base_dir).resolve()
+    if not base_dir.is_dir():
+        raise PathSecurityError(f"基路径不是有效目录: {base_dir}")
+
+    # 解析输入路径
+    input_path_obj = Path(path_str)
+
+    # 如果是绝对路径且不允许外部路径
+    if input_path_obj.is_absolute() and not allow_outside:
+        raise PathSecurityError("禁止使用绝对路径（需在工作目录内）")
+
+    # 组合并解析为绝对路径
+    if input_path_obj.is_absolute():
+        full_path = input_path_obj.resolve()
+    else:
+        full_path = (base_dir / input_path_obj).resolve()
+
+    # 检查路径遍历 — 确保最终路径在 base_dir 内
+    if not allow_outside:
+        try:
+            full_path.relative_to(base_dir)
+        except ValueError:
+            raise PathSecurityError(f"路径遍历检测：路径 '{path_str}' 超出基路径 '{base_dir}' 范围")
+
+    # 扩展名校验
+    if allowed_extensions is not None:
+        ext = full_path.suffix.lower()
+        if ext and ext not in allowed_extensions:
+            raise PathSecurityError(f"不允许的文件类型: {ext}")
+
+    # 检查危险路径模式
+    _check_dangerous_patterns(full_path)
+
+    return full_path
+
+
+def _check_dangerous_patterns(path: Path) -> None:
+    """检查危险路径模式."""
+    path_str = str(path)
+
+    # 检查空字节
+    if "\x00" in path_str:
+        raise PathSecurityError("路径包含空字节")
+
+    # 检查特殊设备文件（Linux）
+    dangerous_prefixes = [
+        "/proc/",
+        "/sys/",
+        "/dev/",
+        "/etc/passwd",
+        "/etc/shadow",
+        "/root/",
+        "/boot/",
+        "/var/run/",
+    ]
+    for prefix in dangerous_prefixes:
+        if path_str.startswith(prefix):
+            raise PathSecurityError(f"禁止访问系统路径: {prefix}")
+
+
+def is_path_safe(
+    input_path: str | Path,
+    base_dir: str | Path,
+    *,
+    allow_outside: bool = False,
+) -> bool:
+    """便捷函数：检查路径是否安全，不抛异常."""
+    try:
+        safe_resolve_path(input_path, base_dir, allow_outside=allow_outside)
+        return True
+    except PathSecurityError:
+        return False
+
+
+def validate_local_schema_path(
+    schema_path: str,
+    work_dir: str | Path,
+) -> Path:
+    """校验 local:// schema 路径，返回安全的本地路径.
+
+    local:// 路径规则：
+    - 必须以 local:// 开头
+    - 后面必须是相对路径
+    - 最终解析后必须在 work_dir 内
+    - 不允许 ../ 遍历
+
+    Args:
+        schema_path: local:// 开头的路径
+        work_dir: 工作目录
+
+    Returns:
+        解析后的安全路径
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not schema_path.startswith(LOCAL_SCHEMA_PREFIX):
+        raise PathSecurityError(f"路径必须以 {LOCAL_SCHEMA_PREFIX} 开头")
+
+    return safe_resolve_path(schema_path, work_dir, allow_outside=False)
+
+
+def sanitize_filename(filename: str) -> str:
+    """清理文件名，移除危险字符.
+
+    保留：字母、数字、下划线、连字符、点、中文字符
+    移除：路径分隔符、控制字符、特殊符号等
+    """
+    import re
+
+    if not filename:
+        return "unnamed"
+
+    # 移除路径分隔符和危险字符
+    # 保留: 字母数字、中文字符、下划线、连字符、点、空格
+    sanitized = re.sub(r'[\\/\x00-\x1f\x7f<>:"|?*]', "_", filename)
+
+    # 移除开头的点和连续的点（防止隐藏文件和路径遍历）
+    while sanitized.startswith("."):
+        sanitized = sanitized[1:]
+
+    # 限制长度
+    if len(sanitized) > 255:
+        name, ext = os.path.splitext(sanitized)
+        sanitized = name[: 255 - len(ext)] + ext
+
+    # 空文件名兜底
+    if not sanitized or sanitized == ".":
+        sanitized = "unnamed"
+
+    return sanitized
+
+
+# ── 允许目录配置 ──────────────────────────────────────────────────────────────
+
+
+def get_allowed_local_dirs() -> list[Path]:
+    """获取允许的本地素材目录列表（从环境变量读取）.
+
+    环境变量 ASSET_ALLOWED_DIRS，多个目录用冒号分隔（Linux）或分号分隔（Windows）。
+    默认包含 /tmp。
+
+    用于：
+    - resolve_asset_path 本地绝对路径白名单
+    - PiP local_path 类型白名单
+    - 贴纸本地路径白名单
+    """
+    env_dirs = os.environ.get("ASSET_ALLOWED_DIRS", "")
+    dirs: list[Path] = []
+    if env_dirs:
+        import re
+
+        sep = ";" if os.name == "nt" else ":"
+        for d in re.split(f"[{sep}]", env_dirs):
+            d = d.strip()
+            if d:
+                try:
+                    dirs.append(Path(d).resolve())
+                except OSError:
+                    pass
+    # 默认允许 /tmp
+    if not dirs:
+        try:
+            dirs.append(Path("/tmp").resolve())  # nosec B108
+        except OSError:
+            pass
+    return dirs
+
+
+def is_in_allowed_dirs(path: str | Path, allowed_dirs: list[Path] | None = None) -> bool:
+    """检查路径是否在允许的目录列表内.
+
+    Args:
+        path: 待检查的路径
+        allowed_dirs: 允许的目录列表，None 则使用默认配置
+
+    Returns:
+        True 表示在允许目录内
+    """
+    if allowed_dirs is None:
+        allowed_dirs = get_allowed_local_dirs()
+
+    try:
+        resolved = Path(path).resolve()
+        for allowed in allowed_dirs:
+            try:
+                resolved.relative_to(allowed)
+                return True
+            except ValueError:
+                continue
+        return False
+    except OSError:
+        return False

apps/worker/video_processing/subtitle_render_engine.py

@@ -28,6 +28,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
+from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
 from video_processing.render_subtitles import generate_ass_subtitles
 from video_processing.subtitle_generator import generate_ass_from_timeline
 
@@ -36,6 +37,8 @@ logger = logging.getLogger(__name__)
 
 # ── 常量 ──────────────────────────────────────────────────────────────────────
 
+ALLOWED_SUBTITLE_EXTENSIONS = {".srt", ".ass", ".vtt", ".sub"}
+
 # 9宫格位置映射（ASS alignment 编号）
 POSITION_ALIGNMENT = {
     "top_left": 7,
@@ -614,6 +617,7 @@ def build_subtitle_filter(
     *,
     video_input_label: str = "0:v",
     output_label: str = "subtitled",
+    work_dir: Path | str | None = None,
 ) -> str:
     """生成 FFmpeg subtitles 滤镜字符串.
 
@@ -621,13 +625,63 @@ def build_subtitle_filter(
         ass_path: ASS 字幕文件路径
         video_input_label: 视频输入标签（如 "0:v" 或 "[v_out]"）
         output_label: 输出标签
+        work_dir: 工作目录（必填，用于路径安全校验，防止路径遍历绕过）
 
     Returns:
         filter_complex 片段，如 "[0:v]subtitles=xxx.ass[subtitled]"
+
+    Raises:
+        PathSecurityError: 字幕路径不安全或 work_dir 未提供
     """
+    # ── 安全校验：字幕文件路径白名单 ──
+    ass_path_str = str(ass_path)
+    if work_dir is None or not str(work_dir).strip():
+        raise PathSecurityError("work_dir 必须提供，不能为 None 或空")
+
+    _validate_subtitle_path(ass_path_str, Path(work_dir))
+
     # FFmpeg subtitles filter 的路径需要转义：
     # - Windows 路径的 \ → /
     # - 冒号 : → \:
     # - 单引号 ' → '\''
-    safe_path = str(ass_path).replace("\\", "/").replace(":", "\\:").replace("'", "'\\''")
+    safe_path = ass_path_str.replace("\\", "/").replace(":", "\\:").replace("'", "'\\''")
     return f"{video_input_label}subtitles='{safe_path}'[{output_label}]"
+
+
+def _validate_subtitle_path(subtitle_path: str, work_dir: Path) -> None:
+    """校验字幕文件路径安全性.
+
+    规则：
+    - 必须是本地路径（不支持远程URL字幕）
+    - local:// schema → 必须在 work_dir 内
+    - 相对路径 → 必须在 work_dir 内
+    - 绝对路径 → 必须在允许目录白名单内
+    - 扩展名必须是字幕格式
+
+    Raises:
+        PathSecurityError: 路径不安全
+    """
+    if not subtitle_path or not isinstance(subtitle_path, str):
+        raise PathSecurityError("字幕路径不能为空")
+
+    # 不允许远程URL字幕（subtitles滤镜不支持远程加载，且有SSRF风险）
+    if subtitle_path.startswith(("http://", "https://", "oss://")):
+        raise PathSecurityError("不允许使用远程URL字幕文件")
+
+    is_abs = subtitle_path.startswith("/") and not subtitle_path.startswith("local://")
+
+    resolved_path = safe_resolve_path(
+        subtitle_path,
+        work_dir,
+        allow_outside=is_abs,
+        allowed_extensions=ALLOWED_SUBTITLE_EXTENSIONS,
+    )
+
+    # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过）
+    if is_abs:
+        resolved_work_dir = work_dir.resolve()
+        try:
+            resolved_path.relative_to(resolved_work_dir)
+        except ValueError:
+            if not is_in_allowed_dirs(resolved_path):
+                raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}")

packages/shared/url_security.py

@@ -0,0 +1,402 @@
+"""URL 安全校验工具 — SSRF 防护.
+
+统一的外部 URL 安全校验方案，覆盖所有渲染管线和 TTS 中的外部下载场景。
+放在 packages/shared/ 作为单一来源，worker 和 application 层都可引用。
+
+防护要点：
+1. Scheme 白名单：仅允许 http/https
+2. 主机 SSRF 防护：禁止内网 IP、回环地址、链路本地地址、元数据服务
+3. 端口白名单：仅允许 80/443（标准 HTTP/HTTPS）
+4. 域名校验：禁止 IP 直接访问（除非在白名单中）
+5. 重定向防护：手动跟随重定向，每次跳转前重新校验目标 URL
+6. 文件大小限制：流式下载，超过上限立即中断
+7. MIME 类型白名单：可选的内容类型校验
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import logging
+import os
+import socket
+import urllib.error
+import urllib.request
+from urllib.parse import urljoin, urlparse
+
+logger = logging.getLogger(__name__)
+
+# 允许的 URL scheme
+ALLOWED_SCHEMES = {"http", "https"}
+
+# 允许的端口（标准 HTTP/HTTPS）
+ALLOWED_PORTS = {80, 443}
+
+# 可信域名白名单（可根据实际 OSS/CDN 域名配置）
+# 从环境变量读取，格式："oss-cn-hangzhou.aliyuncs.com,cdn.example.com"
+# 默认空表示所有公网域名都允许，但仍会做 SSRF 检查
+TRUSTED_DOMAINS: set[str] = set()
+_env_trusted = os.environ.get("URL_SECURITY_TRUSTED_DOMAINS", "")
+if _env_trusted:
+    TRUSTED_DOMAINS = {d.strip() for d in _env_trusted.split(",") if d.strip()}
+
+# 是否允许 IP 直接访问（默认禁止，防止绕过 DNS 校验）
+ALLOW_DIRECT_IP = os.environ.get("URL_SECURITY_ALLOW_DIRECT_IP", "false").lower() == "true"
+
+# 最大 URL 长度
+MAX_URL_LENGTH = 2048
+
+# 单次下载最大文件大小（默认 200MB）
+DEFAULT_MAX_DOWNLOAD_SIZE = int(os.environ.get("URL_SECURITY_MAX_DOWNLOAD_MB", "200")) * 1024 * 1024
+
+# 允许的音频 MIME 类型白名单
+ALLOWED_AUDIO_MIME_TYPES = {
+    "audio/mpeg",
+    "audio/mp3",
+    "audio/wav",
+    "audio/x-wav",
+    "audio/pcm",
+    "audio/ogg",
+    "audio/opus",
+    "audio/flac",
+    "audio/aac",
+    "audio/m4a",
+    "audio/x-m4a",
+    "audio/mp4",
+    "application/octet-stream",  # 兼容一些 CDN 返回通用类型
+}
+
+# 允许的视频 MIME 类型白名单
+ALLOWED_VIDEO_MIME_TYPES = {
+    "video/mp4",
+    "video/quicktime",
+    "video/x-matroska",
+    "video/webm",
+    "video/avi",
+    "video/x-msvideo",
+    "video/mpeg",
+    "application/octet-stream",
+}
+
+# 允许的图片 MIME 类型白名单
+ALLOWED_IMAGE_MIME_TYPES = {
+    "image/jpeg",
+    "image/png",
+    "image/gif",
+    "image/webp",
+    "image/bmp",
+}
+
+# 下载块大小
+_DOWNLOAD_CHUNK_SIZE = 8192
+
+# 最大重定向次数
+_MAX_REDIRECTS = 5
+
+
+class UrlSecurityError(ValueError):
+    """URL 安全校验失败."""
+
+    pass
+
+
+class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """禁止自动重定向的 handler，用于手动控制重定向以做安全校验."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: N802
+        return None
+
+
+def validate_url_safety(url: str, *, purpose: str = "download") -> str:
+    """校验 URL 安全性，返回标准化后的 URL（供下游使用）.
+
+    Args:
+        url: 待校验的 URL
+        purpose: 用途描述（用于日志），如 "bgm_download"、"tts_download"
+
+    Returns:
+        标准化后的 URL
+
+    Raises:
+        UrlSecurityError: URL 不安全
+    """
+    if not url:
+        raise UrlSecurityError("URL 为空")
+
+    if len(url) > MAX_URL_LENGTH:
+        raise UrlSecurityError(f"URL 过长 ({len(url)} > {MAX_URL_LENGTH})")
+
+    # 解析 URL
+    try:
+        parsed = urlparse(url)
+    except Exception as e:
+        raise UrlSecurityError(f"URL 解析失败: {e}") from e
+
+    # 1. Scheme 校验
+    if not parsed.scheme or parsed.scheme.lower() not in ALLOWED_SCHEMES:
+        raise UrlSecurityError(f"不允许的 URL scheme: {parsed.scheme}")
+
+    # 2. 主机名校验
+    hostname = parsed.hostname
+    if not hostname:
+        raise UrlSecurityError("URL 缺少主机名")
+
+    # 2.1 常见内网主机名前置拦截（防止 DNS rebinding 绕过）
+    _check_internal_hostnames(hostname)
+
+    # 3. 端口校验
+    port = parsed.port
+    if port is not None and port not in ALLOWED_PORTS:
+        raise UrlSecurityError(f"不允许的端口: {port}")
+
+    # 4. SSRF 防护 - 解析 IP 并检查
+    try:
+        # 先判断是否是 IP 地址
+        ip_obj = None
+        try:
+            ip_obj = ipaddress.ip_address(hostname)
+        except ValueError:
+            pass  # 不是 IP，继续走域名解析
+
+        if ip_obj is not None:
+            # 是直接 IP 访问
+            if not ALLOW_DIRECT_IP and not _is_trusted_ip(ip_obj):
+                raise UrlSecurityError(f"禁止直接 IP 访问: {hostname}")
+            _check_ssrf_ip(ip_obj)
+        else:
+            # 域名 — 解析 DNS 检查 SSRF
+            _check_ssrf_domain(hostname)
+    except UrlSecurityError:
+        raise
+    except Exception as e:
+        logger.warning("URL 安全校验异常: url=%s purpose=%s error=%s", url[:80], purpose, e)
+        raise UrlSecurityError(f"URL 安全校验异常: {e}") from e
+
+    # 5. 可信域名校验（如果配置了白名单）
+    if TRUSTED_DOMAINS and not _is_trusted_domain(hostname):
+        raise UrlSecurityError(f"域名不在可信白名单中: {hostname}")
+
+    logger.debug("URL 安全校验通过: url=%s purpose=%s", url[:80], purpose)
+    return url
+
+
+def _check_internal_hostnames(hostname: str) -> None:
+    """前置检查常见内网/敏感主机名，防止 DNS 解析层绕过."""
+    hostname_lower = hostname.lower()
+    internal_hostnames = {
+        "localhost",
+        "localhost.localdomain",
+        "ip6-localhost",
+        "ip6-loopback",
+        "metadata",
+        "metadata.google.internal",
+        "169.254.169.254",  # 云元数据服务
+    }
+    if hostname_lower in internal_hostnames:
+        raise UrlSecurityError(f"禁止访问内部主机名: {hostname}")
+
+    # 检查以 .local / .internal 结尾的主机名
+    if hostname_lower.endswith((".local", ".internal", ".localdomain")):
+        raise UrlSecurityError(f"禁止访问内网域名: {hostname}")
+
+
+def _check_ssrf_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
+    """检查 IP 是否属于 SSRF 风险范围."""
+    # 回环地址
+    if ip_obj.is_loopback:
+        raise UrlSecurityError(f"禁止访问回环地址: {ip_obj}")
+
+    # 私有地址（内网）
+    if ip_obj.is_private:
+        raise UrlSecurityError(f"禁止访问内网地址: {ip_obj}")
+
+    # 链路本地地址
+    if ip_obj.is_link_local:
+        raise UrlSecurityError(f"禁止访问链路本地地址: {ip_obj}")
+
+    # 组播地址
+    if ip_obj.is_multicast:
+        raise UrlSecurityError(f"禁止访问组播地址: {ip_obj}")
+
+    # 未指定地址（0.0.0.0 / ::）
+    if ip_obj.is_unspecified:
+        raise UrlSecurityError(f"禁止访问未指定地址: {ip_obj}")
+
+    # 保留地址
+    if ip_obj.is_reserved:
+        raise UrlSecurityError(f"禁止访问保留地址: {ip_obj}")
+
+
+def _check_ssrf_domain(hostname: str) -> None:
+    """对域名做 DNS 解析并检查所有解析结果的 IP 是否安全.
+
+    注意：这不能完全防止 DNS rebinding，但能防御大部分 SSRF 场景。
+    """
+    try:
+        # 解析所有地址
+        infos = socket.getaddrinfo(hostname, None)
+        if not infos:
+            raise UrlSecurityError(f"域名解析失败: {hostname}")
+
+        for info in infos:
+            ip_str = info[4][0]
+            try:
+                ip_obj = ipaddress.ip_address(ip_str)
+                _check_ssrf_ip(ip_obj)
+            except ValueError:
+                # 无法解析为 IP，跳过（不应该发生）
+                continue
+    except socket.gaierror as e:
+        raise UrlSecurityError(f"域名解析失败: {hostname} ({e})") from e
+
+
+def _is_trusted_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    """检查 IP 是否在可信列表中（目前通过环境变量配置域名，IP 级信任暂不开放）."""
+    return False
+
+
+def _is_trusted_domain(hostname: str) -> bool:
+    """检查域名是否在可信白名单中（支持子域名匹配）."""
+    hostname_lower = hostname.lower()
+    if hostname_lower in TRUSTED_DOMAINS:
+        return True
+    # 检查子域名
+    for domain in TRUSTED_DOMAINS:
+        if hostname_lower.endswith("." + domain.lower()):
+            return True
+    return False
+
+
+def is_url_safe(url: str, *, purpose: str = "download") -> bool:
+    """便捷函数：检查 URL 是否安全，不抛异常."""
+    try:
+        validate_url_safety(url, purpose=purpose)
+        return True
+    except UrlSecurityError:
+        return False
+
+
+# ── 安全下载 ───────────────────────────────────────────────────────────────────
+
+
+def safe_download_file(
+    url: str,
+    dest_path: str,
+    *,
+    purpose: str = "download",
+    max_size: int = DEFAULT_MAX_DOWNLOAD_SIZE,
+    allowed_mime_types: set[str] | None = None,
+    timeout: float = 60.0,
+) -> int:
+    """安全下载 URL 到本地文件。
+
+    包含防护：
+    - SSRF 校验（初始 URL + 每次重定向后都校验）
+    - 重定向次数限制 + 手动跟随（避免重定向绕过 SSRF）
+    - 文件大小限制（流式读取，超过立即中断）
+    - MIME 类型白名单（可选）
+
+    Args:
+        url: 下载 URL
+        dest_path: 目标文件路径
+        purpose: 用途描述（日志用）
+        max_size: 最大下载字节数，超过则中断并抛出 UrlSecurityError
+        allowed_mime_types: 允许的 Content-Type 集合，None 表示不校验
+        timeout: 单次请求超时（秒）
+
+    Returns:
+        实际下载的字节数
+
+    Raises:
+        UrlSecurityError: 安全校验失败
+    """
+    current_url = url
+    redirect_count = 0
+    total_bytes = 0
+
+    # 使用不自动跟随重定向的 opener
+    no_redirect_opener = urllib.request.build_opener(NoRedirectHandler())
+
+    while True:
+        # 每次请求前都做 SSRF 校验（重定向目标也会校验）
+        validate_url_safety(current_url, purpose=purpose)
+
+        req = urllib.request.Request(current_url, method="GET")
+        req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
+
+        try:
+            resp = no_redirect_opener.open(req, timeout=timeout)  # nosec B310
+        except urllib.error.HTTPError as e:
+            # 3xx 重定向
+            if 300 <= e.code < 400 and e.headers.get("Location"):
+                if redirect_count >= _MAX_REDIRECTS:
+                    raise UrlSecurityError(f"重定向次数超过限制 ({_MAX_REDIRECTS})") from e
+                redirect_count += 1
+                current_url = urljoin(current_url, e.headers["Location"])
+                continue
+            raise UrlSecurityError(f"HTTP 错误: {e.code} {e.reason}") from e
+        except urllib.error.URLError as e:
+            raise UrlSecurityError(f"URL 错误: {e.reason}") from e
+
+        try:
+            # Content-Type 校验
+            if allowed_mime_types is not None:
+                content_type = resp.headers.get("Content-Type", "").split(";")[0].strip().lower()
+                if content_type and content_type not in allowed_mime_types:
+                    raise UrlSecurityError(
+                        f"不允许的 Content-Type: {content_type}, " f"允许: {sorted(allowed_mime_types)}"
+                    )
+
+            # Content-Length 预检
+            content_length = resp.headers.get("Content-Length")
+            if content_length and int(content_length) > max_size:
+                raise UrlSecurityError(f"文件过大: {content_length} bytes > {max_size} bytes 上限")
+
+            # 流式下载，实时检查大小
+            with open(dest_path, "wb") as f:
+                while True:
+                    chunk = resp.read(_DOWNLOAD_CHUNK_SIZE)
+                    if not chunk:
+                        break
+                    total_bytes += len(chunk)
+                    if total_bytes > max_size:
+                        raise UrlSecurityError(f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes")
+                    f.write(chunk)
+
+            return total_bytes
+        finally:
+            resp.close()
+
+
+def safe_download_bytes(
+    url: str,
+    *,
+    purpose: str = "download",
+    max_size: int = DEFAULT_MAX_DOWNLOAD_SIZE,
+    allowed_mime_types: set[str] | None = None,
+    timeout: float = 60.0,
+) -> bytes:
+    """安全下载 URL 并返回字节内容。
+
+    防护同 safe_download_file，但结果返回在内存中（适合小文件）。
+    """
+    import tempfile
+
+    fd, tmp_path = tempfile.mkstemp()
+    os.close(fd)
+
+    try:
+        safe_download_file(
+            url,
+            tmp_path,
+            purpose=purpose,
+            max_size=max_size,
+            allowed_mime_types=allowed_mime_types,
+            timeout=timeout,
+        )
+        with open(tmp_path, "rb") as f:
+            return f.read()
+    finally:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass

tests/unit/test_multi_track_subtitle_concat.py

@@ -716,18 +716,29 @@ class TestBuildSubtitlesFromPlan:
 class TestSubtitleFilter:
     """字幕滤镜构建测试."""
 
-    def test_build_subtitle_filter(self):
+    def test_build_subtitle_filter(self, work_dir):
         from video_processing.subtitle_render_engine import build_subtitle_filter
 
-        result = build_subtitle_filter("/tmp/test.ass", video_input_label="[v_in]", output_label="out")
+        ass_file = work_dir / "test.ass"
+        ass_file.write_text("test", encoding="utf-8")
+
+        result = build_subtitle_filter(
+            str(ass_file),
+            video_input_label="[v_in]",
+            output_label="out",
+            work_dir=work_dir,
+        )
         assert "subtitles=" in result
         assert "[v_in]" in result
         assert "[out]" in result
 
-    def test_default_labels(self):
+    def test_default_labels(self, work_dir):
         from video_processing.subtitle_render_engine import build_subtitle_filter
 
-        result = build_subtitle_filter("/tmp/sub.ass")
+        ass_file = work_dir / "sub.ass"
+        ass_file.write_text("test", encoding="utf-8")
+
+        result = build_subtitle_filter(str(ass_file), work_dir=work_dir)
         assert "0:v" in result
         assert "[subtitled]" in result
 

tests/unit/test_pr312_security_debt.py

@@ -0,0 +1,484 @@
+"""PR #312 安全债务修复 单元测试.
+
+测试4个P1安全修复：
+1. 多轨道混音：audio_path 路径安全 + 轨道数量上限
+2. 视频拼接：video_path 路径安全 + 段数上限
+3. 字幕渲染：字幕文件路径白名单校验
+"""
+
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+import pytest
+from video_processing.path_security import PathSecurityError
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def work_dir(tmp_path):
+    return tmp_path
+
+
+@pytest.fixture
+def sample_audio(work_dir):
+    """生成一个测试音频文件."""
+    import subprocess
+
+    path = work_dir / "test.aac"
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=1:sample_rate=44100",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            str(path),
+        ],
+        capture_output=True,
+        check=True,
+        timeout=30,
+    )
+    return path
+
+
+@pytest.fixture
+def sample_video(work_dir):
+    """生成一个测试视频文件."""
+    import subprocess
+
+    path = work_dir / "test.mp4"
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "testsrc=duration=1:size=320x240:rate=30",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=1:sample_rate=44100",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "ultrafast",
+            "-c:a",
+            "aac",
+            "-b:a",
+            "128k",
+            "-shortest",
+            str(path),
+        ],
+        capture_output=True,
+        check=True,
+        timeout=60,
+    )
+    return path
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# 1. 多轨道混音安全测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestMultiTrackSecurity:
+    """多轨道混音安全测试."""
+
+    def test_track_count_limit_exceeded(self, work_dir, sample_audio):
+        """超过最大轨道数时应截断到上限."""
+        from unittest.mock import MagicMock, patch
+
+        from video_processing.multi_track_mixer import (
+            MAX_AUDIO_TRACKS,
+            AudioTrack,
+            MultiTrackMixConfig,
+            mix_multi_track,
+        )
+
+        # 创建超过上限的轨道数
+        tracks = []
+        for i in range(MAX_AUDIO_TRACKS + 5):
+            tracks.append(
+                AudioTrack(
+                    track_id=f"track_{i}",
+                    track_type="sfx",
+                    audio_path=str(sample_audio),
+                    volume=0.5,
+                )
+            )
+
+        config = MultiTrackMixConfig(tracks=tracks)
+
+        ctx = MagicMock()
+        ctx.work_dir = work_dir
+        ctx.plan_id = "test_plan"
+
+        # mock _prepare_single_track 避免实际跑ffmpeg
+        with patch("video_processing.multi_track_mixer._prepare_single_track", return_value=True):
+            with patch("video_processing.multi_track_mixer.run_ffmpeg"):
+                import shutil
+
+                with patch("shutil.copy2"):
+                    result = mix_multi_track(ctx, sample_audio, config, 10.0)
+
+        # 验证轨道被截断到上限
+        assert len(config.tracks) == MAX_AUDIO_TRACKS
+        assert result is not None
+
+    def test_track_count_within_limit(self, work_dir, sample_audio):
+        """轨道数在限制内时正常处理."""
+        from unittest.mock import MagicMock, patch
+
+        from video_processing.multi_track_mixer import (
+            MAX_AUDIO_TRACKS,
+            AudioTrack,
+            MultiTrackMixConfig,
+            mix_multi_track,
+        )
+
+        tracks = []
+        for i in range(3):
+            tracks.append(
+                AudioTrack(
+                    track_id=f"track_{i}",
+                    track_type="sfx",
+                    audio_path=str(sample_audio),
+                    volume=0.5,
+                )
+            )
+
+        config = MultiTrackMixConfig(tracks=tracks)
+
+        ctx = MagicMock()
+        ctx.work_dir = work_dir
+        ctx.plan_id = "test_plan"
+
+        with patch("video_processing.multi_track_mixer._prepare_single_track", return_value=True):
+            with patch("video_processing.multi_track_mixer.run_ffmpeg"):
+                result = mix_multi_track(ctx, sample_audio, config, 10.0)
+
+        assert len(config.tracks) == 3
+        assert result is not None
+
+    def test_audio_path_traversal_attack(self, work_dir, sample_audio):
+        """路径遍历攻击应被拦截."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        # 路径遍历
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path("../../../etc/passwd", work_dir)
+
+        # local:// 路径遍历
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path("local://../../../etc/passwd", work_dir)
+
+    def test_audio_path_allowed_extension(self, work_dir, sample_audio):
+        """允许的音频扩展名应通过校验."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        # 在work_dir内的音频文件
+        test_file = work_dir / "test.mp3"
+        test_file.touch()
+        _validate_audio_path(str(test_file), work_dir)  # 不应抛异常
+
+        test_file2 = work_dir / "test.wav"
+        test_file2.touch()
+        _validate_audio_path(str(test_file2), work_dir)  # 不应抛异常
+
+    def test_audio_path_disallowed_extension(self, work_dir):
+        """不允许的文件扩展名应被拦截."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        test_file = work_dir / "test.exe"
+        test_file.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path(str(test_file), work_dir)
+
+        test_file2 = work_dir / "test.php"
+        test_file2.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path(str(test_file2), work_dir)
+
+    def test_audio_path_empty(self, work_dir):
+        """空路径应被拦截."""
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path("", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_audio_path(None, work_dir)
+
+    def test_audio_path_traversal_bypass_startswith(self, work_dir):
+        """【P1绕过】用../构造伪work_dir前缀路径，真实路径逃逸，必须被拦截.
+
+        漏洞：旧代码用 startswith(str(work_dir)) 比原始字符串，
+        /tmp/work/../../opt/secret.aac 会通过 startswith 检查，跳过白名单校验。
+        修复：用 realpath 规范化后再比较。
+        """
+        from video_processing.multi_track_mixer import _validate_audio_path
+
+        evil_path = str(work_dir / "../../../../opt/secret.aac")
+        with pytest.raises(PathSecurityError, match="不在允许目录"):
+            _validate_audio_path(evil_path, work_dir)
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# 2. 视频拼接安全测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestConcatSecurity:
+    """视频拼接安全测试."""
+
+    def test_segment_count_limit_exceeded(self, work_dir, sample_video):
+        """超过最大段数时应报错."""
+        from video_processing.concat_engine import (
+            MAX_CONCAT_SEGMENTS,
+            ConcatConfig,
+            ConcatEngine,
+            ConcatSegment,
+        )
+
+        # 创建超过上限的段数
+        segments = []
+        for i in range(MAX_CONCAT_SEGMENTS + 5):
+            segments.append(ConcatSegment(video_path=str(sample_video)))
+
+        config = ConcatConfig(segments=segments)
+        engine = ConcatEngine(work_dir)
+        output_path = work_dir / "output.mp4"
+
+        with pytest.raises(ValueError, match="Too many concat segments"):
+            engine.concat_videos(config, output_path)
+
+    def test_segment_count_within_limit(self, work_dir, sample_video):
+        """段数在限制内时正常处理."""
+        from unittest.mock import patch
+
+        from video_processing.concat_engine import (
+            MAX_CONCAT_SEGMENTS,
+            ConcatConfig,
+            ConcatEngine,
+            ConcatSegment,
+        )
+
+        segments = [
+            ConcatSegment(video_path=str(sample_video)),
+            ConcatSegment(video_path=str(sample_video)),
+            ConcatSegment(video_path=str(sample_video)),
+        ]
+
+        config = ConcatConfig(segments=segments)
+        engine = ConcatEngine(work_dir)
+        output_path = work_dir / "output.mp4"
+
+        # mock ffmpeg执行
+        with patch.object(engine, "_concat_filter", return_value=output_path):
+            with patch.object(engine, "_can_use_stream_copy", return_value=False):
+                result = engine.concat_videos(config, output_path)
+
+        assert result == output_path
+
+    def test_video_path_traversal_attack(self, work_dir):
+        """路径遍历攻击应被拦截."""
+        from video_processing.concat_engine import _validate_video_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path("../../../etc/passwd", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path("local://../../../etc/passwd", work_dir)
+
+    def test_video_path_allowed_extension(self, work_dir):
+        """允许的视频扩展名应通过校验."""
+        from video_processing.concat_engine import _validate_video_path
+
+        for ext in [".mp4", ".mov", ".avi", ".mkv", ".webm"]:
+            test_file = work_dir / f"test{ext}"
+            test_file.touch()
+            _validate_video_path(str(test_file), work_dir)  # 不应抛异常
+
+    def test_video_path_disallowed_extension(self, work_dir):
+        """不允许的文件扩展名应被拦截."""
+        from video_processing.concat_engine import _validate_video_path
+
+        test_file = work_dir / "test.exe"
+        test_file.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_video_path(str(test_file), work_dir)
+
+        test_file2 = work_dir / "test.js"
+        test_file2.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_video_path(str(test_file2), work_dir)
+
+    def test_video_path_empty(self, work_dir):
+        """空路径应被拦截."""
+        from video_processing.concat_engine import _validate_video_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path("", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_video_path(None, work_dir)
+
+    def test_video_path_traversal_bypass_startswith(self, work_dir):
+        """【P1绕过】视频路径../遍历绕过startswith检查，必须被拦截.
+
+        漏洞：旧代码用 startswith(str(work_dir)) 比原始字符串，
+        /tmp/work/../../opt/secret.mp4 会通过 startswith 检查，跳过白名单校验。
+        修复：用 realpath 规范化后再比较。
+        """
+        from video_processing.concat_engine import _validate_video_path
+
+        evil_path = str(work_dir / "../../../../opt/secret.mp4")
+        with pytest.raises(PathSecurityError, match="不在允许目录"):
+            _validate_video_path(evil_path, work_dir)
+
+    def test_invalid_segments_skipped(self, work_dir, sample_video):
+        """路径不安全的片段应被跳过."""
+        from unittest.mock import patch
+
+        from video_processing.concat_engine import (
+            ConcatConfig,
+            ConcatEngine,
+            ConcatSegment,
+        )
+
+        segments = [
+            ConcatSegment(video_path=str(sample_video)),
+            ConcatSegment(video_path="../../../etc/passwd"),  # 不安全路径
+            ConcatSegment(video_path=str(sample_video)),
+        ]
+
+        config = ConcatConfig(segments=segments)
+        engine = ConcatEngine(work_dir)
+        output_path = work_dir / "output.mp4"
+
+        with patch.object(engine, "_concat_filter", return_value=output_path):
+            with patch.object(engine, "_can_use_stream_copy", return_value=False):
+                result = engine.concat_videos(config, output_path)
+
+        # 验证只有2个安全片段保留
+        assert len(config.segments) == 2
+        assert result == output_path
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# 3. 字幕渲染安全测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestSubtitleSecurity:
+    """字幕渲染安全测试."""
+
+    def test_subtitle_path_traversal_attack(self, work_dir):
+        """路径遍历攻击应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path("../../../etc/passwd", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path("local://../../../etc/shadow", work_dir)
+
+    def test_subtitle_path_allowed_extension(self, work_dir):
+        """允许的字幕扩展名应通过校验."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        for ext in [".srt", ".ass", ".vtt", ".sub"]:
+            test_file = work_dir / f"test{ext}"
+            test_file.touch()
+            _validate_subtitle_path(str(test_file), work_dir)  # 不应抛异常
+
+    def test_subtitle_path_disallowed_extension(self, work_dir):
+        """不允许的文件扩展名应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        test_file = work_dir / "test.exe"
+        test_file.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path(str(test_file), work_dir)
+
+        test_file2 = work_dir / "test.mp4"
+        test_file2.touch()
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path(str(test_file2), work_dir)
+
+    def test_subtitle_remote_url_blocked(self, work_dir):
+        """远程URL字幕应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        with pytest.raises(PathSecurityError, match="远程URL"):
+            _validate_subtitle_path("http://evil.com/evil.ass", work_dir)
+
+        with pytest.raises(PathSecurityError, match="远程URL"):
+            _validate_subtitle_path("https://evil.com/evil.srt", work_dir)
+
+    def test_subtitle_path_empty(self, work_dir):
+        """空路径应被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path("", work_dir)
+
+        with pytest.raises(PathSecurityError):
+            _validate_subtitle_path(None, work_dir)
+
+    def test_build_filter_with_safe_path(self, work_dir):
+        """安全路径应正常生成滤镜字符串."""
+        from video_processing.subtitle_render_engine import build_subtitle_filter
+
+        ass_file = work_dir / "subtitle.ass"
+        ass_file.write_text("test", encoding="utf-8")
+
+        result = build_subtitle_filter(ass_file, work_dir=work_dir)
+        assert "subtitles=" in result
+        assert "subtitle.ass" in result
+        assert "[subtitled]" in result
+
+    def test_build_filter_with_unsafe_path_raises(self, work_dir):
+        """不安全路径应抛出异常."""
+        from video_processing.subtitle_render_engine import build_subtitle_filter
+
+        with pytest.raises(PathSecurityError):
+            build_subtitle_filter("../../../etc/passwd", work_dir=work_dir)
+
+    def test_build_filter_work_dir_required(self, work_dir):
+        """不传work_dir时必须报错（防止自证清白绕过）."""
+        from video_processing.subtitle_render_engine import build_subtitle_filter
+
+        ass_file = work_dir / "sub.ass"
+        ass_file.write_text("test", encoding="utf-8")
+
+        # 不传 work_dir 必须报错
+        with pytest.raises(PathSecurityError, match="work_dir"):
+            build_subtitle_filter(ass_file)  # type: ignore[call-arg]
+
+        # 传 None 也必须报错
+        with pytest.raises(PathSecurityError, match="work_dir"):
+            build_subtitle_filter(ass_file, work_dir=None)  # type: ignore[arg-type]
+
+        # 传空字符串也必须报错
+        with pytest.raises(PathSecurityError, match="work_dir"):
+            build_subtitle_filter(ass_file, work_dir="")
+
+    def test_subtitle_path_traversal_bypass_startswith(self, work_dir):
+        """【P1绕过】字幕路径../遍历绕过startswith检查，必须被拦截."""
+        from video_processing.subtitle_render_engine import _validate_subtitle_path
+
+        evil_path = str(work_dir / "../../../../opt/secret.srt")
+        with pytest.raises(PathSecurityError, match="不在允许目录"):
+            _validate_subtitle_path(evil_path, work_dir)