Fix DLS writes in ingestion paths

Author: phact

src/models/processors.py

@@ -312,13 +312,15 @@ async def process_document_standard(
 
         dimensions = len(embeddings[0])
 
-        # Mapping updates are index-admin operations and fail under document-level
-        # security when attempted with a user-scoped OpenSearch client.
-        mapping_client = clients.opensearch or opensearch_client
+        # Mapping and write operations are index-admin operations and fail under
+        # document-level security when attempted with a user-scoped OpenSearch
+        # client. Reads still use the scoped client above so duplicate checks
+        # respect the caller's visibility.
+        write_client = clients.opensearch or opensearch_client
 
         # Ensure the embedding field exists for this model
         embedding_field_name = await ensure_embedding_field_exists(
-            mapping_client, embedding_model, get_index_name(), dimensions
+            write_client, embedding_model, get_index_name(), dimensions
         )
 
         # Index each chunk as a separate document
@@ -364,7 +366,7 @@ async def process_document_standard(
                 chunk_doc["is_sample_data"] = "true"
             chunk_id = f"{file_hash}_{i}"
             try:
-                await opensearch_client.index(index=get_index_name(), id=chunk_id, body=chunk_doc)
+                await write_client.index(index=get_index_name(), id=chunk_id, body=chunk_doc)
             except Exception as e:
                 logger.error(
                     "OpenSearch indexing failed for chunk",

src/services/knowledge_filter_service.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Optional
+from typing import Any
 
 KNOWLEDGE_FILTERS_INDEX_NAME = "knowledge_filters"
 
@@ -7,15 +7,23 @@ class KnowledgeFilterService:
     def __init__(self, session_manager=None):
         self.session_manager = session_manager
 
+    def _user_client(self, user_id: str = None, jwt_token: str = None):
+        return self.session_manager.get_user_opensearch_client(user_id, jwt_token)
+
+    def _write_client(self, user_id: str = None, jwt_token: str = None):
+        # OpenSearch rejects write requests on indices protected by filter-level
+        # DLS. The app enforces ownership/visibility with the scoped client, then
+        # performs trusted writes with the admin client.
+        from config.settings import clients
+
+        return clients.opensearch or self._user_client(user_id, jwt_token)
+
     async def create_knowledge_filter(
-        self, filter_doc: Dict[str, Any], user_id: str = None, jwt_token: str = None
-    ) -> Dict[str, Any]:
+        self, filter_doc: dict[str, Any], user_id: str = None, jwt_token: str = None
+    ) -> dict[str, Any]:
         """Create a new knowledge filter"""
         try:
-            # Get user's OpenSearch client with JWT for OIDC auth
-            opensearch_client = self.session_manager.get_user_opensearch_client(
-                user_id, jwt_token
-            )
+            opensearch_client = self._write_client(user_id, jwt_token)
 
             # Index the knowledge filter document
             result = await opensearch_client.index(
@@ -40,13 +48,10 @@ async def create_knowledge_filter(
 
     async def search_knowledge_filters(
         self, query: str, user_id: str = None, jwt_token: str = None, limit: int = 20
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Search for knowledge filters by name, description, or query content"""
         try:
-            # Get user's OpenSearch client with JWT for OIDC auth
-            opensearch_client = self.session_manager.get_user_opensearch_client(
-                user_id, jwt_token
-            )
+            opensearch_client = self._user_client(user_id, jwt_token)
 
             if query.strip():
                 # Search across name, description, and query_data fields
@@ -109,17 +114,12 @@ async def search_knowledge_filters(
 
     async def get_knowledge_filter(
         self, filter_id: str, user_id: str = None, jwt_token: str = None
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Get a specific knowledge filter by ID"""
         try:
-            # Get user's OpenSearch client with JWT for OIDC auth
-            opensearch_client = self.session_manager.get_user_opensearch_client(
-                user_id, jwt_token
-            )
+            opensearch_client = self._user_client(user_id, jwt_token)
 
-            result = await opensearch_client.get(
-                index=KNOWLEDGE_FILTERS_INDEX_NAME, id=filter_id
-            )
+            result = await opensearch_client.get(index=KNOWLEDGE_FILTERS_INDEX_NAME, id=filter_id)
 
             if result.get("found"):
                 knowledge_filter = result["_source"]
@@ -133,16 +133,17 @@ async def get_knowledge_filter(
     async def update_knowledge_filter(
         self,
         filter_id: str,
-        updates: Dict[str, Any],
+        updates: dict[str, Any],
         user_id: str = None,
         jwt_token: str = None,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Update an existing knowledge filter"""
         try:
-            # Get user's OpenSearch client with JWT for OIDC auth
-            opensearch_client = self.session_manager.get_user_opensearch_client(
-                user_id, jwt_token
-            )
+            existing = await self.get_knowledge_filter(filter_id, user_id, jwt_token)
+            if not existing.get("success"):
+                return existing
+
+            opensearch_client = self._write_client(user_id, jwt_token)
 
             # Update the document
             result = await opensearch_client.update(
@@ -159,10 +160,10 @@ async def update_knowledge_filter(
                     await opensearch_client.indices.refresh(index=KNOWLEDGE_FILTERS_INDEX_NAME)
                 except Exception:
                     pass
-                updated_doc = await opensearch_client.get(
-                    index=KNOWLEDGE_FILTERS_INDEX_NAME, id=filter_id
-                )
-                return {"success": True, "filter": updated_doc["_source"]}
+                updated_doc = await self.get_knowledge_filter(filter_id, user_id, jwt_token)
+                if updated_doc.get("success"):
+                    return updated_doc
+                return {"success": False, "error": "Failed to read updated knowledge filter"}
             else:
                 return {"success": False, "error": "Failed to update knowledge filter"}
 
@@ -171,13 +172,17 @@ async def update_knowledge_filter(
 
     async def delete_knowledge_filter(
         self, filter_id: str, user_id: str = None, jwt_token: str = None
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Delete a knowledge filter"""
         try:
-            # Get user's OpenSearch client with JWT for OIDC auth
-            opensearch_client = self.session_manager.get_user_opensearch_client(
-                user_id, jwt_token
-            )
+            existing = await self.get_knowledge_filter(filter_id, user_id, jwt_token)
+            if not existing.get("success"):
+                return {
+                    "success": False,
+                    "error": "Knowledge filter not found or access denied",
+                }
+
+            opensearch_client = self._write_client(user_id, jwt_token)
 
             result = await opensearch_client.delete(
                 index=KNOWLEDGE_FILTERS_INDEX_NAME,
@@ -219,20 +224,14 @@ async def delete_knowledge_filter(
     async def add_subscription(
         self,
         filter_id: str,
-        subscription_data: Dict[str, Any],
+        subscription_data: dict[str, Any],
         user_id: str = None,
         jwt_token: str = None,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Add a subscription to a knowledge filter"""
         try:
-            opensearch_client = self.session_manager.get_user_opensearch_client(
-                user_id, jwt_token
-            )
-
             # Get the current filter document
-            filter_result = await self.get_knowledge_filter(
-                filter_id, user_id, jwt_token
-            )
+            filter_result = await self.get_knowledge_filter(filter_id, user_id, jwt_token)
             if not filter_result.get("success"):
                 return filter_result
 
@@ -246,12 +245,11 @@ async def add_subscription(
             update_body = {
                 "doc": {
                     "subscriptions": subscriptions,
-                    "updated_at": subscription_data[
-                        "created_at"
-                    ],  # Use the same timestamp
+                    "updated_at": subscription_data["created_at"],  # Use the same timestamp
                 }
             }
 
+            opensearch_client = self._write_client(user_id, jwt_token)
             result = await opensearch_client.update(
                 index=KNOWLEDGE_FILTERS_INDEX_NAME,
                 id=filter_id,
@@ -273,17 +271,11 @@ async def remove_subscription(
         subscription_id: str,
         user_id: str = None,
         jwt_token: str = None,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Remove a subscription from a knowledge filter"""
         try:
-            opensearch_client = self.session_manager.get_user_opensearch_client(
-                user_id, jwt_token
-            )
-
             # Get the current filter document
-            filter_result = await self.get_knowledge_filter(
-                filter_id, user_id, jwt_token
-            )
+            filter_result = await self.get_knowledge_filter(filter_id, user_id, jwt_token)
             if not filter_result.get("success"):
                 return filter_result
 
@@ -292,9 +284,7 @@ async def remove_subscription(
             # Remove subscription from the subscriptions array
             subscriptions = filter_doc.get("subscriptions", [])
             updated_subscriptions = [
-                sub
-                for sub in subscriptions
-                if sub.get("subscription_id") != subscription_id
+                sub for sub in subscriptions if sub.get("subscription_id") != subscription_id
             ]
 
             if len(updated_subscriptions) == len(subscriptions):
@@ -310,6 +300,7 @@ async def remove_subscription(
                 }
             }
 
+            opensearch_client = self._write_client(user_id, jwt_token)
             result = await opensearch_client.update(
                 index=KNOWLEDGE_FILTERS_INDEX_NAME, id=filter_id, body=update_body
             )
@@ -324,12 +315,10 @@ async def remove_subscription(
 
     async def get_filter_subscriptions(
         self, filter_id: str, user_id: str = None, jwt_token: str = None
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """Get all subscriptions for a knowledge filter"""
         try:
-            filter_result = await self.get_knowledge_filter(
-                filter_id, user_id, jwt_token
-            )
+            filter_result = await self.get_knowledge_filter(filter_id, user_id, jwt_token)
             if not filter_result.get("success"):
                 return filter_result
 

tests/integration/core/test_runtime_migration.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import json
+import sys
 from pathlib import Path
 
 import httpx
@@ -12,6 +13,31 @@
 import yaml
 from sqlalchemy import func, select
 
+_RELOAD_MODULES = [
+    "api",
+    "api.connector_router",
+    "api.router",
+    "app.container",
+    "app.factory",
+    "app.lifespan",
+    "app.routes",
+    "app.routes.internal",
+    "auth_middleware",
+    "config.settings",
+    "db.engine",
+    "db.migrations_runtime",
+    "dependencies",
+    "main",
+    "services",
+    "services.conversation_persistence_service",
+    "services.default_docs_service",
+    "services.rbac_service",
+    "services.search_service",
+    "services.session_ownership_service",
+    "services.startup_orchestrator",
+    "utils.opensearch_init",
+]
+
 
 @pytest_asyncio.fixture
 async def legacy_migration_workspace(tmp_path: Path, monkeypatch):
@@ -35,6 +61,12 @@ async def legacy_migration_workspace(tmp_path: Path, monkeypatch):
     monkeypatch.setenv("DISABLE_STARTUP_INGEST", "true")
     monkeypatch.setenv("FETCH_OPENRAG_DOCS_AT_STARTUP", "false")
 
+    from db.engine import dispose_engine as dispose_existing_engine
+
+    await dispose_existing_engine()
+    for mod in _RELOAD_MODULES:
+        sys.modules.pop(mod, None)
+
     from config.config_manager import config_manager
     from db.engine import dispose_engine
     from dependencies import invalidate_user_ensured_cache
@@ -80,6 +112,8 @@ async def legacy_migration_workspace(tmp_path: Path, monkeypatch):
         session_ownership_service.ownership_file = old_ownership_file
         session_ownership_service.ownership_data = old_ownership_data
         session_ownership_service._session_factory = old_ownership_session_factory
+        for mod in _RELOAD_MODULES:
+            sys.modules.pop(mod, None)
 
 
 def _write_legacy_files(*, config_dir: Path, data_dir: Path) -> None:
@@ -210,12 +244,8 @@ async def _db_snapshot() -> dict[str, int]:
     assert SessionLocal is not None
     async with SessionLocal() as session:
         users = await session.scalar(select(func.count()).select_from(User))
-        conversations = await session.scalar(
-            select(func.count()).select_from(Conversation)
-        )
-        ownership = await session.scalar(
-            select(func.count()).select_from(SessionOwnership)
-        )
+        conversations = await session.scalar(select(func.count()).select_from(Conversation))
+        ownership = await session.scalar(select(func.count()).select_from(SessionOwnership))
         statuses = await session.scalar(select(func.count()).select_from(MigrationStatus))
     return {
         "users": int(users or 0),
@@ -287,18 +317,15 @@ async def test_legacy_file_state_migrates_on_backend_startup_and_is_idempotent()
         first_snapshot = await _db_snapshot()
 
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(
-            transport=transport, base_url="http://testserver"
-        ) as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
             onboarding = await client.get("/onboarding-status")
             assert onboarding.status_code == 200, onboarding.text
             assert onboarding.json() == {"onboarded": True, "current_step": 4}
 
             history = await client.get("/chat/history")
             assert history.status_code == 200, history.text
             conversations = {
-                item["response_id"]: item
-                for item in history.json().get("conversations", [])
+                item["response_id"]: item for item in history.json().get("conversations", [])
             }
             assert conversations["legacy-session"]["title"] == "Migrated chat"
             assert conversations["legacy-session"]["total_messages"] == 3
@@ -310,9 +337,7 @@ async def test_legacy_file_state_migrates_on_backend_startup_and_is_idempotent()
     try:
         assert await _db_snapshot() == first_snapshot
         transport = httpx.ASGITransport(app=app)
-        async with httpx.AsyncClient(
-            transport=transport, base_url="http://testserver"
-        ) as client:
+        async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
             onboarding = await client.get("/onboarding-status")
             assert onboarding.status_code == 200, onboarding.text
             assert onboarding.json()["onboarded"] is True