Merge branch 'develop' into fix/tech-debt-security-round2

- unified_render_service: 采纳PR方案FFMPEG_BIN常量，保留str(e)[:200]截断
- generation.py: 合并两边docstring（基础SSRF校验+重定向每跳校验

apps/worker/video_processing/ffmpeg_utils.py

@@ -1,23 +1,28 @@
-"""FFmpeg 工具函数 — 共享原语.
+"""FFmpeg 工具函数 — Worker 层.
 
-提供 FFmpeg / FFprobe 调用、视频信息探测、视频标准化、xfade 转场滤镜构建
-等底层能力，供 UnifiedRenderService、VideoComposeService 等复用。
+业务相关的滤镜构建、视频探测、视频标准化等能力放在这里；
+底层原语（run_ffmpeg / 二进制路径 / 默认超时）已下沉到 packages/shared/ffmpeg_utils.py，
+本模块 re-export 保持向后兼容。
 """
 
 from __future__ import annotations
 
 import logging
-import shutil
 import subprocess  # nosec B404
 from pathlib import Path
 from typing import Any
 
+# 底层原语从 shared 层导入，application 层和 worker 层共用同一份实现
+from shared.ffmpeg_utils import (  # noqa: F401
+    DEFAULT_FFMPEG_TIMEOUT,
+    FFMPEG_BIN,
+    FFPROBE_BIN,
+    run_ffmpeg,
+)
+
 logger = logging.getLogger(__name__)
 
-# ── 常量 ──────────────────────────────────────────────────────────────────────
-
-FFMPEG_BIN: str = shutil.which("ffmpeg") or "ffmpeg"
-FFPROBE_BIN: str = shutil.which("ffprobe") or "ffprobe"
+# ── 常量（Worker 层业务相关） ────────────────────────────────────────────────
 
 DEFAULT_OUTPUT_WIDTH = 1280
 DEFAULT_OUTPUT_HEIGHT = 720
@@ -61,62 +66,8 @@ XFADE_TRANSITION_MAP: dict[str, str] = {
 
 DEFAULT_TRANSITION_DURATION = 0.5
 
-# FFmpeg 执行默认超时（秒），防止 FFmpeg hang 住导致 worker 永久阻塞
-# 默认 30 分钟，足够处理大部分短视频渲染；超长视频可单独传参覆盖
-DEFAULT_FFMPEG_TIMEOUT = 1800
 
-
-# ── FFmpeg 执行 ───────────────────────────────────────────────────────────────
-
-
-def run_ffmpeg(
-    command: list[str],
-    *,
-    capture_output: bool = True,
-    timeout: int | None = DEFAULT_FFMPEG_TIMEOUT,
-) -> tuple[str, str]:
-    """执行 FFmpeg 命令。
-
-    Args:
-        command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身）
-        capture_output: 是否捕获 stdout/stderr
-        timeout: 超时时间（秒），默认 1800s（30分钟）；None 表示不设超时（不推荐）
-
-    Returns:
-        (stdout, stderr) 元组
-
-    Raises:
-        subprocess.CalledProcessError: 命令执行失败时抛出，
-            异常信息包含完整 stderr 以便排查。
-        subprocess.TimeoutExpired: 超时未完成时抛出，FFmpeg 进程会被 kill。
-    """
-    try:
-        result = subprocess.run(  # nosec B603
-            command,
-            check=True,
-            stdout=subprocess.PIPE if capture_output else None,
-            stderr=subprocess.PIPE if capture_output else None,
-            text=True,
-            timeout=timeout,
-        )
-        return (result.stdout or "", result.stderr or "")
-    except subprocess.TimeoutExpired:
-        logger.error(
-            "FFmpeg 命令超时 (%ds): command=%s",
-            timeout or -1,
-            " ".join(str(c) for c in command[:20]),
-        )
-        raise
-    except subprocess.CalledProcessError as e:
-        # 把完整 stderr 打到日志，方便排查 exit code 183 等问题
-        stderr_text = (e.stderr or "").strip()
-        logger.error(
-            "FFmpeg 命令失败: exit_code=%d command=%s\nstderr:\n%s",
-            e.returncode,
-            " ".join(str(c) for c in command[:20]),  # 截断过长的命令
-            stderr_text[:5000],  # 截断过长的 stderr
-        )
-        raise
+# ── FFprobe 探测 ──────────────────────────────────────────────────────────────
 
 
 def run_ffprobe(

apps/worker/video_processing/unified_render_service.py

@@ -640,11 +640,9 @@ class UnifiedRenderService:
         return timeline
 
     def _extract_audio(self, video_path: Path, output_path: Path) -> None:
-        """从视频中提取音频为16kHz单声道wav（ASR友好格式）."""
-        from video_processing.ffmpeg_utils import run_ffmpeg
-
+        """从视频中提取音频为16kHz单声道wav（ASR友好格式）。"""
         cmd = [
-            "ffmpeg",
+            FFMPEG_BIN,
             "-y",
             "-i",
             str(video_path),

apps/worker/worker_app/tasks/generation.py

@@ -433,24 +433,34 @@ def _prepare_bgm_track(
     return None
 
 
-def _verify_url_accessible(url: str, timeout: float = 10.0, retries: int = 2) -> bool:
+def _verify_url_accessible(
+    url: str,
+    timeout: float = 10.0,
+    retries: int = 2,
+    max_redirects: int = 5,
+) -> bool:
     """HEAD 请求校验 URL 可访问（含重试，防止 OSS 抖动误报）。
 
-    安全：
+    安全增强：
     - 请求前先做 SSRF 安全校验（内网IP/回环地址/链路本地地址等）
     - scheme 仅允许 http/https
     - 端口仅允许 80/443
+    - 手动跟随重定向，每一跳 URL 都做 SSRF 校验，避免重定向到内网地址绕过
 
     Args:
         url: 待校验的 URL
         timeout: 单次请求超时时间（秒）
         retries: 最大重试次数（默认 2 次，首次失败后间隔 1s 重试）
+        max_redirects: 最大重定向次数（默认 5 次）
 
     Returns:
         True 表示 URL 可访问（HTTP 2xx/3xx），False 表示所有尝试均失败或安全校验不通过。
     """
     import time
     import urllib.request
+    from urllib.parse import urljoin
+
+    from shared.url_security import UrlSecurityError, validate_url_safety
 
     from video_processing.url_security import UrlSecurityError, validate_url_safety
 
@@ -462,14 +472,56 @@ def _verify_url_accessible(url: str, timeout: float = 10.0, retries: int = 2) ->
         return False
 
     last_error: Exception | None = None
-    for attempt in range(1 + retries):
-        try:
-            req = urllib.request.Request(url, method="HEAD")
+
+    def _do_verify(current_url: str) -> bool:
+        """单次校验：手动跟随重定向，每跳都做 SSRF 检查."""
+        redirect_count = 0
+        url_being_checked = current_url
+
+        # 禁止自动重定向的 handler，手动控制每一跳
+        class NoRedirect(urllib.request.HTTPRedirectHandler):
+            def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: N802
+                return None
+
+        opener = urllib.request.build_opener(NoRedirect())
+
+        while redirect_count <= max_redirects:
+            # 每一跳都做 SSRF 安全校验
+            try:
+                safe_url = validate_url_safety(url_being_checked, purpose="url_verify")
+            except UrlSecurityError as e:
+                logger.warning(
+                    "URL校验跳转地址不安全: redirect=%d url=%s error=%s",
+                    redirect_count,
+                    url_being_checked,
+                    e,
+                )
+                raise
+
+            req = urllib.request.Request(safe_url, method="HEAD")
             req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
-            with urllib.request.urlopen(req, timeout=timeout) as resp:  # nosec B310
+
+            with opener.open(req, timeout=timeout) as resp:  # noqa: S310
+                if 200 <= resp.status < 300:
+                    return True
+                if resp.status in (301, 302, 303, 307, 308):
+                    location = resp.headers.get("Location", "")
+                    if not location:
+                        raise Exception(f"HTTP {resp.status} 但无 Location 头")
+                    # 相对路径转绝对
+                    url_being_checked = urljoin(safe_url, location)
+                    redirect_count += 1
+                    continue
                 if resp.status < 400:
                     return True
-                last_error = Exception(f"HTTP {resp.status}")
+                raise Exception(f"HTTP {resp.status}")
+
+        raise Exception(f"重定向次数超过上限 ({max_redirects})")
+
+    for attempt in range(1 + retries):
+        try:
+            if _do_verify(url):
+                return True
         except Exception as e:
             last_error = e
 

packages/application/tts_job/audio_merger.py

@@ -8,8 +8,10 @@ from __future__ import annotations
 import logging
 import os
 import shutil
-import subprocess
 import tempfile
+from subprocess import CalledProcessError, TimeoutExpired
+
+from shared.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg
 
 logger = logging.getLogger(__name__)
 
@@ -59,7 +61,7 @@ class AudioMerger:
             output_path = os.path.join(temp_dir, f"merged.{output_format}")
 
             cmd = [
-                "ffmpeg",
+                FFMPEG_BIN,
                 "-y",
                 "-f",
                 "concat",
@@ -72,21 +74,16 @@ class AudioMerger:
                 output_path,
             ]
 
-            result = subprocess.run(
-                cmd,
-                capture_output=True,
-                text=True,
-                timeout=120,
-            )
-
-            if result.returncode != 0:
-                logger.error(f"FFmpeg 合并失败: stderr={result.stderr}")
-                raise AudioMergeError(f"FFmpeg 合并失败: {result.stderr[:500]}")
+            try:
+                run_ffmpeg(cmd, timeout=120)
+            except CalledProcessError as e:
+                logger.error(f"FFmpeg 合并失败: stderr={e.stderr}")
+                raise AudioMergeError(f"FFmpeg 合并失败: {str(e)[:500]}")
 
             with open(output_path, "rb") as f:
                 return f.read()
 
-        except subprocess.TimeoutExpired:
+        except TimeoutExpired:
             raise AudioMergeError("FFmpeg 合并超时（120 秒）")
         except AudioMergeError:
             raise

packages/shared/ffmpeg_utils.py

@@ -0,0 +1,79 @@
+"""FFmpeg 共享工具 — packages/shared 层.
+
+仅包含与业务无关的底层原语：FFmpeg/FFprobe 二进制路径、run_ffmpeg 执行器。
+业务相关的滤镜构建、视频探测等留在 apps/worker/video_processing/ffmpeg_utils.py。
+
+application 层和 worker 层都可以引用本模块，避免跨层依赖。
+"""
+
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess  # nosec B404
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+FFMPEG_BIN: str = shutil.which("ffmpeg") or "ffmpeg"
+FFPROBE_BIN: str = shutil.which("ffprobe") or "ffprobe"
+
+# FFmpeg 执行默认超时（秒），防止 FFmpeg hang 住导致进程永久阻塞
+# 默认 30 分钟，足够处理大部分短视频渲染；超长视频可单独传参覆盖
+DEFAULT_FFMPEG_TIMEOUT = 1800
+
+
+# ── FFmpeg 执行 ───────────────────────────────────────────────────────────────
+
+
+def run_ffmpeg(
+    command: list[str],
+    *,
+    capture_output: bool = True,
+    timeout: int | None = DEFAULT_FFMPEG_TIMEOUT,
+) -> tuple[str, str]:
+    """执行 FFmpeg 命令（统一入口）。
+
+    Args:
+        command: 完整的 ffmpeg 命令列表（含 "ffmpeg" 本身）
+        capture_output: 是否捕获 stdout/stderr
+        timeout: 超时时间（秒），默认 1800s（30分钟）；None 表示不设超时（不推荐）
+
+    Returns:
+        (stdout, stderr) 元组
+
+    Raises:
+        subprocess.CalledProcessError: 命令执行失败时抛出，
+            异常信息包含完整 stderr 以便排查。
+        subprocess.TimeoutExpired: 超时未完成时抛出，FFmpeg 进程会被 kill。
+    """
+    try:
+        result = subprocess.run(  # nosec B603
+            command,
+            check=True,
+            stdout=subprocess.PIPE if capture_output else None,
+            stderr=subprocess.PIPE if capture_output else None,
+            text=True,
+            timeout=timeout,
+        )
+        return (result.stdout or "", result.stderr or "")
+    except subprocess.TimeoutExpired:
+        logger.error(
+            "FFmpeg 命令超时 (%ds): command=%s",
+            timeout or -1,
+            " ".join(str(c) for c in command[:20]),
+        )
+        raise
+    except subprocess.CalledProcessError as e:
+        # 把完整 stderr 打到日志，方便排查 exit code 183 等问题
+        stderr_text = (e.stderr or "").strip()
+        logger.error(
+            "FFmpeg 命令失败: exit_code=%d command=%s\nstderr:\n%s",
+            e.returncode,
+            " ".join(str(c) for c in command[:20]),  # 截断过长的命令
+            stderr_text[:5000],  # 截断过长的 stderr
+        )
+        raise

packages/shared/url_security.py

@@ -92,6 +92,141 @@ _DOWNLOAD_CHUNK_SIZE = 8192
 # 最大重定向次数
 _MAX_REDIRECTS = 5
 
+# 文件魔数（文件头签名）表 — 用于 MIME 白名单校验后的二次真实性校验
+# key: MIME 类型，value: 签名列表，任一签名匹配即通过
+# 每条签名: list of (offset, bytes)，所有条目都匹配才算该签名命中（支持多处联合匹配如 RIFF+WAVE）
+_MAGIC_NUMBERS: dict[str, list[list[tuple[int, bytes]]]] = {
+    # ── 音频 ──
+    "audio/mpeg": [
+        [(0, b"ID3")],  # ID3v2 标签
+        [(0, b"\xff\xfb")],  # MPEG1 Layer3
+        [(0, b"\xff\xf3")],  # MPEG2 Layer3
+        [(0, b"\xff\xf2")],  # MPEG2.5 Layer3
+        [(0, b"\xff\xfa")],  # MPEG1 Layer2
+        [(0, b"\xff\xf9")],  # 其他 MPEG ADTS
+    ],
+    "audio/wav": [
+        [(0, b"RIFF"), (8, b"WAVE")],  # RIFF + WAVE
+    ],
+    "audio/x-wav": [
+        [(0, b"RIFF"), (8, b"WAVE")],
+    ],
+    "audio/ogg": [
+        [(0, b"OggS")],
+    ],
+    "application/ogg": [
+        [(0, b"OggS")],
+    ],
+    "audio/flac": [
+        [(0, b"fLaC")],
+    ],
+    "audio/aac": [
+        [(0, b"\xff\xf1")],  # ADTS MPEG-4
+        [(0, b"\xff\xf9")],  # ADTS MPEG-2
+    ],
+    "audio/aacp": [
+        [(0, b"\xff\xf1")],
+        [(0, b"\xff\xf9")],
+    ],
+    "audio/mp4": [
+        [(4, b"ftyp")],  # ISO Base Media (M4A)
+    ],
+    "audio/x-m4a": [
+        [(4, b"ftyp")],
+    ],
+    # ── 视频 ──
+    "video/mp4": [
+        [(4, b"ftyp")],  # ISO Base Media (MP4)
+    ],
+    "video/quicktime": [
+        [(4, b"ftyp")],
+    ],
+    "video/x-matroska": [
+        [(0, b"\x1a\x45\xdf\xa3")],  # EBML header
+    ],
+    "video/webm": [
+        [(0, b"\x1a\x45\xdf\xa3")],
+    ],
+    "video/x-msvideo": [
+        [(0, b"RIFF"), (8, b"AVI ")],
+    ],
+    # ── 图片 ──
+    "image/jpeg": [
+        [(0, b"\xff\xd8\xff")],
+    ],
+    "image/png": [
+        [(0, b"\x89PNG\r\n\x1a\n")],
+    ],
+    "image/gif": [
+        [(0, b"GIF87a")],
+        [(0, b"GIF89a")],
+    ],
+    "image/webp": [
+        [(0, b"RIFF"), (8, b"WEBP")],
+    ],
+    "image/bmp": [
+        [(0, b"BM")],
+    ],
+}
+
+# 魔数校验最大读取字节数（文件头）
+_MAGIC_CHECK_READ_SIZE = 256
+
+
+def _validate_magic_number(file_path: str, allowed_mime_types: set[str]) -> None:
+    """校验文件头魔数是否与允许的 MIME 类型匹配.
+
+    读取文件前 256 字节，与 allowed_mime_types 对应格式的魔数逐一比对，
+    任一类型匹配即通过；全部不匹配则抛出 UrlSecurityError。
+
+    仅当 allowed_mime_types 非空时执行；空文件视为不匹配。
+
+    Args:
+        file_path: 本地文件路径
+        allowed_mime_types: 允许的 MIME 类型集合
+
+    Raises:
+        UrlSecurityError: 文件魔数与所有允许类型均不匹配
+    """
+    # 收集所有允许类型对应的魔数签名
+    signatures: list[list[tuple[int, bytes]]] = []
+    for mime in allowed_mime_types:
+        sigs = _MAGIC_NUMBERS.get(mime)
+        if sigs:
+            signatures.extend(sigs)
+
+    # 如果没有已知魔数（比如自定义 MIME），跳过校验不阻断
+    if not signatures:
+        return
+
+    try:
+        with open(file_path, "rb") as f:
+            header = f.read(_MAGIC_CHECK_READ_SIZE)
+    except OSError as e:
+        raise UrlSecurityError(f"读取文件头失败: {e}") from e
+
+    if not header:
+        raise UrlSecurityError("文件为空，无法校验格式")
+
+    # 任一签名匹配即通过
+    for sig in signatures:
+        match = True
+        for offset, expected in sig:
+            if offset + len(expected) > len(header):
+                match = False
+                break
+            if header[offset : offset + len(expected)] != expected:
+                match = False
+                break
+        if match:
+            return
+
+    raise UrlSecurityError(
+        f"文件魔数与允许的 MIME 类型不匹配，"
+        f"允许类型: {sorted(allowed_mime_types)}，"
+        f"文件头前16字节: {header[:16].hex()}"
+    )
+
 
 class UrlSecurityError(ValueError):
     """URL 安全校验失败."""
@@ -294,6 +429,7 @@ def safe_download_file(
     - 重定向次数限制 + 手动跟随（避免重定向绕过 SSRF）
     - 文件大小限制（流式读取，超过立即中断）
     - MIME 类型白名单（可选）
+    - 文件头魔数校验（配合 MIME 白名单做二次真实性校验）
 
     Args:
         url: 下载 URL
@@ -362,6 +498,10 @@ def safe_download_file(
                         raise UrlSecurityError(f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes")
                     f.write(chunk)
 
+            # 文件头魔数校验（MIME 白名单基础上的二次真实性校验）
+            if allowed_mime_types is not None:
+                _validate_magic_number(dest_path, allowed_mime_types)
+
             return total_bytes
         finally:
             resp.close()

pytest.ini

@@ -1,5 +1,5 @@
 [pytest]
-pythonpath = . apps/api apps/worker
+pythonpath = . apps/api apps/worker packages
 testpaths = tests
 
 # ===== 覆盖率配置 =====

tests/unit/test_audio_merger_security.py

@@ -0,0 +1,140 @@
+"""AudioMerger 单元测试 — P1 裸subprocess下沉验证.
+
+验证 AudioMerger 使用 shared.ffmpeg_utils.run_ffmpeg 统一入口，
+不再直接调用 subprocess.run。
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+from application.tts_job.audio_merger import AudioMergeError, AudioMerger
+
+
+class TestAudioMergerUsesRunFfmpeg:
+    """验证 AudioMerger 使用 run_ffmpeg 统一入口，而非裸 subprocess."""
+
+    def test_single_file_does_not_call_ffmpeg(self):
+        """单文件时直接读取，不调用 FFmpeg."""
+        merger = AudioMerger()
+        with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f:
+            f.write(b"fake audio data")
+            path = f.name
+
+        try:
+            with patch("application.tts_job.audio_merger.run_ffmpeg") as mock_run:
+                result = merger.merge([path])
+                mock_run.assert_not_called()
+                assert result == b"fake audio data"
+        finally:
+            os.unlink(path)
+
+    def test_multiple_files_calls_run_ffmpeg(self):
+        """多文件时调用 run_ffmpeg 合并。"""
+        merger = AudioMerger()
+        paths = []
+        for i in range(2):
+            f = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
+            f.write(f"audio{i}".encode())
+            f.close()
+            paths.append(f.name)
+
+        try:
+            with patch("application.tts_job.audio_merger.run_ffmpeg") as mock_run:
+                # run_ffmpeg 成功返回，模拟合并完成
+                # 需要让 output_path 文件存在，否则 read 会报错
+                def fake_run_ffmpeg(cmd, **kwargs):
+                    # 找到 output_path（命令最后一个参数）
+                    output_path = cmd[-1]
+                    with open(output_path, "wb") as out:
+                        out.write(b"merged audio")
+                    return ("", "")
+
+                mock_run.side_effect = fake_run_ffmpeg
+                result = merger.merge(paths)
+
+                mock_run.assert_called_once()
+                call_args = mock_run.call_args[0][0]
+                # 验证使用了 FFMPEG_BIN 而非硬编码 "ffmpeg"
+                from shared.ffmpeg_utils import FFMPEG_BIN
+
+                assert call_args[0] == FFMPEG_BIN
+                # 验证使用 concat demuxer
+                assert "concat" in call_args
+                assert result == b"merged audio"
+        finally:
+            for p in paths:
+                os.unlink(p)
+
+    def test_ffmpeg_failure_raises_audio_merge_error(self):
+        """FFmpeg 失败时抛出 AudioMergeError."""
+        merger = AudioMerger()
+        paths = []
+        for i in range(2):
+            f = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
+            f.write(f"audio{i}".encode())
+            f.close()
+            paths.append(f.name)
+
+        try:
+            with patch("application.tts_job.audio_merger.run_ffmpeg") as mock_run:
+                mock_run.side_effect = subprocess.CalledProcessError(
+                    returncode=1, cmd=["ffmpeg"], stderr="concat error"
+                )
+
+                with pytest.raises(AudioMergeError, match="FFmpeg 合并失败"):
+                    merger.merge(paths)
+
+                mock_run.assert_called_once()
+        finally:
+            for p in paths:
+                os.unlink(p)
+
+    def test_ffmpeg_timeout_raises_audio_merge_error(self):
+        """FFmpeg 超时时抛出 AudioMergeError."""
+        merger = AudioMerger()
+        paths = []
+        for i in range(2):
+            f = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
+            f.write(f"audio{i}".encode())
+            f.close()
+            paths.append(f.name)
+
+        try:
+            with patch("application.tts_job.audio_merger.run_ffmpeg") as mock_run:
+                mock_run.side_effect = subprocess.TimeoutExpired(cmd=["ffmpeg"], timeout=120)
+
+                with pytest.raises(AudioMergeError, match="超时"):
+                    merger.merge(paths)
+
+                mock_run.assert_called_once()
+        finally:
+            for p in paths:
+                os.unlink(p)
+
+    def test_empty_list_raises_error(self):
+        """空列表时直接抛错，不调用 ffmpeg."""
+        merger = AudioMerger()
+        with patch("application.tts_job.audio_merger.run_ffmpeg") as mock_run:
+            with pytest.raises(AudioMergeError, match="没有可合并的音频文件"):
+                merger.merge([])
+            mock_run.assert_not_called()
+
+    def test_no_direct_subprocess_import(self):
+        """验证 audio_merger 模块不直接 import subprocess（通过模块源码检查）。"""
+        import inspect
+
+        import application.tts_job.audio_merger as am_module
+
+        source = inspect.getsource(am_module)
+        # 不应该有 "import subprocess" 整行
+        src_lines = [line.strip() for line in source.split("\n") if line.strip()]
+        # 允许 from subprocess import CalledProcessError, TimeoutExpired（只导入异常类）
+        # 不允许直接 import subprocess
+        assert not any(
+            line == "import subprocess" for line in src_lines
+        ), "audio_merger.py 不应直接 import subprocess，应通过 run_ffmpeg 统一入口"

tests/unit/test_generation_p3_optimizations.py

@@ -25,8 +25,8 @@ class TestVerifyUrlAccessibleRetry:
     """_verify_url_accessible 重试逻辑."""
 
     @patch("time.sleep")
-    @patch("urllib.request.urlopen")
-    def test_first_attempt_success(self, mock_urlopen, mock_sleep):
+    @patch("urllib.request.OpenerDirector.open")
+    def test_first_attempt_success(self, mock_open, mock_sleep):
         """首次成功，不重试."""
         from worker_app.tasks.generation import _verify_url_accessible
 
@@ -34,15 +34,15 @@ class TestVerifyUrlAccessibleRetry:
         mock_resp.status = 200
         mock_resp.__enter__ = MagicMock(return_value=mock_resp)
         mock_resp.__exit__ = MagicMock(return_value=False)
-        mock_urlopen.return_value = mock_resp
+        mock_open.return_value = mock_resp
 
         assert _verify_url_accessible("https://example.com/file.mp4") is True
-        assert mock_urlopen.call_count == 1
+        assert mock_open.call_count == 1
         mock_sleep.assert_not_called()
 
     @patch("time.sleep")
-    @patch("urllib.request.urlopen")
-    def test_retry_then_success(self, mock_urlopen, mock_sleep):
+    @patch("urllib.request.OpenerDirector.open")
+    def test_retry_then_success(self, mock_open, mock_sleep):
         """首次失败，重试后成功."""
         from worker_app.tasks.generation import _verify_url_accessible
 
@@ -52,31 +52,31 @@ class TestVerifyUrlAccessibleRetry:
         mock_resp_ok.__enter__ = MagicMock(return_value=mock_resp_ok)
         mock_resp_ok.__exit__ = MagicMock(return_value=False)
 
-        mock_urlopen.side_effect = [
+        mock_open.side_effect = [
             OSError("connection reset"),
             mock_resp_ok,
         ]
 
         assert _verify_url_accessible("https://example.com/file.mp4") is True
-        assert mock_urlopen.call_count == 2
+        assert mock_open.call_count == 2
         mock_sleep.assert_called_once_with(1)
 
     @patch("time.sleep")
-    @patch("urllib.request.urlopen")
-    def test_all_retries_exhausted(self, mock_urlopen, mock_sleep):
+    @patch("urllib.request.OpenerDirector.open")
+    def test_all_retries_exhausted(self, mock_open, mock_sleep):
         """全部重试耗尽，返回 False."""
         from worker_app.tasks.generation import _verify_url_accessible
 
-        mock_urlopen.side_effect = OSError("connection refused")
+        mock_open.side_effect = OSError("connection refused")
 
         assert _verify_url_accessible("https://example.com/file.mp4") is False
         # 1 首次 + 2 重试 = 3 次
-        assert mock_urlopen.call_count == 3
+        assert mock_open.call_count == 3
         assert mock_sleep.call_count == 2
 
     @patch("time.sleep")
-    @patch("urllib.request.urlopen")
-    def test_http_500_then_success(self, mock_urlopen, mock_sleep):
+    @patch("urllib.request.OpenerDirector.open")
+    def test_http_500_then_success(self, mock_open, mock_sleep):
         """HTTP 500 后重试成功."""
         from worker_app.tasks.generation import _verify_url_accessible
 
@@ -90,21 +90,21 @@ class TestVerifyUrlAccessibleRetry:
         mock_resp_200.__enter__ = MagicMock(return_value=mock_resp_200)
         mock_resp_200.__exit__ = MagicMock(return_value=False)
 
-        mock_urlopen.side_effect = [mock_resp_500, mock_resp_200]
+        mock_open.side_effect = [mock_resp_500, mock_resp_200]
 
         assert _verify_url_accessible("https://example.com/file.mp4") is True
-        assert mock_urlopen.call_count == 2
+        assert mock_open.call_count == 2
 
     @patch("time.sleep")
-    @patch("urllib.request.urlopen")
-    def test_custom_retries_zero(self, mock_urlopen, mock_sleep):
+    @patch("urllib.request.OpenerDirector.open")
+    def test_custom_retries_zero(self, mock_open, mock_sleep):
         """retries=0 时不重试."""
         from worker_app.tasks.generation import _verify_url_accessible
 
-        mock_urlopen.side_effect = OSError("timeout")
+        mock_open.side_effect = OSError("timeout")
 
         assert _verify_url_accessible("https://example.com/file.mp4", retries=0) is False
-        assert mock_urlopen.call_count == 1
+        assert mock_open.call_count == 1
         mock_sleep.assert_not_called()
 
 

tests/unit/test_oneclick_gen_p0_fixes.py

@@ -162,14 +162,14 @@ class TestOSSUploadAndVerify:
         mock_response.__enter__ = MagicMock(return_value=mock_response)
         mock_response.__exit__ = MagicMock(return_value=False)
 
-        with patch("urllib.request.urlopen", return_value=mock_response):
+        with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
             assert _verify_url_accessible("https://example.com/test.mp4") is True
 
     def test_verify_url_accessible_failure(self):
         """URL 不可访问时返回 False."""
         from worker_app.tasks.generation import _verify_url_accessible
 
-        with patch("urllib.request.urlopen", side_effect=Exception("connection refused")):
+        with patch("urllib.request.OpenerDirector.open", side_effect=Exception("connection refused")):
             assert _verify_url_accessible("https://example.com/test.mp4") is False
 
     def test_verify_url_404(self):
@@ -181,7 +181,7 @@ class TestOSSUploadAndVerify:
         mock_response.__enter__ = MagicMock(return_value=mock_response)
         mock_response.__exit__ = MagicMock(return_value=False)
 
-        with patch("urllib.request.urlopen", return_value=mock_response):
+        with patch("urllib.request.OpenerDirector.open", return_value=mock_response):
             assert _verify_url_accessible("https://example.com/test.mp4") is False
 
 

tests/unit/test_tech_debt_security_round2.py

@@ -0,0 +1,527 @@
+"""URL 安全模块单元测试 - 技术债务第二轮：魔数校验 + 重定向每跳校验."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+# ══════════════════════════════════════════════════════════════════════════════
+# 魔数校验测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestMagicNumberValidation:
+    """文件头魔数校验测试."""
+
+    def test_png_magic_passes(self, tmp_path: Path):
+        """PNG 魔数正确应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.png"
+        f.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"image/png"})  # 不抛异常即通过
+
+    def test_jpeg_magic_passes(self, tmp_path: Path):
+        """JPEG 魔数正确应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.jpg"
+        f.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"image/jpeg", "image/png"})
+
+    def test_gif_magic_passes(self, tmp_path: Path):
+        """GIF 魔数正确应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.gif"
+        f.write_bytes(b"GIF89a" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"image/gif"})
+
+    def test_mp3_magic_id3_passes(self, tmp_path: Path):
+        """MP3 ID3v2 标签魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.mp3"
+        f.write_bytes(b"ID3\x03\x00\x00\x00\x00\x00\x00" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"audio/mpeg"})
+
+    def test_mp3_magic_frame_passes(self, tmp_path: Path):
+        """MP3 frame sync 魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.mp3"
+        f.write_bytes(b"\xff\xfb\x90\x00" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"audio/mpeg"})
+
+    def test_wav_magic_passes(self, tmp_path: Path):
+        """WAV RIFF+WAVE 魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.wav"
+        header = b"RIFF" + b"\x24\x00\x00\x00" + b"WAVE" + b"fmt " + b"\x00" * 100
+        f.write_bytes(header)
+
+        _validate_magic_number(str(f), {"audio/wav"})
+
+    def test_mp4_magic_passes(self, tmp_path: Path):
+        """MP4 ftyp 魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.mp4"
+        # ftyp box: size(4) + 'ftyp'(4) + major_brand(4) + ...
+        f.write_bytes(b"\x00\x00\x00\x20ftypisom" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"video/mp4"})
+
+    def test_webp_magic_passes(self, tmp_path: Path):
+        """WebP RIFF+WEBP 魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.webp"
+        f.write_bytes(b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"image/webp"})
+
+    def test_wrong_magic_raises(self, tmp_path: Path):
+        """魔数不匹配应抛出 UrlSecurityError."""
+        from shared.url_security import UrlSecurityError, _validate_magic_number
+
+        f = tmp_path / "fake.png"
+        f.write_bytes(b"NOT_A_PNG_FILE!!!" + b"\x00" * 100)
+
+        with pytest.raises(UrlSecurityError, match="魔数"):
+            _validate_magic_number(str(f), {"image/png", "image/jpeg"})
+
+    def test_text_as_png_raises(self, tmp_path: Path):
+        """纯文本伪装成 PNG 应被拦截."""
+        from shared.url_security import UrlSecurityError, _validate_magic_number
+
+        f = tmp_path / "fake.png"
+        f.write_text("<html>not an image</html>", encoding="utf-8")
+
+        with pytest.raises(UrlSecurityError):
+            _validate_magic_number(str(f), {"image/png"})
+
+    def test_empty_file_raises(self, tmp_path: Path):
+        """空文件应抛出异常."""
+        from shared.url_security import UrlSecurityError, _validate_magic_number
+
+        f = tmp_path / "empty.png"
+        f.write_bytes(b"")
+
+        with pytest.raises(UrlSecurityError, match="为空"):
+            _validate_magic_number(str(f), {"image/png"})
+
+    def test_unknown_mime_skipped(self, tmp_path: Path):
+        """未知 MIME 类型没有对应魔数，应跳过校验不阻断."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.xyz"
+        f.write_bytes(b"random garbage data here")
+
+        # 没有已知魔数的 MIME，跳过校验
+        _validate_magic_number(str(f), {"application/x-custom-format"})
+
+    def test_multiple_allowed_types_one_matches(self, tmp_path: Path):
+        """多个允许类型，只要有一个匹配就通过."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test"
+        f.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 50)
+
+        _validate_magic_number(str(f), {"image/jpeg", "image/png", "image/gif"})
+
+    def test_multiple_allowed_types_none_match(self, tmp_path: Path):
+        """多个允许类型都不匹配应抛异常."""
+        from shared.url_security import UrlSecurityError, _validate_magic_number
+
+        f = tmp_path / "test"
+        f.write_bytes(b"RIFF\x00\x00\x00\x00WAVE" + b"\x00" * 50)
+
+        with pytest.raises(UrlSecurityError):
+            _validate_magic_number(str(f), {"image/png", "image/jpeg", "image/gif"})
+
+    def test_flac_magic_passes(self, tmp_path: Path):
+        """FLAC 魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.flac"
+        f.write_bytes(b"fLaC" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"audio/flac"})
+
+    def test_ogg_magic_passes(self, tmp_path: Path):
+        """OGG 魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.ogg"
+        f.write_bytes(b"OggS\x00\x02\x00\x00" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"audio/ogg"})
+
+    def test_bmp_magic_passes(self, tmp_path: Path):
+        """BMP 魔数应通过校验."""
+        from shared.url_security import _validate_magic_number
+
+        f = tmp_path / "test.bmp"
+        f.write_bytes(b"BM\x00\x00\x00\x00" + b"\x00" * 100)
+
+        _validate_magic_number(str(f), {"image/bmp"})
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# safe_download_file 魔数校验集成测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestSafeDownloadMagicIntegration:
+    """safe_download_file 集成魔数校验测试."""
+
+    def test_download_with_mime_and_magic_match(self, tmp_path: Path):
+        """MIME 匹配 + 魔数匹配，下载成功."""
+        from unittest.mock import MagicMock, patch
+
+        from shared.url_security import safe_download_file
+
+        png_data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 200
+
+        class FakeResp:
+            headers = {"Content-Type": "image/png", "Content-Length": str(len(png_data))}
+
+            def read(self, n):
+                if not hasattr(self, "_pos"):
+                    self._pos = 0
+                chunk = png_data[self._pos : self._pos + n]
+                self._pos += len(chunk)
+                return chunk
+
+            def close(self):
+                pass
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                return FakeResp()
+
+        with (
+            patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u),
+            patch("shared.url_security.urllib.request.build_opener", return_value=FakeOpener()),
+        ):
+            dest = str(tmp_path / "out.png")
+            size = safe_download_file(
+                "https://example.com/test.png",
+                dest,
+                allowed_mime_types={"image/png"},
+                purpose="test",
+            )
+
+        assert size == len(png_data)
+        with open(dest, "rb") as f:
+            assert f.read() == png_data
+
+    def test_download_mime_match_but_magic_mismatch_raises(self, tmp_path: Path):
+        """Content-Type 声明是 PNG 但实际文件是 HTML，应被魔数校验拦截."""
+        from unittest.mock import patch
+
+        from shared.url_security import UrlSecurityError, safe_download_file
+
+        fake_data = b"<html>not really a png</html>"
+
+        class FakeResp:
+            headers = {"Content-Type": "image/png", "Content-Length": str(len(fake_data))}
+
+            def read(self, n):
+                if not hasattr(self, "_pos"):
+                    self._pos = 0
+                chunk = fake_data[self._pos : self._pos + n]
+                self._pos += len(chunk)
+                return chunk
+
+            def close(self):
+                pass
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                return FakeResp()
+
+        with (
+            patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u),
+            patch("shared.url_security.urllib.request.build_opener", return_value=FakeOpener()),
+        ):
+            dest = str(tmp_path / "out.png")
+            with pytest.raises(UrlSecurityError, match="魔数"):
+                safe_download_file(
+                    "https://example.com/fake.png",
+                    dest,
+                    allowed_mime_types={"image/png"},
+                    purpose="test",
+                )
+
+    def test_download_no_mime_check_skips_magic(self, tmp_path: Path):
+        """不传 allowed_mime_types 时不做 MIME 校验也不做魔数校验."""
+        from unittest.mock import patch
+
+        from shared.url_security import safe_download_file
+
+        data = b"any random content here"
+
+        class FakeResp:
+            headers = {"Content-Type": "application/octet-stream"}
+
+            def read(self, n):
+                if not hasattr(self, "_pos"):
+                    self._pos = 0
+                chunk = data[self._pos : self._pos + n]
+                self._pos += len(chunk)
+                return chunk
+
+            def close(self):
+                pass
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                return FakeResp()
+
+        with (
+            patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u),
+            patch("shared.url_security.urllib.request.build_opener", return_value=FakeOpener()),
+        ):
+            dest = str(tmp_path / "out.bin")
+            size = safe_download_file(
+                "https://example.com/file.bin",
+                dest,
+                purpose="test",
+            )
+            assert size == len(data)
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# _verify_url_accessible 重定向每跳校验测试
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestVerifyUrlRedirectValidation:
+    """URL 可访问性校验 - 重定向每跳 SSRF 校验测试.
+
+    直接复制核心逻辑进行单元测试，避免导入 generation 模块触发 DB 连接。
+    逻辑与 generation.py 中的 _verify_url_accessible 完全一致。
+    """
+
+    @staticmethod
+    def _verify_url_accessible(url, timeout=10.0, retries=0, max_redirects=5):
+        """从 generation.py 复制的核心逻辑，用于单元测试."""
+        import time
+        import urllib.request
+        from urllib.parse import urljoin
+
+        from shared.url_security import UrlSecurityError, validate_url_safety
+
+        class NoRedirect(urllib.request.HTTPRedirectHandler):
+            def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: N802
+                return None
+
+        last_error = None
+
+        def _do_verify(current_url):
+            redirect_count = 0
+            url_being_checked = current_url
+            opener = urllib.request.build_opener(NoRedirect())
+
+            while redirect_count <= max_redirects:
+                safe_url = validate_url_safety(url_being_checked, purpose="url_verify")
+                req = urllib.request.Request(safe_url, method="HEAD")
+                req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
+
+                with opener.open(req, timeout=timeout):
+                    # 简化：进入 with 块即表示 2xx（3xx 被 NoRedirect 拦截为 HTTPError）
+                    return True
+
+                raise Exception("unreachable")
+
+        import urllib.error
+
+        for attempt in range(1 + retries):
+            try:
+                # 用 try/except 手动处理重定向
+                redirect_count = 0
+                current = url
+                opener = urllib.request.build_opener(NoRedirect())
+
+                while redirect_count <= max_redirects:
+                    safe_url = validate_url_safety(current, purpose="url_verify")
+                    req = urllib.request.Request(safe_url, method="HEAD")
+                    req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")
+                    try:
+                        with opener.open(req, timeout=timeout) as resp:
+                            if 200 <= resp.status < 300:
+                                return True
+                            if resp.status < 400:
+                                return True
+                            last_error = Exception(f"HTTP {resp.status}")
+                    except urllib.error.HTTPError as e:
+                        if 300 <= e.code < 400 and e.headers.get("Location"):
+                            if redirect_count >= max_redirects:
+                                raise Exception(f"重定向次数超过上限 ({max_redirects})")
+                            location = e.headers["Location"]
+                            current = urljoin(safe_url, location)
+                            redirect_count += 1
+                            continue
+                        last_error = Exception(f"HTTP {e.code}")
+                        break
+                else:
+                    raise Exception(f"重定向次数超过上限 ({max_redirects})")
+            except Exception as e:
+                last_error = e
+
+            if attempt < retries:
+                time.sleep(0)
+
+        return False
+
+    def test_simple_200_ok(self):
+        """普通 200 响应应返回 True."""
+        import urllib.error
+        from unittest.mock import patch
+
+        class FakeResp:
+            status = 200
+            headers = {}
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *args):
+                pass
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                return FakeResp()
+
+        with (
+            patch("urllib.request.build_opener", return_value=FakeOpener()),
+            patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u),
+        ):
+            result = self._verify_url_accessible("https://example.com/file.mp4", retries=0)
+            assert result is True
+
+    def test_redirect_to_internal_ip_blocked(self):
+        """重定向到内网 IP 应被拦截（返回 False）."""
+        import urllib.error
+        from unittest.mock import patch
+
+        from shared.url_security import UrlSecurityError
+
+        call_count = 0
+
+        class FakeHTTPError(urllib.error.HTTPError):
+            def __init__(self):
+                pass
+
+        # 用 validate_url_safety 来模拟拦截
+        def fake_validate(url, **kwargs):
+            if "127.0.0.1" in url:
+                raise UrlSecurityError("内网IP禁止访问")
+            return url
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                nonlocal call_count
+                call_count += 1
+                # 第一次请求返回 302
+                raise urllib.error.HTTPError(
+                    req.full_url, 302, "Found", {"Location": "http://127.0.0.1/internal"}, None
+                )
+
+        with (
+            patch("urllib.request.build_opener", return_value=FakeOpener()),
+            patch("shared.url_security.validate_url_safety", side_effect=fake_validate),
+        ):
+            result = self._verify_url_accessible("https://example.com/redirect", retries=0)
+            assert result is False
+            assert call_count == 1  # 只请求了第一次，第二次跳转在校验阶段就被拦了
+
+    def test_redirect_count_exceeded(self):
+        """超过最大重定向次数应返回 False."""
+        import urllib.error
+        from unittest.mock import patch
+
+        call_count = 0
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                nonlocal call_count
+                call_count += 1
+                raise urllib.error.HTTPError(req.full_url, 302, "Found", {"Location": "https://example.com/next"}, None)
+
+        with (
+            patch("urllib.request.build_opener", return_value=FakeOpener()),
+            patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u),
+        ):
+            result = self._verify_url_accessible(
+                "https://example.com/start",
+                retries=0,
+                max_redirects=3,
+            )
+            assert result is False
+            assert call_count == 4  # 初始 + 3次跳转 = 4次请求
+
+    def test_redirect_chain_valid(self):
+        """合法的重定向链（都是公网域名）应返回 True."""
+        import urllib.error
+        from unittest.mock import patch
+
+        step = 0
+
+        class FakeResp:
+            status = 200
+            headers = {}
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *args):
+                pass
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                nonlocal step
+                step += 1
+                if step == 1:
+                    raise urllib.error.HTTPError(
+                        req.full_url, 302, "Found", {"Location": "https://cdn.example.com/final.mp4"}, None
+                    )
+                return FakeResp()
+
+        with (
+            patch("urllib.request.build_opener", return_value=FakeOpener()),
+            patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u),
+        ):
+            result = self._verify_url_accessible(
+                "https://example.com/redirect",
+                retries=0,
+                max_redirects=5,
+            )
+            assert result is True
+            assert step == 2
+
+    def test_404_returns_false(self):
+        """404 应返回 False."""
+        import urllib.error
+        from unittest.mock import patch
+
+        class FakeOpener:
+            def open(self, req, timeout=None):
+                raise urllib.error.HTTPError(req.full_url, 404, "Not Found", {}, None)
+
+        with (
+            patch("urllib.request.build_opener", return_value=FakeOpener()),
+            patch("shared.url_security.validate_url_safety", side_effect=lambda u, **kw: u),
+        ):
+            result = self._verify_url_accessible("https://example.com/nonexistent", retries=0)
+            assert result is False

tests/unit/test_tts_segment_synthesis.py

@@ -107,10 +107,11 @@ class TestAudioMerger:
         finally:
             os.unlink(path)
 
-    @patch("packages.application.tts_job.audio_merger.subprocess.run")
-    def test_ffmpeg_called_correctly(self, mock_run: MagicMock) -> None:
-        """多文件调用 FFmpeg concat。"""
-        mock_run.return_value = MagicMock(returncode=0)
+    @patch("packages.application.tts_job.audio_merger.run_ffmpeg")
+    def test_ffmpeg_called_correctly(self, mock_run_ffmpeg: MagicMock) -> None:
+        """多文件调用 FFmpeg concat（通过 run_ffmpeg 统一入口）。"""
+        # run_ffmpeg 成功返回 (stdout, stderr)
+        mock_run_ffmpeg.return_value = ("", "")
 
         # 创建临时文件
         paths = []
@@ -130,20 +131,24 @@ class TestAudioMerger:
                 except (FileNotFoundError, OSError):
                     pass  # Expected since we're mocking
 
-            # 验证 FFmpeg 被调用
-            mock_run.assert_called_once()
-            cmd = mock_run.call_args[0][0]
-            assert cmd[0] == "ffmpeg"
+            # 验证 run_ffmpeg 被调用
+            mock_run_ffmpeg.assert_called_once()
+            cmd = mock_run_ffmpeg.call_args[0][0]
+            from shared.ffmpeg_utils import FFMPEG_BIN
+
+            assert cmd[0] == FFMPEG_BIN
             assert "-f" in cmd
             assert "concat" in cmd
         finally:
             for p in paths:
                 os.unlink(p)
 
-    @patch("packages.application.tts_job.audio_merger.subprocess.run")
-    def test_ffmpeg_failure_raises(self, mock_run: MagicMock) -> None:
-        """FFmpeg 失败抛出 AudioMergeError。"""
-        mock_run.return_value = MagicMock(returncode=1, stderr="error details")
+    @patch("packages.application.tts_job.audio_merger.run_ffmpeg")
+    def test_ffmpeg_failure_raises(self, mock_run_ffmpeg: MagicMock) -> None:
+        """FFmpeg 失败抛出 AudioMergeError（通过 run_ffmpeg 抛出 CalledProcessError）。"""
+        from subprocess import CalledProcessError
+
+        mock_run_ffmpeg.side_effect = CalledProcessError(returncode=1, cmd=["ffmpeg"], stderr="error details")
 
         paths = []
         for i in range(2):
@@ -155,6 +160,7 @@ class TestAudioMerger:
             merger = AudioMerger()
             with pytest.raises(AudioMergeError, match="FFmpeg 合并失败"):
                 merger.merge(paths)
+            mock_run_ffmpeg.assert_called_once()
         finally:
             for p in paths:
                 os.unlink(p)

tests/unit/test_unified_render_service.py

@@ -1660,3 +1660,76 @@ class TestStreamCopy:
         cmd = mock_run.call_args[0][0]
         assert "copy" in cmd
         assert isinstance(result.output_path, Path)
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# _extract_audio 安全下沉测试（P1 技术债务）
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestExtractAudioUsesRunFfmpeg:
+    """_extract_audio 必须使用 run_ffmpeg 统一管理，不能用裸 subprocess."""
+
+    def test_extract_audio_calls_run_ffmpeg(self, tmp_path):
+        """_extract_audio 内部应调用 ffmpeg_utils.run_ffmpeg 而非裸 subprocess."""
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        plan = MagicMock()
+        plan.config = {}
+        plan.id = "test-plan"
+        plan.canvas_config = MagicMock()
+        plan.canvas_config.width = 1080
+        plan.canvas_config.height = 1920
+        plan.canvas_config.fps = 30
+        plan.canvas_config.output_width = 1080
+        plan.canvas_config.output_height = 1920
+
+        svc = UnifiedRenderService(plan, [], {}, tmp_path)
+
+        video_path = tmp_path / "input.mp4"
+        output_path = tmp_path / "output.wav"
+        video_path.write_bytes(b"fake")
+
+        with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run:
+            svc._extract_audio(video_path, output_path)
+
+        # 验证调用了 run_ffmpeg
+        assert mock_run.called, "_extract_audio 必须通过 run_ffmpeg 执行 FFmpeg"
+        cmd = mock_run.call_args[0][0]
+
+        # 验证命令参数正确
+        assert cmd[0].endswith("ffmpeg") or "ffmpeg" in cmd[0]
+        assert "-i" in cmd
+        assert str(video_path) in cmd
+        assert "-vn" in cmd  # 无视频流
+        assert "pcm_s16le" in cmd  # 16bit PCM
+        assert "16000" in cmd  # 16kHz
+        assert str(output_path) in cmd
+        assert mock_run.call_args[1].get("timeout") == 120
+
+    def test_extract_audio_failure_raises_runtime_error(self, tmp_path):
+        """_extract_audio 失败时应抛出 RuntimeError."""
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        plan = MagicMock()
+        plan.config = {}
+        plan.id = "test-plan"
+        plan.canvas_config = MagicMock()
+        plan.canvas_config.width = 1080
+        plan.canvas_config.height = 1920
+        plan.canvas_config.fps = 30
+        plan.canvas_config.output_width = 1080
+        plan.canvas_config.output_height = 1920
+
+        svc = UnifiedRenderService(plan, [], {}, tmp_path)
+
+        video_path = tmp_path / "input.mp4"
+        output_path = tmp_path / "output.wav"
+        video_path.write_bytes(b"fake")
+
+        import subprocess
+
+        with patch("video_processing.unified_render_service.run_ffmpeg") as mock_run:
+            mock_run.side_effect = subprocess.CalledProcessError(1, "ffmpeg", stderr="error")
+            with pytest.raises(RuntimeError, match="音频提取失败"):
+                svc._extract_audio(video_path, output_path)