diff --git a/app/api/routers/repositories.py b/app/api/routers/repositories.py index 0fd68d4..623d054 100644 --- a/app/api/routers/repositories.py +++ b/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 = [] diff --git a/app/api/routers/settings.py b/app/api/routers/settings.py index 0fdedfa..7239e20 100644 --- a/app/api/routers/settings.py +++ b/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() diff --git a/app/services/file_reader.py b/app/services/file_reader.py index 3c6a0cb..bb51324 100644 --- a/app/services/file_reader.py +++ b/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): diff --git a/app/services/git_manager.py b/app/services/git_manager.py index 4deb31f..68deef0 100644 --- a/app/services/git_manager.py +++ b/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."} diff --git a/app/services/summarizer.py b/app/services/summarizer.py index 4b08372..0e24135 100644 --- a/app/services/summarizer.py +++ b/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() diff --git a/app/services/vector_store/chroma_store.py b/app/services/vector_store/chroma_store.py index 886b065..49b5197 100644 --- a/app/services/vector_store/chroma_store.py +++ b/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.""" diff --git a/app/services/vector_store/manager.py b/app/services/vector_store/manager.py index a4e3627..94f7abc 100644 --- a/app/services/vector_store/manager.py +++ b/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: diff --git a/app/services/vector_store/pgvector_store.py b/app/services/vector_store/pgvector_store.py index 928d7aa..0236c1b 100644 --- a/app/services/vector_store/pgvector_store.py +++ b/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.""" diff --git a/app/services/vector_store/qdrant_store.py b/app/services/vector_store/qdrant_store.py index f6bb8fd..9e0ae27 100644 --- a/app/services/vector_store/qdrant_store.py +++ b/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.""" diff --git a/tests/backend/test_git_manager.py b/tests/backend/test_git_manager.py index 9f20811..7b694df 100644 --- a/tests/backend/test_git_manager.py +++ b/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__": diff --git a/tests/test_pgvector_store.py b/tests/test_pgvector_store.py index 58e4b4a..0ef41ff 100644 --- a/tests/test_pgvector_store.py +++ b/tests/test_pgvector_store.py @@ -101,7 +101,7 @@ def test_health_check_failure(self): store = PgVectorStore(engine=mock_engine, auto_init=False) healthy, msg = store.health_check() assert healthy is False - assert "DB Connection Refused" in msg + assert "health check failed" in msg.lower() def test_get_stats_success(self): from app.services.vector_store.pgvector_store import PgVectorStore