From 8a1c9ba5f30abfb5af80c7aacdbf0d442b6433b2 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Tue, 14 Jul 2026 11:15:43 +0200 Subject: [PATCH] fix(security): scope every HybridRAG leg and ingestion path to the caller's tenant A live /query/hybrid probe as user=llm_tester returned jpmschweitzer pages. Audit of all legs (vector, graph, web-persistence, volatile, documents) plus enrichment/persistence found and fixed these unscoped paths: - vector_service.update_from_page and graph_service.update_from_page now refuse pages outside users/{user}/ - previously any tenant could ingest any wiki page (incl. another tenant's) into its own collection and graph labels, which is how foreign content entered the vector leg. - ingestion_service.ingest_all_pages clamps path_prefix to the caller's namespace (segment-exact, sanitized comparison) and defaults to users/{user}; /ingest/all returns 400 on cross-tenant prefixes. - hybrid_rag_service._persist_search_for_librarian linked SearchQuery nodes to unscoped (d:Document {page_id}); now matches only User_{Tenant}_Document nodes. - graph_service: _get_entity_mention_count, entity-stub mention/related queries, generate_entity_stubs, find/purge_orphan_entities matched unscoped Document nodes; cleanup_broken_relationships matched all tenants' SearchQuery nodes; _entity_has_wiki_page listed all wiki pages. All are now tenant-label / namespace scoped. - volatile_service collection names now use the sanitized user id. - is_path_in_user_namespace enforces a path-segment boundary (users/llm_tester2 is not llm_tester's namespace) and treats hyphen/underscore tenant spellings as the same sanitized tenant. - New offline unit tests per leg (mocked clients) assert the tenant-scoped collection/label/path is used and cross-tenant access is refused. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 8 + src/core/multi_tenancy.py | 14 +- src/routers/ingestion.py | 15 +- src/services/graph_service.py | 45 +++- src/services/hybrid_rag_service.py | 7 +- src/services/ingestion_service.py | 28 +- src/services/vector_service.py | 12 +- src/services/volatile_service.py | 11 +- tests/test_tenant_scoping.py | 412 +++++++++++++++++++++++++++++ 9 files changed, 522 insertions(+), 30 deletions(-) create mode 100644 tests/test_tenant_scoping.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a585c5..cac3eb1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed (security) +- **Cross-tenant leaks in HybridRAG legs and ingestion closed** - A live probe as `user=llm_tester` returned `jpmschweitzer` pages. Root causes fixed: + - **Ingestion namespace enforcement**: `vector` and `graph` `update_from_page` now refuse pages whose wiki path is outside `users/{user}/` (previously any tenant could ingest any page id — including another tenant's — into its own collection/labels, which is how foreign content ended up in the vector leg). `/ingest/all` clamps `path_prefix` to the caller's namespace (400 on cross-tenant prefixes) and defaults to `users/{user}`. + - **Search persistence**: the `FOUND` link in HybridRAG phase 6 matched `(d:Document {page_id})` unscoped, attaching the caller's SearchQuery to other tenants' Document nodes; it now matches only `User_{Tenant}_Document` nodes. + - **Graph enrichment/consolidation queries scoped**: `_get_entity_mention_count`, entity-stub generation, orphan-entity find/purge, and `cleanup_broken_relationships` matched unscoped `Document`/`SearchQuery` nodes; all now use the tenant's labels. Entity-page existence checks list only the tenant's wiki namespace. + - **Volatile collections sanitized**: `volatile_{user}` collection names now use the sanitized user id (same scheme as document collections). + - **Namespace matching hardened**: `is_path_in_user_namespace` now enforces a path-segment boundary (`users/llm_tester2` is no longer inside `llm_tester`'s namespace) and compares sanitized tenant segments. + - Offline unit tests added per leg (vector, graph, volatile, documents, enrichment, persistence, ingestion) asserting the tenant-scoped collection/label/path is used. + - **`/query/graph` and `/graph/query` hardened to read-only** - The documented "automatic user scoping" was a no-op (a live probe confirmed any user string could read the whole graph) and the client permitted writes. Raw Cypher queries are now (1) rejected with 400 when they contain write clauses (`CREATE`/`MERGE`/`DELETE`/`DETACH`/`SET`/`REMOVE`/`DROP`/`FOREACH`/`LOAD CSV`) or any `CALL` procedure (conservative denylist on the uppercased query), and (2) executed through a Neo4j session opened with `default_access_mode=READ_ACCESS` so the database itself refuses writes as a backstop. The endpoints are now honestly documented as **admin/debug, unscoped read-only**: results are not restricted to the caller's tenant labels — use `/graph/nodes` for tenant-scoped access. ### Added diff --git a/src/core/multi_tenancy.py b/src/core/multi_tenancy.py index 330c806..de6f8f7 100644 --- a/src/core/multi_tenancy.py +++ b/src/core/multi_tenancy.py @@ -243,6 +243,16 @@ def is_path_in_user_namespace(path: str, user_id: str) -> bool: False >>> is_path_in_user_namespace("/public/docs", "jpmschweitzer") False + >>> is_path_in_user_namespace("/users/llm_tester2/x", "llm_tester") + False + >>> is_path_in_user_namespace("users/llm-tester/x", "llm_tester") + True """ - namespace = get_wikijs_namespace(user_id) - return path.startswith(namespace) + # Compare the tenant path segment exactly (after sanitization, since + # canonical wiki namespaces use sanitized user ids). This enforces a + # segment boundary — "users/llm_tester2" is NOT in "llm_tester"'s + # namespace — and treats "llm-tester"/"llm_tester" as the same tenant. + parts = str(path).lstrip("/").split("/") + if len(parts) < 2 or parts[0] != "users": + return False + return sanitize_user_id(parts[1]) == sanitize_user_id(user_id) diff --git a/src/routers/ingestion.py b/src/routers/ingestion.py index 09f2920..b18df36 100644 --- a/src/routers/ingestion.py +++ b/src/routers/ingestion.py @@ -159,11 +159,14 @@ async def ingest_all_pages( -H "Authorization: Bearer $API_KEY" ``` """ - result = await ingestion.ingest_all_pages( - user=user, - path_prefix=path_prefix, - force_refresh=force_refresh, - max_concurrent=max_concurrent - ) + try: + result = await ingestion.ingest_all_pages( + user=user, + path_prefix=path_prefix, + force_refresh=force_refresh, + max_concurrent=max_concurrent + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) return result diff --git a/src/services/graph_service.py b/src/services/graph_service.py index e383cec..06dfd32 100644 --- a/src/services/graph_service.py +++ b/src/services/graph_service.py @@ -12,7 +12,7 @@ import logging from src.clients.neo4j_client import Neo4jClient from src.clients.wikijs_client import WikiJSClient -from src.core.multi_tenancy import get_neo4j_user_label +from src.core.multi_tenancy import get_neo4j_user_label, is_path_in_user_namespace from src.models.graph import ( GraphNode, GraphRelationship, GraphNodeDetail, CypherQueryResponse, GraphUpdateSummary, EntityMention, @@ -424,6 +424,14 @@ class GraphService: if not page: raise ValueError(f"Page {page_id} not found") + # TENANT ISOLATION: only pages inside the user's own wiki + # namespace may be written into that user's graph labels. + if not is_path_in_user_namespace(page.get("path", ""), user): + raise ValueError( + f"Page {page_id} (path: {page.get('path')!r}) is outside " + f"user '{user}' namespace - refusing cross-tenant ingestion" + ) + # PROTECTION: Skip entity extraction on auto-generated entity stub pages tags = page.get("tags", []) if "entity-stub" in tags or "auto-generated" in tags: @@ -675,10 +683,11 @@ class GraphService: """ from src.core.multi_tenancy import get_neo4j_user_base_label user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) query = f""" MATCH (e:{user_base_label}:{entity_type} {{name: $name}}) - MATCH (d:Document)-[:MENTIONS]->(e) + MATCH (d:{user_doc_label}:Document)-[:MENTIONS]->(e) RETURN count(distinct d) as mention_count """ @@ -714,8 +723,8 @@ class GraphService: entity_path = f"{user_namespace}/entities/{entity_type.lower()}/{entity_name.lower().replace(' ', '-')}" try: - # Search for page by path - pages = await self.wiki.list_pages(limit=1000) + # Search for page by path (scoped to the user's namespace) + pages = await self.wiki.list_pages(path_prefix=user_namespace, limit=1000) # list_pages returns a list directly, not a dict for page in pages: if page.get("path", "") == entity_path: @@ -822,11 +831,12 @@ Feel free to expand it with more details! try: from src.core.multi_tenancy import get_neo4j_user_base_label user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) - # Get mentioning documents + # Get mentioning documents (scoped to this tenant's documents) mention_query = f""" MATCH (e:{user_base_label}:{entity_type} {{name: $name}}) - MATCH (d:Document)-[:MENTIONS]->(e) + MATCH (d:{user_doc_label}:Document)-[:MENTIONS]->(e) RETURN d.title as title, d.path as path, d.page_id as page_id """ @@ -844,7 +854,7 @@ Feel free to expand it with more details! # Only match entity nodes (not Document nodes) related_query = f""" MATCH (e1:{user_base_label}:{entity_type} {{name: $name}}) - MATCH (d:Document)-[:MENTIONS]->(e1) + MATCH (d:{user_doc_label}:Document)-[:MENTIONS]->(e1) MATCH (d)-[:MENTIONS]->(e2:{user_base_label}) WHERE e2 <> e1 AND NOT (e2:Document) RETURN DISTINCT e2.name as name, labels(e2) as labels, @@ -934,15 +944,16 @@ Feel free to expand it with more details! from src.core.multi_tenancy import get_neo4j_user_base_label user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) pages_created = [] pages_skipped = [] try: - # Query for entities with sufficient mentions + # Query for entities with sufficient mentions (tenant-scoped) for entity_type in entity_types: query = f""" MATCH (e:{user_base_label}:{entity_type}) - MATCH (d:Document)-[:MENTIONS]->(e) + MATCH (d:{user_doc_label}:Document)-[:MENTIONS]->(e) WITH e, count(distinct d) as mention_count WHERE mention_count >= $min_mentions RETURN e.name as name, mention_count @@ -1427,12 +1438,13 @@ Feel free to expand it with more details! from src.core.multi_tenancy import get_neo4j_user_base_label user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) query = f""" MATCH (e:{user_base_label}) WHERE NOT e:Document AND NOT e:DocumentCollection - AND NOT EXISTS {{ (d:Document)-[:MENTIONS]->(e) }} + AND NOT EXISTS {{ (d:{user_doc_label}:Document)-[:MENTIONS]->(e) }} RETURN elementId(e) as id, e.name as name, labels(e) as labels """ @@ -1475,12 +1487,13 @@ Feel free to expand it with more details! from src.core.multi_tenancy import get_neo4j_user_base_label user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) query = f""" MATCH (e:{user_base_label}) WHERE NOT e:Document AND NOT e:DocumentCollection - AND NOT EXISTS {{ (d:Document)-[:MENTIONS]->(e) }} + AND NOT EXISTS {{ (d:{user_doc_label}:Document)-[:MENTIONS]->(e) }} DETACH DELETE e RETURN count(e) as purged_count """ @@ -1608,9 +1621,13 @@ Feel free to expand it with more details! Returns: Number of relationships cleaned """ - query = """ - MATCH (sq:SearchQuery)-[r:FOUND]->(d) - WHERE NOT EXISTS { (d) } + from src.core.multi_tenancy import get_neo4j_user_base_label + + user_base_label = get_neo4j_user_base_label(user) + + query = f""" + MATCH (sq:{user_base_label}_SearchQuery:SearchQuery)-[r:FOUND]->(d) + WHERE NOT EXISTS {{ (d) }} DELETE r RETURN count(r) as cleaned_count """ diff --git a/src/services/hybrid_rag_service.py b/src/services/hybrid_rag_service.py index 7ef48e2..1187035 100644 --- a/src/services/hybrid_rag_service.py +++ b/src/services/hybrid_rag_service.py @@ -916,6 +916,7 @@ Ranking:""" """ try: user_base_label = get_neo4j_user_base_label(user) + user_doc_label = get_neo4j_user_label(user) search_id = str(uuid.uuid4()) # Create SearchQuery node @@ -956,9 +957,13 @@ Ranking:""" page_id = result.get("page_id") if page_id: + # TENANT ISOLATION: only link to this tenant's Document + # nodes. An unscoped (d:Document {page_id}) match would + # attach FOUND relationships to other tenants' documents + # that share the same Wiki.js page id. link_doc_query = f""" MATCH (sq:{user_base_label}_SearchQuery:SearchQuery {{id: $search_id}}) - MATCH (d:Document {{page_id: $page_id}}) + MATCH (d:{user_doc_label}:Document {{page_id: $page_id}}) MERGE (sq)-[f:FOUND]->(d) SET f.source = $source, f.rank = $rank, diff --git a/src/services/ingestion_service.py b/src/services/ingestion_service.py index 6944578..bc51cc3 100644 --- a/src/services/ingestion_service.py +++ b/src/services/ingestion_service.py @@ -386,12 +386,32 @@ class IngestionService: Returns: BatchIngestionResult """ - logger.info(f"Finding all pages for user {user} (prefix: {path_prefix or 'all'})") + from src.core.multi_tenancy import sanitize_user_id + + # TENANT ISOLATION: the listing prefix is clamped to the user's own + # wiki namespace. A caller-supplied prefix outside users/{user}/ + # would otherwise ingest another tenant's pages into this tenant's + # collection and graph labels. + if path_prefix: + parts = path_prefix.strip("/").split("/") + if ( + len(parts) < 2 + or parts[0] != "users" + or sanitize_user_id(parts[1]) != sanitize_user_id(user) + ): + raise ValueError( + f"path_prefix {path_prefix!r} is outside user '{user}' " + f"namespace (users/{sanitize_user_id(user)}/) - refusing " + "cross-tenant ingestion" + ) + effective_prefix = path_prefix.strip("/") + else: + effective_prefix = f"users/{sanitize_user_id(user)}" + + logger.info(f"Finding all pages for user {user} (prefix: {effective_prefix})") # List all pages (not search - search requires a query and may have stale index) - pages = await self.wiki.list_all_pages( - path_prefix=path_prefix or f"users/{user}" - ) + pages = await self.wiki.list_all_pages(path_prefix=effective_prefix) if not pages: logger.warning(f"No pages found for user {user}") diff --git a/src/services/vector_service.py b/src/services/vector_service.py index b4fe385..65a017f 100644 --- a/src/services/vector_service.py +++ b/src/services/vector_service.py @@ -14,7 +14,7 @@ import logging from src.clients.qdrant_client import QdrantClientWrapper from src.clients.wikijs_client import WikiJSClient from src.clients.ollama_client import OllamaClient -from src.core.multi_tenancy import get_qdrant_collection_name +from src.core.multi_tenancy import get_qdrant_collection_name, is_path_in_user_namespace from src.models.vector import ( SearchResult, SearchResponse, VectorUpdateSummary, DocumentChunk, CollectionInfo, CollectionListResponse @@ -120,6 +120,16 @@ class VectorService: if not page: raise ValueError(f"Page {page_id} not found") + # TENANT ISOLATION: only pages inside the user's own wiki + # namespace may be embedded into that user's collection. + # Without this check any tenant could ingest (and then read) + # another tenant's wiki content. + if not is_path_in_user_namespace(page.get("path", ""), user): + raise ValueError( + f"Page {page_id} (path: {page.get('path')!r}) is outside " + f"user '{user}' namespace - refusing cross-tenant ingestion" + ) + # Get collection name for user collection_name = get_qdrant_collection_name(user) diff --git a/src/services/volatile_service.py b/src/services/volatile_service.py index 274a5df..760ebc3 100644 --- a/src/services/volatile_service.py +++ b/src/services/volatile_service.py @@ -58,8 +58,15 @@ class VolatileCacheService: logger.info("Initialized VolatileCacheService (Qdrant backend)") def _collection_name(self, user: str) -> str: - """Get volatile collection name for user.""" - return f"{self.COLLECTION_PREFIX}{user}" + """ + Get volatile collection name for user. + + The user id is sanitized (same rules as the document collections) + so raw identifiers cannot alias or escape the per-tenant + collection naming scheme. + """ + from src.core.multi_tenancy import sanitize_user_id + return f"{self.COLLECTION_PREFIX}{sanitize_user_id(user)}" def _make_vector_id(self, namespace: str, key: str) -> str: """ diff --git a/tests/test_tenant_scoping.py b/tests/test_tenant_scoping.py new file mode 100644 index 0000000..c04ba01 --- /dev/null +++ b/tests/test_tenant_scoping.py @@ -0,0 +1,412 @@ +""" +Offline tenant-isolation unit tests for every HybridRAG leg (mocked clients). + +For each retrieval leg (vector, graph, volatile, documents) plus the +enrichment and persistence phases, assert that the tenant-scoped +collection / label / path is used and that cross-tenant access is refused. + +Live probes showed /query/hybrid as user=llm_tester returning jpmschweitzer +pages: the root cause was unscoped ingestion (any tenant could ingest any +page id / any path prefix into its own collection) and an unscoped +Document match in search persistence. These tests pin the fixes. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from src.models.hybrid_rag import HybridRAGConfig +from src.services.graph_service import GraphService +from src.services.hybrid_rag_service import HybridRAGService +from src.services.ingestion_service import IngestionService +from src.services.vector_service import VectorService +from src.services.volatile_service import VolatileCacheService + +TENANT = "llm_tester" +TENANT_COLLECTION = "library_desk_llm_tester" +TENANT_BASE_LABEL = "User_Llm_Tester" +TENANT_DOC_LABEL = "User_Llm_Tester_Document" + + +# ============================================================================= +# Fixtures +# ============================================================================= + + +@pytest.fixture +def mock_qdrant(): + qdrant = MagicMock() + qdrant.collection_exists = AsyncMock(return_value=True) + qdrant.ensure_collection = AsyncMock() + qdrant.search_vectors = AsyncMock(return_value=[]) + qdrant.search_with_expiry_filter = AsyncMock(return_value=[]) + qdrant.upsert_vector = AsyncMock(return_value=True) + qdrant.delete_by_filter = AsyncMock(return_value=0) + qdrant.client = MagicMock() + qdrant.client.search = MagicMock(return_value=[]) + return qdrant + + +@pytest.fixture +def mock_ollama(): + ollama = MagicMock() + ollama.embed = AsyncMock(return_value=[0.1] * 768) + ollama.embed_text = AsyncMock(return_value=[0.1] * 768) + ollama.generate_text = AsyncMock(return_value="{}") + return ollama + + +@pytest.fixture +def mock_neo4j(): + neo4j = MagicMock() + neo4j.execute_query = AsyncMock(return_value=[]) + neo4j.execute_read = AsyncMock(return_value=[]) + return neo4j + + +@pytest.fixture +def mock_wiki(): + wiki = MagicMock() + wiki.get_page = AsyncMock(return_value=None) + wiki.list_all_pages = AsyncMock(return_value=[]) + wiki.list_pages = AsyncMock(return_value=[]) + return wiki + + +@pytest.fixture +def vector_service(mock_qdrant, mock_wiki, mock_ollama): + return VectorService(mock_qdrant, mock_wiki, mock_ollama) + + +@pytest.fixture +def graph_service(mock_neo4j, mock_wiki): + return GraphService(mock_neo4j, mock_wiki) + + +@pytest.fixture +def volatile_service(mock_qdrant, mock_ollama): + return VolatileCacheService(mock_qdrant, mock_ollama, MagicMock()) + + +@pytest.fixture +def hybrid_service(vector_service, graph_service, volatile_service, mock_ollama): + settings = MagicMock() + settings.ollama_llm_model = "test-model" + settings.vector_similarity_threshold = 0.7 + return HybridRAGService( + vector_service=vector_service, + graph_service=graph_service, + searxng_client=MagicMock(), + ollama_client=mock_ollama, + content_extractor=MagicMock(), + settings=settings, + volatile_service=volatile_service, + ) + + +def _all_cypher(mock_neo4j) -> str: + """Concatenate all Cypher sent to the mocked Neo4j client.""" + return "\n".join( + str(call.args[0]) for call in mock_neo4j.execute_query.await_args_list + ) + + +# ============================================================================= +# Vector leg +# ============================================================================= + + +@pytest.mark.unit +class TestVectorLegScoping: + async def test_search_uses_tenant_collection(self, vector_service, mock_qdrant): + await vector_service.search(query="q", user=TENANT) + + mock_qdrant.collection_exists.assert_awaited_with(TENANT_COLLECTION) + assert ( + mock_qdrant.search_vectors.await_args.kwargs["collection_name"] + == TENANT_COLLECTION + ) + + async def test_delete_page_chunks_uses_tenant_collection( + self, vector_service, mock_qdrant + ): + await vector_service.delete_page_chunks(page_id=1, user=TENANT) + + assert ( + mock_qdrant.delete_by_filter.await_args.kwargs["collection_name"] + == TENANT_COLLECTION + ) + + async def test_update_from_page_rejects_foreign_namespace( + self, vector_service, mock_qdrant, mock_wiki + ): + """Ingesting another tenant's page into our collection must fail.""" + mock_wiki.get_page = AsyncMock(return_value={ + "id": 42, "title": "T", "path": "users/jpmschweitzer/secret", + "content": "secret content", "tags": [], + }) + + summary = await vector_service.update_from_page(page_id=42, user=TENANT) + + assert summary.success is False + assert "outside user" in (summary.error_message or "") + mock_qdrant.upsert_vector.assert_not_awaited() + mock_qdrant.delete_by_filter.assert_not_awaited() + + async def test_update_from_page_rejects_sibling_prefix_namespace( + self, vector_service, mock_qdrant, mock_wiki + ): + """users/llm_tester2 is NOT inside llm_tester's namespace.""" + mock_wiki.get_page = AsyncMock(return_value={ + "id": 43, "title": "T", "path": "users/llm_tester2/page", + "content": "content", "tags": [], + }) + + summary = await vector_service.update_from_page(page_id=43, user=TENANT) + + assert summary.success is False + mock_qdrant.upsert_vector.assert_not_awaited() + + async def test_update_from_page_accepts_own_namespace( + self, vector_service, mock_qdrant, mock_wiki + ): + mock_wiki.get_page = AsyncMock(return_value={ + "id": 44, "title": "T", "path": "users/llm_tester/page", + "content": "hello world", "tags": [], + }) + + summary = await vector_service.update_from_page(page_id=44, user=TENANT) + + assert summary.success is True + assert ( + mock_qdrant.upsert_vector.await_args.kwargs["collection_name"] + == TENANT_COLLECTION + ) + + +# ============================================================================= +# Graph leg +# ============================================================================= + + +@pytest.mark.unit +class TestGraphLegScoping: + async def test_search_documents_uses_tenant_labels( + self, graph_service, mock_neo4j + ): + await graph_service.search_documents( + query="docker", user=TENANT, keywords_data={"core_keywords": ["docker"]} + ) + + cypher = _all_cypher(mock_neo4j) + assert f"(e:{TENANT_BASE_LABEL})" in cypher + assert f"(d:{TENANT_DOC_LABEL}:Document)" in cypher + + async def test_update_from_page_rejects_foreign_namespace( + self, graph_service, mock_neo4j, mock_wiki + ): + mock_wiki.get_page = AsyncMock(return_value={ + "id": 42, "title": "T", "path": "users/jpmschweitzer/secret", + "content": "secret", "tags": [], + }) + + summary = await graph_service.update_from_page(page_id=42, user=TENANT) + + assert summary.success is False + assert "outside user" in (summary.error_message or "") + mock_neo4j.execute_query.assert_not_awaited() + + async def test_update_from_page_writes_tenant_labels( + self, graph_service, mock_neo4j, mock_wiki + ): + mock_wiki.get_page = AsyncMock(return_value={ + "id": 44, "title": "T", "path": "users/llm_tester/page", + "content": "Uses Docker daily", "tags": [], + }) + + summary = await graph_service.update_from_page(page_id=44, user=TENANT) + + assert summary.success is True + cypher = _all_cypher(mock_neo4j) + assert TENANT_DOC_LABEL in cypher + # No unscoped Document writes + assert "MERGE (d:Document {" not in cypher + + async def test_entity_mention_count_scoped(self, graph_service, mock_neo4j): + await graph_service._get_entity_mention_count("Docker", "Technology", TENANT) + + cypher = _all_cypher(mock_neo4j) + assert f"(d:{TENANT_DOC_LABEL}:Document)" in cypher + assert "MATCH (d:Document)" not in cypher + + async def test_orphan_queries_scoped(self, graph_service, mock_neo4j): + await graph_service.find_orphan_entities(TENANT) + await graph_service.purge_orphan_entities(TENANT) + + cypher = _all_cypher(mock_neo4j) + assert f"(d:{TENANT_DOC_LABEL}:Document)" in cypher + assert "(d:Document)-[:MENTIONS]" not in cypher + + async def test_cleanup_broken_relationships_scoped( + self, graph_service, mock_neo4j + ): + await graph_service.cleanup_broken_relationships(TENANT) + + cypher = _all_cypher(mock_neo4j) + assert f"{TENANT_BASE_LABEL}_SearchQuery" in cypher + assert "MATCH (sq:SearchQuery)" not in cypher + + +# ============================================================================= +# Volatile leg +# ============================================================================= + + +@pytest.mark.unit +class TestVolatileLegScoping: + async def test_search_uses_tenant_collection(self, volatile_service, mock_qdrant): + await volatile_service.search(user=TENANT, query="weather") + + mock_qdrant.collection_exists.assert_awaited_with("volatile_llm_tester") + assert ( + mock_qdrant.search_with_expiry_filter.await_args.kwargs["collection_name"] + == "volatile_llm_tester" + ) + + async def test_store_uses_tenant_collection(self, volatile_service, mock_qdrant): + await volatile_service.store( + user=TENANT, namespace="weather", key="rotterdam", data={"t": 1} + ) + + assert ( + mock_qdrant.upsert_vector.await_args.kwargs["collection_name"] + == "volatile_llm_tester" + ) + + def test_collection_name_is_sanitized(self, volatile_service): + """Raw user strings cannot alias/escape the collection scheme.""" + assert volatile_service._collection_name("llm-tester") == "volatile_llm_tester" + assert volatile_service._collection_name("Evil User!") == "volatile_evil_user" + + +# ============================================================================= +# Documents leg (Paperless chunks in the tenant collection) +# ============================================================================= + + +@pytest.mark.unit +class TestDocumentLegScoping: + async def test_document_search_uses_tenant_collection( + self, hybrid_service, mock_qdrant + ): + config = HybridRAGConfig( + enable_vector=False, enable_graph=False, enable_web=False, + enable_volatile=False, enable_documents=True, + ) + + await hybrid_service._retrieve_parallel("q", TENANT, config, {}) + + assert ( + mock_qdrant.client.search.call_args.kwargs["collection_name"] + == TENANT_COLLECTION + ) + + +# ============================================================================= +# Enrichment phase +# ============================================================================= + + +@pytest.mark.unit +class TestEnrichmentScoping: + async def test_related_documents_scoped_to_tenant( + self, graph_service, mock_neo4j + ): + await graph_service.get_related_documents(page_id=1, user=TENANT) + + cypher = _all_cypher(mock_neo4j) + assert f"(d1:{TENANT_DOC_LABEL}:Document" in cypher + assert f"(d2:{TENANT_DOC_LABEL}:Document)" in cypher + + +# ============================================================================= +# Persistence phase (SearchQuery / FOUND / WebResult) +# ============================================================================= + + +@pytest.mark.unit +class TestPersistenceScoping: + async def test_persist_scopes_searchquery_links_and_webresults( + self, hybrid_service, mock_neo4j + ): + final_results = [ + { + "result": {"page_id": 42, "title": "wiki hit"}, + "source_type": "wiki", "sources": ["vector"], + "rrf_score": 0.5, "final_rank": 1, + }, + { + "result": {"url": "http://example.com", "title": "web hit", + "content": "c"}, + "source_type": "web", "sources": ["web"], + "rrf_score": 0.4, "final_rank": 2, + }, + ] + + await hybrid_service._persist_search_for_librarian( + query="q", user=TENANT, keywords_data={}, + raw_results={}, final_results=final_results, timing={}, + ) + + cypher = _all_cypher(mock_neo4j) + # SearchQuery + WebResult nodes carry the tenant label + assert f"{TENANT_BASE_LABEL}_SearchQuery" in cypher + assert f"{TENANT_BASE_LABEL}_WebResult" in cypher + # FOUND link matches only this tenant's Document nodes + assert f"MATCH (d:{TENANT_DOC_LABEL}:Document {{page_id: $page_id}})" in cypher + # The old cross-tenant match must be gone + assert "MATCH (d:Document {page_id: $page_id})" not in cypher + + +# ============================================================================= +# Ingestion (feeds the vector/graph legs) +# ============================================================================= + + +@pytest.mark.unit +class TestIngestAllPagesScoping: + @pytest.fixture + def ingestion_service(self, vector_service, graph_service, mock_wiki): + return IngestionService(vector_service, graph_service, mock_wiki) + + async def test_defaults_to_own_namespace(self, ingestion_service, mock_wiki): + await ingestion_service.ingest_all_pages(user=TENANT) + + mock_wiki.list_all_pages.assert_awaited_once_with( + path_prefix="users/llm_tester" + ) + + async def test_rejects_foreign_prefix(self, ingestion_service, mock_wiki): + with pytest.raises(ValueError, match="outside user"): + await ingestion_service.ingest_all_pages( + user=TENANT, path_prefix="users/jpmschweitzer" + ) + + mock_wiki.list_all_pages.assert_not_awaited() + + async def test_rejects_sibling_prefix(self, ingestion_service, mock_wiki): + with pytest.raises(ValueError, match="outside user"): + await ingestion_service.ingest_all_pages( + user=TENANT, path_prefix="users/llm_tester2" + ) + + async def test_accepts_subtree_of_own_namespace( + self, ingestion_service, mock_wiki + ): + await ingestion_service.ingest_all_pages( + user=TENANT, path_prefix="users/llm_tester/tech" + ) + + mock_wiki.list_all_pages.assert_awaited_once_with( + path_prefix="users/llm_tester/tech" + )