"""Sandboxed host filesystem operations for the web file browser. All paths are *logical* host paths (what the user sees, e.g. ``/opt/foo``). They are validated against ``ALLOWED_BROWSE_ROOTS`` and then mapped into the container's view via ``HOST_ROOT_PREFIX`` before any I/O. Directory listing is provided by :func:`device_service.browse`; this module adds the read/write, upload/download and management operations needed for a full browser. """ from __future__ import annotations import os import shutil from services.device_service import BrowseError, _is_allowed, _real_root # Largest file we will load into the in-browser text editor. MAX_EDIT_BYTES = 2 * 1024 * 1024 # 2 MiB def _safe_real(path: str) -> str: """Validate a logical path against the sandbox and return its real path.""" path = os.path.normpath(path or "/") if not path.startswith("/"): raise BrowseError("Path must be absolute") if not _is_allowed(path): raise BrowseError("Path is outside the allowed browse roots") return _real_root(path) def _child(path: str, name: str) -> str: """Return the logical path of ``name`` directly inside ``path``. ``name`` must be a single path component (no separators, no traversal). """ if not name or name in (".", "..") or "/" in name or "\\" in name: raise BrowseError("Invalid name") base = "" if path == "/" else path.rstrip("/") return f"{base}/{name}" def _looks_binary(chunk: bytes) -> bool: return b"\x00" in chunk # --------------------------------------------------------------------------- # # Read / write text # --------------------------------------------------------------------------- # def read_file(path: str) -> dict: real = _safe_real(path) if not os.path.isfile(real): raise BrowseError(f"Not a file: {path}") size = os.path.getsize(real) if size > MAX_EDIT_BYTES: return { "path": path, "content": None, "size": size, "binary": False, "too_large": True, } try: with open(real, "rb") as fh: raw = fh.read() except PermissionError as exc: raise BrowseError(f"Permission denied: {path}") from exc if _looks_binary(raw[:8192]): return {"path": path, "content": None, "size": size, "binary": True, "too_large": False} try: content = raw.decode("utf-8") except UnicodeDecodeError: return {"path": path, "content": None, "size": size, "binary": True, "too_large": False} return {"path": path, "content": content, "size": size, "binary": False, "too_large": False} def write_file(path: str, content: str) -> dict: real = _safe_real(path) if os.path.isdir(real): raise BrowseError(f"Is a directory: {path}") parent = os.path.dirname(real) if not os.path.isdir(parent): raise BrowseError("Parent directory does not exist") try: with open(real, "w", encoding="utf-8") as fh: fh.write(content) except PermissionError as exc: raise BrowseError(f"Permission denied: {path}") from exc return {"path": path, "size": os.path.getsize(real)} # --------------------------------------------------------------------------- # # Management # --------------------------------------------------------------------------- # def create_dir(path: str, name: str) -> dict: child = _child(path, name) real = _safe_real(child) if os.path.exists(real): raise BrowseError(f"Already exists: {name}") try: os.mkdir(real) except PermissionError as exc: raise BrowseError(f"Permission denied: {path}") from exc return {"path": child} def create_file(path: str, name: str) -> dict: child = _child(path, name) real = _safe_real(child) if os.path.exists(real): raise BrowseError(f"Already exists: {name}") try: with open(real, "x", encoding="utf-8"): pass except PermissionError as exc: raise BrowseError(f"Permission denied: {path}") from exc return {"path": child} def rename(path: str, new_name: str) -> dict: real = _safe_real(path) if not os.path.lexists(real): raise BrowseError(f"No such path: {path}") parent = os.path.dirname(path) or "/" dest = _child(parent, new_name) dest_real = _safe_real(dest) if os.path.lexists(dest_real): raise BrowseError(f"Already exists: {new_name}") try: os.rename(real, dest_real) except PermissionError as exc: raise BrowseError(f"Permission denied: {path}") from exc return {"path": dest} def delete(path: str, recursive: bool = False) -> dict: real = _safe_real(path) norm = os.path.normpath(path) if norm == "/" or norm in {os.path.normpath(r) for r in _root_paths()}: raise BrowseError("Refusing to delete a browse root") if not os.path.lexists(real): raise BrowseError(f"No such path: {path}") try: if os.path.isdir(real) and not os.path.islink(real): if recursive: shutil.rmtree(real) else: os.rmdir(real) # fails if non-empty else: os.remove(real) except OSError as exc: raise BrowseError(f"Could not delete {path}: {exc.strerror or exc}") from exc return {"path": path} def _root_paths() -> list[str]: from config import settings return settings.ALLOWED_BROWSE_ROOTS # --------------------------------------------------------------------------- # # Download / upload # --------------------------------------------------------------------------- # def resolve_download(path: str) -> tuple[str, str]: """Return (real_path, filename) for a file download, or raise BrowseError.""" real = _safe_real(path) if not os.path.isfile(real): raise BrowseError(f"Not a file: {path}") return real, os.path.basename(path) def upload_target(dir_path: str, filename: str, overwrite: bool = False) -> str: """Validate an upload destination and return the real path to write to.""" real_dir = _safe_real(dir_path) if not os.path.isdir(real_dir): raise BrowseError(f"Not a directory: {dir_path}") name = os.path.basename(filename or "") child = _child(dir_path, name) real = _safe_real(child) if os.path.exists(real) and not overwrite: raise BrowseError(f"Already exists: {name}") return real