stackpilot/backend/services/file_service.py

"""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,
    rel_path: str | None = None,
) -> str:
    """Validate an upload destination and return the real path to write to.

    When ``rel_path`` is given (a folder-upload's relative path such as
    ``photos/2024/img.jpg``) the intermediate directories are created under
    ``dir_path`` and the file lands at their leaf. Each path component is
    validated to block traversal. Otherwise the file lands directly in
    ``dir_path`` under ``filename``.
    """
    real_dir = _safe_real(dir_path)
    if not os.path.isdir(real_dir):
        raise BrowseError(f"Not a directory: {dir_path}")

    components: list[str]
    if rel_path:
        # Normalise separators, drop empty segments, validate each component.
        components = [p for p in rel_path.replace("\\", "/").split("/") if p not in ("", ".")]
        if not components:
            raise BrowseError("Invalid upload path")
    else:
        components = [os.path.basename(filename or "")]

    # Build the logical path one component at a time; _child rejects "..".
    logical = dir_path
    for comp in components:
        logical = _child(logical, comp)
    real = _safe_real(logical)

    # Create intermediate directories (mkdir -p), staying inside the sandbox.
    parent = os.path.dirname(real)
    try:
        os.makedirs(parent, exist_ok=True)
    except PermissionError as exc:
        raise BrowseError(f"Permission denied: {dir_path}") from exc

    if os.path.exists(real) and not overwrite:
        raise BrowseError(f"Already exists: {os.path.basename(logical)}")
    return real


# --------------------------------------------------------------------------- #
# Copy / move
# --------------------------------------------------------------------------- #


def _transfer_dest(src: str, dest_dir: str, overwrite: bool) -> tuple[str, str, str]:
    """Validate a copy/move and return (src_real, dest_real, dest_logical)."""
    src_real = _safe_real(src)
    if not os.path.lexists(src_real):
        raise BrowseError(f"No such path: {src}")
    real_dest_dir = _safe_real(dest_dir)
    if not os.path.isdir(real_dest_dir):
        raise BrowseError(f"Not a directory: {dest_dir}")

    name = os.path.basename(src.rstrip("/"))
    dest_logical = _child(dest_dir, name)
    dest_real = _safe_real(dest_logical)

    # Refuse to copy/move a directory into itself or its own subtree.
    src_norm = os.path.normpath(src_real)
    dest_norm = os.path.normpath(dest_real)
    if dest_norm == src_norm or dest_norm.startswith(src_norm + os.sep):
        raise BrowseError("Cannot move or copy a folder into itself")
    if os.path.exists(dest_real) and not overwrite:
        raise BrowseError(f"Already exists: {name}")
    return src_real, dest_real, dest_logical


def copy(src: str, dest_dir: str, overwrite: bool = False) -> dict:
    src_real, dest_real, dest_logical = _transfer_dest(src, dest_dir, overwrite)
    try:
        if os.path.isdir(src_real) and not os.path.islink(src_real):
            if os.path.exists(dest_real):
                shutil.rmtree(dest_real)
            shutil.copytree(src_real, dest_real, symlinks=True)
        else:
            shutil.copy2(src_real, dest_real, follow_symlinks=False)
    except OSError as exc:
        raise BrowseError(f"Could not copy {src}: {exc.strerror or exc}") from exc
    return {"path": dest_logical}


def move(src: str, dest_dir: str, overwrite: bool = False) -> dict:
    src_real, dest_real, dest_logical = _transfer_dest(src, dest_dir, overwrite)
    try:
        if os.path.exists(dest_real) and overwrite:
            if os.path.isdir(dest_real) and not os.path.islink(dest_real):
                shutil.rmtree(dest_real)
            else:
                os.remove(dest_real)
        shutil.move(src_real, dest_real)
    except OSError as exc:
        raise BrowseError(f"Could not move {src}: {exc.strerror or exc}") from exc
    return {"path": dest_logical}