fix(local_image_loader): refuse filesystem roots and home as saved paths
save_paths is reachable without auth, so saving "/" made every media file on the machine servable. Reject filesystem/drive roots, the home directory and its ancestors (realpath-resolved, so symlinks count), and ignore such entries already in config.json.
This commit is contained in:
@@ -48,6 +48,25 @@ def is_within_roots(path: Any, roots: Iterable[Any]) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def is_too_broad_root(path: Any, home: Optional[str] = None) -> bool:
|
||||
"""Return True if path is unusable or too broad to be a saved root.
|
||||
|
||||
Refuses filesystem/drive roots, the home directory itself and any
|
||||
ancestor of it (e.g. /home), since those would expose every media file
|
||||
a user owns. Folders inside home are fine.
|
||||
"""
|
||||
real = resolve_path(path)
|
||||
if real is None:
|
||||
return True
|
||||
if os.path.dirname(real) == real: # "/" or a drive root like "C:\\"
|
||||
return True
|
||||
real_home = os.path.realpath(home or os.path.expanduser("~"))
|
||||
try:
|
||||
return os.path.commonpath([real, real_home]) == real
|
||||
except ValueError: # different drives
|
||||
return False
|
||||
|
||||
|
||||
def is_allowed_media_path(
|
||||
path: Any, roots: Iterable[Any], kinds: Iterable[str] = ("image",)
|
||||
) -> bool:
|
||||
|
||||
@@ -13,6 +13,7 @@ from .logic import (
|
||||
create_empty_tensor,
|
||||
is_allowed_image_path,
|
||||
is_allowed_media_path,
|
||||
is_too_broad_root,
|
||||
is_within_roots,
|
||||
list_subdirectories,
|
||||
load_image_from_path,
|
||||
@@ -82,7 +83,10 @@ def get_allowed_roots() -> List[str]:
|
||||
roots: List[str] = []
|
||||
saved = load_config().get("saved_paths", [])
|
||||
if isinstance(saved, list):
|
||||
roots.extend(p for p in saved if isinstance(p, str) and p)
|
||||
# Skip broad roots saved before save_paths refused them
|
||||
roots.extend(
|
||||
p for p in saved if isinstance(p, str) and p and not is_too_broad_root(p)
|
||||
)
|
||||
try:
|
||||
import folder_paths
|
||||
|
||||
@@ -235,6 +239,13 @@ async def save_paths(request):
|
||||
paths = data.get("paths", []) if isinstance(data, dict) else None
|
||||
if not isinstance(paths, list) or not all(isinstance(p, str) for p in paths):
|
||||
return _error(400, "paths must be a list of strings.")
|
||||
too_broad = [p for p in paths if is_too_broad_root(p)]
|
||||
if too_broad:
|
||||
return _error(
|
||||
400,
|
||||
f"Folder too broad to share: {', '.join(too_broad)}. "
|
||||
"Pick a folder inside your home directory or another media folder.",
|
||||
)
|
||||
config = load_config()
|
||||
config["saved_paths"] = paths
|
||||
save_config(config)
|
||||
|
||||
@@ -6,6 +6,7 @@ import pytest
|
||||
|
||||
from kikotools.tools.local_image_loader.logic import (
|
||||
is_allowed_image_path,
|
||||
is_too_broad_root,
|
||||
is_within_roots,
|
||||
list_subdirectories,
|
||||
resolve_path,
|
||||
@@ -124,3 +125,38 @@ class TestListSubdirectories:
|
||||
got = list_subdirectories("/e")
|
||||
assert all(os.path.isdir(p) for p in got)
|
||||
assert all(p.startswith("/e") for p in got)
|
||||
|
||||
|
||||
class TestTooBroadRoot:
|
||||
"""Saved roots that would expose everything are refused."""
|
||||
|
||||
@pytest.fixture
|
||||
def home(self, tmp_path):
|
||||
home = tmp_path / "home" / "user"
|
||||
home.mkdir(parents=True)
|
||||
return str(home)
|
||||
|
||||
def test_filesystem_root(self, home):
|
||||
assert is_too_broad_root(os.path.abspath(os.sep), home=home)
|
||||
|
||||
def test_home_itself(self, home):
|
||||
assert is_too_broad_root(home, home=home)
|
||||
assert is_too_broad_root(home + os.sep, home=home)
|
||||
|
||||
def test_ancestor_of_home(self, home):
|
||||
assert is_too_broad_root(os.path.dirname(home), home=home)
|
||||
|
||||
def test_folder_inside_home_ok(self, home):
|
||||
assert not is_too_broad_root(os.path.join(home, "Pictures"), home=home)
|
||||
|
||||
def test_unrelated_folder_ok(self, home, tmp_path):
|
||||
assert not is_too_broad_root(str(tmp_path / "srv" / "media"), home=home)
|
||||
|
||||
def test_symlink_to_home_refused(self, home, tmp_path):
|
||||
link = tmp_path / "link"
|
||||
link.symlink_to(home)
|
||||
assert is_too_broad_root(str(link), home=home)
|
||||
|
||||
@pytest.mark.parametrize("bad", [None, "", 5, "a\0b"])
|
||||
def test_unusable_values_refused(self, home, bad):
|
||||
assert is_too_broad_root(bad, home=home)
|
||||
|
||||
@@ -75,6 +75,16 @@ class TestAllowedRoots:
|
||||
with patch.dict("sys.modules", {"folder_paths": fp}):
|
||||
assert node_mod.get_allowed_roots() == []
|
||||
|
||||
def test_too_broad_saved_paths_ignored(self, tmp_path, monkeypatch):
|
||||
# A "/" saved before the guard existed must not keep working.
|
||||
monkeypatch.setattr(node_mod, "CONFIG_FILE", str(tmp_path / "c.json"))
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
node_mod.save_config(
|
||||
{"saved_paths": [os.path.abspath(os.sep), str(tmp_path), "/saved"]}
|
||||
)
|
||||
with patch.dict("sys.modules", {"folder_paths": None}):
|
||||
assert node_mod.get_allowed_roots() == ["/saved"]
|
||||
|
||||
def test_non_list_saved_paths(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(node_mod, "CONFIG_FILE", str(tmp_path / "c.json"))
|
||||
node_mod.save_config({"saved_paths": "oops"})
|
||||
@@ -335,6 +345,25 @@ class TestSelectionAndConfig:
|
||||
def test_save_paths_bad_input_400(self, env, payload):
|
||||
assert run(node_mod.save_paths(req(body=payload))).status == 400
|
||||
|
||||
@pytest.mark.parametrize("broad", ["root", "home", "home_parent"])
|
||||
def test_save_paths_refuses_too_broad_roots(
|
||||
self, env, monkeypatch, tmp_path, broad
|
||||
):
|
||||
home = tmp_path / "h" / "user"
|
||||
home.mkdir(parents=True)
|
||||
monkeypatch.setenv("HOME", str(home))
|
||||
path = {
|
||||
"root": os.path.abspath(os.sep),
|
||||
"home": str(home),
|
||||
"home_parent": str(home.parent),
|
||||
}[broad]
|
||||
node_mod.save_config({"saved_paths": ["/keep"]})
|
||||
r = run(node_mod.save_paths(req(body={"paths": ["/ok", path]})))
|
||||
assert r.status == 400
|
||||
assert path in body(r)["error"]
|
||||
# Nothing was saved
|
||||
assert node_mod.load_config()["saved_paths"] == ["/keep"]
|
||||
|
||||
def test_save_paths_bad_json_400(self, env):
|
||||
assert run(node_mod.save_paths(req(body=ValueError()))).status == 400
|
||||
|
||||
|
||||
Reference in New Issue
Block a user