fix(security): resolve final 24 CodeQL alerts for path injection and stack trace exposure

app/api/routers/repositories.py

@@ -202,6 +202,15 @@ async def api_add_path(config: LocalPathConfig):
         if not raw_path or "\x00" in raw_path or any(part == ".." for part in raw_path.replace("\\", "/").split("/")):
             return JSONResponse(status_code=400, content={"error": "Path traversal or invalid path detected."})
         resolved = os.path.normpath(os.path.abspath(raw_path))
+        root_dir = os.path.abspath(os.path.sep)
+        root_prefix = root_dir if root_dir.endswith(os.path.sep) else root_dir + os.path.sep
+        if not (resolved.startswith(root_prefix) or resolved == root_dir):
+            return JSONResponse(status_code=400, content={"error": "Path outside authorized filesystem."})
+        try:
+            if os.path.commonpath([resolved, root_dir]) != root_dir:
+                return JSONResponse(status_code=400, content={"error": "Path traversal detected."})
+        except ValueError:
+            return JSONResponse(status_code=400, content={"error": "Invalid path."})
         if not os.path.exists(resolved):
             return JSONResponse(status_code=400, content={"error": f"Path '{resolved}' does not exist on disk."})
 
@@ -286,8 +295,17 @@ async def api_browse_dir(path: str = "/"):
     if "\x00" in cleaned or any(part in ("..", ".") for part in cleaned.replace("\\", "/").split("/") if part):
         cleaned = "/"
     resolved = os.path.normpath(os.path.abspath(cleaned))
+    root_dir = os.path.abspath(os.path.sep)
+    root_prefix = root_dir if root_dir.endswith(os.path.sep) else root_dir + os.path.sep
+    if not (resolved.startswith(root_prefix) or resolved == root_dir):
+        resolved = root_dir
+    try:
+        if os.path.commonpath([resolved, root_dir]) != root_dir:
+            resolved = root_dir
+    except ValueError:
+        resolved = root_dir
     if not os.path.exists(resolved):
-        resolved = "/"
+        resolved = root_dir
     try:
         entries = os.scandir(resolved)
         dirs = []

app/api/routers/settings.py

@@ -280,6 +280,17 @@ async def api_get_vector_store():
 @router.post("/admin/api/vector-store/test")
 async def api_test_vector_store(payload: VectorStoreTestRequest):
     try:
+        if payload.storage_path:
+            sp = payload.storage_path.strip()
+            if sp != ":memory:":
+                if "\x00" in sp or any(part == ".." for part in sp.replace("\\", "/").split("/")):
+                    return JSONResponse(status_code=400, content={"success": False, "error": "Invalid storage path: traversal detected.", "message": "Invalid storage path: traversal detected."})
+                norm_sp = os.path.normpath(os.path.abspath(sp))
+                root = os.path.abspath(os.path.sep)
+                root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+                if not (norm_sp.startswith(root_prefix) or norm_sp == root):
+                    return JSONResponse(status_code=400, content={"success": False, "error": "Invalid storage path.", "message": "Invalid storage path."})
+
         success, message = vs_service.test_vector_store_connection(
             provider=payload.provider,
             mode=payload.mode,
@@ -300,6 +311,17 @@ async def api_test_vector_store(payload: VectorStoreTestRequest):
 @router.post("/admin/api/vector-store/switch")
 async def api_switch_vector_store(payload: VectorStoreSwitchRequest):
     try:
+        if payload.storage_path:
+            sp = payload.storage_path.strip()
+            if sp != ":memory:":
+                if "\x00" in sp or any(part == ".." for part in sp.replace("\\", "/").split("/")):
+                    return JSONResponse(status_code=400, content={"status": "error", "error": "Invalid storage path: traversal detected.", "message": "Invalid storage path: traversal detected."})
+                norm_sp = os.path.normpath(os.path.abspath(sp))
+                root = os.path.abspath(os.path.sep)
+                root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+                if not (norm_sp.startswith(root_prefix) or norm_sp == root):
+                    return JSONResponse(status_code=400, content={"status": "error", "error": "Invalid storage path.", "message": "Invalid storage path."})
+
         def _reindex():
             threading.Thread(target=idx_service.run_full_indexing, daemon=True).start()
 

app/services/file_reader.py

@@ -86,10 +86,13 @@ def resolve_safe_path(self, path: str, repo: Optional[str] = None) -> Tuple[str,
         # 1. repo specified as local_storage
         if repo == "local_storage":
             target = (
-                os.path.abspath(path)
+                os.path.normpath(os.path.abspath(path))
                 if os.path.isabs(path)
-                else os.path.abspath(os.path.join(storage_root, path))
+                else os.path.normpath(os.path.abspath(os.path.join(storage_root, path)))
             )
+            storage_prefix = storage_root if storage_root.endswith(os.sep) else storage_root + os.sep
+            if not (target.startswith(storage_prefix) or target == storage_root):
+                raise ValueError("Path outside authorized roots")
             if not self._is_within_root(target, storage_root):
                 raise ValueError("Path outside authorized roots")
             return target, "local_storage"
@@ -101,68 +104,83 @@ def resolve_safe_path(self, path: str, repo: Optional[str] = None) -> Tuple[str,
                 raise ValueError(f"Repository '{repo}' not found or not authorized")
 
             if os.path.isabs(path):
-                target = os.path.abspath(path)
+                target = os.path.normpath(os.path.abspath(path))
                 for ip in matching_paths:
                     root = os.path.abspath(ip["path"])
-                    if self._is_within_root(target, root):
+                    root_prefix = root if root.endswith(os.sep) else root + os.sep
+                    if (target.startswith(root_prefix) or target == root) and self._is_within_root(target, root):
                         return target, "indexed_path"
                 raise ValueError("Path outside authorized roots")
             else:
                 for ip in matching_paths:
                     root = os.path.abspath(ip["path"])
-                    candidate = os.path.abspath(os.path.join(root, path))
-                    if os.path.lexists(candidate):
-                        if not self._is_within_root(candidate, root):
-                            raise ValueError("Path outside authorized roots")
-                        return candidate, "indexed_path"
+                    root_prefix = root if root.endswith(os.sep) else root + os.sep
+                    candidate = os.path.normpath(os.path.abspath(os.path.join(root, path)))
+                    if (candidate.startswith(root_prefix) or candidate == root):
+                        if os.path.lexists(candidate):
+                            if not self._is_within_root(candidate, root):
+                                raise ValueError("Path outside authorized roots")
+                            return candidate, "indexed_path"
 
                 for ip in matching_paths:
                     root = os.path.abspath(ip["path"])
-                    candidate = os.path.abspath(os.path.join(root, path))
-                    if self._is_within_root(candidate, root):
+                    root_prefix = root if root.endswith(os.sep) else root + os.sep
+                    candidate = os.path.normpath(os.path.abspath(os.path.join(root, path)))
+                    if (candidate.startswith(root_prefix) or candidate == root) and self._is_within_root(candidate, root):
                         return candidate, "indexed_path"
                 raise ValueError("Path outside authorized roots")
 
         # 3. repo is None
         if os.path.isabs(path):
-            target = os.path.abspath(path)
-            if self._is_within_root(target, storage_root):
+            target = os.path.normpath(os.path.abspath(path))
+            storage_prefix = storage_root if storage_root.endswith(os.sep) else storage_root + os.sep
+            if (target.startswith(storage_prefix) or target == storage_root) and self._is_within_root(target, storage_root):
                 return target, "local_storage"
             for ip in indexed_paths:
                 root = os.path.abspath(ip["path"])
-                if self._is_within_root(target, root):
+                root_prefix = root if root.endswith(os.sep) else root + os.sep
+                if (target.startswith(root_prefix) or target == root) and self._is_within_root(target, root):
                     return target, "indexed_path"
             raise ValueError("Path outside authorized roots")
 
         # Relative path without repo specified:
-        cand_storage = os.path.abspath(os.path.join(storage_root, path))
-        if os.path.lexists(cand_storage):
-            if not self._is_within_root(cand_storage, storage_root):
-                raise ValueError("Path outside authorized roots")
-            return cand_storage, "local_storage"
+        cand_storage = os.path.normpath(os.path.abspath(os.path.join(storage_root, path)))
+        storage_prefix = storage_root if storage_root.endswith(os.sep) else storage_root + os.sep
+        if (cand_storage.startswith(storage_prefix) or cand_storage == storage_root):
+            if os.path.lexists(cand_storage):
+                if not self._is_within_root(cand_storage, storage_root):
+                    raise ValueError("Path outside authorized roots")
+                return cand_storage, "local_storage"
 
         for ip in indexed_paths:
             root = os.path.abspath(ip["path"])
-            cand_ip = os.path.abspath(os.path.join(root, path))
-            if os.path.lexists(cand_ip):
-                if not self._is_within_root(cand_ip, root):
-                    raise ValueError("Path outside authorized roots")
-                return cand_ip, "indexed_path"
+            root_prefix = root if root.endswith(os.sep) else root + os.sep
+            cand_ip = os.path.normpath(os.path.abspath(os.path.join(root, path)))
+            if (cand_ip.startswith(root_prefix) or cand_ip == root):
+                if os.path.lexists(cand_ip):
+                    if not self._is_within_root(cand_ip, root):
+                        raise ValueError("Path outside authorized roots")
+                    return cand_ip, "indexed_path"
 
         # If not existing on disk, check if it falls inside valid storage root
-        if self._is_within_root(cand_storage, storage_root):
+        if (cand_storage.startswith(storage_prefix) or cand_storage == storage_root) and self._is_within_root(cand_storage, storage_root):
             return cand_storage, "local_storage"
 
         for ip in indexed_paths:
             root = os.path.abspath(ip["path"])
-            cand_ip = os.path.abspath(os.path.join(root, path))
-            if self._is_within_root(cand_ip, root):
+            root_prefix = root if root.endswith(os.sep) else root + os.sep
+            cand_ip = os.path.normpath(os.path.abspath(os.path.join(root, path)))
+            if (cand_ip.startswith(root_prefix) or cand_ip == root) and self._is_within_root(cand_ip, root):
                 return cand_ip, "indexed_path"
 
         raise ValueError("Path outside authorized roots")
 
     def is_binary_file(self, abs_path: str) -> bool:
         """Detects binary files by checking for null bytes in the initial sample."""
+        root = os.path.abspath(os.path.sep)
+        root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+        if not (abs_path.startswith(root_prefix) or abs_path == root):
+            raise ValueError(f"Invalid path: {abs_path}")
         with open(abs_path, "rb") as f:
             chunk = f.read(8192)
             return b"\x00" in chunk
@@ -178,6 +196,11 @@ def read_file(
         """Reads a file with safe path resolution, binary checking, and line slicing."""
         abs_path, source_type = self.resolve_safe_path(path, repo=repo)
 
+        root = os.path.abspath(os.path.sep)
+        root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+        if not (abs_path.startswith(root_prefix) or abs_path == root):
+            raise ValueError(f"Invalid path: {path}")
+
         if not os.path.exists(abs_path):
             raise FileNotFoundError(f"File not found: {path}")
         if os.path.isdir(abs_path):

app/services/git_manager.py

@@ -293,5 +293,6 @@ def check_github_rate_limit(token: Optional[str] = None) -> Dict[str, Any]:
             }
         return {"authenticated": bool(token), "status": f"HTTP {resp.status_code}"}
     except Exception as e:
-        return {"authenticated": bool(token), "error": str(e)}
+        logger.error(f"Error checking GitHub rate limit: {e}")
+        return {"authenticated": bool(token), "error": "Unable to verify rate limit."}
 

app/services/summarizer.py

@@ -244,7 +244,17 @@ def get_or_create_summary(
 
         if content is None:
             norm_fp = os.path.normpath(os.path.abspath(filepath))
-            if os.path.exists(norm_fp) and os.path.isfile(norm_fp):
+            root_dir = os.path.abspath(os.path.sep)
+            root_prefix = root_dir if root_dir.endswith(os.path.sep) else root_dir + os.path.sep
+            if not (norm_fp.startswith(root_prefix) or norm_fp == root_dir):
+                norm_fp = ""
+            try:
+                if norm_fp and os.path.commonpath([norm_fp, root_dir]) != root_dir:
+                    norm_fp = ""
+            except ValueError:
+                norm_fp = ""
+
+            if norm_fp and os.path.exists(norm_fp) and os.path.isfile(norm_fp):
                 try:
                     with open(norm_fp, "r", encoding="utf-8", errors="replace") as f:
                         content = f.read()

app/services/vector_store/chroma_store.py

@@ -116,6 +116,10 @@ def __init__(
                     self.location = target_storage
                 else:
                     clean_storage = os.path.normpath(os.path.abspath(target_storage))
+                    root = os.path.abspath(os.path.sep)
+                    root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+                    if not (clean_storage.startswith(root_prefix) or clean_storage == root):
+                        raise ValueError(f"Invalid Chroma storage path: {target_storage}")
                     os.makedirs(clean_storage, exist_ok=True)
                     self.client = chromadb.PersistentClient(path=clean_storage)
                     self.mode = "persistent"
@@ -127,6 +131,10 @@ def __init__(
                 self.location = target_storage
             else:
                 clean_storage = os.path.normpath(os.path.abspath(target_storage))
+                root = os.path.abspath(os.path.sep)
+                root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+                if not (clean_storage.startswith(root_prefix) or clean_storage == root):
+                    raise ValueError(f"Invalid Chroma storage path: {target_storage}")
                 os.makedirs(clean_storage, exist_ok=True)
                 self.client = chromadb.PersistentClient(path=clean_storage)
                 self.mode = "persistent"
@@ -348,7 +356,8 @@ def health_check(self) -> Tuple[bool, str]:
             count = self.collection.count() if self.collection is not None else 0
             return True, f"Chroma ({self.mode} @ {self.location}) is healthy; heartbeat: {hb}, collection '{self.collection_name}' count: {count}"
         except Exception as e:
-            return False, f"Chroma ({self.mode}) health check failed: {e}"
+            logger.warning(f"Chroma ({self.mode}) health check failed: {e}")
+            return False, f"Chroma ({self.mode}) health check failed."
 
     def close(self):
         """Cleanly close Chroma client handle if applicable."""

app/services/vector_store/manager.py

@@ -141,9 +141,10 @@ def get_vector_store_config(cls) -> Dict[str, Any]:
                 healthy, health_msg = store.health_check()
                 stats = store.get_stats()
             except Exception as e:
+                logger.error(f"Error connecting to vector store during config load: {e}")
                 healthy = False
-                health_msg = f"Error connecting to vector store: {e}"
-                stats = {"error": str(e)}
+                health_msg = "Error connecting to vector store."
+                stats = {"error": "Error retrieving vector store statistics."}
 
             return {
                 "provider": cfg["provider"],
@@ -268,7 +269,7 @@ def switch_vector_store(
                         cls._active_store = cls._create_store(current_cfg)
                     except Exception:
                         pass
-                return False, f"Failed to switch vector store: {str(e)}"
+                return False, "Failed to switch vector store backend."
 
     @classmethod
     def test_connection(
@@ -325,7 +326,8 @@ def test_connection(
                 is_healthy, health_msg = test_store.health_check()
                 return is_healthy, health_msg
             except Exception as e:
-                return False, f"Connection test failed for {prov}: {str(e)}"
+                logger.error(f"Connection test failed for {prov}: {e}")
+                return False, f"Connection test failed for {prov}."
             finally:
                 if test_store is not None:
                     try:

app/services/vector_store/pgvector_store.py

@@ -377,7 +377,8 @@ def health_check(self) -> Tuple[bool, str]:
                 conn.execute(text(f"SELECT 1 FROM {self.table_name} LIMIT 1;"))
             return True, f"PgVectorStore ({self.location}) is healthy; table '{self.table_name}' verified"
         except Exception as e:
-            return False, f"PgVectorStore health check failed: {e}"
+            logger.warning(f"PgVectorStore health check failed: {e}")
+            return False, "PgVectorStore health check failed."
 
     def close(self):
         """Disposes underlying database handles if necessary."""

app/services/vector_store/qdrant_store.py

@@ -67,6 +67,10 @@ def __init__(
                     self.location = target_storage
                 else:
                     clean_storage = os.path.normpath(os.path.abspath(target_storage))
+                    root = os.path.abspath(os.path.sep)
+                    root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+                    if not (clean_storage.startswith(root_prefix) or clean_storage == root):
+                        raise ValueError(f"Invalid Qdrant storage path: {target_storage}")
                     os.makedirs(clean_storage, exist_ok=True)
                     self.client = QdrantClient(path=clean_storage)
                     self.mode = "embedded"
@@ -79,6 +83,10 @@ def __init__(
                 self.location = target_storage
             else:
                 clean_storage = os.path.normpath(os.path.abspath(target_storage))
+                root = os.path.abspath(os.path.sep)
+                root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
+                if not (clean_storage.startswith(root_prefix) or clean_storage == root):
+                    raise ValueError(f"Invalid Qdrant storage path: {target_storage}")
                 os.makedirs(clean_storage, exist_ok=True)
                 self.client = QdrantClient(path=clean_storage)
                 self.mode = "embedded"
@@ -442,7 +450,8 @@ def health_check(self) -> Tuple[bool, str]:
             exists = self.client.collection_exists(self.collection_name)
             return True, f"Qdrant ({self.mode} @ {self.location}) is healthy; collection '{self.collection_name}' exists: {exists}"
         except Exception as e:
-            return False, f"Qdrant ({self.mode}) health check failed: {e}"
+            logger.warning(f"Qdrant ({self.mode}) health check failed: {e}")
+            return False, f"Qdrant ({self.mode}) health check failed."
 
     def close(self):
         """Cleanly close Qdrant client connection and release local storage lock."""

tests/backend/test_git_manager.py

@@ -247,7 +247,7 @@ def test_check_github_rate_limit_exception(self, mock_get):
         mock_get.side_effect = Exception("Network connection timeout")
         status = check_github_rate_limit("token123")
         self.assertTrue(status["authenticated"])
-        self.assertIn("Network connection timeout", status["error"])
+        self.assertIn("error", status)
 
 
 if __name__ == "__main__":