refactor(core): reuse shared path containment check
This commit is contained in:
@@ -73,6 +73,7 @@ def register_routes(self):
|
||||
get_comfy_root_path,
|
||||
get_filename_from_path,
|
||||
is_path_in_configured_model_roots,
|
||||
is_path_within,
|
||||
prefer_local_base_directory,
|
||||
read_json_safe,
|
||||
split_path_segments,
|
||||
|
||||
@@ -23,6 +23,7 @@ class DirectoryService:
|
||||
"get_enabled_download_categories"
|
||||
)
|
||||
self.get_local_path_identity = context.require("get_local_path_identity")
|
||||
self.is_path_within = context.require("is_path_within")
|
||||
self.is_civarchive_available = context.require("is_civarchive_available")
|
||||
self.is_lora_manager_archive_available = context.require(
|
||||
"is_lora_manager_archive_available"
|
||||
@@ -232,19 +233,9 @@ class DirectoryService:
|
||||
def find_base_dir(full_path):
|
||||
if not full_path:
|
||||
return ""
|
||||
full_path_identity = self.get_local_path_identity(full_path)
|
||||
for base_dir in base_dirs:
|
||||
base_identity = self.get_local_path_identity(base_dir)
|
||||
try:
|
||||
if (
|
||||
os.path.commonpath(
|
||||
[full_path_identity, base_identity]
|
||||
)
|
||||
== base_identity
|
||||
):
|
||||
return base_dir
|
||||
except Exception:
|
||||
continue
|
||||
if self.is_path_within(full_path, base_dir):
|
||||
return base_dir
|
||||
return ""
|
||||
|
||||
for folder_key in available_folder_keys:
|
||||
|
||||
@@ -9,6 +9,7 @@ from unittest.mock import MagicMock, patch
|
||||
import pytest
|
||||
from aiohttp import web
|
||||
|
||||
from core.path_utils import is_path_within
|
||||
from core.routes.context import RouteContext
|
||||
from core.routes.directories import register_directory_routes
|
||||
from core.type_utils import to_bool
|
||||
@@ -57,6 +58,7 @@ def _build_routes(**overrides):
|
||||
"get_local_path_identity": lambda path: os.path.normcase(
|
||||
os.path.abspath(path)
|
||||
),
|
||||
"is_path_within": is_path_within,
|
||||
"get_model_files": MagicMock(return_value=[]),
|
||||
"infer_download_path_templates": MagicMock(return_value=[]),
|
||||
"invalidate_local_hash_match_cache": MagicMock(),
|
||||
|
||||
@@ -36,6 +36,7 @@ from core.path_utils import (
|
||||
get_safe_metadata_sidecar_path,
|
||||
get_safe_model_resolver_sidecar_path,
|
||||
is_path_in_configured_model_roots,
|
||||
is_path_within,
|
||||
)
|
||||
|
||||
|
||||
@@ -51,6 +52,17 @@ class DummyFolderPaths:
|
||||
|
||||
|
||||
class PathSecurityTests(unittest.TestCase):
|
||||
def test_is_path_within_rejects_sibling_directory_with_shared_prefix(self):
|
||||
with tempfile.TemporaryDirectory() as temp_root:
|
||||
model_root = os.path.join(temp_root, "models")
|
||||
sibling_root = os.path.join(temp_root, "models2")
|
||||
os.makedirs(model_root)
|
||||
os.makedirs(sibling_root)
|
||||
|
||||
self.assertTrue(is_path_within(model_root, model_root))
|
||||
self.assertTrue(is_path_within(os.path.join(model_root, "model.safetensors"), model_root))
|
||||
self.assertFalse(is_path_within(os.path.join(sibling_root, "model.safetensors"), model_root))
|
||||
|
||||
def test_external_model_root_is_allowed(self):
|
||||
with tempfile.TemporaryDirectory() as model_root:
|
||||
model_path = os.path.join(model_root, "model.safetensors")
|
||||
|
||||
@@ -6,6 +6,7 @@ from unittest.mock import MagicMock, patch
|
||||
import pytest
|
||||
from aiohttp import web
|
||||
|
||||
from core.path_utils import is_path_within
|
||||
from core.routes.context import RouteContext
|
||||
from core.routes.directories import register_directory_routes
|
||||
from core.routes.metadata import register_metadata_routes
|
||||
@@ -81,6 +82,7 @@ def _build_directory_routes():
|
||||
side_effect=lambda categories: categories
|
||||
),
|
||||
"get_local_path_identity": MagicMock(),
|
||||
"is_path_within": is_path_within,
|
||||
"get_model_files": MagicMock(
|
||||
return_value=[{"filename": "model.safetensors"}]
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user