menzeljandClaude Opus 4.8
f1782eca0e
The file browser/editor could only download individual files. Add a recursive directory download that streams the folder as a zip archive, on the local host and on every remote agent. - file_service.archive_dir(): zip a directory recursively into a temp file, preserving the folder name as the archive root and empty subdirectories; symlinks are skipped (no sandbox escape / loops). - /api/files/download and /agent/files/download branch on directories and return application/zip, cleaning up the temp file afterwards. - Files page: show the download button for folders too (as <name>.zip). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""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
|
|
import tempfile
|
|
import zipfile
|
|
|
|
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 is_dir(path: str) -> bool:
|
|
"""Whether ``path`` points at a directory inside the sandbox."""
|
|
return os.path.isdir(_safe_real(path))
|
|
|
|
|
|
def archive_dir(path: str) -> tuple[str, str]:
|
|
"""Zip a directory (recursively) into a temp file.
|
|
|
|
Returns ``(tmp_zip_path, download_filename)``. The caller is responsible
|
|
for deleting the temp file once it has been streamed to the client.
|
|
Symlinks are skipped so the archive cannot escape the sandbox or loop.
|
|
"""
|
|
real = _safe_real(path)
|
|
if not os.path.isdir(real):
|
|
raise BrowseError(f"Not a directory: {path}")
|
|
name = os.path.basename(path.rstrip("/")) or "root"
|
|
|
|
fd, tmp = tempfile.mkstemp(suffix=".zip")
|
|
os.close(fd)
|
|
try:
|
|
with zipfile.ZipFile(tmp, "w", zipfile.ZIP_DEFLATED) as zf:
|
|
for root, dirs, files in os.walk(real):
|
|
# Don't follow symlinked directories (avoids loops / escapes).
|
|
dirs[:] = [d for d in dirs if not os.path.islink(os.path.join(root, d))]
|
|
rel_root = os.path.relpath(root, real)
|
|
if not files and not dirs and rel_root != ".":
|
|
# Preserve otherwise-empty directories.
|
|
zf.writestr(os.path.join(name, rel_root) + "/", "")
|
|
for f in files:
|
|
full = os.path.join(root, f)
|
|
if os.path.islink(full):
|
|
continue
|
|
zf.write(full, os.path.join(name, rel_root, f) if rel_root != "."
|
|
else os.path.join(name, f))
|
|
except OSError:
|
|
if os.path.exists(tmp):
|
|
os.unlink(tmp)
|
|
raise
|
|
return tmp, f"{name}.zip"
|
|
|
|
|
|
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}
|