"""Module for path validator.""" import re from typing import Any from apps.core.exceptions import ValidationException from apps.core.utils.path import Path class FolderPathValidator: def is_network_path(self, path: str) -> bool: value = (path or "").strip() return value.lower().startswith("smb://") or value.startswith("\\\\") def validate_folder_path(self, path: str) -> tuple[bool, str | None]: if not path or not path.strip(): return False, "请输入文件夹路径" value = path.strip() if value.startswith("/"): if re.search(r'[<>:"|?*]', value): return False, "路径包含非法字符" return True, None if re.match(r"^[A-Za-z]:[\\/]", value): rest = value[2:] if ":" in rest: return False, "路径包含非法字符" if re.search(r'[<>"|?*]', rest): return False, "路径包含非法字符" return True, None if value.lower().startswith("smb://") or value.startswith("\\\\"): return True, None return False, "请输入有效的文件夹路径" def sanitize_file_name(self, file_name: str) -> str: raw = "" if file_name is None else str(file_name) if not raw or not raw.strip(): raise ValidationException( message="文件名不能为空", code="INVALID_FILE_NAME", errors={"file_name": "文件名不能为空"} ) name = raw if "/" in name or "\\" in name: raise ValidationException( message="文件名包含非法路径分隔符", code="INVALID_FILE_NAME", errors={"file_name": "文件名不能包含路径分隔符"}, ) base = Path(name).name if not str(base).strip() or base in {".", ".."}: raise ValidationException( message="文件名无效", code="INVALID_FILE_NAME", errors={"file_name": "文件名无效"} ) return str(base) def sanitize_relative_dir(self, dir_path: str) -> list[str]: if not dir_path or not str(dir_path).strip(): raise ValidationException( message="子目录路径无效", code="INVALID_SUBDIR", errors={"subdir": "子目录路径不能为空"} ) raw = str(dir_path).strip().replace("\\", "/") if raw.startswith("/"): raise ValidationException( message="子目录路径必须为相对路径", code="INVALID_SUBDIR", errors={"subdir": "子目录路径必须为相对路径"} ) if re.match(r"^[A-Za-z]:[\\/]", raw): raise ValidationException( message="子目录路径必须为相对路径", code="INVALID_SUBDIR", errors={"subdir": "子目录路径必须为相对路径"} ) parts = [p for p in raw.split("/") if p not in {"", "."}] if not parts or any(part == ".." for part in parts): raise ValidationException( message="子目录路径包含非法片段", code="INVALID_SUBDIR", errors={"subdir": "子目录路径不能包含 .."} ) return parts def normalize_relative_path(self, relative_path: str) -> str: if relative_path is None: raise ValidationException( message="相对路径不能为空", code="INVALID_RELATIVE_PATH", errors={"relative_path": relative_path} ) raw = str(relative_path).strip().replace("\\", "/") if raw.startswith("/") or raw.startswith("~"): raise ValidationException( message="禁止使用绝对路径", code="INVALID_RELATIVE_PATH", errors={"relative_path": relative_path} ) if re.match(r"^[A-Za-z]:[\\/]", raw): raise ValidationException( message="禁止使用绝对路径", code="INVALID_RELATIVE_PATH", errors={"relative_path": relative_path} ) parts = [p for p in raw.split("/") if p not in {"", "."}] if any(p == ".." for p in parts): raise ValidationException( message="路径包含非法跳转", code="INVALID_RELATIVE_PATH", errors={"relative_path": relative_path} ) for p in parts: if re.search(r'[<>:"|?*]', p): raise ValidationException( message="路径包含非法字符", code="INVALID_RELATIVE_PATH", errors={"relative_path": relative_path} ) return "/".join(parts) def sanitize_zip_member_path(self, member_name: str) -> list[str]: if not member_name: raise ValidationException(message="压缩包条目无效", code="INVALID_ZIP_ENTRY", errors={"entry": "空条目"}) name = member_name.replace("\\", "/") while name.startswith("/"): name = name[1:] if not name or name in {".", ".."}: raise ValidationException(message="压缩包条目无效", code="INVALID_ZIP_ENTRY", errors={"entry": member_name}) if ":" in name.split("/", 1)[0]: raise ValidationException( message="压缩包条目包含非法路径", code="INVALID_ZIP_ENTRY", errors={"entry": member_name} ) parts = [p for p in name.split("/") if p not in {"", "."}] if not parts or any(p == ".." for p in parts): raise ValidationException( message="压缩包条目包含非法片段", code="INVALID_ZIP_ENTRY", errors={"entry": member_name} ) return parts def ensure_within_base(self, base_dir: Path, target_path: Path) -> None: base_abs = str(Path(str(base_dir))).rstrip("/\\") target_abs = str(Path(str(target_path))) if target_abs == base_abs: return if not target_abs.startswith(base_abs + "/") and not target_abs.startswith(base_abs + "\\"): raise ValidationException(message="路径越界", code="PATH_TRAVERSAL", errors={"path": str(target_path)}) def mkdirs(self, path_obj: Any) -> None: if hasattr(path_obj, "makedirs_p"): path_obj.makedirs_p() return if hasattr(path_obj, "mkdir"): try: path_obj.mkdir(parents=True, exist_ok=True) except TypeError: path_obj.mkdir() return raise ValidationException(message="无法创建目录", code="MKDIR_FAILED", errors={"path": str(path_obj)})