fix: 渲染引擎全链路安全加固 P0+P1
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 23s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 15m19s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 15m19s
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1537h55m59s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1537h56m1s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1537h56m2s
CI/CD Pipeline / Build Production Runtime Images (pull_request) Failing after 1537h56m3s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Failing after 1537h56m3s
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1538h27m38s

P0-1 SSRF防护:统一URL安全校验方案
- 新增 url_security.py:SSRF防护工具(scheme白名单、端口白名单、内网IP拦截、内网主机名拦截、DNS解析校验)
- generation.py URL校验函数增加SSRF前置校验
- PiP引擎URL类型增加安全校验
- 贴纸引擎URL类型增加安全校验

P0-2 路径遍历防护:输入路径校验强化
- 新增 path_security.py:路径安全工具(路径遍历防护、local:// schema校验、文件名清理、允许目录白名单)
- resolve_asset_path 加固:本地绝对路径白名单、空字节检测、存储键路径遍历拦截
- PiP引擎local_path类型:增加允许目录校验
- 贴纸引擎:区分本地路径/URL,分别增加安全校验

P1 裸subprocess统一:3处FFmpeg调用走run_ffmpeg
- unified_render_service.py _extract_audio:改用 run_ffmpeg
- compose_video.py:改用 run_ffmpeg
- edit_plan_generation.py:改用 run_ffmpeg

新增测试:
- test_url_security.py:29个SSRF防护单测
- test_path_security.py:27个路径遍历防护单测
- 原有171+105=276个单测零回归
This commit is contained in:
CI Bot
2026-07-14 13:18:16 +08:00
parent 2b5b650b9e
commit 9e10e31a29
11 changed files with 1062 additions and 55 deletions
+277
View File
@@ -0,0 +1,277 @@
"""路径安全校验工具 — 路径遍历防护.
统一的文件路径安全校验方案,覆盖所有渲染管线中的路径处理场景:
- 本地素材路径校验
- 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())
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