diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index 10ee0e752..8ac5f493a 100755 --- a/apps/worker/video_processing/ffmpeg_utils.py +++ b/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( diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py old mode 100644 new mode 100755 index 76b751052..ad614e2c3 --- a/apps/worker/video_processing/unified_render_service.py +++ b/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), diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py old mode 100644 new mode 100755 index 1af503185..99e94a614 --- a/apps/worker/worker_app/tasks/generation.py +++ b/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 diff --git a/packages/application/tts_job/audio_merger.py b/packages/application/tts_job/audio_merger.py old mode 100644 new mode 100755 index 0d583f821..7c432486a --- a/packages/application/tts_job/audio_merger.py +++ b/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 diff --git a/packages/shared/ffmpeg_utils.py b/packages/shared/ffmpeg_utils.py new file mode 100755 index 000000000..6518001ca --- /dev/null +++ b/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 diff --git a/packages/shared/url_security.py b/packages/shared/url_security.py old mode 100644 new mode 100755 index 75d64e9ed..d2220376f --- a/packages/shared/url_security.py +++ b/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() diff --git a/pytest.ini b/pytest.ini old mode 100644 new mode 100755 index 06ff5b0f9..92a1f25d3 --- a/pytest.ini +++ b/pytest.ini @@ -1,5 +1,5 @@ [pytest] -pythonpath = . apps/api apps/worker +pythonpath = . apps/api apps/worker packages testpaths = tests # ===== 覆盖率配置 ===== diff --git a/tests/unit/test_audio_merger_security.py b/tests/unit/test_audio_merger_security.py new file mode 100755 index 000000000..d34d12b58 --- /dev/null +++ b/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 统一入口" diff --git a/tests/unit/test_generation_p3_optimizations.py b/tests/unit/test_generation_p3_optimizations.py index d4c379b2a..ff4229083 100644 --- a/tests/unit/test_generation_p3_optimizations.py +++ b/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() diff --git a/tests/unit/test_oneclick_gen_p0_fixes.py b/tests/unit/test_oneclick_gen_p0_fixes.py index efde21b16..9f259f484 100644 --- a/tests/unit/test_oneclick_gen_p0_fixes.py +++ b/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 diff --git a/tests/unit/test_tech_debt_security_round2.py b/tests/unit/test_tech_debt_security_round2.py new file mode 100644 index 000000000..3372b38e6 --- /dev/null +++ b/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("not an image", 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"not really a png" + + 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 diff --git a/tests/unit/test_tts_segment_synthesis.py b/tests/unit/test_tts_segment_synthesis.py index 6e9a4c2c3..e729d0665 100755 --- a/tests/unit/test_tts_segment_synthesis.py +++ b/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) diff --git a/tests/unit/test_unified_render_service.py b/tests/unit/test_unified_render_service.py index a9dc6d5a2..ad78b8da3 100755 --- a/tests/unit/test_unified_render_service.py +++ b/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)