mirror of
https://github.com/unslothai/unsloth.git
synced 2026-08-07 07:54:10 +00:00
Studio: expose Windows drive roots in the folder browser (#7082)
* Studio: expose Windows drive roots in the folder browser The model-selection folder browser bounds navigation to the roots returned by _build_browse_allowlist(), which exposed Linux removable-media mounts via linux_run_media_mount_roots() but had no Windows analog. As a result a user on C: could not browse to D:/E: to pick a model directory. Add windows_drive_roots(), a Windows-only companion to linux_run_media_mount_roots() that lists readable logical drive roots, and wire it into both browse-allowlist builders and their suggestion chips so other drives are both navigable and offered as quick-picks. The helper is a no-op on Linux/macOS, so existing platforms are unaffected. Closes #6368 * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: cover the Windows drive-root browse wiring with an integration test Add an allowlist integration test mirroring the Linux side's test_legacy_browse_allowlist_includes_linux_run_media_mounts: it extracts _build_browse_allowlist from routes/models.py, stubs external_media so windows_drive_roots() yields a fake drive root, and asserts that root becomes browsable through the built allowlist. Proves the wiring, not just the helper. * Studio: skip inactive drives via GetLogicalDrives before probing Resolve active logical drives from GetLogicalDrives() before probing each letter with os.path.isdir. Probing a drive letter mapped to a disconnected network share can otherwise block the async backend for tens of seconds per letter. The call degrades gracefully (falls back to probing all letters) when ctypes/windll is unavailable, so behavior is unchanged on Linux/macOS. Tests override the bitmask source to stay deterministic on real Windows hosts. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: allow browsing descendants of a drive-root allowlist entry routes/models.py _is_path_inside_allowlist() checked descendants with startswith(root_real + os.sep). A drive root ("D:\") already ends in a separator, so the prefix became "D:\\" and a child like "D:\models" was rejected with 403 after the browser opened the drive root. Only append a separator when the root does not already end in one. folder_browser.py already uses commonpath and was unaffected. Adds a regression test covering the separator-terminated-root descendant case. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: enforce the system-directory denylist during folder browsing Exposing whole Windows drive roots (and any legacy-registered filesystem root) widened the browse allowlist above system directories, but the browse resolvers only re-applied the credential/config denylist, not the _denied_path_prefixes() system-dir denylist that scan-folder registration enforces. That let browse-folders enumerate C:\Windows, C:\Program Files, /etc and /proc. - Add is_denied_system_path() to both storage modules and enforce it in both browse resolvers (legacy routes/models.py and hub folder_browser.py), on each resolved child and on the final target, keeping the /run/media carve-out. - Rework the legacy _is_path_inside_allowlist to use splitdrive + commonpath so a Windows drive root authorizes its descendants while a bare POSIX / does not, and to compare case-insensitively like the hub browser. - Reject the filesystem root in the legacy add_scan_folder, matching the hub. - Hide denied system dirs from browse listings and suggestion chips. - Add tests/test_browse_denylist.py and update the external-media path tests. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: make browse-denylist tests OS-portable The browse-time denylist tests used real /etc and tmp_path locations; on macOS tmp lives under the (legitimately denied) /private/var and /etc resolves to /private/etc, so three tests failed there. Pin the platform / use a tmp-based denied prefix so they assert the same behavior on Linux, macOS and Windows. * Studio: apply the bare POSIX-root guard to the hub folder browser too The _is_path_inside_allowlist guard that stops a legacy-registered '/' scan folder from authorizing every absolute path lived only in the legacy browser. The hub browser used commonpath without it, so a stale '/' row let it descend into /var, /root, /home -- which the system-directory denylist (/proc /sys /dev /etc /boot /run) does not cover, while the legacy browser blocked them. Mirror the legacy guard so both browsers treat '/' identically. Also resolve each directory entry before the denylist check in both listing loops, so a symlink or junction pointing into a denied dir is hidden instead of rendered as a row that 403s on descent. Adds legacy-vs-hub parity tests. * Studio: bound Windows drive probing so a disconnected mapping can't stall the browser GetLogicalDrives includes mapped network drives, so a disconnected but still mapped drive (e.g. Z: -> \\nas\share) stays set in the bitmask and reaches os.path.isdir, which can block for tens of seconds while Windows tries to reconnect. Because windows_drive_roots() runs synchronously while building both folder-browser responses, one stale mapping stalled every browse request. Probe each surviving drive in a daemon thread bounded by a short timeout and skip it if it does not answer in time, so a hung mapping is dropped instead of blocking the caller. Connected drives (local or network) still respond well within the timeout, so drive discovery is unchanged. Corrects the GetLogicalDrives docstring, which claimed the bitmask alone prevented the stall. * Studio: probe drive/media roots once per browse request, not twice Both folder browsers called windows_drive_roots() (and linux_run_media_mount_roots()) twice per browse request: once to seed the allowlist in _build_browse_allowlist() and again to build the suggestion chips. With the bounded drive probe, a disconnected mapped network drive then paid the timeout twice per folder click. Probe both once in the request handler and pass the results into _build_browse_allowlist(), reusing them for the chips, in both the legacy and hub browsers. Adds a test asserting the roots are reused, not re-probed. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: run the legacy browse endpoint in the threadpool, fix its stale test Two follow-ups from review of the drive-probe changes: - browse_folders was 'async def' but does only blocking filesystem I/O (the timeout-bounded drive probe, iterdir, realpath). On the event loop a disconnected mapped drive waiting out its probe timeout stalled every other request. Declare it sync 'def' so FastAPI runs it in the threadpool, matching the hub browse endpoint. No await was used in the body. - test_browse_folders_hides_sensitive_dirs monkeypatched _build_browse_allowlist with a zero-arg lambda; the once-per-request refactor now calls it with (media_roots, drive_roots), so the lambda raised TypeError. Accept and ignore the args. * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: probe Windows drive roots concurrently so multiple dead mappings don't stack timeouts windows_drive_roots() probed each candidate serially, so N disconnected-but-mapped network drives each paid the full per-drive timeout in turn (e.g. four stale mappings added ~8s to every folder-browser request). Collect the candidate roots first, then probe them all at once under a single overall deadline, so the added delay stays at ~one timeout regardless of how many drives are disconnected. _readable_dir_within stays as a thin single-path wrapper for its existing callers/tests. * Studio: tighten comments in the folder-browser drive-root changes Condense the comments and docstrings added by the Windows drive-root and system-directory denylist work to be shorter and clearer while keeping the security and correctness rationale intact. Comment and docstring text only; no code changes. * Studio: iterate the input, not the results dict, when collecting readable drive probes _readable_dirs_within returned {path for path, ok in results.items()...}, but a probe thread that exceeded the join deadline is still alive and can insert its key into results during that iteration, raising 'dictionary changed size during iteration' -- reachable exactly in the disconnected-mapped-drive case the probe exists for. Iterate the fixed input list and read results.get(path) (an atomic read) instead. * Studio: keep the browse-route containment tests denylist-inert so they pass on macOS test_browse_folders_route.py exercises allowlist containment and the file-vs-directory guard, not the system-directory denylist. On macOS pytest tmp_path resolves under /private/var, a denied prefix, so _resolve_browse_target 403s the fixture dirs before the containment logic runs (4 failures). Add an autouse fixture that makes is_denied_system_path inert in this file; the denylist keeps its own coverage in test_browse_denylist.py. * Studio: keep the hub browse tests denylist-inert so they pass on macOS * Studio: register a UNC share root; only reject local filesystem roots * Studio: reject device drive roots and browse a registered UNC share root * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Studio: treat device-namespace volume GUID roots as local filesystem roots --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: danielhanchen <danielhanchen@gmail.com>
This commit is contained in:
Gaurav Dubey
GitHub
parent
f9aa818ca8
commit
dc65638b7d
10 changed files with 1117 additions and 28 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
||||
|
|
|
|||
371
studio/backend/tests/test_browse_denylist.py
Normal file
371
studio/backend/tests/test_browse_denylist.py
Normal file
|
|
@ -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, "<extracted studio_db.py>", "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/<user>/<volume> 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, "<extracted routes/models.py>", "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"<extracted {'/'.join(rel_parts)}>", "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
|
||||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
354
studio/backend/tests/test_windows_external_drive_paths.py
Normal file
354
studio/backend/tests/test_windows_external_drive_paths.py
Normal file
|
|
@ -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, "<extracted routes/models.py>", "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, "<extracted routes/models.py>", "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, "<extracted routes/models.py>", "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
|
||||
|
|
@ -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]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue