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