mirror of
https://github.com/unslothai/unsloth.git
synced 2026-07-10 00:08:58 +00:00
Fix Studio custom folders on Linux external drives (#6799)
* Fix external drive custom folder selection * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Update studio/backend/tests/test_linux_external_media_paths.py Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Keep legacy media scan validation strict * Apply sensitive-dir denylist to legacy folder browser for PR #6799 The legacy /api/models browse endpoint gained the new /run/media mount roots in its allowlist but not the credential/config guard that scan-folder registration and the Hub browser already enforce. Filter sensitive names during enumeration and reject them in _resolve_browse_target so .ssh, .aws, .config, etc. under allowlisted roots stay unbrowseable, matching the Hub browser. Add a public contains_sensitive_path_component helper and cover the legacy resolver with a regression test. * Trim redundant comments in PR #6799 changes * Skip sensitive Linux media roots * Reject sensitive dirs at exact browse roots for PR #6799 Both _resolve_browse_target functions only checked contains_sensitive_path_component while walking descendant parts, so requesting an allowlisted root itself (empty relative path) returned it unchecked. A pre-existing scan-folder row under ~/.ssh, ~/.aws, ~/.config, etc. (registerable before the denylist was added) is re-added to the allowlist on upgrade and could then be browsed. Check the resolved target once before returning in both the legacy and Hub browsers, and cover the root case in both test suites. * fix: avoid unused path helper reexports * fix: import sensitive path helpers directly * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> Co-authored-by: Daniel Han <danielhanchen@gmail.com> Co-authored-by: imagineer99 <samleejackson0@gmail.com> Co-authored-by: Lee Jackson <130007945+Imagineer99@users.noreply.github.com>
This commit is contained in:
parent
9c2eacc35e
commit
01f7e14988
8 changed files with 520 additions and 31 deletions
|
|
@ -27,6 +27,7 @@ from hub.utils.paths import (
|
|||
studio_root,
|
||||
well_known_model_dirs,
|
||||
)
|
||||
from utils.paths.external_media import linux_run_media_mount_roots
|
||||
from hub.services.models.common import _safe_is_dir
|
||||
from hub.services.models.local_inventory import _resolve_hf_cache_dir
|
||||
|
||||
|
|
@ -175,6 +176,8 @@ def _build_browse_allowlist() -> list[Path]:
|
|||
candidates.append(resolved)
|
||||
|
||||
_add(Path.home())
|
||||
for p in linux_run_media_mount_roots():
|
||||
_add(p)
|
||||
_add(_resolve_hf_cache_dir())
|
||||
try:
|
||||
_add(hf_default_cache_dir())
|
||||
|
|
@ -346,6 +349,11 @@ def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Pa
|
|||
)
|
||||
current = resolved_child
|
||||
|
||||
if contains_sensitive_path_component(str(current)):
|
||||
raise HTTPException(
|
||||
status_code = 403,
|
||||
detail = "Credential or configuration directories are not browseable.",
|
||||
)
|
||||
if not current.is_dir():
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
|
|
@ -485,6 +493,8 @@ def browse_folders_response(
|
|||
|
||||
# Home first as the safe fallback.
|
||||
_add_sug(Path.home())
|
||||
for p in linux_run_media_mount_roots():
|
||||
_add_sug(p)
|
||||
# The HF cache root in use (honors HF_HOME / HF_HUB_CACHE), then the default.
|
||||
try:
|
||||
_add_sug(_resolve_hf_cache_dir())
|
||||
|
|
|
|||
|
|
@ -16,37 +16,14 @@ 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.sensitive import (
|
||||
contains_sensitive_path_component as _shared_contains_sensitive_path_component,
|
||||
)
|
||||
|
||||
|
||||
_schema_lock = threading.Lock()
|
||||
_schema_ready = False
|
||||
_SENSITIVE_PATH_COMPONENTS = {
|
||||
".aws",
|
||||
".azure",
|
||||
".config",
|
||||
".docker",
|
||||
".gcloud",
|
||||
".gnupg",
|
||||
".huggingface",
|
||||
".kaggle",
|
||||
".kube",
|
||||
".modelscope",
|
||||
".ngc",
|
||||
".local",
|
||||
".mozilla",
|
||||
".pki",
|
||||
".thunderbird",
|
||||
".ssh",
|
||||
".1password",
|
||||
".bitwarden",
|
||||
".password-store",
|
||||
"1password",
|
||||
"bitwarden",
|
||||
"keychains",
|
||||
"keyrings",
|
||||
"mozilla",
|
||||
"thunderbird",
|
||||
}
|
||||
|
||||
|
||||
def _denied_path_prefixes() -> list[str]:
|
||||
|
|
@ -76,8 +53,7 @@ def _denied_path_prefixes() -> list[str]:
|
|||
|
||||
|
||||
def _contains_sensitive_path_component(path: str) -> bool:
|
||||
parts = os.path.normpath(path).split(os.sep)
|
||||
return any(part.lower() in _SENSITIVE_PATH_COMPONENTS for part in parts)
|
||||
return _shared_contains_sensitive_path_component(path)
|
||||
|
||||
|
||||
def contains_sensitive_path_component(path: str) -> bool:
|
||||
|
|
@ -142,6 +118,8 @@ def add_scan_folder(path: str) -> dict:
|
|||
check = os.path.normcase(normalized) if is_win else normalized
|
||||
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
|
||||
raise ValueError(f"Path under {prefix} is not allowed")
|
||||
|
||||
conn = get_connection()
|
||||
|
|
|
|||
|
|
@ -168,6 +168,16 @@ def test_resolve_browse_target_rejects_sensitive_dir(tmp_path):
|
|||
assert exc_info.value.status_code == 403
|
||||
|
||||
|
||||
def test_resolve_browse_target_rejects_sensitive_root(tmp_path):
|
||||
ssh = tmp_path / "home" / ".ssh"
|
||||
ssh.mkdir(parents = True)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
folder_browser._resolve_browse_target(str(ssh), [ssh])
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
|
||||
def test_browse_folders_hides_sensitive_dirs(monkeypatch, tmp_path):
|
||||
home = tmp_path / "home"
|
||||
(home / ".ssh").mkdir(parents = True)
|
||||
|
|
@ -181,6 +191,24 @@ def test_browse_folders_hides_sensitive_dirs(monkeypatch, tmp_path):
|
|||
assert ".ssh" not in names
|
||||
|
||||
|
||||
def test_browse_allowlist_includes_linux_run_media_mounts(monkeypatch, tmp_path):
|
||||
home = tmp_path / "home"
|
||||
media_root = tmp_path / "run" / "media" / "dspofu" / "nvmeB"
|
||||
model_dir = media_root / "modelsAI" / "gguf" / "qwen3.6"
|
||||
home.mkdir()
|
||||
model_dir.mkdir(parents = True)
|
||||
monkeypatch.setattr(folder_browser.Path, "home", lambda: home)
|
||||
monkeypatch.setattr(folder_browser, "linux_run_media_mount_roots", lambda: [media_root])
|
||||
monkeypatch.setattr(folder_browser, "_resolve_hf_cache_dir", lambda: tmp_path / "missing-hf")
|
||||
monkeypatch.setattr(scan_folders, "list_scan_folders", lambda: [])
|
||||
monkeypatch.setattr(folder_browser, "well_known_model_dirs", lambda: [])
|
||||
|
||||
allowlist = folder_browser._build_browse_allowlist()
|
||||
|
||||
assert media_root.resolve() in allowlist
|
||||
assert folder_browser._resolve_browse_target(str(model_dir), allowlist) == model_dir.resolve()
|
||||
|
||||
|
||||
def test_get_models_folder_response_creates_and_returns_dir(monkeypatch, tmp_path):
|
||||
# The endpoint creates the cache dir on demand so the desktop "Open folder"
|
||||
# action works even before the first download.
|
||||
|
|
|
|||
|
|
@ -1202,6 +1202,7 @@ def _build_browse_allowlist() -> list[Path]:
|
|||
legacy_hf_cache_dir,
|
||||
well_known_model_dirs,
|
||||
)
|
||||
from utils.paths.external_media import linux_run_media_mount_roots
|
||||
from storage.studio_db import list_scan_folders
|
||||
|
||||
candidates: list[Path] = []
|
||||
|
|
@ -1217,6 +1218,8 @@ def _build_browse_allowlist() -> list[Path]:
|
|||
candidates.append(resolved)
|
||||
|
||||
_add(Path.home())
|
||||
for p in linux_run_media_mount_roots():
|
||||
_add(p)
|
||||
_add(_resolve_hf_cache_dir())
|
||||
try:
|
||||
_add(hf_default_cache_dir())
|
||||
|
|
@ -1336,6 +1339,8 @@ 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
|
||||
|
||||
requested_path = _normalize_browse_request_path(path)
|
||||
resolved_roots: list[Path] = []
|
||||
seen_roots: set[str] = set()
|
||||
|
|
@ -1386,8 +1391,18 @@ def _resolve_browse_target(path: Optional[str], allowed_roots: list[Path]) -> Pa
|
|||
"under your home folder."
|
||||
),
|
||||
)
|
||||
if contains_sensitive_path_component(str(resolved_child)):
|
||||
raise HTTPException(
|
||||
status_code = 403,
|
||||
detail = "Credential or configuration directories are not browseable.",
|
||||
)
|
||||
current = resolved_child
|
||||
|
||||
if contains_sensitive_path_component(str(current)):
|
||||
raise HTTPException(
|
||||
status_code = 403,
|
||||
detail = "Credential or configuration directories are not browseable.",
|
||||
)
|
||||
if not current.is_dir():
|
||||
raise HTTPException(
|
||||
status_code = 400,
|
||||
|
|
@ -1435,7 +1450,8 @@ async def browse_folders(
|
|||
then hidden (if ``show_hidden=true``).
|
||||
"""
|
||||
from utils.paths import hf_default_cache_dir, well_known_model_dirs
|
||||
from storage.studio_db import list_scan_folders
|
||||
from utils.paths.external_media import linux_run_media_mount_roots
|
||||
from storage.studio_db import contains_sensitive_path_component, list_scan_folders
|
||||
|
||||
# Build once; the sandbox check and suggestion chips share it.
|
||||
allowed_roots = _build_browse_allowlist()
|
||||
|
|
@ -1488,6 +1504,8 @@ async def browse_folders(
|
|||
is_hidden = name.startswith(".")
|
||||
if is_hidden and not show_hidden:
|
||||
continue
|
||||
if contains_sensitive_path_component(name):
|
||||
continue
|
||||
entries.append(
|
||||
BrowseEntry(
|
||||
name = name,
|
||||
|
|
@ -1541,6 +1559,8 @@ async def browse_folders(
|
|||
|
||||
# Home first -- the safe fallback when everything else is cold.
|
||||
_add_sug(Path.home())
|
||||
for p in linux_run_media_mount_roots():
|
||||
_add_sug(p)
|
||||
# The HF cache root the process is actually using.
|
||||
try:
|
||||
_add_sug(hf_default_cache_dir())
|
||||
|
|
|
|||
|
|
@ -22,7 +22,15 @@ logger = logging.getLogger(__name__)
|
|||
from typing import Any, Iterable, Optional
|
||||
|
||||
|
||||
from utils.paths import project_workspaces_root, studio_db_path, ensure_dir
|
||||
from utils.paths import (
|
||||
ensure_dir,
|
||||
project_workspaces_root,
|
||||
studio_db_path,
|
||||
)
|
||||
from utils.paths.external_media import is_linux_run_media_path
|
||||
from utils.paths.sensitive import (
|
||||
contains_sensitive_path_component as _shared_contains_sensitive_path_component,
|
||||
)
|
||||
from utils.training_runs import extract_project_name
|
||||
|
||||
|
||||
|
|
@ -61,6 +69,14 @@ def _denied_path_prefixes() -> list[str]:
|
|||
return []
|
||||
|
||||
|
||||
def _contains_sensitive_path_component(path: str) -> bool:
|
||||
return _shared_contains_sensitive_path_component(path)
|
||||
|
||||
|
||||
def contains_sensitive_path_component(path: str) -> bool:
|
||||
return _contains_sensitive_path_component(path)
|
||||
|
||||
|
||||
_schema_lock = threading.Lock()
|
||||
_schema_ready = False
|
||||
_SQLITE_IN_CHUNK_SIZE = 900
|
||||
|
|
@ -896,6 +912,8 @@ 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 _contains_sensitive_path_component(normalized):
|
||||
raise ValueError("Credential or configuration directories are not allowed")
|
||||
|
||||
# Windows: normcase for the denylist check but store original casing
|
||||
# so consumers see the native drive-letter casing (e.g. C:\Models).
|
||||
|
|
@ -903,6 +921,8 @@ def add_scan_folder(path: str) -> dict:
|
|||
check = os.path.normcase(normalized) if is_win else normalized
|
||||
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
|
||||
raise ValueError(f"Path under {prefix} is not allowed")
|
||||
|
||||
conn = get_connection()
|
||||
|
|
|
|||
287
studio/backend/tests/test_linux_external_media_paths.py
Normal file
287
studio/backend/tests/test_linux_external_media_paths.py
Normal file
|
|
@ -0,0 +1,287 @@
|
|||
# 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
|
||||
|
||||
import pytest
|
||||
|
||||
from hub.storage import scan_folders
|
||||
from storage import studio_db
|
||||
from utils.paths import external_media
|
||||
|
||||
|
||||
_BACKEND_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
class _ExistingScanFolderConn:
|
||||
def __init__(self):
|
||||
self.params = ()
|
||||
|
||||
def execute(
|
||||
self,
|
||||
_sql,
|
||||
params = (),
|
||||
):
|
||||
self.params = params
|
||||
return self
|
||||
|
||||
def fetchone(self):
|
||||
return {"id": 1, "path": self.params[0], "created_at": "fake"}
|
||||
|
||||
def commit(self):
|
||||
pass
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
class _HTTPException(Exception):
|
||||
def __init__(self, status_code: int, detail: str):
|
||||
super().__init__(detail)
|
||||
self.status_code = status_code
|
||||
self.detail = detail
|
||||
|
||||
|
||||
def _stub_linux_path_checks(monkeypatch, module):
|
||||
monkeypatch.setattr(module.platform, "system", lambda: "Linux")
|
||||
monkeypatch.setattr(module.os.path, "realpath", os.path.normpath)
|
||||
monkeypatch.setattr(module.os.path, "expanduser", lambda p: p)
|
||||
monkeypatch.setattr(module.os.path, "exists", lambda _p: True)
|
||||
monkeypatch.setattr(module.os.path, "isdir", lambda _p: True)
|
||||
monkeypatch.setattr(module.os, "access", lambda _p, _mode: True)
|
||||
|
||||
|
||||
def _stub_hub_scan_folder_db(monkeypatch):
|
||||
monkeypatch.setattr(scan_folders, "_ensure_schema", lambda _conn: None)
|
||||
monkeypatch.setattr(scan_folders, "get_connection", _ExistingScanFolderConn)
|
||||
|
||||
|
||||
def _stub_legacy_scan_folder_db(monkeypatch):
|
||||
monkeypatch.setattr(studio_db, "get_connection", _ExistingScanFolderConn)
|
||||
|
||||
|
||||
def test_linux_run_media_policy_accepts_mounted_volume_descendants(monkeypatch):
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
|
||||
assert external_media.is_linux_run_media_path("/run/media/dspofu/nvmeB")
|
||||
assert external_media.is_linux_run_media_path("/run/media/dspofu/nvmeB/modelsAI/gguf/qwen3.6")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"/run",
|
||||
"/run/media",
|
||||
"/run/media/dspofu",
|
||||
"/run/user/1000/models",
|
||||
"/run/systemd/private",
|
||||
"/run/not-media/dspofu/nvmeB",
|
||||
],
|
||||
)
|
||||
def test_linux_run_media_policy_rejects_unrelated_run_paths(monkeypatch, path):
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
|
||||
assert not external_media.is_linux_run_media_path(path)
|
||||
|
||||
|
||||
def test_linux_run_media_mount_roots_lists_readable_volume_roots(monkeypatch, tmp_path):
|
||||
base = tmp_path / "run" / "media"
|
||||
mount = base / "dspofu" / "nvmeB"
|
||||
sensitive_mount = base / "dspofu" / ".ssh"
|
||||
sensitive_aws_mount = base / "dspofu" / ".aws"
|
||||
other_user_mount = base / "other" / "backup"
|
||||
incomplete = base / "dspofu-only"
|
||||
mount.mkdir(parents = True)
|
||||
sensitive_mount.mkdir()
|
||||
sensitive_aws_mount.mkdir()
|
||||
other_user_mount.mkdir(parents = True)
|
||||
incomplete.mkdir()
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
|
||||
roots = external_media.linux_run_media_mount_roots(base, user = "dspofu")
|
||||
|
||||
assert roots == [mount.resolve()]
|
||||
|
||||
|
||||
def test_linux_run_media_mount_roots_skips_sensitive_resolved_volume_name(monkeypatch, tmp_path):
|
||||
base = tmp_path / "run" / "media"
|
||||
normal_mount = base / "dspofu" / "nvmeB"
|
||||
sensitive_target = base / "dspofu" / ".config"
|
||||
normal_mount.mkdir(parents = True)
|
||||
sensitive_target.mkdir()
|
||||
alias = base / "dspofu" / "config-alias"
|
||||
alias.symlink_to(sensitive_target, target_is_directory = True)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
|
||||
roots = external_media.linux_run_media_mount_roots(base, user = "dspofu")
|
||||
|
||||
assert roots == [normal_mount.resolve()]
|
||||
|
||||
|
||||
def test_linux_run_media_mount_roots_skips_sensitive_resolved_descendant(monkeypatch, tmp_path):
|
||||
base = tmp_path / "run" / "media"
|
||||
normal_mount = base / "dspofu" / "nvmeB"
|
||||
sensitive_descendant = normal_mount / ".ssh" / "models"
|
||||
sensitive_descendant.mkdir(parents = True)
|
||||
alias = base / "dspofu" / "models-alias"
|
||||
alias.symlink_to(sensitive_descendant, target_is_directory = True)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
|
||||
roots = external_media.linux_run_media_mount_roots(base, user = "dspofu")
|
||||
|
||||
assert roots == [normal_mount.resolve()]
|
||||
|
||||
|
||||
def test_hub_scan_folder_accepts_linux_run_media_mount(monkeypatch):
|
||||
_stub_linux_path_checks(monkeypatch, scan_folders)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
_stub_hub_scan_folder_db(monkeypatch)
|
||||
target = "/run/media/dspofu/nvmeB/modelsAI/gguf/qwen3.6"
|
||||
|
||||
row = scan_folders.add_scan_folder(target)
|
||||
|
||||
assert row["path"] == target
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target",
|
||||
[
|
||||
"/run",
|
||||
"/run/media",
|
||||
"/run/media/dspofu",
|
||||
"/run/user/1000/models",
|
||||
"/run/systemd/private",
|
||||
"/run/not-media/dspofu/nvmeB",
|
||||
],
|
||||
)
|
||||
def test_hub_scan_folder_keeps_unrelated_run_paths_blocked(monkeypatch, target):
|
||||
_stub_linux_path_checks(monkeypatch, scan_folders)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
_stub_hub_scan_folder_db(monkeypatch)
|
||||
|
||||
with pytest.raises(ValueError, match = "Path under /run is not allowed"):
|
||||
scan_folders.add_scan_folder(target)
|
||||
|
||||
|
||||
def test_hub_scan_folder_keeps_sensitive_dirs_blocked_under_run_media(monkeypatch):
|
||||
_stub_linux_path_checks(monkeypatch, scan_folders)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
_stub_hub_scan_folder_db(monkeypatch)
|
||||
|
||||
with pytest.raises(ValueError, match = "Credential or configuration"):
|
||||
scan_folders.add_scan_folder("/run/media/dspofu/nvmeB/.ssh/models")
|
||||
|
||||
|
||||
def test_legacy_scan_folder_accepts_linux_run_media_mount(monkeypatch):
|
||||
_stub_linux_path_checks(monkeypatch, studio_db)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
_stub_legacy_scan_folder_db(monkeypatch)
|
||||
target = "/run/media/dspofu/nvmeB/modelsAI/gguf/qwen3.6"
|
||||
|
||||
row = studio_db.add_scan_folder(target)
|
||||
|
||||
assert row["path"] == target
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target",
|
||||
[
|
||||
"/run",
|
||||
"/run/media",
|
||||
"/run/media/dspofu",
|
||||
"/run/user/1000/models",
|
||||
"/run/systemd/private",
|
||||
"/run/not-media/dspofu/nvmeB",
|
||||
],
|
||||
)
|
||||
def test_legacy_scan_folder_keeps_unrelated_run_paths_blocked(monkeypatch, target):
|
||||
_stub_linux_path_checks(monkeypatch, studio_db)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
_stub_legacy_scan_folder_db(monkeypatch)
|
||||
|
||||
with pytest.raises(ValueError, match = "Path under /run is not allowed"):
|
||||
studio_db.add_scan_folder(target)
|
||||
|
||||
|
||||
def test_legacy_scan_folder_keeps_sensitive_dirs_blocked_under_run_media(monkeypatch):
|
||||
_stub_linux_path_checks(monkeypatch, studio_db)
|
||||
monkeypatch.setattr(external_media.platform, "system", lambda: "Linux")
|
||||
_stub_legacy_scan_folder_db(monkeypatch)
|
||||
|
||||
with pytest.raises(ValueError, match = "Credential or configuration"):
|
||||
studio_db.add_scan_folder("/run/media/dspofu/nvmeB/.aws/models")
|
||||
|
||||
|
||||
def test_legacy_browse_allowlist_includes_linux_run_media_mounts(monkeypatch, tmp_path):
|
||||
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"
|
||||
media_root = tmp_path / "run" / "media" / "dspofu" / "nvmeB"
|
||||
model_dir = media_root / "modelsAI" / "gguf" / "qwen3.6"
|
||||
home.mkdir()
|
||||
model_dir.mkdir(parents = True)
|
||||
(media_root / ".ssh").mkdir()
|
||||
|
||||
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: [media_root])
|
||||
fake_studio_db = SimpleNamespace(
|
||||
list_scan_folders = lambda: [],
|
||||
contains_sensitive_path_component = studio_db.contains_sensitive_path_component,
|
||||
)
|
||||
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"]()
|
||||
|
||||
assert media_root.resolve() in allowlist
|
||||
assert ns["_resolve_browse_target"](str(model_dir), allowlist) == model_dir.resolve()
|
||||
|
||||
with pytest.raises(_HTTPException) as exc:
|
||||
ns["_resolve_browse_target"](str(media_root / ".ssh"), allowlist)
|
||||
assert exc.value.status_code == 403
|
||||
|
||||
ssh_root = media_root / ".ssh"
|
||||
with pytest.raises(_HTTPException) as exc_root:
|
||||
ns["_resolve_browse_target"](str(ssh_root), [ssh_root])
|
||||
assert exc_root.value.status_code == 403
|
||||
100
studio/backend/utils/paths/external_media.py
Normal file
100
studio/backend/utils/paths/external_media.py
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
||||
|
||||
"""External media path helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import getpass
|
||||
import os
|
||||
import platform
|
||||
from pathlib import Path
|
||||
|
||||
from utils.paths.sensitive import (
|
||||
contains_sensitive_path_component,
|
||||
is_sensitive_path_component,
|
||||
)
|
||||
|
||||
|
||||
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))))
|
||||
try:
|
||||
rel = os.path.relpath(normalized, root)
|
||||
except ValueError:
|
||||
return False
|
||||
if rel == "." or rel == ".." or rel.startswith(f"..{os.sep}"):
|
||||
return False
|
||||
parts = [part for part in rel.split(os.sep) if part]
|
||||
return len(parts) >= 2 and all(part not in (".", "..") for part in parts[:2])
|
||||
|
||||
|
||||
def is_linux_run_media_path(path: str) -> bool:
|
||||
"""True for Linux removable-media paths under /run/media/<user>/<volume>."""
|
||||
if platform.system() != "Linux":
|
||||
return False
|
||||
return _is_linux_media_mount_path(path, "/run/media")
|
||||
|
||||
|
||||
def _current_username() -> str | None:
|
||||
try:
|
||||
user = getpass.getuser().strip()
|
||||
except Exception:
|
||||
return None
|
||||
return user or None
|
||||
|
||||
|
||||
def _contains_sensitive_media_component(path: Path, media_root: Path) -> bool:
|
||||
try:
|
||||
rel = path.relative_to(media_root)
|
||||
except ValueError:
|
||||
rel = path
|
||||
return contains_sensitive_path_component(str(rel))
|
||||
|
||||
|
||||
def linux_run_media_mount_roots(
|
||||
base: Path | str = "/run/media", *, user: str | None = None
|
||||
) -> list[Path]:
|
||||
"""Readable /run/media/<user>/<volume> roots for the folder browser."""
|
||||
if platform.system() != "Linux":
|
||||
return []
|
||||
user = user or _current_username()
|
||||
if not user or user in (".", "..") or os.sep in user:
|
||||
return []
|
||||
base_path = Path(base)
|
||||
try:
|
||||
resolved_base = base_path.resolve()
|
||||
except (OSError, RuntimeError, ValueError):
|
||||
return []
|
||||
|
||||
roots: list[Path] = []
|
||||
seen: set[str] = set()
|
||||
user_dir = base_path / user
|
||||
try:
|
||||
if not user_dir.is_dir():
|
||||
return []
|
||||
volume_dirs = list(user_dir.iterdir())
|
||||
except (OSError, RuntimeError, ValueError):
|
||||
return []
|
||||
for volume_dir in volume_dirs:
|
||||
if is_sensitive_path_component(volume_dir.name):
|
||||
continue
|
||||
try:
|
||||
resolved = volume_dir.resolve()
|
||||
except (OSError, RuntimeError, ValueError):
|
||||
continue
|
||||
if not _is_linux_media_mount_path(str(resolved), resolved_base):
|
||||
continue
|
||||
if _contains_sensitive_media_component(resolved, resolved_base):
|
||||
continue
|
||||
key = os.path.normcase(os.path.realpath(str(resolved)))
|
||||
if key in seen:
|
||||
continue
|
||||
try:
|
||||
is_dir = resolved.is_dir()
|
||||
except OSError:
|
||||
continue
|
||||
if is_dir and os.access(resolved, os.R_OK | os.X_OK):
|
||||
seen.add(key)
|
||||
roots.append(resolved)
|
||||
return roots
|
||||
46
studio/backend/utils/paths/sensitive.py
Normal file
46
studio/backend/utils/paths/sensitive.py
Normal file
|
|
@ -0,0 +1,46 @@
|
|||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
||||
|
||||
"""Shared sensitive path-component policy."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
|
||||
SENSITIVE_PATH_COMPONENTS = {
|
||||
".aws",
|
||||
".azure",
|
||||
".config",
|
||||
".docker",
|
||||
".gcloud",
|
||||
".gnupg",
|
||||
".huggingface",
|
||||
".kaggle",
|
||||
".kube",
|
||||
".modelscope",
|
||||
".ngc",
|
||||
".local",
|
||||
".mozilla",
|
||||
".pki",
|
||||
".thunderbird",
|
||||
".ssh",
|
||||
".1password",
|
||||
".bitwarden",
|
||||
".password-store",
|
||||
"1password",
|
||||
"bitwarden",
|
||||
"keychains",
|
||||
"keyrings",
|
||||
"mozilla",
|
||||
"thunderbird",
|
||||
}
|
||||
|
||||
|
||||
def is_sensitive_path_component(name: str) -> bool:
|
||||
return name.lower() in SENSITIVE_PATH_COMPONENTS
|
||||
|
||||
|
||||
def contains_sensitive_path_component(path: str) -> bool:
|
||||
parts = os.path.normpath(path).split(os.sep)
|
||||
return any(is_sensitive_path_component(part) for part in parts)
|
||||
Loading…
Add table
Add a link
Reference in a new issue