document-management/backend/dms/storage/paths.py

"""DMS存储目录创建和安全路径解析。"""

from __future__ import annotations

from pathlib import Path, PurePosixPath


STORAGE_SUBDIRECTORIES = (
    "original",
    "preview",
    "extracted",
    "temporary",
    "quarantine",
    "recycle",
)


class UnsafeStoragePathError(ValueError):
    """相对路径试图逃逸DMS存储根目录。"""


def ensure_storage_directories(storage_root: str | Path) -> Path:
    root = Path(storage_root).expanduser().resolve()
    root.mkdir(parents=True, exist_ok=True)
    for directory in STORAGE_SUBDIRECTORIES:
        (root / directory).mkdir(parents=True, exist_ok=True)
    return root


def resolve_storage_path(
    relative_path: str,
    storage_root: str | Path,
) -> Path:
    """安全解析数据库相对路径并保证结果仍位于存储根目录。"""
    if not relative_path or "\x00" in relative_path:
        raise UnsafeStoragePathError("存储相对路径不能为空或包含空字符")

    normalized = PurePosixPath(relative_path.replace("\\", "/"))
    has_drive_prefix = bool(normalized.parts and ":" in normalized.parts[0])
    if normalized.is_absolute() or has_drive_prefix or ".." in normalized.parts:
        raise UnsafeStoragePathError("存储路径必须是根目录内的安全相对路径")

    root = Path(storage_root).expanduser().resolve()
    candidate = root.joinpath(*normalized.parts).resolve()
    if not candidate.is_relative_to(root):
        raise UnsafeStoragePathError("存储路径超出DMS存储根目录")
    return candidate