Security Hardening: Fixed path traversal, BM25 RCE, IDOR, and removed backdoor

RAG_FULL_APPLICATION_BACKEND/app/main.py

@@ -7,18 +7,24 @@ from .utils.ws_manager import ws_manager
 import logging
 import os
 
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.util import get_remote_address
+from slowapi.errors import RateLimitExceeded
+from .utils.auth_utils import decode_token
+
 # Setup Logger
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+limiter = Limiter(key_func=get_remote_address, default_limits=[f"{settings.RATE_LIMIT_PER_MINUTE}/minute"])
 app = FastAPI(title="RAG Pipeline API", version="3.0.0")
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 
 # CORS
-# origins = settings.CORS_ORIGINS.split(",")
-origins = settings.CORS_ORIGINS.split(",") if settings.CORS_ORIGINS else ["*"]
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=origins,
+    allow_origins=["http://localhost:5174", "http://127.0.0.1:5174"],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
@@ -40,9 +46,14 @@ async def health_check():
 
 @app.websocket("/ws/pipeline/{job_id}")
 async def pipeline_ws(websocket: WebSocket, job_id: str, token: str):
-    # JWT verification logic will go here
-    # For now, just connect
-    await ws_manager.connect(job_id, websocket, "anonymous")
+    # JWT verification
+    payload = decode_token(token)
+    if not payload:
+        await websocket.close(code=1008, reason="Invalid token")
+        return
+        
+    user_id = payload.get("id", "anonymous")
+    await ws_manager.connect(job_id, websocket, user_id)
     try:
         while True:
             data = await websocket.receive_text()
@@ -50,7 +61,7 @@ async def pipeline_ws(websocket: WebSocket, job_id: str, token: str):
     except Exception as e:
         logger.error(f"WebSocket error for job {job_id}: {e}")
     finally:
-        await ws_manager.disconnect(job_id, "anonymous")
+        await ws_manager.disconnect(job_id, user_id)
 
 # Serve frontend static files in production monolith
 static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "static")

RAG_FULL_APPLICATION_BACKEND/app/routers/auth.py

@@ -44,18 +44,3 @@ async def login(form_data: OAuth2PasswordRequestForm = Depends()):
     access_token = create_access_token(data={"sub": user["username"], "id": user["id"]})
     return {"access_token": access_token, "token_type": "bearer"}
 
-@router.post("/seed_admin")
-async def seed_admin():
-    import traceback
-    try:
-        hashed = get_password_hash("admin123")
-        supabase_service.client.table("users").delete().eq("username", "admin").execute()
-        supabase_service.client.table("users").insert({
-            "username": "admin",
-            "password_hash": hashed
-        }).execute()
-        logger.info("Admin user seeded successfully.")
-        return {"msg": "Admin user created/reset (admin / admin123)"}
-    except Exception as e:
-        logger.error(f"Seeding failed: {e}")
-        return {"error": str(e), "traceback": traceback.format_exc()}

RAG_FULL_APPLICATION_BACKEND/app/routers/ingest.py

@@ -33,10 +33,11 @@ async def upload_file(
     strategy: str = Form("fixed"),
     user: dict = Depends(get_current_user)
 ):
+    safe_filename = os.path.basename(file.filename)
     job_id = str(uuid.uuid4())
     temp_dir = Path("./data/uploads") / user["id"]
     temp_dir.mkdir(parents=True, exist_ok=True)
-    file_path = temp_dir / file.filename
+    file_path = temp_dir / safe_filename
     
     with open(file_path, "wb") as f:
         f.write(await file.read())
@@ -45,7 +46,7 @@ async def upload_file(
     background_tasks.add_task(
         process_ingestion, 
         str(file_path), 
-        file.filename, 
+        safe_filename, 
         chunk_size, 
         overlap, 
         strategy, 

RAG_FULL_APPLICATION_BACKEND/app/services/bm25_service.py

@@ -1,6 +1,3 @@
-import pickle
-import os
-from pathlib import Path
 from typing import List, Dict, Any
 from rank_bm25 import BM25Okapi
 import logging
@@ -8,60 +5,48 @@ import logging
 logger = logging.getLogger(__name__)
 
 class BM25Service:
-    def __init__(self, data_dir: str = "./data/bm25_indexes"):
-        self.data_dir = Path(data_dir)
-        self.data_dir.mkdir(parents=True, exist_ok=True)
-
-    def _get_index_path(self, document_id: str) -> Path:
-        return self.data_dir / f"{document_id}.pkl"
+    def __init__(self):
+        # We no longer store pickle files to prevent RCE and memory bottlenecks.
+        pass
 
     def index_chunks(self, document_id: str, chunks: List[Dict[str, Any]]):
-        """Build and save BM25 index for a document."""
-        texts = [c["text"] for c in chunks]
-        tokenized_corpus = [text.lower().split() for text in texts]
-        bm25 = BM25Okapi(tokenized_corpus)
-        
-        # Save both the bm25 object and the chunk mapping
-        with open(self._get_index_path(document_id), "wb") as f:
-            pickle.dump({"bm25": bm25, "chunks": chunks}, f)
+        """Mock method for API compatibility. Chunks are now indexed dynamically on search."""
+        pass
+
+    def delete_document(self, document_id: str):
+        """Mock method for API compatibility. No files to delete."""
+        pass
 
     def search(self, document_id: str, query: str, top_n: int = 10) -> List[Dict[str, Any]]:
-        """Search using BM25."""
-        path = self._get_index_path(document_id)
-        if not path.exists():
-            logger.warning(f"BM25 index not found for {document_id}. Attempting to rebuild from Supabase...")
-            try:
-                from .supabase_client import supabase_service
-                result = supabase_service.client.table("chunks").select("*").eq("document_id", document_id).execute()
-                chunks = result.data
-                if chunks:
-                    logger.info(f"Rebuilding BM25 index for {document_id} with {len(chunks)} chunks.")
-                    self.index_chunks(document_id, chunks)
-                else:
-                    logger.warning(f"No chunks found in Supabase for {document_id}. Cannot rebuild index.")
-                    return []
-            except Exception as e:
-                logger.error(f"Failed to rebuild BM25 index: {e}")
+        """Search using BM25 by dynamically fetching chunks from DB."""
+        try:
+            from .supabase_client import supabase_service
+            result = supabase_service.client.table("chunks").select("*").eq("document_id", document_id).execute()
+            chunks = result.data
+            if not chunks:
                 return []
-
-        with open(path, "rb") as f:
-            data = pickle.load(f)
-            bm25 = data["bm25"]
-            chunks = data["chunks"]
-
-        tokenized_query = query.lower().split()
-        scores = bm25.get_scores(tokenized_query)
-        
-        # Add score to chunks
-        results = []
-        for i, score in enumerate(scores):
-            if score > 0:
-                chunk = chunks[i].copy()
-                chunk["bm25_score"] = float(score)
-                results.append(chunk)
-        
-        # Sort by score
-        results.sort(key=lambda x: x["bm25_score"], reverse=True)
-        return results[:top_n]
+                
+            texts = [c["text"] for c in chunks]
+            tokenized_corpus = [text.lower().split() for text in texts]
+            bm25 = BM25Okapi(tokenized_corpus)
+            
+            tokenized_query = query.lower().split()
+            scores = bm25.get_scores(tokenized_query)
+            
+            # Add score to chunks
+            results = []
+            for i, score in enumerate(scores):
+                if score > 0:
+                    chunk = chunks[i].copy()
+                    chunk["bm25_score"] = float(score)
+                    results.append(chunk)
+            
+            # Sort by score
+            results.sort(key=lambda x: x["bm25_score"], reverse=True)
+            return results[:top_n]
+            
+        except Exception as e:
+            logger.error(f"BM25 Search failed for {document_id}: {e}")
+            return []
 
 bm25_service = BM25Service()

RAG_FULL_APPLICATION_BACKEND/app/services/supabase_client.py

@@ -109,7 +109,7 @@ class SupabaseService:
         """Delete document + chunks + vectors (CASCADE)"""
         try:
             # 1. Chunks (will cascade to vectors)
-            self.client.table("chunks").delete().eq("document_id", document_id).execute()
+            self.client.table("chunks").delete().eq("document_id", document_id).eq("user_id", user_id).execute()
             # 2. Document
             self.client.table("documents").delete().eq("id", document_id).eq("user_id", user_id).execute()
         except Exception as e: