fix(security): resolve vector store and summarizer path injection sinks

app/api/routers/settings.py

@@ -3,6 +3,7 @@
 import json
 import sqlite3
 import logging
+import tempfile
 from typing import Optional
 from urllib.parse import urlsplit
 from fastapi import APIRouter, Request
@@ -277,24 +278,42 @@
         logger.error(f"Error reading vector store config: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to read vector store config."})
 
+def _validate_vector_storage_path(storage_path: Optional[str]) -> Optional[str]:
+    """Validates vector store storage path to ensure it resides within authorized data directories."""
+    if not storage_path:
+        return None
+    sp = storage_path.strip()
+    if sp == ":memory:":
+        return sp
+    if "\x00" in sp or any(part == ".." for part in sp.replace("\\", "/").split("/")):
+        raise ValueError("Invalid storage path: traversal detected.")
+
+    data_dir = os.path.abspath(os.getenv("DATA_DIR", "/app/data"))
+    repo_data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "data"))
+    temp_dir = os.path.abspath(tempfile.gettempdir())
+    allowed_roots = [data_dir, repo_data_dir, temp_dir, "/tmp"]
+
+    norm_sp = os.path.normpath(os.path.abspath(sp)) if os.path.isabs(sp) else os.path.normpath(os.path.abspath(os.path.join(data_dir, sp)))
+    for safe_root in allowed_roots:
+        if norm_sp.startswith(safe_root):
+            return norm_sp
+    raise ValueError(f"Storage path '{sp}' is outside authorized directories.")
+
+
 @router.post("/admin/api/vector-store/test")
 async def api_test_vector_store(payload: VectorStoreTestRequest):
     try:
+        safe_sp = payload.storage_path
         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."})
+            try:
+                safe_sp = _validate_vector_storage_path(payload.storage_path)
+            except ValueError as ve:
+                return JSONResponse(status_code=400, content={"success": False, "error": str(ve), "message": str(ve)})
 
         success, message = vs_service.test_vector_store_connection(
             provider=payload.provider,
             mode=payload.mode,
-            storage_path=payload.storage_path,
+            storage_path=safe_sp,
             url=payload.url,
             collection=payload.collection
         )
@@ -311,24 +330,20 @@
 @router.post("/admin/api/vector-store/switch")
 async def api_switch_vector_store(payload: VectorStoreSwitchRequest):
     try:
+        safe_sp = payload.storage_path
         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."})
+            try:
+                safe_sp = _validate_vector_storage_path(payload.storage_path)
+            except ValueError as ve:
+                return JSONResponse(status_code=400, content={"status": "error", "error": str(ve), "message": str(ve)})
 
         def _reindex():
             threading.Thread(target=idx_service.run_full_indexing, daemon=True).start()
 
         success, message = vs_service.switch_vector_store(
             provider=payload.provider,
             mode=payload.mode,
-            storage_path=payload.storage_path,
+            storage_path=safe_sp,
             url=payload.url,
             collection=payload.collection,
             reindex_callback=_reindex

app/services/summarizer.py

@@ -236,22 +236,21 @@ def get_or_create_summary(
         try:
             from app.services.file_reader import get_file_reader_service
             reader = get_file_reader_service()
-            read_res = reader.read_file(filepath, repo=resolved_repo)
+            read_res = reader.read_file(filepath, repo=repo)
             if isinstance(read_res, dict) and "content" in read_res:
                 content = read_res["content"]
         except Exception:
             content = None
 
-        if content is None:
-            # Fallback to direct read if safe file exists on disk
+        if content is None and repo:
             try:
-                norm_fp = os.path.normpath(os.path.abspath(filepath))
-                if os.path.exists(norm_fp) and os.path.isfile(norm_fp):
-                    with open(norm_fp, "r", encoding="utf-8", errors="replace") as f:
-                        content = f.read()
+                from app.services.file_reader import get_file_reader_service
+                reader = get_file_reader_service()
+                read_res = reader.read_file(filepath, repo=None)
+                if isinstance(read_res, dict) and "content" in read_res:
+                    content = read_res["content"]
             except Exception:
-                pass
-
+                content = None
         if content is None:
             # Fallback to local_storage
             try:

app/services/vector_store/chroma_store.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import uuid
+import tempfile
 from typing import List, Dict, Any, Optional, Tuple, Union
 from urllib.parse import urlparse
 import chromadb
@@ -48,6 +49,17 @@ def get_default_chroma_storage_path() -> str:
     return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "chroma_storage")
 
 
+def validate_chroma_storage_path(target_storage: str) -> str:
+    clean = os.path.normpath(os.path.abspath(target_storage))
+    data_dir = os.path.abspath(os.getenv("DATA_DIR", "/app/data"))
+    repo_data_dir = os.path.abspath(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data"))
+    allowed_roots = [data_dir, repo_data_dir, os.path.abspath(tempfile.gettempdir()), "/tmp"]
+    for root in allowed_roots:
+        if clean.startswith(root):
+            return clean
+    raise ValueError(f"Invalid Chroma storage path outside authorized directories: {target_storage}")
+
+
 class ChromaVectorStore(VectorStore):
     """ChromaDB vector store backend supporting persistent disk, in-memory, and remote HTTP modes with auto-fallback."""
 
@@ -115,11 +127,7 @@ def __init__(
                     self.mode = "memory"
                     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}")
+                    clean_storage = validate_chroma_storage_path(target_storage)
                     os.makedirs(clean_storage, exist_ok=True)
                     self.client = chromadb.PersistentClient(path=clean_storage)
                     self.mode = "persistent"
@@ -130,11 +138,7 @@ def __init__(
                 self.mode = "memory"
                 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}")
+                clean_storage = validate_chroma_storage_path(target_storage)
                 os.makedirs(clean_storage, exist_ok=True)
                 self.client = chromadb.PersistentClient(path=clean_storage)
                 self.mode = "persistent"

app/services/vector_store/qdrant_store.py

@@ -1,6 +1,7 @@
 import os
 import uuid
 import logging
+import tempfile
 from typing import List, Dict, Any, Optional, Tuple, Union
 from qdrant_client import QdrantClient
 from qdrant_client.http import models as qmodels
@@ -20,6 +21,17 @@ def get_default_qdrant_storage_path() -> str:
     return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "qdrant_storage")
 
 
+def validate_qdrant_storage_path(target_storage: str) -> str:
+    clean = os.path.normpath(os.path.abspath(target_storage))
+    data_dir = os.path.abspath(os.getenv("DATA_DIR", "/app/data"))
+    repo_data_dir = os.path.abspath(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data"))
+    allowed_roots = [data_dir, repo_data_dir, os.path.abspath(tempfile.gettempdir()), "/tmp"]
+    for root in allowed_roots:
+        if clean.startswith(root):
+            return clean
+    raise ValueError(f"Invalid Qdrant storage path outside authorized directories: {target_storage}")
+
+
 class QdrantVectorStore(VectorStore):
     """Qdrant vector store backend supporting embedded disk, in-memory, and remote server modes with automatic fallback."""
 
@@ -66,11 +78,7 @@ def __init__(
                     self.mode = "memory"
                     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}")
+                    clean_storage = validate_qdrant_storage_path(target_storage)
                     os.makedirs(clean_storage, exist_ok=True)
                     self.client = QdrantClient(path=clean_storage)
                     self.mode = "embedded"
@@ -82,11 +90,7 @@ def __init__(
                 self.mode = "memory"
                 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}")
+                clean_storage = validate_qdrant_storage_path(target_storage)
                 os.makedirs(clean_storage, exist_ok=True)
                 self.client = QdrantClient(path=clean_storage)
                 self.mode = "embedded"

tests/test_summarizer.py

@@ -21,6 +21,12 @@ def test_db(tmp_path, monkeypatch):
     monkeypatch.setattr("app.services.database.CACHE_DB_PATH", str(db_file), raising=False)
     engine = get_db_engine(db_url, reset=True)
     init_db(engine=engine)
+    with get_db_connection() as conn:
+        conn.execute(
+            "INSERT INTO indexed_paths (path, type, enabled, repo, category) VALUES (?, 'directory', 1, 'test-repo', 'code')",
+            (str(tmp_path),)
+        )
+        conn.commit()
     return engine