From 63bd21c2c4778f1704f4ad84d1d93ea03778348d Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 16:29:25 +0800 Subject: [PATCH 1/7] =?UTF-8?q?fix:=20PR#312=E5=AE=89=E5=85=A8=E5=80=BA?= =?UTF-8?q?=E5=8A=A1=204=E4=B8=AAP1=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 多轨道混音: 轨道数量上限8轨 + audio_path路径白名单校验 - 视频拼接: 段数上限50段 + video_path路径白名单校验 - 字幕渲染: subtitles滤镜路径白名单校验 + 禁止远程URL字幕 - 复用path_security.py统一安全校验模块 - 新增21个安全单元测试，52个原有测试零回归 --- apps/worker/video_processing/concat_engine.py | 68 +++ .../video_processing/multi_track_mixer.py | 81 ++++ apps/worker/video_processing/path_security.py | 303 ++++++++++++ .../subtitle_render_engine.py | 52 ++- packages/shared/url_security.py | 413 ++++++++++++++++ tests/unit/test_pr312_security_debt.py | 439 ++++++++++++++++++ 6 files changed, 1355 insertions(+), 1 deletion(-) create mode 100644 apps/worker/video_processing/path_security.py create mode 100644 packages/shared/url_security.py create mode 100755 tests/unit/test_pr312_security_debt.py diff --git a/apps/worker/video_processing/concat_engine.py b/apps/worker/video_processing/concat_engine.py index dc58a6723..d18afe533 100755 --- a/apps/worker/video_processing/concat_engine.py +++ b/apps/worker/video_processing/concat_engine.py @@ -24,12 +24,17 @@ from pathlib import Path from typing import Any from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, probe_video_info, run_ffmpeg +from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path logger = logging.getLogger(__name__) # ── 常量 ────────────────────────────────────────────────────────────────────── +MAX_CONCAT_SEGMENTS = 50 # 最大拼接段数（安全上限，防止OOM） + +ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"} + # concat demuxer 要求一致的参数列表 CONCAT_DEMUXER_REQUIRED_PARAMS = [ "codec_name", # 视频编码 @@ -144,6 +149,46 @@ class ConcatConfig: return len([s for s in self.segments if s.video_path]) +# ── 路径安全校验 ──────────────────────────────────────────────────────────── + + +def _validate_video_path(video_path: str, work_dir: Path) -> None: + """校验视频文件路径安全性. + + 规则： + - local:// schema → 必须在 work_dir 内 + - 相对路径 → 必须在 work_dir 内 + - 绝对路径 → 必须在允许目录白名单内 + - 扩展名必须是视频格式 + + Raises: + PathSecurityError: 路径不安全 + """ + if not video_path or not isinstance(video_path, str): + raise PathSecurityError("视频路径不能为空") + + # 本地路径（local:// 或相对路径 / 绝对路径） + if video_path.startswith("local://") or not video_path.startswith(("http://", "https://", "oss://")): + is_abs = video_path.startswith("/") and not video_path.startswith("local://") + safe_resolve_path( + video_path, + work_dir, + allow_outside=is_abs, + allowed_extensions=ALLOWED_VIDEO_EXTENSIONS, + ) + # 绝对路径额外检查白名单目录 + if is_abs and not video_path.startswith(str(work_dir)): + if not is_in_allowed_dirs(video_path): + raise PathSecurityError(f"视频路径不在允许目录内: {video_path[:80]}") + # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责） + # 但检查扩展名 + else: + path_part = video_path.split("?")[0].split("#")[0] + ext = Path(path_part).suffix.lower() + if ext and ext not in ALLOWED_VIDEO_EXTENSIONS: + raise PathSecurityError(f"不允许的视频文件类型: {ext}") + + # ── 视频拼接引擎 ────────────────────────────────────────────────────────────── @@ -179,6 +224,29 @@ class ConcatEngine: if not valid_segments: raise ValueError("No valid video segments to concat") + # ── 安全校验：段数上限 ── + if len(valid_segments) > MAX_CONCAT_SEGMENTS: + raise ValueError( + f"Too many concat segments: {len(valid_segments)} > {MAX_CONCAT_SEGMENTS}" + ) + + # ── 安全校验：所有视频路径白名单校验 ── + safe_segments = [] + for seg in valid_segments: + try: + _validate_video_path(seg.video_path, self.work_dir) + safe_segments.append(seg) + except PathSecurityError as e: + logger.warning("[concat] skip segment: path security check failed: %s", e) + + if len(safe_segments) != len(valid_segments): + valid_segments = safe_segments + config.segments = safe_segments + logger.info("[concat] %d segments passed security check", len(safe_segments)) + + if not valid_segments: + raise ValueError("No valid video segments after security check") + if len(valid_segments) == 1: # 只有一段，直接复制 import shutil diff --git a/apps/worker/video_processing/multi_track_mixer.py b/apps/worker/video_processing/multi_track_mixer.py index b626d0a73..764cedfe6 100755 --- a/apps/worker/video_processing/multi_track_mixer.py +++ b/apps/worker/video_processing/multi_track_mixer.py @@ -21,6 +21,7 @@ from pathlib import Path from typing import TYPE_CHECKING from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg +from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path if TYPE_CHECKING: from video_processing.render_audio import RenderContext @@ -36,6 +37,8 @@ TRACK_TYPE_VOICEOVER = "voiceover" # 配音（TTS/人声） TRACK_TYPE_SFX = "sfx" # 音效 TRACK_TYPE_AMBIENT = "ambient" # 环境音 +MAX_AUDIO_TRACKS = 8 # 最大混音轨道数（安全上限，防止资源耗尽） + # 各轨道默认音量（相对主音频） DEFAULT_VOLUMES = { TRACK_TYPE_MAIN: 1.0, @@ -153,6 +156,51 @@ class MultiTrackMixConfig: return len([t for t in self.tracks if t.enabled and t.audio_path]) > 0 +# ── 路径安全校验 ──────────────────────────────────────────────────────────── + + +ALLOWED_AUDIO_EXTENSIONS = {".mp3", ".wav", ".aac", ".ogg", ".flac", ".m4a", ".wma"} + + +def _validate_audio_path(audio_path: str, work_dir: Path) -> None: + """校验音频文件路径安全性. + + 规则： + - local:// schema → 必须在 work_dir 内 + - 相对路径 → 必须在 work_dir 内 + - 绝对路径 → 必须在允许目录白名单内 + - 扩展名必须是音频格式 + + Raises: + PathSecurityError: 路径不安全 + """ + if not audio_path or not isinstance(audio_path, str): + raise PathSecurityError("音频路径不能为空") + + # 本地路径（local:// 或相对路径） + if audio_path.startswith("local://") or not audio_path.startswith(("http://", "https://", "oss://")): + safe_resolve_path( + audio_path, + work_dir, + allow_outside=audio_path.startswith(("/", "local://")) and not audio_path.startswith("local://"), + allowed_extensions=ALLOWED_AUDIO_EXTENSIONS, + ) + # 绝对路径额外检查白名单目录 + if audio_path.startswith("/") and not audio_path.startswith(str(work_dir)): + if not is_in_allowed_dirs(audio_path): + raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}") + # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责） + # 但检查扩展名 + else: + # URL路径，检查扩展名白名单（取 ? 之前的部分） + path_part = audio_path.split("?")[0].split("#")[0] + from pathlib import Path as _P + + ext = _P(path_part).suffix.lower() + if ext and ext not in ALLOWED_AUDIO_EXTENSIONS: + raise PathSecurityError(f"不允许的音频文件类型: {ext}") + + # ── 单轨道预处理 ──────────────────────────────────────────────────────────── @@ -289,6 +337,39 @@ def mix_multi_track( if target_duration <= 0: target_duration = 5.0 + # ── 安全校验：轨道数量上限 ── + enabled_tracks = [t for t in config.tracks if t.enabled and t.audio_path] + if len(enabled_tracks) > MAX_AUDIO_TRACKS: + logger.warning( + "[multi-track] too many tracks: %d > %d, truncating to max", + len(enabled_tracks), + MAX_AUDIO_TRACKS, + ) + enabled_tracks = enabled_tracks[:MAX_AUDIO_TRACKS] + # 更新 config.tracks 为截断后的列表 + config.tracks = enabled_tracks + + # ── 安全校验：所有音频路径白名单校验 ── + # 主音频路径 + try: + _validate_audio_path(str(main_audio_path), ctx.work_dir) + except PathSecurityError as e: + logger.error("[multi-track] main audio path security check failed: %s", e) + raise + + # 各轨道音频路径 + valid_tracks = [] + for track in enabled_tracks: + try: + _validate_audio_path(track.audio_path, ctx.work_dir) + valid_tracks.append(track) + except PathSecurityError as e: + logger.warning("[multi-track] skip track %s: path security check failed: %s", track.track_id, e) + + if len(valid_tracks) != len(enabled_tracks): + config.tracks = valid_tracks + logger.info("[multi-track] %d tracks passed security check", len(valid_tracks)) + # 收集所有有效轨道（已预处理好的） prepared_tracks: list[Path] = [] diff --git a/apps/worker/video_processing/path_security.py b/apps/worker/video_processing/path_security.py new file mode 100644 index 000000000..f393457de --- /dev/null +++ b/apps/worker/video_processing/path_security.py @@ -0,0 +1,303 @@ +"""路径安全校验工具 — 路径遍历防护. + +统一的文件路径安全校验方案，覆盖所有渲染管线中的路径处理场景： +- 本地素材路径校验 +- local:// 路径 schema 校验 +- 工作目录内路径安全约束 +- 防止路径遍历攻击 (../) + +防护要点： +1. 所有用户可控路径必须在允许的目录内 +2. 解析符号链接后的真实路径仍需在允许目录内 +3. 禁止空路径、相对路径遍历、绝对路径逃逸 +4. 路径字符限制与规范化 +""" + +from __future__ import annotations + +import logging +import os +from pathlib import Path + +logger = logging.getLogger(__name__) + +# 最大路径长度 +MAX_PATH_LENGTH = 4096 + +# 允许的文件扩展名（渲染相关） +ALLOWED_MEDIA_EXTENSIONS = { + ".mp4", + ".mov", + ".avi", + ".mkv", + ".webm", + ".flv", + ".wmv", # 视频 + ".mp3", + ".wav", + ".aac", + ".ogg", + ".flac", + ".m4a", + ".wma", # 音频 + ".jpg", + ".jpeg", + ".png", + ".gif", + ".bmp", + ".webp", + ".tiff", # 图片 + ".srt", + ".ass", + ".vtt", + ".sub", # 字幕 + ".txt", + ".json", # 文本/配置 +} + +# local:// schema 前缀 +LOCAL_SCHEMA_PREFIX = "local://" + + +class PathSecurityError(ValueError): + """路径安全校验失败.""" + + pass + + +def safe_resolve_path( + input_path: str | Path, + base_dir: str | Path, + *, + allow_outside: bool = False, + allowed_extensions: set[str] | None = None, +) -> Path: + """安全解析路径，确保最终路径在 base_dir 内. + + Args: + input_path: 输入路径（相对或绝对） + base_dir: 基路径目录，解析后的路径必须在此目录内 + allow_outside: 是否允许路径在 base_dir 外（默认禁止） + allowed_extensions: 允许的文件扩展名集合（None 表示不限制） + + Returns: + 解析后的绝对路径 Path 对象 + + Raises: + PathSecurityError: 路径不安全 + """ + if input_path is None: + raise PathSecurityError("路径不能为空") + + path_str = str(input_path).strip() + if not path_str: + raise PathSecurityError("路径不能为空") + + if len(path_str) > MAX_PATH_LENGTH: + raise PathSecurityError(f"路径过长 ({len(path_str)} > {MAX_PATH_LENGTH})") + + # 空字节检测（必须在 Path() 之前） + if "\x00" in path_str: + raise PathSecurityError("路径包含空字节") + + # 处理 local:// schema + if path_str.startswith(LOCAL_SCHEMA_PREFIX): + path_str = path_str[len(LOCAL_SCHEMA_PREFIX) :] + # local:// 后必须是相对路径（相对于 base_dir），不能是绝对路径 + if os.path.isabs(path_str): + raise PathSecurityError("local:// 路径不能是绝对路径") + + # 规范化 base_dir + base_dir = Path(base_dir).resolve() + if not base_dir.is_dir(): + raise PathSecurityError(f"基路径不是有效目录: {base_dir}") + + # 解析输入路径 + input_path_obj = Path(path_str) + + # 如果是绝对路径且不允许外部路径 + if input_path_obj.is_absolute() and not allow_outside: + raise PathSecurityError("禁止使用绝对路径（需在工作目录内）") + + # 组合并解析为绝对路径 + if input_path_obj.is_absolute(): + full_path = input_path_obj.resolve() + else: + full_path = (base_dir / input_path_obj).resolve() + + # 检查路径遍历 — 确保最终路径在 base_dir 内 + if not allow_outside: + try: + full_path.relative_to(base_dir) + except ValueError: + raise PathSecurityError(f"路径遍历检测：路径 '{path_str}' 超出基路径 '{base_dir}' 范围") + + # 扩展名校验 + if allowed_extensions is not None: + ext = full_path.suffix.lower() + if ext and ext not in allowed_extensions: + raise PathSecurityError(f"不允许的文件类型: {ext}") + + # 检查危险路径模式 + _check_dangerous_patterns(full_path) + + return full_path + + +def _check_dangerous_patterns(path: Path) -> None: + """检查危险路径模式.""" + path_str = str(path) + + # 检查空字节 + if "\x00" in path_str: + raise PathSecurityError("路径包含空字节") + + # 检查特殊设备文件（Linux） + dangerous_prefixes = [ + "/proc/", + "/sys/", + "/dev/", + "/etc/passwd", + "/etc/shadow", + "/root/", + "/boot/", + "/var/run/", + ] + for prefix in dangerous_prefixes: + if path_str.startswith(prefix): + raise PathSecurityError(f"禁止访问系统路径: {prefix}") + + +def is_path_safe( + input_path: str | Path, + base_dir: str | Path, + *, + allow_outside: bool = False, +) -> bool: + """便捷函数：检查路径是否安全，不抛异常.""" + try: + safe_resolve_path(input_path, base_dir, allow_outside=allow_outside) + return True + except PathSecurityError: + return False + + +def validate_local_schema_path( + schema_path: str, + work_dir: str | Path, +) -> Path: + """校验 local:// schema 路径，返回安全的本地路径. + + local:// 路径规则： + - 必须以 local:// 开头 + - 后面必须是相对路径 + - 最终解析后必须在 work_dir 内 + - 不允许 ../ 遍历 + + Args: + schema_path: local:// 开头的路径 + work_dir: 工作目录 + + Returns: + 解析后的安全路径 + + Raises: + PathSecurityError: 路径不安全 + """ + if not schema_path.startswith(LOCAL_SCHEMA_PREFIX): + raise PathSecurityError(f"路径必须以 {LOCAL_SCHEMA_PREFIX} 开头") + + return safe_resolve_path(schema_path, work_dir, allow_outside=False) + + +def sanitize_filename(filename: str) -> str: + """清理文件名，移除危险字符. + + 保留：字母、数字、下划线、连字符、点、中文字符 + 移除：路径分隔符、控制字符、特殊符号等 + """ + import re + + if not filename: + return "unnamed" + + # 移除路径分隔符和危险字符 + # 保留: 字母数字、中文字符、下划线、连字符、点、空格 + sanitized = re.sub(r'[\\/\x00-\x1f\x7f<>:"|?*]', "_", filename) + + # 移除开头的点和连续的点（防止隐藏文件和路径遍历） + while sanitized.startswith("."): + sanitized = sanitized[1:] + + # 限制长度 + if len(sanitized) > 255: + name, ext = os.path.splitext(sanitized) + sanitized = name[: 255 - len(ext)] + ext + + # 空文件名兜底 + if not sanitized or sanitized == ".": + sanitized = "unnamed" + + return sanitized + + +# ── 允许目录配置 ────────────────────────────────────────────────────────────── + + +def get_allowed_local_dirs() -> list[Path]: + """获取允许的本地素材目录列表（从环境变量读取）. + + 环境变量 ASSET_ALLOWED_DIRS，多个目录用冒号分隔（Linux）或分号分隔（Windows）。 + 默认包含 /tmp。 + + 用于： + - resolve_asset_path 本地绝对路径白名单 + - PiP local_path 类型白名单 + - 贴纸本地路径白名单 + """ + env_dirs = os.environ.get("ASSET_ALLOWED_DIRS", "") + dirs: list[Path] = [] + if env_dirs: + import re + + sep = ";" if os.name == "nt" else ":" + for d in re.split(f"[{sep}]", env_dirs): + d = d.strip() + if d: + try: + dirs.append(Path(d).resolve()) + except OSError: + pass + # 默认允许 /tmp + if not dirs: + try: + dirs.append(Path("/tmp").resolve()) # nosec B108 + except OSError: + pass + return dirs + + +def is_in_allowed_dirs(path: str | Path, allowed_dirs: list[Path] | None = None) -> bool: + """检查路径是否在允许的目录列表内. + + Args: + path: 待检查的路径 + allowed_dirs: 允许的目录列表，None 则使用默认配置 + + Returns: + True 表示在允许目录内 + """ + if allowed_dirs is None: + allowed_dirs = get_allowed_local_dirs() + + try: + resolved = Path(path).resolve() + for allowed in allowed_dirs: + try: + resolved.relative_to(allowed) + return True + except ValueError: + continue + return False + except OSError: + return False diff --git a/apps/worker/video_processing/subtitle_render_engine.py b/apps/worker/video_processing/subtitle_render_engine.py index 931b2240a..ff6174ee3 100755 --- a/apps/worker/video_processing/subtitle_render_engine.py +++ b/apps/worker/video_processing/subtitle_render_engine.py @@ -30,12 +30,15 @@ from typing import Any from video_processing.render_subtitles import generate_ass_subtitles from video_processing.subtitle_generator import generate_ass_from_timeline +from video_processing.path_security import PathSecurityError, safe_resolve_path, is_in_allowed_dirs logger = logging.getLogger(__name__) # ── 常量 ────────────────────────────────────────────────────────────────────── +ALLOWED_SUBTITLE_EXTENSIONS = {".srt", ".ass", ".vtt", ".sub"} + # 9宫格位置映射（ASS alignment 编号） POSITION_ALIGNMENT = { "top_left": 7, @@ -614,6 +617,7 @@ def build_subtitle_filter( *, video_input_label: str = "0:v", output_label: str = "subtitled", + work_dir: Path | str | None = None, ) -> str: """生成 FFmpeg subtitles 滤镜字符串. @@ -621,13 +625,59 @@ def build_subtitle_filter( ass_path: ASS 字幕文件路径 video_input_label: 视频输入标签（如 "0:v" 或 "[v_out]"） output_label: 输出标签 + work_dir: 工作目录（用于路径安全校验，不传则用 ass_path 父目录） Returns: filter_complex 片段，如 "[0:v]subtitles=xxx.ass[subtitled]" + + Raises: + PathSecurityError: 字幕路径不安全 """ + # ── 安全校验：字幕文件路径白名单 ── + ass_path_str = str(ass_path) + if work_dir is None: + work_dir = str(Path(ass_path_str).parent) + + _validate_subtitle_path(ass_path_str, Path(work_dir)) + # FFmpeg subtitles filter 的路径需要转义： # - Windows 路径的 \ → / # - 冒号 : → \: # - 单引号 ' → '\'' - safe_path = str(ass_path).replace("\\", "/").replace(":", "\\:").replace("'", "'\\''") + safe_path = ass_path_str.replace("\\", "/").replace(":", "\\:").replace("'", "'\\''") return f"{video_input_label}subtitles='{safe_path}'[{output_label}]" + + +def _validate_subtitle_path(subtitle_path: str, work_dir: Path) -> None: + """校验字幕文件路径安全性. + + 规则： + - 必须是本地路径（不支持远程URL字幕） + - local:// schema → 必须在 work_dir 内 + - 相对路径 → 必须在 work_dir 内 + - 绝对路径 → 必须在允许目录白名单内 + - 扩展名必须是字幕格式 + + Raises: + PathSecurityError: 路径不安全 + """ + if not subtitle_path or not isinstance(subtitle_path, str): + raise PathSecurityError("字幕路径不能为空") + + # 不允许远程URL字幕（subtitles滤镜不支持远程加载，且有SSRF风险） + if subtitle_path.startswith(("http://", "https://", "oss://")): + raise PathSecurityError("不允许使用远程URL字幕文件") + + is_abs = subtitle_path.startswith("/") and not subtitle_path.startswith("local://") + + safe_resolve_path( + subtitle_path, + work_dir, + allow_outside=is_abs, + allowed_extensions=ALLOWED_SUBTITLE_EXTENSIONS, + ) + + # 绝对路径额外检查白名单目录 + if is_abs and not subtitle_path.startswith(str(work_dir)): + if not is_in_allowed_dirs(subtitle_path): + raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}") diff --git a/packages/shared/url_security.py b/packages/shared/url_security.py new file mode 100644 index 000000000..c96a43e3f --- /dev/null +++ b/packages/shared/url_security.py @@ -0,0 +1,413 @@ +"""URL 安全校验工具 — SSRF 防护. + +统一的外部 URL 安全校验方案，覆盖所有渲染管线和 TTS 中的外部下载场景。 +放在 packages/shared/ 作为单一来源，worker 和 application 层都可引用。 + +防护要点： +1. Scheme 白名单：仅允许 http/https +2. 主机 SSRF 防护：禁止内网 IP、回环地址、链路本地地址、元数据服务 +3. 端口白名单：仅允许 80/443（标准 HTTP/HTTPS） +4. 域名校验：禁止 IP 直接访问（除非在白名单中） +5. 重定向防护：手动跟随重定向，每次跳转前重新校验目标 URL +6. 文件大小限制：流式下载，超过上限立即中断 +7. MIME 类型白名单：可选的内容类型校验 +""" + +from __future__ import annotations + +import ipaddress +import logging +import os +import socket +import urllib.error +import urllib.request +from urllib.parse import urljoin, urlparse + +logger = logging.getLogger(__name__) + +# 允许的 URL scheme +ALLOWED_SCHEMES = {"http", "https"} + +# 允许的端口（标准 HTTP/HTTPS） +ALLOWED_PORTS = {80, 443} + +# 可信域名白名单（可根据实际 OSS/CDN 域名配置） +# 从环境变量读取，格式："oss-cn-hangzhou.aliyuncs.com,cdn.example.com" +# 默认空表示所有公网域名都允许，但仍会做 SSRF 检查 +TRUSTED_DOMAINS: set[str] = set() +_env_trusted = os.environ.get("URL_SECURITY_TRUSTED_DOMAINS", "") +if _env_trusted: + TRUSTED_DOMAINS = {d.strip() for d in _env_trusted.split(",") if d.strip()} + +# 是否允许 IP 直接访问（默认禁止，防止绕过 DNS 校验） +ALLOW_DIRECT_IP = os.environ.get("URL_SECURITY_ALLOW_DIRECT_IP", "false").lower() == "true" + +# 最大 URL 长度 +MAX_URL_LENGTH = 2048 + +# 单次下载最大文件大小（默认 200MB） +DEFAULT_MAX_DOWNLOAD_SIZE = int( + os.environ.get("URL_SECURITY_MAX_DOWNLOAD_MB", "200") +) * 1024 * 1024 + +# 允许的音频 MIME 类型白名单 +ALLOWED_AUDIO_MIME_TYPES = { + "audio/mpeg", + "audio/mp3", + "audio/wav", + "audio/x-wav", + "audio/pcm", + "audio/ogg", + "audio/opus", + "audio/flac", + "audio/aac", + "audio/m4a", + "audio/x-m4a", + "audio/mp4", + "application/octet-stream", # 兼容一些 CDN 返回通用类型 +} + +# 允许的视频 MIME 类型白名单 +ALLOWED_VIDEO_MIME_TYPES = { + "video/mp4", + "video/quicktime", + "video/x-matroska", + "video/webm", + "video/avi", + "video/x-msvideo", + "video/mpeg", + "application/octet-stream", +} + +# 允许的图片 MIME 类型白名单 +ALLOWED_IMAGE_MIME_TYPES = { + "image/jpeg", + "image/png", + "image/gif", + "image/webp", + "image/bmp", +} + +# 下载块大小 +_DOWNLOAD_CHUNK_SIZE = 8192 + +# 最大重定向次数 +_MAX_REDIRECTS = 5 + + +class UrlSecurityError(ValueError): + """URL 安全校验失败.""" + + pass + + +class NoRedirectHandler(urllib.request.HTTPRedirectHandler): + """禁止自动重定向的 handler，用于手动控制重定向以做安全校验.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: N802 + return None + + +def validate_url_safety(url: str, *, purpose: str = "download") -> str: + """校验 URL 安全性，返回标准化后的 URL（供下游使用）. + + Args: + url: 待校验的 URL + purpose: 用途描述（用于日志），如 "bgm_download"、"tts_download" + + Returns: + 标准化后的 URL + + Raises: + UrlSecurityError: URL 不安全 + """ + if not url: + raise UrlSecurityError("URL 为空") + + if len(url) > MAX_URL_LENGTH: + raise UrlSecurityError(f"URL 过长 ({len(url)} > {MAX_URL_LENGTH})") + + # 解析 URL + try: + parsed = urlparse(url) + except Exception as e: + raise UrlSecurityError(f"URL 解析失败: {e}") from e + + # 1. Scheme 校验 + if not parsed.scheme or parsed.scheme.lower() not in ALLOWED_SCHEMES: + raise UrlSecurityError(f"不允许的 URL scheme: {parsed.scheme}") + + # 2. 主机名校验 + hostname = parsed.hostname + if not hostname: + raise UrlSecurityError("URL 缺少主机名") + + # 2.1 常见内网主机名前置拦截（防止 DNS rebinding 绕过） + _check_internal_hostnames(hostname) + + # 3. 端口校验 + port = parsed.port + if port is not None and port not in ALLOWED_PORTS: + raise UrlSecurityError(f"不允许的端口: {port}") + + # 4. SSRF 防护 - 解析 IP 并检查 + try: + # 先判断是否是 IP 地址 + ip_obj = None + try: + ip_obj = ipaddress.ip_address(hostname) + except ValueError: + pass # 不是 IP，继续走域名解析 + + if ip_obj is not None: + # 是直接 IP 访问 + if not ALLOW_DIRECT_IP and not _is_trusted_ip(ip_obj): + raise UrlSecurityError(f"禁止直接 IP 访问: {hostname}") + _check_ssrf_ip(ip_obj) + else: + # 域名 — 解析 DNS 检查 SSRF + _check_ssrf_domain(hostname) + except UrlSecurityError: + raise + except Exception as e: + logger.warning("URL 安全校验异常: url=%s purpose=%s error=%s", url[:80], purpose, e) + raise UrlSecurityError(f"URL 安全校验异常: {e}") from e + + # 5. 可信域名校验（如果配置了白名单） + if TRUSTED_DOMAINS and not _is_trusted_domain(hostname): + raise UrlSecurityError(f"域名不在可信白名单中: {hostname}") + + logger.debug("URL 安全校验通过: url=%s purpose=%s", url[:80], purpose) + return url + + +def _check_internal_hostnames(hostname: str) -> None: + """前置检查常见内网/敏感主机名，防止 DNS 解析层绕过.""" + hostname_lower = hostname.lower() + internal_hostnames = { + "localhost", + "localhost.localdomain", + "ip6-localhost", + "ip6-loopback", + "metadata", + "metadata.google.internal", + "169.254.169.254", # 云元数据服务 + } + if hostname_lower in internal_hostnames: + raise UrlSecurityError(f"禁止访问内部主机名: {hostname}") + + # 检查以 .local / .internal 结尾的主机名 + if hostname_lower.endswith((".local", ".internal", ".localdomain")): + raise UrlSecurityError(f"禁止访问内网域名: {hostname}") + + +def _check_ssrf_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None: + """检查 IP 是否属于 SSRF 风险范围.""" + # 回环地址 + if ip_obj.is_loopback: + raise UrlSecurityError(f"禁止访问回环地址: {ip_obj}") + + # 私有地址（内网） + if ip_obj.is_private: + raise UrlSecurityError(f"禁止访问内网地址: {ip_obj}") + + # 链路本地地址 + if ip_obj.is_link_local: + raise UrlSecurityError(f"禁止访问链路本地地址: {ip_obj}") + + # 组播地址 + if ip_obj.is_multicast: + raise UrlSecurityError(f"禁止访问组播地址: {ip_obj}") + + # 未指定地址（0.0.0.0 / ::） + if ip_obj.is_unspecified: + raise UrlSecurityError(f"禁止访问未指定地址: {ip_obj}") + + # 保留地址 + if ip_obj.is_reserved: + raise UrlSecurityError(f"禁止访问保留地址: {ip_obj}") + + +def _check_ssrf_domain(hostname: str) -> None: + """对域名做 DNS 解析并检查所有解析结果的 IP 是否安全. + + 注意：这不能完全防止 DNS rebinding，但能防御大部分 SSRF 场景。 + """ + try: + # 解析所有地址 + infos = socket.getaddrinfo(hostname, None) + if not infos: + raise UrlSecurityError(f"域名解析失败: {hostname}") + + for info in infos: + ip_str = info[4][0] + try: + ip_obj = ipaddress.ip_address(ip_str) + _check_ssrf_ip(ip_obj) + except ValueError: + # 无法解析为 IP，跳过（不应该发生） + continue + except socket.gaierror as e: + raise UrlSecurityError(f"域名解析失败: {hostname} ({e})") from e + + +def _is_trusted_ip(ip_obj: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + """检查 IP 是否在可信列表中（目前通过环境变量配置域名，IP 级信任暂不开放）.""" + return False + + +def _is_trusted_domain(hostname: str) -> bool: + """检查域名是否在可信白名单中（支持子域名匹配）.""" + hostname_lower = hostname.lower() + if hostname_lower in TRUSTED_DOMAINS: + return True + # 检查子域名 + for domain in TRUSTED_DOMAINS: + if hostname_lower.endswith("." + domain.lower()): + return True + return False + + +def is_url_safe(url: str, *, purpose: str = "download") -> bool: + """便捷函数：检查 URL 是否安全，不抛异常.""" + try: + validate_url_safety(url, purpose=purpose) + return True + except UrlSecurityError: + return False + + +# ── 安全下载 ─────────────────────────────────────────────────────────────────── + + +def safe_download_file( + url: str, + dest_path: str, + *, + purpose: str = "download", + max_size: int = DEFAULT_MAX_DOWNLOAD_SIZE, + allowed_mime_types: set[str] | None = None, + timeout: float = 60.0, +) -> int: + """安全下载 URL 到本地文件。 + + 包含防护： + - SSRF 校验（初始 URL + 每次重定向后都校验） + - 重定向次数限制 + 手动跟随（避免重定向绕过 SSRF） + - 文件大小限制（流式读取，超过立即中断） + - MIME 类型白名单（可选） + + Args: + url: 下载 URL + dest_path: 目标文件路径 + purpose: 用途描述（日志用） + max_size: 最大下载字节数，超过则中断并抛出 UrlSecurityError + allowed_mime_types: 允许的 Content-Type 集合，None 表示不校验 + timeout: 单次请求超时（秒） + + Returns: + 实际下载的字节数 + + Raises: + UrlSecurityError: 安全校验失败 + """ + current_url = url + redirect_count = 0 + total_bytes = 0 + + # 使用不自动跟随重定向的 opener + no_redirect_opener = urllib.request.build_opener(NoRedirectHandler()) + + while True: + # 每次请求前都做 SSRF 校验（重定向目标也会校验） + validate_url_safety(current_url, purpose=purpose) + + req = urllib.request.Request(current_url, method="GET") + req.add_header("User-Agent", "xiaoxia-saas-worker/1.0") + + try: + resp = no_redirect_opener.open(req, timeout=timeout) # nosec B310 + except urllib.error.HTTPError as e: + # 3xx 重定向 + if 300 <= e.code < 400 and e.headers.get("Location"): + if redirect_count >= _MAX_REDIRECTS: + raise UrlSecurityError( + f"重定向次数超过限制 ({_MAX_REDIRECTS})" + ) from e + redirect_count += 1 + current_url = urljoin(current_url, e.headers["Location"]) + continue + raise UrlSecurityError(f"HTTP 错误: {e.code} {e.reason}") from e + except urllib.error.URLError as e: + raise UrlSecurityError(f"URL 错误: {e.reason}") from e + + try: + # Content-Type 校验 + if allowed_mime_types is not None: + content_type = ( + resp.headers.get("Content-Type", "").split(";")[0].strip().lower() + ) + if content_type and content_type not in allowed_mime_types: + raise UrlSecurityError( + f"不允许的 Content-Type: {content_type}, " + f"允许: {sorted(allowed_mime_types)}" + ) + + # Content-Length 预检 + content_length = resp.headers.get("Content-Length") + if content_length and int(content_length) > max_size: + raise UrlSecurityError( + f"文件过大: {content_length} bytes > {max_size} bytes 上限" + ) + + # 流式下载，实时检查大小 + with open(dest_path, "wb") as f: + while True: + chunk = resp.read(_DOWNLOAD_CHUNK_SIZE) + if not chunk: + break + total_bytes += len(chunk) + if total_bytes > max_size: + raise UrlSecurityError( + f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes" + ) + f.write(chunk) + + return total_bytes + finally: + resp.close() + + +def safe_download_bytes( + url: str, + *, + purpose: str = "download", + max_size: int = DEFAULT_MAX_DOWNLOAD_SIZE, + allowed_mime_types: set[str] | None = None, + timeout: float = 60.0, +) -> bytes: + """安全下载 URL 并返回字节内容。 + + 防护同 safe_download_file，但结果返回在内存中（适合小文件）。 + """ + import tempfile + + fd, tmp_path = tempfile.mkstemp() + os.close(fd) + + try: + safe_download_file( + url, + tmp_path, + purpose=purpose, + max_size=max_size, + allowed_mime_types=allowed_mime_types, + timeout=timeout, + ) + with open(tmp_path, "rb") as f: + return f.read() + finally: + try: + os.unlink(tmp_path) + except OSError: + pass diff --git a/tests/unit/test_pr312_security_debt.py b/tests/unit/test_pr312_security_debt.py new file mode 100755 index 000000000..0c9a3c2e1 --- /dev/null +++ b/tests/unit/test_pr312_security_debt.py @@ -0,0 +1,439 @@ +"""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, + MultiTrackMixConfig, + AudioTrack, + 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, + MultiTrackMixConfig, + AudioTrack, + 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) + + +# ══════════════════════════════════════════════════════════════════════════════ +# 2. 视频拼接安全测试 +# ══════════════════════════════════════════════════════════════════════════════ + + +class TestConcatSecurity: + """视频拼接安全测试.""" + + def test_segment_count_limit_exceeded(self, work_dir, sample_video): + """超过最大段数时应报错.""" + from video_processing.concat_engine import ( + MAX_CONCAT_SEGMENTS, + ConcatConfig, + ConcatSegment, + ConcatEngine, + ) + + # 创建超过上限的段数 + 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, + ConcatSegment, + ConcatEngine, + ) + + 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_invalid_segments_skipped(self, work_dir, sample_video): + """路径不安全的片段应被跳过.""" + from unittest.mock import patch + + from video_processing.concat_engine import ( + ConcatConfig, + ConcatSegment, + ConcatEngine, + ) + + 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_default_work_dir(self, work_dir): + """不传work_dir时默认用ass_path父目录.""" + 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，应该用父目录作为 base + result = build_subtitle_filter(ass_file) + assert "subtitles=" in result -- 2.54.0 From 4792acb5d9502f9058a87aba403899af14df1d62 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 14 Jul 2026 16:40:58 +0800 Subject: [PATCH 2/7] style: black + isort format --- apps/worker/video_processing/concat_engine.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/apps/worker/video_processing/concat_engine.py b/apps/worker/video_processing/concat_engine.py index d18afe533..af9bd2a5f 100755 --- a/apps/worker/video_processing/concat_engine.py +++ b/apps/worker/video_processing/concat_engine.py @@ -226,9 +226,7 @@ class ConcatEngine: # ── 安全校验：段数上限 ── if len(valid_segments) > MAX_CONCAT_SEGMENTS: - raise ValueError( - f"Too many concat segments: {len(valid_segments)} > {MAX_CONCAT_SEGMENTS}" - ) + raise ValueError(f"Too many concat segments: {len(valid_segments)} > {MAX_CONCAT_SEGMENTS}") # ── 安全校验：所有视频路径白名单校验 ── safe_segments = [] -- 2.54.0 From 439420d44c00693214eedf713774ad772e882d59 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 14 Jul 2026 16:40:58 +0800 Subject: [PATCH 3/7] style: black + isort format --- packages/shared/url_security.py | 23 ++++++----------------- 1 file changed, 6 insertions(+), 17 deletions(-) diff --git a/packages/shared/url_security.py b/packages/shared/url_security.py index c96a43e3f..75d64e9ed 100644 --- a/packages/shared/url_security.py +++ b/packages/shared/url_security.py @@ -46,9 +46,7 @@ ALLOW_DIRECT_IP = os.environ.get("URL_SECURITY_ALLOW_DIRECT_IP", "false").lower( MAX_URL_LENGTH = 2048 # 单次下载最大文件大小（默认 200MB） -DEFAULT_MAX_DOWNLOAD_SIZE = int( - os.environ.get("URL_SECURITY_MAX_DOWNLOAD_MB", "200") -) * 1024 * 1024 +DEFAULT_MAX_DOWNLOAD_SIZE = int(os.environ.get("URL_SECURITY_MAX_DOWNLOAD_MB", "200")) * 1024 * 1024 # 允许的音频 MIME 类型白名单 ALLOWED_AUDIO_MIME_TYPES = { @@ -331,9 +329,7 @@ def safe_download_file( # 3xx 重定向 if 300 <= e.code < 400 and e.headers.get("Location"): if redirect_count >= _MAX_REDIRECTS: - raise UrlSecurityError( - f"重定向次数超过限制 ({_MAX_REDIRECTS})" - ) from e + raise UrlSecurityError(f"重定向次数超过限制 ({_MAX_REDIRECTS})") from e redirect_count += 1 current_url = urljoin(current_url, e.headers["Location"]) continue @@ -344,21 +340,16 @@ def safe_download_file( try: # Content-Type 校验 if allowed_mime_types is not None: - content_type = ( - resp.headers.get("Content-Type", "").split(";")[0].strip().lower() - ) + content_type = resp.headers.get("Content-Type", "").split(";")[0].strip().lower() if content_type and content_type not in allowed_mime_types: raise UrlSecurityError( - f"不允许的 Content-Type: {content_type}, " - f"允许: {sorted(allowed_mime_types)}" + f"不允许的 Content-Type: {content_type}, " f"允许: {sorted(allowed_mime_types)}" ) # Content-Length 预检 content_length = resp.headers.get("Content-Length") if content_length and int(content_length) > max_size: - raise UrlSecurityError( - f"文件过大: {content_length} bytes > {max_size} bytes 上限" - ) + raise UrlSecurityError(f"文件过大: {content_length} bytes > {max_size} bytes 上限") # 流式下载，实时检查大小 with open(dest_path, "wb") as f: @@ -368,9 +359,7 @@ def safe_download_file( break total_bytes += len(chunk) if total_bytes > max_size: - raise UrlSecurityError( - f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes" - ) + raise UrlSecurityError(f"下载超过大小限制: {total_bytes} bytes > {max_size} bytes") f.write(chunk) return total_bytes -- 2.54.0 From 7e9bcbc3ae202bfef3d3b348e4da6f9f271a8a92 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 14 Jul 2026 16:40:59 +0800 Subject: [PATCH 4/7] style: black + isort format --- tests/unit/test_pr312_security_debt.py | 41 ++++++++++++++------------ 1 file changed, 22 insertions(+), 19 deletions(-) diff --git a/tests/unit/test_pr312_security_debt.py b/tests/unit/test_pr312_security_debt.py index 0c9a3c2e1..4e82ea757 100755 --- a/tests/unit/test_pr312_security_debt.py +++ b/tests/unit/test_pr312_security_debt.py @@ -14,10 +14,8 @@ 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 ────────────────────────────────────────────────────────────────── @@ -103,20 +101,22 @@ class TestMultiTrackSecurity: from video_processing.multi_track_mixer import ( MAX_AUDIO_TRACKS, - MultiTrackMixConfig, 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, - )) + tracks.append( + AudioTrack( + track_id=f"track_{i}", + track_type="sfx", + audio_path=str(sample_audio), + volume=0.5, + ) + ) config = MultiTrackMixConfig(tracks=tracks) @@ -128,6 +128,7 @@ class TestMultiTrackSecurity: 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) @@ -141,19 +142,21 @@ class TestMultiTrackSecurity: from video_processing.multi_track_mixer import ( MAX_AUDIO_TRACKS, - MultiTrackMixConfig, 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, - )) + tracks.append( + AudioTrack( + track_id=f"track_{i}", + track_type="sfx", + audio_path=str(sample_audio), + volume=0.5, + ) + ) config = MultiTrackMixConfig(tracks=tracks) @@ -231,8 +234,8 @@ class TestConcatSecurity: from video_processing.concat_engine import ( MAX_CONCAT_SEGMENTS, ConcatConfig, - ConcatSegment, ConcatEngine, + ConcatSegment, ) # 创建超过上限的段数 @@ -254,8 +257,8 @@ class TestConcatSecurity: from video_processing.concat_engine import ( MAX_CONCAT_SEGMENTS, ConcatConfig, - ConcatSegment, ConcatEngine, + ConcatSegment, ) segments = [ @@ -324,8 +327,8 @@ class TestConcatSecurity: from video_processing.concat_engine import ( ConcatConfig, - ConcatSegment, ConcatEngine, + ConcatSegment, ) segments = [ -- 2.54.0 From 47bf6bfa5f4fef5228e22f9c22521c88b74c5d01 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 16:55:03 +0800 Subject: [PATCH 5/7] =?UTF-8?q?fix(security):=20#322=20=E4=BF=AE=E5=A4=8D2?= =?UTF-8?q?=E4=B8=AAP1=E7=BB=95=E8=BF=87=E6=BC=8F=E6=B4=9E=20-=20=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E6=A0=A1=E9=AA=8Crealpath=E8=A7=84=E8=8C=83=E5=8C=96?= =?UTF-8?q?=20+=20work=5Fdir=E5=BF=85=E5=A1=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1-1: 路径校验startswith绕过 - multi_track_mixer: 改用resolved_path.relative_to()替代原始字符串startswith - concat_engine: 同上 - subtitle_render_engine: 同上 - 原因: /tmp/work/../../opt/secret.aac 原始字符串以work_dir开头， 旧代码误判为在目录内从而跳过白名单校验 P1-2: build_subtitle_filter work_dir默认值绕过 - work_dir=None时不再用ass_path父目录自证清白 - 不传/None/空字符串一律抛PathSecurityError - 调用方必须显式传入可信的work_dir 新增4个单测验证绕过漏洞已修复，24个测试全绿 --- apps/worker/video_processing/concat_engine.py | 14 +++-- .../video_processing/multi_track_mixer.py | 17 +++--- .../subtitle_render_engine.py | 22 ++++---- tests/unit/test_pr312_security_debt.py | 52 +++++++++++++++++-- 4 files changed, 80 insertions(+), 25 deletions(-) mode change 100755 => 100644 tests/unit/test_pr312_security_debt.py diff --git a/apps/worker/video_processing/concat_engine.py b/apps/worker/video_processing/concat_engine.py index af9bd2a5f..93f414eb6 100755 --- a/apps/worker/video_processing/concat_engine.py +++ b/apps/worker/video_processing/concat_engine.py @@ -170,16 +170,20 @@ def _validate_video_path(video_path: str, work_dir: Path) -> None: # 本地路径（local:// 或相对路径 / 绝对路径） if video_path.startswith("local://") or not video_path.startswith(("http://", "https://", "oss://")): is_abs = video_path.startswith("/") and not video_path.startswith("local://") - safe_resolve_path( + resolved_path = safe_resolve_path( video_path, work_dir, allow_outside=is_abs, allowed_extensions=ALLOWED_VIDEO_EXTENSIONS, ) - # 绝对路径额外检查白名单目录 - if is_abs and not video_path.startswith(str(work_dir)): - if not is_in_allowed_dirs(video_path): - raise PathSecurityError(f"视频路径不在允许目录内: {video_path[:80]}") + # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过） + if is_abs: + resolved_work_dir = work_dir.resolve() + try: + resolved_path.relative_to(resolved_work_dir) + except ValueError: + if not is_in_allowed_dirs(resolved_path): + raise PathSecurityError(f"视频路径不在允许目录内: {video_path[:80]}") # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责） # 但检查扩展名 else: diff --git a/apps/worker/video_processing/multi_track_mixer.py b/apps/worker/video_processing/multi_track_mixer.py index 764cedfe6..15bea7ef1 100755 --- a/apps/worker/video_processing/multi_track_mixer.py +++ b/apps/worker/video_processing/multi_track_mixer.py @@ -179,16 +179,21 @@ def _validate_audio_path(audio_path: str, work_dir: Path) -> None: # 本地路径（local:// 或相对路径） if audio_path.startswith("local://") or not audio_path.startswith(("http://", "https://", "oss://")): - safe_resolve_path( + is_abs = audio_path.startswith("/") and not audio_path.startswith("local://") + resolved_path = safe_resolve_path( audio_path, work_dir, - allow_outside=audio_path.startswith(("/", "local://")) and not audio_path.startswith("local://"), + allow_outside=is_abs, allowed_extensions=ALLOWED_AUDIO_EXTENSIONS, ) - # 绝对路径额外检查白名单目录 - if audio_path.startswith("/") and not audio_path.startswith(str(work_dir)): - if not is_in_allowed_dirs(audio_path): - raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}") + # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过） + if is_abs: + resolved_work_dir = work_dir.resolve() + try: + resolved_path.relative_to(resolved_work_dir) + except ValueError: + if not is_in_allowed_dirs(resolved_path): + raise PathSecurityError(f"音频路径不在允许目录内: {audio_path[:80]}") # URL类型路径不做本地路径校验（由下载阶段的SSRF防护负责） # 但检查扩展名 else: diff --git a/apps/worker/video_processing/subtitle_render_engine.py b/apps/worker/video_processing/subtitle_render_engine.py index ff6174ee3..71575b7d2 100755 --- a/apps/worker/video_processing/subtitle_render_engine.py +++ b/apps/worker/video_processing/subtitle_render_engine.py @@ -625,18 +625,18 @@ def build_subtitle_filter( ass_path: ASS 字幕文件路径 video_input_label: 视频输入标签（如 "0:v" 或 "[v_out]"） output_label: 输出标签 - work_dir: 工作目录（用于路径安全校验，不传则用 ass_path 父目录） + work_dir: 工作目录（必填，用于路径安全校验，防止路径遍历绕过） Returns: filter_complex 片段，如 "[0:v]subtitles=xxx.ass[subtitled]" Raises: - PathSecurityError: 字幕路径不安全 + PathSecurityError: 字幕路径不安全或 work_dir 未提供 """ # ── 安全校验：字幕文件路径白名单 ── ass_path_str = str(ass_path) - if work_dir is None: - work_dir = str(Path(ass_path_str).parent) + if work_dir is None or not str(work_dir).strip(): + raise PathSecurityError("work_dir 必须提供，不能为 None 或空") _validate_subtitle_path(ass_path_str, Path(work_dir)) @@ -670,14 +670,18 @@ def _validate_subtitle_path(subtitle_path: str, work_dir: Path) -> None: is_abs = subtitle_path.startswith("/") and not subtitle_path.startswith("local://") - safe_resolve_path( + resolved_path = safe_resolve_path( subtitle_path, work_dir, allow_outside=is_abs, allowed_extensions=ALLOWED_SUBTITLE_EXTENSIONS, ) - # 绝对路径额外检查白名单目录 - if is_abs and not subtitle_path.startswith(str(work_dir)): - if not is_in_allowed_dirs(subtitle_path): - raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}") + # 绝对路径额外检查白名单目录（用realpath规范化后的真实路径比较，防止 ../ 遍历绕过） + if is_abs: + resolved_work_dir = work_dir.resolve() + try: + resolved_path.relative_to(resolved_work_dir) + except ValueError: + if not is_in_allowed_dirs(resolved_path): + raise PathSecurityError(f"字幕路径不在允许目录内: {subtitle_path[:80]}") diff --git a/tests/unit/test_pr312_security_debt.py b/tests/unit/test_pr312_security_debt.py old mode 100755 new mode 100644 index 4e82ea757..14225835d --- a/tests/unit/test_pr312_security_debt.py +++ b/tests/unit/test_pr312_security_debt.py @@ -220,6 +220,19 @@ class TestMultiTrackSecurity: 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. 视频拼接安全测试 @@ -321,6 +334,19 @@ class TestConcatSecurity: 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 @@ -430,13 +456,29 @@ class TestSubtitleSecurity: with pytest.raises(PathSecurityError): build_subtitle_filter("../../../etc/passwd", work_dir=work_dir) - def test_build_filter_default_work_dir(self, work_dir): - """不传work_dir时默认用ass_path父目录.""" + 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，应该用父目录作为 base - result = build_subtitle_filter(ass_file) - assert "subtitles=" in result + # 不传 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) -- 2.54.0 From 73f3334f85db9f3c568516ff9a486447e8587fd9 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 14 Jul 2026 17:19:45 +0800 Subject: [PATCH 6/7] style: black + isort format --- apps/worker/video_processing/subtitle_render_engine.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/worker/video_processing/subtitle_render_engine.py b/apps/worker/video_processing/subtitle_render_engine.py index 71575b7d2..cc640a324 100755 --- a/apps/worker/video_processing/subtitle_render_engine.py +++ b/apps/worker/video_processing/subtitle_render_engine.py @@ -28,9 +28,9 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Any +from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path from video_processing.render_subtitles import generate_ass_subtitles from video_processing.subtitle_generator import generate_ass_from_timeline -from video_processing.path_security import PathSecurityError, safe_resolve_path, is_in_allowed_dirs logger = logging.getLogger(__name__) -- 2.54.0 From d7b37ce59a06434572f226053ac0f1f64979ab51 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 17:21:23 +0800 Subject: [PATCH 7/7] =?UTF-8?q?fix:=20TestSubtitleFilter=E8=A1=A5work=5Fdi?= =?UTF-8?q?r=E5=8F=82=E6=95=B0=E9=80=82=E9=85=8D=E6=96=B0=E5=AE=89?= =?UTF-8?q?=E5=85=A8=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../unit/test_multi_track_subtitle_concat.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/tests/unit/test_multi_track_subtitle_concat.py b/tests/unit/test_multi_track_subtitle_concat.py index 3d80e565d..f24c83296 100755 --- a/tests/unit/test_multi_track_subtitle_concat.py +++ b/tests/unit/test_multi_track_subtitle_concat.py @@ -716,18 +716,29 @@ class TestBuildSubtitlesFromPlan: class TestSubtitleFilter: """字幕滤镜构建测试.""" - def test_build_subtitle_filter(self): + def test_build_subtitle_filter(self, work_dir): from video_processing.subtitle_render_engine import build_subtitle_filter - result = build_subtitle_filter("/tmp/test.ass", video_input_label="[v_in]", output_label="out") + ass_file = work_dir / "test.ass" + ass_file.write_text("test", encoding="utf-8") + + result = build_subtitle_filter( + str(ass_file), + video_input_label="[v_in]", + output_label="out", + work_dir=work_dir, + ) assert "subtitles=" in result assert "[v_in]" in result assert "[out]" in result - def test_default_labels(self): + def test_default_labels(self, work_dir): from video_processing.subtitle_render_engine import build_subtitle_filter - result = build_subtitle_filter("/tmp/sub.ass") + ass_file = work_dir / "sub.ass" + ass_file.write_text("test", encoding="utf-8") + + result = build_subtitle_filter(str(ass_file), work_dir=work_dir) assert "0:v" in result assert "[subtitled]" in result -- 2.54.0