"""路径安全校验工具 — 路径遍历防护. 统一的文件路径安全校验方案,覆盖所有渲染管线中的路径处理场景: - 本地素材路径校验 - 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 as _e: raise PathSecurityError(f"路径遍历检测:路径 '{path_str}' 超出基路径 '{base_dir}' 范围") from _e # 扩展名校验 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