diff --git a/studio/backend/hub/services/models/folder_browser.py b/studio/backend/hub/services/models/folder_browser.py index eb137127f..d56b62c31 100644 --- a/studio/backend/hub/services/models/folder_browser.py +++ b/studio/backend/hub/services/models/folder_browser.py @@ -15,6 +15,7 @@ from loggers import get_logger from hub.schemas.inventory import BrowseEntry, BrowseFoldersResponse from hub.storage.scan_folders import ( contains_sensitive_path_component, + is_denied_system_path, list_scan_folders, ) from hub.utils.paths import ( @@ -27,7 +28,10 @@ from hub.utils.paths import ( studio_root, well_known_model_dirs, ) -from utils.paths.external_media import linux_run_media_mount_roots +from utils.paths.external_media import ( + linux_run_media_mount_roots, + windows_drive_roots, +) from hub.services.models.common import _safe_is_dir from hub.services.models.local_inventory import _resolve_hf_cache_dir @@ -158,8 +162,14 @@ def _looks_like_model_dir(directory: Path) -> bool: return False -def _build_browse_allowlist() -> list[Path]: - """Root directories the browser may walk (also seeds the suggestion chips): HOME, resolved HF cache dirs, Studio outputs/exports/root, registered scan folders, and well-known local-LLM dirs. Each is added only if it resolves to a real directory so the sandbox has no dead boundary.""" +def _build_browse_allowlist( + media_roots: Optional[list[Path]] = None, drive_roots: Optional[list[Path]] = None +) -> list[Path]: + """Root directories the browser may walk (also seeds the suggestion chips): HOME, resolved HF cache dirs, Studio outputs/exports/root, registered scan folders, and well-known local-LLM dirs. Each is added only if it resolves to a real directory so the sandbox has no dead boundary. + + *media_roots* / *drive_roots* let the caller pass already-probed + removable-media and Windows drive roots so they aren't scanned again (a + disconnected mapped drive can make each probe slow); probed here when ``None``.""" from hub.storage.scan_folders import list_scan_folders candidates: list[Path] = [] @@ -176,7 +186,13 @@ def _build_browse_allowlist() -> list[Path]: candidates.append(resolved) _add(Path.home()) - for p in linux_run_media_mount_roots(): + if media_roots is None: + media_roots = linux_run_media_mount_roots() + if drive_roots is None: + drive_roots = windows_drive_roots() + for p in media_roots: + _add(p) + for p in drive_roots: _add(p) _add(_resolve_hf_cache_dir()) try: @@ -218,7 +234,14 @@ def _build_browse_allowlist() -> list[Path]: def _is_path_inside_allowlist(target: Path, allowed_roots: list[Path]) -> bool: - """True if *target* equals or descends from any allowed root; uses ``os.path.realpath`` so symlinks cannot escape the sandbox.""" + """True if *target* equals or descends from any allowed root; uses ``os.path.realpath`` so symlinks cannot escape the sandbox. + + A Windows drive root (``D:\\``) authorizes its descendants, but a bare POSIX + root (``/``) must NOT: a single ``/`` allowlist entry (e.g. a legacy scan + folder) would otherwise authorize every absolute path, reaching ``/var``, + ``/root``, etc. the denylist does not cover. Mirrors the legacy browser so + both treat ``/`` identically. + """ try: target_real = os.path.normcase(os.path.realpath(str(target))) except OSError: @@ -228,13 +251,25 @@ def _is_path_inside_allowlist(target: Path, allowed_roots: list[Path]) -> bool: root_real = os.path.normcase(os.path.realpath(str(root))) except OSError: continue + if target_real == root_real: + return True + drive, tail = os.path.splitdrive(root_real) + if os.path.dirname(root_real) == root_real and not drive: + # Bare POSIX filesystem root ("/"): equality above is the only + # match; do not let it authorize arbitrary descendants. + continue + if drive.startswith(("\\\\", "//")) and not tail: + # Bare UNC share root (\\server\share): os.path.commonpath raises + # "can't mix absolute and relative" on it, so authorize its + # descendants with a boundary-safe prefix test (normcase applied). + if target_real.startswith(root_real.rstrip("\\/") + os.sep): + return True + continue try: if os.path.commonpath([target_real, root_real]) == root_real: return True except ValueError: continue - if target_real == root_real: - return True return False @@ -347,6 +382,11 @@ def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Pa status_code = 403, detail = "Credential or configuration directories are not browseable.", ) + if is_denied_system_path(str(resolved_child)): + raise HTTPException( + status_code = 403, + detail = "System directories are not browseable.", + ) current = resolved_child if contains_sensitive_path_component(str(current)): @@ -354,6 +394,13 @@ def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Pa status_code = 403, detail = "Credential or configuration directories are not browseable.", ) + # Zero-component case: the requested path IS an allowlist root + # (e.g. a legacy-registered "/" or a Windows drive root). + if is_denied_system_path(str(current)): + raise HTTPException( + status_code = 403, + detail = "System directories are not browseable.", + ) if not current.is_dir(): raise HTTPException( status_code = 400, @@ -382,9 +429,13 @@ def browse_folders_response( """ from hub.storage.scan_folders import list_scan_folders + # Probe removable-media and Windows drive roots once; the allowlist and + # chips reuse the result so a disconnected mapped drive isn't scanned twice. + media_roots = linux_run_media_mount_roots() + drive_roots = windows_drive_roots() # Build the allowlist once -- the sandbox check and suggestion chips share # it so chips are always navigable. - allowed_roots = _build_browse_allowlist() + allowed_roots = _build_browse_allowlist(media_roots, drive_roots) try: target = _resolve_browse_target(path, allowed_roots) @@ -440,6 +491,15 @@ def browse_folders_response( # descending into them is refused and registration rejects them. if contains_sensitive_path_component(name): continue + # Same for denied system dirs (C:\Windows, /etc, ...): descent 403s, + # so don't render them as clickable rows. Resolve first so a + # symlink/junction into a denied dir is hidden too, not just a literal name. + try: + resolved_child = os.path.realpath(str(child)) + except (OSError, ValueError): + resolved_child = str(child) + if is_denied_system_path(resolved_child): + continue entries.append( BrowseEntry( name = name, @@ -487,13 +547,22 @@ def browse_folders_response( return if resolved in seen_sug: return + # Drop a denied system dir (e.g. a stale scan-folder row) so it never + # becomes a chip that 403s on click. Drive roots stay: only their + # system subdirectories are denied, not the root itself. + if is_denied_system_path(resolved): + return if _safe_is_dir(resolved): seen_sug.add(resolved) suggestions.append(resolved) # Home first as the safe fallback. _add_sug(Path.home()) - for p in linux_run_media_mount_roots(): + # Reuse the roots probed for the allowlist above (no second drive scan). + for p in media_roots: + _add_sug(p) + # Windows drive roots so the user can hop between C:, D:, E: ... + for p in drive_roots: _add_sug(p) # The HF cache root in use (honors HF_HOME / HF_HUB_CACHE), then the default. try: diff --git a/studio/backend/hub/storage/scan_folders.py b/studio/backend/hub/storage/scan_folders.py index fdb15c7c3..81623d275 100644 --- a/studio/backend/hub/storage/scan_folders.py +++ b/studio/backend/hub/storage/scan_folders.py @@ -16,7 +16,7 @@ from datetime import datetime, timezone from storage.studio_db import get_connection from hub.utils.paths import normalize_path -from utils.paths.external_media import is_linux_run_media_path +from utils.paths.external_media import is_linux_run_media_path, is_local_filesystem_root from utils.paths.sensitive import ( contains_sensitive_path_component as _shared_contains_sensitive_path_component, ) @@ -52,6 +52,25 @@ def _denied_path_prefixes() -> list[str]: return [] +def is_denied_system_path(path: str) -> bool: + """True if *path* is, or descends from, a denied system directory. + + Mirrors the denylist add_scan_folder() enforces at registration so the + browser refuses /etc, /proc, C:\\Windows, etc. even when the allowlist holds + a broad root (a Windows drive root C:\\ or a legacy-registered / root). The + /run carve-out keeps Linux removable-media mounts browseable. Expects an + already-resolved (realpath) path so symlinks cannot escape into a denied subtree. + """ + is_win = platform.system() == "Windows" + check = os.path.normcase(path) if is_win else path + for prefix in _denied_path_prefixes(): + if check == prefix or check.startswith(prefix + os.sep): + if prefix == "/run" and is_linux_run_media_path(check): + continue + return True + return False + + def _contains_sensitive_path_component(path: str) -> bool: return _shared_contains_sensitive_path_component(path) @@ -108,8 +127,9 @@ def add_scan_folder(path: str) -> dict: raise ValueError("Path must be a directory, not a file") if not os.access(normalized, os.R_OK | os.X_OK): raise ValueError("Path is not readable") - if os.path.dirname(normalized) == normalized: - # Registering a filesystem root would expose denied system dirs via browse. + if is_local_filesystem_root(normalized): + # A local fs root ("/", "C:\\") would expose denied system dirs via browse; + # a UNC share root (\\server\share) has none under it and stays registerable. raise ValueError("The filesystem root cannot be registered") if _contains_sensitive_path_component(normalized): raise ValueError("Credential or configuration directories are not allowed") diff --git a/studio/backend/hub/tests/test_model_services.py b/studio/backend/hub/tests/test_model_services.py index 44701c0b6..2c33e09b2 100644 --- a/studio/backend/hub/tests/test_model_services.py +++ b/studio/backend/hub/tests/test_model_services.py @@ -36,6 +36,20 @@ from hub.utils import ( from hub.workers import hf_download +@pytest.fixture(autouse = True) +def _denylist_inert(monkeypatch): + # The browse tests here exercise allowlist containment, symlink safety and + # the sensitive-name filter, not the system-directory denylist (which has + # its own suite in tests/test_browse_denylist.py). On macOS tmp_path + # resolves under /private/var, a denied prefix, so _resolve_browse_target + # would 403 the fixture dirs before that logic runs. Keep the denylist inert + # so these assertions hold on every platform. folder_browser binds + # is_denied_system_path at import, so patch it on that module, not on + # scan_folders. The "rejects" cases still 403 via the allowlist/sensitive + # checks, and the non-browse tests never call it. + monkeypatch.setattr(folder_browser, "is_denied_system_path", lambda _p: False) + + def _repo(repo_id: str, files: list[SimpleNamespace], repo_path: Path): return SimpleNamespace( repo_id = repo_id, @@ -228,7 +242,8 @@ def test_browse_folders_hides_sensitive_dirs(monkeypatch, tmp_path): home = tmp_path / "home" (home / ".ssh").mkdir(parents = True) (home / "models").mkdir() - monkeypatch.setattr(folder_browser, "_build_browse_allowlist", lambda: [home]) + # Accept and ignore the optional (media_roots, drive_roots) args the caller now passes. + monkeypatch.setattr(folder_browser, "_build_browse_allowlist", lambda *_a, **_k: [home]) response = folder_browser.browse_folders_response(str(home), show_hidden = True) diff --git a/studio/backend/routes/models.py b/studio/backend/routes/models.py index c23ab1d42..b8526c75e 100644 --- a/studio/backend/routes/models.py +++ b/studio/backend/routes/models.py @@ -1188,7 +1188,9 @@ def _looks_like_model_dir(directory: Path) -> bool: return False -def _build_browse_allowlist() -> list[Path]: +def _build_browse_allowlist( + media_roots: Optional[list[Path]] = None, drive_roots: Optional[list[Path]] = None +) -> list[Path]: """Return the root directories the folder browser may walk. The same list seeds the sidebar suggestion chips, so chip targets are @@ -1196,13 +1198,20 @@ def _build_browse_allowlist() -> list[Path]: outputs/exports/studio root, registered scan folders, and well-known local-LLM dirs (LM Studio, Ollama, ``~/models``); each added only if it resolves to a real directory. + + *media_roots* / *drive_roots* let the caller pass already-probed + removable-media and Windows drive roots so they aren't scanned again (a + disconnected mapped drive can make each probe slow); probed here when ``None``. """ from utils.paths import ( hf_default_cache_dir, legacy_hf_cache_dir, well_known_model_dirs, ) - from utils.paths.external_media import linux_run_media_mount_roots + from utils.paths.external_media import ( + linux_run_media_mount_roots, + windows_drive_roots, + ) from storage.studio_db import list_scan_folders candidates: list[Path] = [] @@ -1218,7 +1227,13 @@ def _build_browse_allowlist() -> list[Path]: candidates.append(resolved) _add(Path.home()) - for p in linux_run_media_mount_roots(): + if media_roots is None: + media_roots = linux_run_media_mount_roots() + if drive_roots is None: + drive_roots = windows_drive_roots() + for p in media_roots: + _add(p) + for p in drive_roots: _add(p) _add(_resolve_hf_cache_dir()) try: @@ -1269,19 +1284,43 @@ def _build_browse_allowlist() -> list[Path]: def _is_path_inside_allowlist(target: Path, allowed_roots: list[Path]) -> bool: """True if *target* equals or descends from any allowed root. - Uses ``os.path.realpath`` so symlinks can't escape the sandbox. + Uses ``os.path.realpath`` (symlinks can't escape the sandbox) and + ``os.path.commonpath`` for a component-wise containment test, so a string + prefix like ``/home/u`` never matches a sibling ``/home/user2`` while a + drive root ``D:\\`` still contains ``D:\\models``. A Windows drive root + authorizes its descendants, but a bare POSIX root ``/`` must NOT, else one + ``/`` allowlist entry would authorize every absolute path. ``normcase`` keeps + the drive-letter comparison case-insensitive, matching the hub browser. """ try: - target_real = os.path.realpath(str(target)) + target_real = os.path.normcase(os.path.realpath(str(target))) except OSError: return False for root in allowed_roots: try: - root_real = os.path.realpath(str(root)) + root_real = os.path.normcase(os.path.realpath(str(root))) except OSError: continue - if target_real == root_real or target_real.startswith(root_real + os.sep): + if target_real == root_real: return True + drive, tail = os.path.splitdrive(root_real) + if os.path.dirname(root_real) == root_real and not drive: + # Bare POSIX filesystem root ("/"): equality above is the only + # match; do not let it authorize arbitrary descendants. + continue + if drive.startswith(("\\\\", "//")) and not tail: + # Bare UNC share root (\\server\share): os.path.commonpath raises + # "can't mix absolute and relative" on it, so authorize its + # descendants with a boundary-safe prefix test (normcase applied). + if target_real.startswith(root_real.rstrip("\\/") + os.sep): + return True + continue + try: + if os.path.commonpath([target_real, root_real]) == root_real: + return True + except ValueError: + # Different drives / mixed absolute-relative: not contained. + continue return False @@ -1339,7 +1378,10 @@ def _match_browse_child(current: Path, name: str) -> Optional[Path]: def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Path: """Resolve a requested browse path by walking from trusted allowlist roots.""" - from storage.studio_db import contains_sensitive_path_component + from storage.studio_db import ( + contains_sensitive_path_component, + is_denied_system_path, + ) requested_path = _normalize_browse_request_path(path) resolved_roots: list[Path] = [] @@ -1396,6 +1438,11 @@ def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Pa status_code = 403, detail = "Credential or configuration directories are not browseable.", ) + if is_denied_system_path(str(resolved_child)): + raise HTTPException( + status_code = 403, + detail = "System directories are not browseable.", + ) current = resolved_child if contains_sensitive_path_component(str(current)): @@ -1403,6 +1450,13 @@ def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Pa status_code = 403, detail = "Credential or configuration directories are not browseable.", ) + # Zero-component case: the requested path IS an allowlist root + # (e.g. a legacy-registered "/" or a Windows drive root). + if is_denied_system_path(str(current)): + raise HTTPException( + status_code = 403, + detail = "System directories are not browseable.", + ) if not current.is_dir(): raise HTTPException( status_code = 400, @@ -1420,8 +1474,12 @@ def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Pa ) +# Sync (def, not async) so FastAPI runs the blocking filesystem I/O (drive +# probes, iterdir, realpath) in the threadpool: a disconnected mapped drive can +# make the probe wait out its timeout, which on the event loop would stall every +# other request. Matches the hub browse endpoint. @router.get("/browse-folders", response_model = BrowseFoldersResponse) -async def browse_folders( +def browse_folders( path: Optional[str] = Query( None, description = ( @@ -1450,11 +1508,22 @@ async def browse_folders( then hidden (if ``show_hidden=true``). """ from utils.paths import hf_default_cache_dir, well_known_model_dirs - from utils.paths.external_media import linux_run_media_mount_roots - from storage.studio_db import contains_sensitive_path_component, list_scan_folders + from utils.paths.external_media import ( + linux_run_media_mount_roots, + windows_drive_roots, + ) + from storage.studio_db import ( + contains_sensitive_path_component, + is_denied_system_path, + list_scan_folders, + ) + # Probe removable-media and Windows drive roots once; the allowlist and + # chips reuse the result so a disconnected mapped drive isn't scanned twice. + media_roots = linux_run_media_mount_roots() + drive_roots = windows_drive_roots() # Build once; the sandbox check and suggestion chips share it. - allowed_roots = _build_browse_allowlist() + allowed_roots = _build_browse_allowlist(media_roots, drive_roots) try: target = _resolve_browse_target(path, allowed_roots) @@ -1506,6 +1575,15 @@ async def browse_folders( continue if contains_sensitive_path_component(name): continue + # Hide denied system dirs (C:\Windows, /etc, ...) so they don't + # render as clickable rows that then 403 on descent. Resolve first + # so a symlink/junction into a denied dir is hidden too, not just a literal name. + try: + resolved_child = os.path.realpath(str(child)) + except (OSError, ValueError): + resolved_child = str(child) + if is_denied_system_path(resolved_child): + continue entries.append( BrowseEntry( name = name, @@ -1553,13 +1631,22 @@ async def browse_folders( return if resolved in seen_sug: return + # Drop a denied system dir (e.g. a stale scan-folder row) so it never + # becomes a chip that 403s on click. Drive roots stay: only their + # system subdirectories are denied, not the root itself. + if is_denied_system_path(resolved): + return if _safe_is_dir(resolved): seen_sug.add(resolved) suggestions.append(resolved) # Home first -- the safe fallback when everything else is cold. _add_sug(Path.home()) - for p in linux_run_media_mount_roots(): + # Reuse the roots probed for the allowlist above (no second drive scan). + for p in media_roots: + _add_sug(p) + # Windows drive roots so the user can hop between C:, D:, E: ... + for p in drive_roots: _add_sug(p) # The HF cache root the process is actually using. try: diff --git a/studio/backend/storage/studio_db.py b/studio/backend/storage/studio_db.py index 87aa50ee2..4e0c711b6 100644 --- a/studio/backend/storage/studio_db.py +++ b/studio/backend/storage/studio_db.py @@ -27,7 +27,7 @@ from utils.paths import ( project_workspaces_root, studio_db_path, ) -from utils.paths.external_media import is_linux_run_media_path +from utils.paths.external_media import is_linux_run_media_path, is_local_filesystem_root from utils.paths.sensitive import ( contains_sensitive_path_component as _shared_contains_sensitive_path_component, ) @@ -69,6 +69,25 @@ def _denied_path_prefixes() -> list[str]: return [] +def is_denied_system_path(path: str) -> bool: + """True if *path* is, or descends from, a denied system directory. + + Mirrors the denylist add_scan_folder() enforces at registration so the + browser refuses /etc, /proc, C:\\Windows, etc. even when the allowlist holds + a broad root (a Windows drive root C:\\ or a legacy-registered / root). The + /run carve-out keeps Linux removable-media mounts browseable. Expects an + already-resolved (realpath) path so symlinks cannot escape into a denied subtree. + """ + is_win = platform.system() == "Windows" + check = os.path.normcase(path) if is_win else path + for prefix in _denied_path_prefixes(): + if check == prefix or check.startswith(prefix + os.sep): + if prefix == "/run" and is_linux_run_media_path(check): + continue + return True + return False + + def _contains_sensitive_path_component(path: str) -> bool: return _shared_contains_sensitive_path_component(path) @@ -931,6 +950,12 @@ def add_scan_folder(path: str) -> dict: raise ValueError("Path must be a directory, not a file") if not os.access(normalized, os.R_OK | os.X_OK): raise ValueError("Path is not readable") + # Reject a local filesystem root ("/", or a bare Windows drive root "C:\\"): + # registering one seeds the browse allowlist with a root above denied system + # dirs. A UNC share root (\\server\share) has none under it and was + # registerable before this guard, so it stays allowed. Mirrors scan_folders.py. + if is_local_filesystem_root(normalized): + raise ValueError("The filesystem root cannot be registered") if _contains_sensitive_path_component(normalized): raise ValueError("Credential or configuration directories are not allowed") diff --git a/studio/backend/tests/test_browse_denylist.py b/studio/backend/tests/test_browse_denylist.py new file mode 100644 index 000000000..e21dc5c5a --- /dev/null +++ b/studio/backend/tests/test_browse_denylist.py @@ -0,0 +1,371 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 + +"""System-directory denylist enforcement for the folder browser. + +Once the allowlist can hold a whole Windows drive root (C:\\) or a legacy / +root, the browse endpoints must re-apply the ``_denied_path_prefixes()`` policy +``add_scan_folder`` enforces, so /etc, /proc, C:\\Windows, C:\\Program Files stay +unbrowseable even under an allowlisted root. Windows/macOS branches run on this +POSIX host by AST-extracting the pure helper with ``ntpath`` / a mocked ``platform``. +""" + +from __future__ import annotations + +import ast +import ntpath +import os +import posixpath +from pathlib import Path +from types import SimpleNamespace +from typing import Optional + +import pytest + +from hub.storage import scan_folders +from storage import studio_db +from utils.paths.external_media import is_local_filesystem_root + + +_BACKEND_ROOT = Path(__file__).resolve().parent.parent + + +class _HTTPException(Exception): + def __init__(self, status_code: int, detail: str): + super().__init__(detail) + self.status_code = status_code + self.detail = detail + + +def _extract_is_denied_windows(): + """is_denied_system_path (+ _denied_path_prefixes) from studio_db.py under Windows semantics (ntpath) on a POSIX host.""" + src = (_BACKEND_ROOT / "storage" / "studio_db.py").read_text(encoding = "utf-8") + tree = ast.parse(src) + funcs = [ + n + for n in tree.body + if isinstance(n, ast.FunctionDef) + and n.name in {"_denied_path_prefixes", "is_denied_system_path"} + ] + module = ast.Module(body = funcs, type_ignores = []) + ast.fix_missing_locations(module) + + win_os = SimpleNamespace( + sep = "\\", + environ = { + "SystemRoot": r"C:\Windows", + "ProgramFiles": r"C:\Program Files", + "ProgramFiles(x86)": r"C:\Program Files (x86)", + }, + path = SimpleNamespace(normcase = ntpath.normcase), + ) + ns = { + "os": win_os, + "platform": SimpleNamespace(system = lambda: "Windows"), + # /run has no Windows analog, so the carve-out is never reached. + "is_linux_run_media_path": lambda _p: False, + } + exec(compile(module, "", "exec"), ns) + return ns["is_denied_system_path"] + + +# is_denied_system_path -- Linux (real helper, this host) +@pytest.mark.parametrize( + "path", + [ + "/etc", + "/etc/ssl/private", + "/proc", + "/proc/1", + "/sys", + "/dev", + "/boot", + "/run", + "/run/systemd/private", + "/run/media", + "/run/media/dspofu", + ], +) +def test_is_denied_system_path_linux_denies_system_dirs(monkeypatch, path): + monkeypatch.setattr(studio_db.platform, "system", lambda: "Linux") + assert studio_db.is_denied_system_path(path) is True + + +@pytest.mark.parametrize( + "path", + ["/run/media/dspofu/nvmeB", "/run/media/dspofu/nvmeB/models"], +) +def test_is_denied_system_path_linux_allows_run_media_mounts(monkeypatch, path): + # The /run/media// carve-out keeps removable media browseable. + monkeypatch.setattr(studio_db.platform, "system", lambda: "Linux") + assert studio_db.is_denied_system_path(path) is False + + +@pytest.mark.parametrize( + "path", + ["/etc-backup", "/etcetera", "/home/u/models", "/mnt/data", "/devices", "/", "/opt/models"], +) +def test_is_denied_system_path_linux_allows_non_system(monkeypatch, path): + monkeypatch.setattr(studio_db.platform, "system", lambda: "Linux") + assert studio_db.is_denied_system_path(path) is False + + +def test_legacy_and_hub_denylist_agree(monkeypatch): + monkeypatch.setattr(studio_db.platform, "system", lambda: "Linux") + monkeypatch.setattr(scan_folders.platform, "system", lambda: "Linux") + for p in ["/etc", "/proc/1", "/home/u", "/boot", "/opt/x"]: + assert studio_db.is_denied_system_path(p) == scan_folders.is_denied_system_path(p) + + +# is_denied_system_path -- Windows (ntpath-backed), case-insensitive + collisions +@pytest.mark.parametrize( + "path", + [ + r"C:\Windows", + r"C:\Windows\System32", + r"c:\windows", + r"C:\WINDOWS\Temp", + r"C:\Program Files", + r"C:\Program Files\x", + r"C:\Program Files (x86)\y", + r"c:\program files", + ], +) +def test_is_denied_system_path_windows_denies_system_dirs(path): + is_denied = _extract_is_denied_windows() + assert is_denied(path) is True + + +@pytest.mark.parametrize( + "path", + [ + r"C:\Models", + r"D:\models", + r"C:\WindowsApps", + r"C:\ProgramData", + r"C:\Program Files Extra", + r"E:\gguf", + r"C:\Users\me\models", + ], +) +def test_is_denied_system_path_windows_allows_non_system(path): + is_denied = _extract_is_denied_windows() + assert is_denied(path) is False + + +# _resolve_browse_target -- real-FS integration (legacy browser) +def _extract_resolver(): + """Extract the legacy browse resolver; its inline imports use the real storage.studio_db policy.""" + src = (_BACKEND_ROOT / "routes" / "models.py").read_text(encoding = "utf-8") + tree = ast.parse(src) + names = { + "_is_path_inside_allowlist", + "_normalize_browse_request_path", + "_browse_relative_parts", + "_match_browse_child", + "_resolve_browse_target", + } + funcs = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in names] + module = ast.Module(body = funcs, type_ignores = []) + ast.fix_missing_locations(module) + ns = { + "os": os, + "Path": Path, + "Optional": Optional, + "HTTPException": _HTTPException, + "logger": SimpleNamespace(warning = lambda *a, **k: None, debug = lambda *a, **k: None), + } + exec(compile(module, "", "exec"), ns) + return ns["_resolve_browse_target"] + + +def test_resolve_browse_target_blocks_etc_via_root(): + # Registering "/" must not make /etc browsable (Codex #3 regression guard). + resolve = _extract_resolver() + with pytest.raises(_HTTPException) as exc: + resolve("/etc", [Path("/")]) + assert exc.value.status_code == 403 + + +def test_resolve_browse_target_blocks_stale_denied_root(tmp_path, monkeypatch): + # A stale scan-folder row pointing at a denied dir is refused by the + # browse-time denylist even though it is its own allowlist root. A tmp-based + # denied prefix (+ Linux compare) keeps the assertion OS-agnostic: on macOS + # tmp lives under the already-denied /private/var, masking the message. + denied = (tmp_path / "sysfake").resolve() + denied.mkdir() + monkeypatch.setattr(studio_db.platform, "system", lambda: "Linux") + monkeypatch.setattr(studio_db, "_denied_path_prefixes", lambda: [str(denied)]) + resolve = _extract_resolver() + with pytest.raises(_HTTPException) as exc: + resolve(str(denied), [denied]) + assert exc.value.status_code == 403 + assert "System directories" in exc.value.detail + + +def test_resolve_browse_target_allows_root_itself(): + resolve = _extract_resolver() + assert resolve("/", [Path("/")]) == Path("/") + + +def test_resolve_browse_target_allows_legit_nested_dir(tmp_path, monkeypatch): + # Force the Linux denylist so the macOS temp location (under the denied + # /private/var) doesn't reject the tmp fixture; a normal nested dir must not be over-blocked. + monkeypatch.setattr(studio_db.platform, "system", lambda: "Linux") + resolve = _extract_resolver() + base = tmp_path / "allowed" + sub = base / "models" / "gguf" + sub.mkdir(parents = True) + assert resolve(str(sub), [base]) == sub.resolve() + + +def test_resolve_browse_target_symlink_escape_blocked(tmp_path): + resolve = _extract_resolver() + base = tmp_path / "allowed" + base.mkdir() + link = base / "escape" + try: + link.symlink_to("/etc", target_is_directory = True) + except OSError: + pytest.skip("symlinks unsupported on this host") + with pytest.raises(_HTTPException) as exc: + resolve(str(link), [base]) + assert exc.value.status_code == 403 + + +# _is_path_inside_allowlist -- bare POSIX root parity (legacy == hub) +def _extract_is_inside(rel_parts, *, os_module = os): + """Extract a standalone _is_path_inside_allowlist (os/Path only) so both browsers' copies compare without importing their heavy modules.""" + src = _BACKEND_ROOT.joinpath(*rel_parts).read_text(encoding = "utf-8") + tree = ast.parse(src) + funcs = [ + n + for n in tree.body + if isinstance(n, ast.FunctionDef) and n.name == "_is_path_inside_allowlist" + ] + module = ast.Module(body = funcs, type_ignores = []) + ast.fix_missing_locations(module) + ns = {"os": os_module, "Path": Path} + exec(compile(module, f"", "exec"), ns) + return ns["_is_path_inside_allowlist"] + + +# ntpath semantics with a no-FS realpath, so UNC containment can be driven on a +# POSIX CI (the real realpath cannot resolve \\server\share off Windows). +_WIN_OS = SimpleNamespace( + sep = ntpath.sep, + path = SimpleNamespace( + realpath = lambda p: ntpath.normpath(str(p)), + normcase = ntpath.normcase, + splitdrive = ntpath.splitdrive, + dirname = ntpath.dirname, + commonpath = ntpath.commonpath, + ), +) + + +def test_legacy_and_hub_allowlist_agree_on_posix_root(): + # A bare "/" allowlist entry must authorize only "/" itself in BOTH + # browsers, never descend into /var, /root, /home (which the denylist does + # not cover). Guards the hub browser against authorizing every absolute path. + legacy = _extract_is_inside(["routes", "models.py"]) + hub = _extract_is_inside(["hub", "services", "models", "folder_browser.py"]) + roots = [Path("/")] + for tgt in ["/var", "/root", "/home", "/usr", "/opt", "/etc"]: + assert legacy(Path(tgt), roots) is False + assert hub(Path(tgt), roots) is False + # "/" itself stays browseable; only its descendants are withheld. + assert legacy(Path("/"), roots) is True + assert hub(Path("/"), roots) is True + + +def test_hub_allowlist_authorizes_normal_nested_dir(tmp_path): + # The bare-root special case must not over-block a normal allowlist root's descendants. + hub = _extract_is_inside(["hub", "services", "models", "folder_browser.py"]) + base = tmp_path / "allowed" + sub = base / "models" / "gguf" + sub.mkdir(parents = True) + assert hub(sub, [base]) is True + assert hub(base, [base]) is True + + +# add_scan_folder -- filesystem-root rejection parity (legacy == hub) +def test_legacy_add_scan_folder_rejects_filesystem_root(monkeypatch): + monkeypatch.setattr(studio_db.platform, "system", lambda: "Linux") + with pytest.raises(ValueError, match = "filesystem root"): + studio_db.add_scan_folder("/") + + +def test_hub_add_scan_folder_rejects_filesystem_root(monkeypatch): + monkeypatch.setattr(scan_folders.platform, "system", lambda: "Linux") + with pytest.raises(ValueError, match = "filesystem root"): + scan_folders.add_scan_folder("/") + + +# is_local_filesystem_root: reject "/" and "C:\\" (roots above denied system dirs), +# but NOT a UNC share root -- registering \\server\share was allowed before this +# guard and has no system dirs under it. _pathmod drives Windows semantics on POSIX CI. +@pytest.mark.parametrize( + "path, pathmod, expected", + [ + # Local filesystem roots -> rejected (True). + ("/", posixpath, True), + ("C:\\", ntpath, True), + ("c:\\", ntpath, True), + ("D:\\", ntpath, True), + # UNC share roots -> NOT a local root, stay registerable (False). + (r"\\server\share", ntpath, False), + (r"\\nas\models", ntpath, False), + ("//server/share", ntpath, False), + # Device / extended-length volume roots -> still local roots (rejected), + # so neither \\?\C:\ nor a drive-letter-less \\?\Volume{GUID}\ can slip + # past the guard as if it were a share root. + (r"\\?\C:" + "\\", ntpath, True), + (r"\\.\C:" + "\\", ntpath, True), + (r"\\?\C:", ntpath, True), + (r"\\.\C:", ntpath, True), + (r"\\?\Volume{2f8e6d31-0000-0000-0000-100000000000}" + "\\", ntpath, True), + (r"\\.\Volume{2f8e6d31-0000-0000-0000-100000000000}", ntpath, True), + # Device-namespace UNC share root -> stays registerable (False). + (r"\\?\UNC\server\share", ntpath, False), + # Non-root paths (incl. deep device / extended-length) -> not a root (False). + ("C:\\Models", ntpath, False), + (r"\\server\share\models", ntpath, False), + (r"\\?\C:\Users\me\models", ntpath, False), + (r"\\?\Volume{2f8e6d31-0000-0000-0000-100000000000}\models", ntpath, False), + ("/home/user", posixpath, False), + ], +) +def test_is_local_filesystem_root(path, pathmod, expected): + assert is_local_filesystem_root(path, _pathmod = pathmod) is expected + + +def test_both_guards_use_the_shared_local_root_helper(): + # Register-root parity: both browsers reject the same roots via one helper, so a + # UNC-share exemption can never drift between the legacy and hub code paths. + legacy_src = (_BACKEND_ROOT / "storage" / "studio_db.py").read_text(encoding = "utf-8") + hub_src = (_BACKEND_ROOT / "hub" / "storage" / "scan_folders.py").read_text(encoding = "utf-8") + assert "is_local_filesystem_root(normalized)" in legacy_src + assert "is_local_filesystem_root(normalized)" in hub_src + + +# A registered UNC share root must authorize its own descendants in both browsers. +# os.path.commonpath raises "can't mix absolute and relative" on a bare +# \\server\share, so containment falls back to a boundary-safe prefix test; without +# it, registering a UNC share (now allowed) would 403 every folder under it. +@pytest.mark.parametrize( + "rel_parts", + [ + ["routes", "models.py"], + ["hub", "services", "models", "folder_browser.py"], + ], +) +def test_unc_share_root_authorizes_its_descendants(rel_parts): + is_inside = _extract_is_inside(rel_parts, os_module = _WIN_OS) + root = [Path(r"\\server\share")] + assert is_inside(Path(r"\\server\share"), root) is True # the root itself + assert is_inside(Path(r"\\server\share\models"), root) is True # direct child + assert is_inside(Path(r"\\server\share\a\b\c"), root) is True # deep descendant + assert is_inside(Path(r"\\SERVER\SHARE\Models"), root) is True # case-insensitive + assert is_inside(Path(r"\\server\share2\models"), root) is False # sibling share + assert is_inside(Path(r"C:\models"), root) is False # different volume diff --git a/studio/backend/tests/test_browse_folders_route.py b/studio/backend/tests/test_browse_folders_route.py index 3a607e6b1..970057f1e 100644 --- a/studio/backend/tests/test_browse_folders_route.py +++ b/studio/backend/tests/test_browse_folders_route.py @@ -22,6 +22,18 @@ if "structlog" not in sys.modules: ) import routes.models as models_route +import storage.studio_db as studio_db + + +@pytest.fixture(autouse = True) +def _denylist_inert(monkeypatch): + # These tests exercise allowlist containment and the file-vs-directory guard, + # not the system-directory denylist (which has its own suite in + # test_browse_denylist.py). On macOS tmp_path resolves under /private/var, a + # denied prefix, so _resolve_browse_target would 403 the fixture dirs before + # the containment logic runs. Keep the denylist inert here so these + # assertions hold on every platform. + monkeypatch.setattr(studio_db, "is_denied_system_path", lambda _p: False) def test_resolve_browse_target_returns_allowed_directory(tmp_path): diff --git a/studio/backend/tests/test_linux_external_media_paths.py b/studio/backend/tests/test_linux_external_media_paths.py index c763248f6..b735bd113 100644 --- a/studio/backend/tests/test_linux_external_media_paths.py +++ b/studio/backend/tests/test_linux_external_media_paths.py @@ -252,10 +252,17 @@ def test_legacy_browse_allowlist_includes_linux_run_media_mounts(monkeypatch, tm outputs_root = lambda: tmp_path / "missing-outputs", exports_root = lambda: tmp_path / "missing-exports", ) - fake_external_media = SimpleNamespace(linux_run_media_mount_roots = lambda: [media_root]) + fake_external_media = SimpleNamespace( + linux_run_media_mount_roots = lambda: [media_root], + windows_drive_roots = lambda: [], + ) fake_studio_db = SimpleNamespace( list_scan_folders = lambda: [], contains_sensitive_path_component = studio_db.contains_sensitive_path_component, + # The media root is a legitimate mount, not denied; the .ssh 403 below + # comes from the credential check. A False stub keeps this OS-independent + # (on macOS tmp_path lives under the denied /private/var). + is_denied_system_path = lambda _p: False, ) monkeypatch.setitem(sys.modules, "utils.paths", fake_paths) monkeypatch.setitem(sys.modules, "utils.paths.external_media", fake_external_media) diff --git a/studio/backend/tests/test_windows_external_drive_paths.py b/studio/backend/tests/test_windows_external_drive_paths.py new file mode 100644 index 000000000..9686d45c9 --- /dev/null +++ b/studio/backend/tests/test_windows_external_drive_paths.py @@ -0,0 +1,354 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0 + +from __future__ import annotations + +import ast +import os +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Optional + +from utils.paths import external_media + + +_BACKEND_ROOT = Path(__file__).resolve().parent.parent + + +class _HTTPException(Exception): + def __init__(self, status_code: int, detail: str): + super().__init__(detail) + self.status_code = status_code + self.detail = detail + + +def _extract_routes_function(name: str, ns_extra: Optional[dict] = None) -> dict: + """Exec one top-level function from routes/models.py without importing the module (which pulls in FastAPI).""" + tree = ast.parse((_BACKEND_ROOT / "routes" / "models.py").read_text(encoding = "utf-8")) + fn = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name) + module = ast.Module(body = [fn], type_ignores = []) + ast.fix_missing_locations(module) + ns = {"os": os, "Path": Path, "Optional": Optional} + if ns_extra: + ns.update(ns_extra) + exec(compile(module, "", "exec"), ns) + return ns + + +def _stub_windows(monkeypatch, existing_drives): + """Simulate Windows exposing only *existing_drives* (e.g. {"C", "D"}) as readable roots, independent of the host FS. + + Overriding _active_windows_drive_bitmask keeps it deterministic even on a + real Windows host, where live GetLogicalDrives would return the actual layout.""" + monkeypatch.setattr(external_media.platform, "system", lambda: "Windows") + mask = sum(1 << (ord(d.upper()) - ord("A")) for d in existing_drives) + monkeypatch.setattr(external_media, "_active_windows_drive_bitmask", lambda: mask) + present = {f"{d.upper()}:\\" for d in existing_drives} + monkeypatch.setattr(external_media.os.path, "isdir", lambda p: str(p) in present) + monkeypatch.setattr(external_media.os, "access", lambda p, _mode: str(p) in present) + + +def test_windows_drive_roots_empty_off_windows(monkeypatch): + # Regression guard: the helper is a no-op on Linux/macOS so it can't change the allowlist on the platforms CI runs on. + monkeypatch.setattr(external_media.platform, "system", lambda: "Linux") + assert external_media.windows_drive_roots() == [] + monkeypatch.setattr(external_media.platform, "system", lambda: "Darwin") + assert external_media.windows_drive_roots() == [] + + +def test_windows_drive_roots_lists_readable_drives(monkeypatch): + _stub_windows(monkeypatch, {"C", "D", "E"}) + + roots = external_media.windows_drive_roots(drive_letters = "CDEF") + + # F is absent, so it is skipped; the rest are exposed in order. + assert roots == [Path("C:\\"), Path("D:\\"), Path("E:\\")] + + +def test_windows_drive_roots_skips_absent_and_unreadable(monkeypatch): + _stub_windows(monkeypatch, {"C"}) + + roots = external_media.windows_drive_roots(drive_letters = "CDE") + + assert roots == [Path("C:\\")] + + +def test_windows_drive_roots_ignores_bad_letters_and_dedupes(monkeypatch): + _stub_windows(monkeypatch, {"C", "D"}) + + roots = external_media.windows_drive_roots( + drive_letters = ["c:", "C", "D", "1", "AB", "", " d "], + ) + + assert roots == [Path("C:\\"), Path("D:\\")] + + +def test_readable_dir_within_times_out(monkeypatch): + # A probe that outlives the timeout is reported not-readable, so a hung + # (disconnected mapped network) drive is skipped instead of blocking. + import time + + monkeypatch.setattr(external_media.os.path, "isdir", lambda p: time.sleep(5) or True) + monkeypatch.setattr(external_media.os, "access", lambda p, _mode: True) + start = time.monotonic() + ok = external_media._readable_dir_within("Z:\\", timeout = 0.2) + elapsed = time.monotonic() - start + assert ok is False + assert elapsed < 3.0 # returned on the timeout, did not wait out the 5s stall + + +def test_readable_dir_within_reports_fast_probe(monkeypatch): + monkeypatch.setattr(external_media.os.path, "isdir", lambda p: True) + monkeypatch.setattr(external_media.os, "access", lambda p, _mode: True) + assert external_media._readable_dir_within("C:\\", timeout = 2.0) is True + + +def test_windows_drive_roots_skips_hung_drive(monkeypatch): + # A disconnected mapped drive stays set in the bitmask and its os.path.isdir + # stalls; it must be skipped without stalling enumeration. C answers, D hangs, + # so only C is listed, bounded by the per-drive timeout, not the stall. + import time + + monkeypatch.setattr(external_media.platform, "system", lambda: "Windows") + monkeypatch.setattr( + external_media, + "_active_windows_drive_bitmask", + lambda: sum(1 << (ord(d) - ord("A")) for d in "CD"), + ) + monkeypatch.setattr(external_media, "_DRIVE_PROBE_TIMEOUT_S", 0.2) + + def _isdir(p): + if str(p) == "D:\\": + time.sleep(5) # simulate the reconnect stall + return True + return str(p) == "C:\\" + + monkeypatch.setattr(external_media.os.path, "isdir", _isdir) + monkeypatch.setattr(external_media.os, "access", lambda p, _mode: True) + + start = time.monotonic() + roots = external_media.windows_drive_roots(drive_letters = "CD") + elapsed = time.monotonic() - start + + assert roots == [Path("C:\\")] + assert elapsed < 3.0 # bounded by the per-drive timeout, not the 5s stall + + +def test_windows_drive_roots_probes_hung_drives_in_parallel(monkeypatch): + # Several disconnected mapped drives must add ~one timeout total, not one + # per drive: C answers fast, D/E/F stall. The concurrent probe stays bounded + # by a single deadline where serial probing would cost ~4x the timeout. + import time + + monkeypatch.setattr(external_media.platform, "system", lambda: "Windows") + monkeypatch.setattr( + external_media, + "_active_windows_drive_bitmask", + lambda: sum(1 << (ord(d) - ord("A")) for d in "CDEF"), + ) + timeout = 0.2 + monkeypatch.setattr(external_media, "_DRIVE_PROBE_TIMEOUT_S", timeout) + + def _isdir(p): + if str(p) == "C:\\": + return True + time.sleep(5) # every other drive simulates a reconnect stall + return True + + monkeypatch.setattr(external_media.os.path, "isdir", _isdir) + monkeypatch.setattr(external_media.os, "access", lambda p, _mode: True) + + start = time.monotonic() + roots = external_media.windows_drive_roots(drive_letters = "CDEF") + elapsed = time.monotonic() - start + + assert roots == [Path("C:\\")] + # 3 stalled drives probed in parallel finish within ~1 timeout, well under the ~3*timeout a serial probe would take. + assert elapsed < 3 * timeout + + +def test_browse_allowlist_includes_windows_drive_roots(monkeypatch, tmp_path): + # End-to-end wiring: windows_drive_roots() output flows into the browse + # allowlist built by routes/models.py, mirroring the Linux media-mounts test. + tree = ast.parse((_BACKEND_ROOT / "routes" / "models.py").read_text(encoding = "utf-8")) + function_names = { + "_build_browse_allowlist", + "_browse_relative_parts", + "_is_path_inside_allowlist", + "_match_browse_child", + "_normalize_browse_request_path", + "_resolve_browse_target", + } + functions = [ + node + for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name in function_names + ] + module = ast.Module(body = functions, type_ignores = []) + ast.fix_missing_locations(module) + + home = tmp_path / "home" + drive_root = tmp_path / "D_drive" + model_dir = drive_root / "modelsAI" / "gguf" + home.mkdir() + model_dir.mkdir(parents = True) + + fake_paths = SimpleNamespace( + hf_default_cache_dir = lambda: tmp_path / "missing-default-hf", + legacy_hf_cache_dir = lambda: tmp_path / "missing-legacy-hf", + well_known_model_dirs = lambda: [], + studio_root = lambda: tmp_path / "missing-studio", + outputs_root = lambda: tmp_path / "missing-outputs", + exports_root = lambda: tmp_path / "missing-exports", + ) + fake_external_media = SimpleNamespace( + linux_run_media_mount_roots = lambda: [], + windows_drive_roots = lambda: [drive_root], + ) + fake_studio_db = SimpleNamespace( + list_scan_folders = lambda: [], + contains_sensitive_path_component = lambda _p: False, + # The simulated D:\ root maps to a tmp_path dir, not a denied system path. + is_denied_system_path = lambda _p: False, + ) + monkeypatch.setitem(sys.modules, "utils.paths", fake_paths) + monkeypatch.setitem(sys.modules, "utils.paths.external_media", fake_external_media) + monkeypatch.setitem(sys.modules, "storage.studio_db", fake_studio_db) + + ns = { + "HTTPException": _HTTPException, + "os": os, + "Path": Path, + "Optional": Optional, + "_safe_is_dir": lambda p: Path(p).is_dir(), + "_resolve_hf_cache_dir": lambda: tmp_path / "missing-hf", + "logger": SimpleNamespace(debug = lambda *_args, **_kwargs: None), + } + exec(compile(module, "", "exec"), ns) + + allowlist = ns["_build_browse_allowlist"]() + + # The simulated Windows drive root is now browsable, and a model dir on it resolves. + assert drive_root.resolve() in allowlist + assert ns["_resolve_browse_target"](str(model_dir), allowlist) == model_dir.resolve() + + +def test_build_browse_allowlist_reuses_passed_roots(monkeypatch, tmp_path): + # Double-probe fix: a browse request probes the drive/media roots once and + # passes them in, so _build_browse_allowlist must NOT scan + # windows_drive_roots() again (a disconnected drive would double the stall). + tree = ast.parse((_BACKEND_ROOT / "routes" / "models.py").read_text(encoding = "utf-8")) + functions = [ + node + for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "_build_browse_allowlist" + ] + module = ast.Module(body = functions, type_ignores = []) + ast.fix_missing_locations(module) + + drive_root = tmp_path / "D_drive" + drive_root.mkdir() + + calls = {"drive": 0, "media": 0} + + def _drive_roots(): + calls["drive"] += 1 + return [drive_root] + + def _media_roots(): + calls["media"] += 1 + return [] + + fake_paths = SimpleNamespace( + hf_default_cache_dir = lambda: tmp_path / "missing-default-hf", + legacy_hf_cache_dir = lambda: tmp_path / "missing-legacy-hf", + well_known_model_dirs = lambda: [], + studio_root = lambda: tmp_path / "missing-studio", + outputs_root = lambda: tmp_path / "missing-outputs", + exports_root = lambda: tmp_path / "missing-exports", + ) + fake_external_media = SimpleNamespace( + linux_run_media_mount_roots = _media_roots, + windows_drive_roots = _drive_roots, + ) + fake_studio_db = SimpleNamespace(list_scan_folders = lambda: []) + monkeypatch.setitem(sys.modules, "utils.paths", fake_paths) + monkeypatch.setitem(sys.modules, "utils.paths.external_media", fake_external_media) + monkeypatch.setitem(sys.modules, "storage.studio_db", fake_studio_db) + + ns = { + "os": os, + "Path": Path, + "Optional": Optional, + "_safe_is_dir": lambda p: Path(p).is_dir(), + "_resolve_hf_cache_dir": lambda: tmp_path / "missing-hf", + "logger": SimpleNamespace(debug = lambda *_args, **_kwargs: None), + } + exec(compile(module, "", "exec"), ns) + build = ns["_build_browse_allowlist"] + + # Roots passed in -> neither helper is probed, but the roots still flow in. + allowlist = build([], [drive_root]) + assert calls == {"drive": 0, "media": 0} + assert drive_root.resolve() in allowlist + + # No args -> each helper is probed exactly once. + build() + assert calls == {"drive": 1, "media": 1} + + +def test_is_path_inside_allowlist_real_descendants_and_siblings(tmp_path): + # Component-wise containment (commonpath): a genuine descendant is allowed, + # but a sibling sharing only a string prefix ("models_root_evil" vs + # "models_root") is not, which the old startswith check could miss. + ns = _extract_routes_function("_is_path_inside_allowlist") + root = tmp_path / "models_root" + child = root / "gguf" / "qwen" + sibling = tmp_path / "models_root_evil" + child.mkdir(parents = True) + sibling.mkdir() + + is_inside = ns["_is_path_inside_allowlist"] + assert is_inside(root, [root]) is True # the root itself + assert is_inside(child, [root]) is True # a genuine descendant + assert is_inside(sibling, [root]) is False # prefix-collision sibling + + +def test_is_path_inside_allowlist_posix_root_does_not_authorize_descendants(monkeypatch): + # Regression for the reported POSIX "/" unlock: a bare filesystem root may + # match itself but must NOT authorize arbitrary descendants such as /etc. + ns = _extract_routes_function("_is_path_inside_allowlist") + monkeypatch.setattr(os.path, "realpath", lambda p: str(p)) # keep "/" intact + + is_inside = ns["_is_path_inside_allowlist"] + assert is_inside("/", ["/"]) is True # the root itself + assert is_inside("/etc", ["/"]) is False # not a licensed descendant + assert is_inside("/root/models", ["/"]) is False + + +def test_is_path_inside_allowlist_windows_drive_root_descendants(): + # Exercise the Windows drive-root branch on a POSIX host by backing os.path + # with ntpath and an identity realpath (the simulated drives don't exist + # here). A drive root authorizes its descendants; a different drive does not. + import ntpath + + win_os = SimpleNamespace( + sep = "\\", + path = SimpleNamespace( + normcase = ntpath.normcase, + realpath = lambda p: str(p), + splitdrive = ntpath.splitdrive, + dirname = ntpath.dirname, + commonpath = ntpath.commonpath, + ), + ) + ns = _extract_routes_function("_is_path_inside_allowlist", {"os": win_os}) + is_inside = ns["_is_path_inside_allowlist"] + + assert is_inside("D:\\", ["D:\\"]) is True # drive root itself + assert is_inside("D:\\models", ["D:\\"]) is True # descendant on the drive + assert is_inside("D:\\models\\gguf", ["D:\\"]) is True # deeper descendant + assert is_inside("d:\\models", ["D:\\"]) is True # case-insensitive drive letter + assert is_inside("C:\\Users", ["D:\\"]) is False # different drive + assert is_inside("D:\\models", ["E:\\"]) is False diff --git a/studio/backend/utils/paths/external_media.py b/studio/backend/utils/paths/external_media.py index 1f1754664..0ea0477cc 100644 --- a/studio/backend/utils/paths/external_media.py +++ b/studio/backend/utils/paths/external_media.py @@ -8,6 +8,10 @@ from __future__ import annotations import getpass import os import platform +import string +import threading +import time +from collections.abc import Iterable from pathlib import Path from utils.paths.sensitive import ( @@ -16,6 +20,33 @@ from utils.paths.sensitive import ( ) +def is_local_filesystem_root(path: str, *, _pathmod = os.path) -> bool: + """True for a bare local filesystem root -- POSIX ``/``, a drive root ``C:\\``, + or a device-namespace volume root like ``\\\\?\\C:\\`` or + ``\\\\?\\Volume{GUID}\\`` -- which sit above denied system dirs, but NOT a UNC + share root (``\\\\server\\share`` or its ``\\\\?\\UNC\\...`` form), which has + none under it and was registerable before this guard. ``splitdrive`` is empty + on POSIX servers, so this reduces to the plain ``dirname == self`` test there. + ``_pathmod`` lets tests drive ``ntpath`` semantics on a POSIX CI. + """ + # Resolve the Windows device / extended-length namespace, where \\?\C:\, + # \\.\C:\ and \\?\Volume{GUID}\ are all bare LOCAL volume roots (rejected) + # while only \\?\UNC\server\share is a UNC share (handled like \\server\share). + if path[:4].lower() in ("\\\\?\\", "\\\\.\\"): + rest = path[4:] + if rest[:4].lower() == "unc\\": + path = "\\\\" + rest[4:] + else: + # A device volume root is just the volume specifier (C:, Volume{GUID}) + # with no further component; a deeper path is an ordinary folder. + core = rest.rstrip("\\/") + return "\\" not in core and "/" not in core + if _pathmod.dirname(path) != path: + return False + drive, _ = _pathmod.splitdrive(path) + return drive[:2] not in ("\\\\", "//") + + def _is_linux_media_mount_path(path: str, media_root: Path | str) -> bool: normalized = os.path.normpath(os.path.realpath(os.path.expanduser(path))) root = os.path.normpath(os.path.realpath(os.path.expanduser(str(media_root)))) @@ -98,3 +129,101 @@ def linux_run_media_mount_roots( seen.add(key) roots.append(resolved) return roots + + +def _active_windows_drive_bitmask() -> int: + """Active-logical-drive bitmask from ``GetLogicalDrives`` (bit 0 = ``A:``), or ``0`` when unavailable. + + A fast non-blocking call that lets :func:`windows_drive_roots` skip the + ``os.path.isdir`` probe on unmapped letters. A disconnected network mapping + stays set here, so it does not guard the reconnect stall on its own; + :func:`windows_drive_roots` bounds each surviving probe too. Returns ``0`` + (probe every letter) when ctypes/``windll`` is missing. + """ + try: + import ctypes + return int(ctypes.windll.kernel32.GetLogicalDrives()) + except Exception: # noqa: BLE001 -- best-effort; fall back to probing all letters + return 0 + + +# A disconnected mapped drive stays set in the GetLogicalDrives bitmask, so +# ``os.path.isdir`` on it can block for tens of seconds. Bound each drive probe +# so one stale mapping cannot stall a whole folder-browser request. +_DRIVE_PROBE_TIMEOUT_S = 2.0 + + +def _readable_dirs_within(paths: Iterable[str], timeout: float) -> set[str]: + """Which of *paths* are readable directories, probed concurrently under one overall *timeout* (seconds). + + Each path is checked (``os.path.isdir`` + ``os.access(R_OK)``) in its own + daemon thread and the call waits at most *timeout* total, not per path, so N + stalled network drives add ~timeout instead of N*timeout. A path not + answering ``True`` by the deadline is treated as unreadable. The daemon + threads are never joined past the deadline, so a stuck OS call cannot delay + interpreter exit or block the caller (``os.path.isdir`` releases the GIL). + """ + paths = list(paths) # fixed input we can iterate twice; one probe per path + results: dict[str, bool] = {} + + def _probe(path: str) -> None: + try: + results[path] = os.path.isdir(path) and os.access(path, os.R_OK) + except OSError: + results[path] = False + + threads: list[threading.Thread] = [] + for path in paths: + thread = threading.Thread(target = _probe, args = (path,), daemon = True) + thread.start() + threads.append(thread) + + deadline = time.monotonic() + timeout + for thread in threads: + thread.join(max(0.0, deadline - time.monotonic())) + + # Iterate the fixed input, not results.items(): a probe that timed out is + # still alive and may insert its key here, which would raise "dictionary + # changed size during iteration". results.get() is an atomic read. + return {path for path in paths if results.get(path)} + + +def _readable_dir_within(path: str, timeout: float) -> bool: + """``os.path.isdir(path) and os.access(path, R_OK)``, bounded by *timeout* seconds; single-path wrapper over :func:`_readable_dirs_within`.""" + return path in _readable_dirs_within((path,), timeout) + + +def windows_drive_roots(drive_letters: Iterable[str] = string.ascii_uppercase) -> list[Path]: + """Readable logical drive roots (``C:\\``, ``D:\\`` ...) for the folder browser; the Windows analog of :func:`linux_run_media_mount_roots`. + + Without it the allowlist and chips only reach the home drive, so a user + cannot navigate from ``C:`` to ``D:``/``E:``. ``GetLogicalDrives`` drops + unmapped letters; the rest are probed concurrently under a single timeout + and kept only if readable in time. A disconnected mapped drive stays active + in the bitmask and its ``os.path.isdir`` can hang for tens of seconds, so + parallel probing bounds the added delay at ~one timeout rather than one per + drive. Returns ``[]`` off Windows. + """ + if platform.system() != "Windows": + return [] + + active_mask = _active_windows_drive_bitmask() + candidates: list[str] = [] + seen: set[str] = set() + for letter in drive_letters: + letter = letter.strip().rstrip(":").upper() + if len(letter) != 1 or letter not in string.ascii_uppercase: + continue + if active_mask and not active_mask & (1 << (ord(letter) - ord("A"))): + continue + root_text = f"{letter}:\\" + key = os.path.normcase(root_text) + if key in seen: + continue + seen.add(key) + candidates.append(root_text) + + # Bounded concurrent probe: an active bitmask bit can still be a + # disconnected mapping whose os.path.isdir blocks, so probe all at once. + readable = _readable_dirs_within(candidates, _DRIVE_PROBE_TIMEOUT_S) + return [Path(root_text) for root_text in candidates if root_text in readable]