From 6f35df33d6d126444dec52b948e82055276ee5a0 Mon Sep 17 00:00:00 2001
From: Dariusz L <dl122@wp.pl>
Date: Thu, 6 Aug 2026 16:58:06 +0200
Subject: [PATCH] refactor(core): reuse shared path containment check

---
 core/routes/registry.py            |  1 +
 core/services/directory_service.py | 15 +++------------
 tests/test_directory_routes.py     |  2 ++
 tests/test_path_security.py        | 12 ++++++++++++
 tests/test_scanner_routes.py       |  2 ++
 5 files changed, 20 insertions(+), 12 deletions(-)

diff --git a/core/routes/registry.py b/core/routes/registry.py
index 4bf289d..1da67fa 100644
--- a/core/routes/registry.py
+++ b/core/routes/registry.py
@@ -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,
diff --git a/core/services/directory_service.py b/core/services/directory_service.py
index 43726a7..2f444a6 100644
--- a/core/services/directory_service.py
+++ b/core/services/directory_service.py
@@ -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:
diff --git a/tests/test_directory_routes.py b/tests/test_directory_routes.py
index 30346bd..0ef65fe 100644
--- a/tests/test_directory_routes.py
+++ b/tests/test_directory_routes.py
@@ -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(),
diff --git a/tests/test_path_security.py b/tests/test_path_security.py
index 2dfef8f..1bb4fe0 100644
--- a/tests/test_path_security.py
+++ b/tests/test_path_security.py
@@ -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")
diff --git a/tests/test_scanner_routes.py b/tests/test_scanner_routes.py
index fe3afa0..b290864 100644
--- a/tests/test_scanner_routes.py
+++ b/tests/test_scanner_routes.py
@@ -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"}]
         ),