"""Linux/POSIX path helpers for remote SSH paths and Nexus-host Unix paths. Use this module whenever manipulating path *strings* for Linux targets. Do not use ``os.path`` for remote paths — on Windows dev hosts ``os.path.join`` produces backslashes and breaks SSH/SFTP commands. For real filesystem I/O on the Nexus host (``open``, ``os.scandir``), keep ``os.path.realpath`` / ``os.walk`` after normalizing with :func:`to_posix`. """ from __future__ import annotations import os import posixpath UPLOAD_STAGING_PREFIX = "/tmp/nexus_upload_" class PosixPathError(ValueError): """Invalid or unsafe POSIX path.""" def to_posix(path: str) -> str: """Normalize slashes to forward slash (does not resolve ``..``).""" return (path or "").strip().replace("\\", "/") def reject_windows_separators(path: str, *, field: str = "path") -> None: if "\\" in path or "\0" in path: raise PosixPathError(f"{field} 必须使用 Linux 路径(不允许反斜杠)") def normalize_remote_abs_path(path: str) -> str: """Validate and normalize a remote absolute path (SSH/SFTP).""" raw_in = (path or "").strip() reject_windows_separators(raw_in, field="path") raw = to_posix(raw_in) if not raw.startswith("/"): raise PosixPathError("路径必须为绝对路径") parts = [p for p in raw.split("/") if p and p != "."] if ".." in parts: raise PosixPathError("路径不允许包含 '..'") normalized = "/" + "/".join(parts) if parts else "/" if len(normalized) > 1 and normalized.endswith("/"): normalized = normalized.rstrip("/") return normalized or "/" def posix_join(*parts: str) -> str: """Join path segments using POSIX rules (safe on any dev OS).""" if not parts: return "/" acc = to_posix(parts[0]) for part in parts[1:]: seg = to_posix(part) if not seg or seg == "/": continue acc = posixpath.join(acc, seg.lstrip("/")) return posixpath.normpath(acc) or "/" def posix_basename(path: str) -> str: return posixpath.basename(to_posix(path.rstrip("/"))) def posix_dirname(path: str) -> str: parent = posixpath.dirname(to_posix(path.rstrip("/"))) return parent or "/" def remote_join(directory: str, name: str) -> str: """Join remote directory + file name (both logical Linux paths).""" base = normalize_remote_abs_path(directory) segment = to_posix(name).strip("/") reject_windows_separators(segment, field="name") if ".." in segment.split("/"): raise PosixPathError("名称不能包含 '..'") if not segment: return base return posix_join(base, segment) def resolve_nexus_host_path(path: str) -> str: """Resolve a path on the Nexus host filesystem (Linux deployment).""" posix = to_posix(path) reject_windows_separators(posix, field="path") return os.path.realpath(posix) def ensure_under_nexus_upload(path: str) -> str: """Resolve *path* and ensure it stays under ``/tmp/nexus_upload_*``.""" real = resolve_nexus_host_path(path) if not to_posix(real).startswith(UPLOAD_STAGING_PREFIX): raise PosixPathError("仅允许访问上传临时目录") return real def assert_zip_member_safe(extract_dir: str, member_name: str) -> str: """Zip-slip check; returns normalized member path under *extract_dir*.""" base = normalize_remote_abs_path(extract_dir) member = to_posix(member_name).lstrip("/") if not member or member == ".": return base if member == ".." or member.startswith("../") or "/../" in f"/{member}/": raise PosixPathError(f"ZIP 包含非法路径: {member_name}") candidate = posixpath.normpath(posixpath.join(base, member)) if candidate != base and not candidate.startswith(base + "/"): raise PosixPathError(f"ZIP 包含非法路径: {member_name}") return candidate def is_path_under_prefix(resolved: str, prefix: str) -> bool: """True if *resolved* is *prefix* or a child (POSIX string compare after norm).""" r = normalize_remote_abs_path(to_posix(resolved)) p = normalize_remote_abs_path(to_posix(prefix)) return r == p or r.startswith(p + "/") def normalize_remote_dest(path: str | None, *, default: str = "/tmp/sync") -> str: """Normalize remote target directory (push/rsync/diagnose); uses *default* when empty.""" raw = (path or "").strip() or default return normalize_remote_abs_path(raw)