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