From f4701f7b39c34aec8597654cce0f03e69c2038df Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 19:14:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(security):=20P1=20audio=5Fmerger=E8=A3=B8su?= =?UTF-8?q?bprocess=E4=B8=8B=E6=B2=89=20+=20ffmpeg=5Futils=E6=9E=B6?= =?UTF-8?q?=E6=9E=84=E4=B8=8B=E6=B2=89=E5=88=B0packages/shared?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 packages/shared/ffmpeg_utils.py: run_ffmpeg/FFMPEG_BIN/FFPROBE_BIN下沉到共享层 - apps/worker/video_processing/ffmpeg_utils.py: 从shared re-export，保持向后兼容 - packages/application/tts_job/audio_merger.py: 改用 shared.ffmpeg_utils.run_ffmpeg - 新增 test_audio_merger_security.py: 6个单测验证下沉正确性 解决架构分层问题：application层不再需要跨层调用worker层的ffmpeg工具 --- apps/worker/video_processing/ffmpeg_utils.py | 77 ++-------- packages/application/tts_job/audio_merger.py | 23 ++- packages/shared/ffmpeg_utils.py | 79 ++++++++++ tests/unit/test_audio_merger_security.py | 147 +++++++++++++++++++ 4 files changed, 250 insertions(+), 76 deletions(-) mode change 100644 => 100755 packages/application/tts_job/audio_merger.py create mode 100755 packages/shared/ffmpeg_utils.py create mode 100755 tests/unit/test_audio_merger_security.py diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py index d40cc2d1f..5dc0ab1fe 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 probe_has_audio(local_path: str | Path) -> bool: 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/tests/unit/test_audio_merger_security.py b/tests/unit/test_audio_merger_security.py new file mode 100755 index 000000000..6fdf32020 --- /dev/null +++ b/tests/unit/test_audio_merger_security.py @@ -0,0 +1,147 @@ +"""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" 整行 + lines = [l.strip() for l in source.split("\n") if l.strip()] + direct_imports = [ + l for l in lines + if l == "import subprocess" or l.startswith("from subprocess") and "run" in l + ] + # 允许 from subprocess import CalledProcessError, TimeoutExpired（只导入异常类） + # 不允许直接 import subprocess + assert not any(l == "import subprocess" for l in lines), ( + "audio_merger.py 不应直接 import subprocess，应通过 run_ffmpeg 统一入口" + )