xiaoxia-saas/apps/worker/video_processing/path_security.py

"""路径安全校验工具 — 路径遍历防护.

统一的文件路径安全校验方案，覆盖所有渲染管线中的路径处理场景：
- 本地素材路径校验
- 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