Enforce tenant isolation via TenantScopedClient + CI guard

README.md

@@ -51,6 +51,22 @@ built via `payload_m`. Every single query carries a mandatory `Filter` on
 that proves it). This is Qdrant's recommended multitenant layout: cheaper than a
 collection-per-tenant and strictly isolated.
 
+**Enforced isolation (copy this).** Isolation used to depend on every call site
+remembering `build_filter(tenant_id, ...)`. That is easy to forget on a new code
+path, and the failure is silent. tenantq now routes all search reads through
+`TenantScopedClient` (`src/tenantq/scoped_client.py`):
+
+1. `tenant_id` is a **required** keyword argument on `query_points` / `scroll` /
+   `retrieve`.
+2. The wrapper **always** injects a `tenant_id` match into the filter (and into
+   every hybrid `Prefetch` branch). A raw filter cannot drop the tenant scope.
+3. A CI test (`tests/test_scoped_access.py`) fails if package code calls
+   `.query_points(`, `.scroll(`, or `.retrieve(` outside that wrapper.
+
+`search()` behaviour is unchanged for callers; only the enforcement mechanism is
+new. If you adapt this layout, copy the wrapper + the grep-style guard — not just
+the filter helper.
+
 **Hybrid retrieval.** Dense vectors come from `fastembed`
 (`sentence-transformers/all-MiniLM-L6-v2`), sparse from fastembed's
 `SparseTextEmbedding` (`Qdrant/bm25`, IDF modifier). Hybrid search uses the Qdrant

src/tenantq/scoped_client.py

@@ -0,0 +1,138 @@
+"""Tenant-scoped Qdrant access.
+
+Raw ``QdrantClient.query_points`` / ``scroll`` / ``retrieve`` accept an optional
+filter. A caller that forgets the tenant condition silently returns other
+tenants' documents. This wrapper is the only code path that may issue those
+reads: it requires ``tenant_id`` and always injects the tenant filter.
+
+``search()`` and any future query helpers must go through :class:`TenantScopedClient`.
+A CI test greps the package for direct client reads outside this module.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from qdrant_client import QdrantClient, models
+
+from .config import TENANT_FIELD
+
+
+def _require_tenant_id(tenant_id: str) -> str:
+    if tenant_id is None or not str(tenant_id).strip():
+        raise ValueError("tenant_id is required and must be non-empty")
+    return str(tenant_id)
+
+
+def _with_tenant(tenant_id: str, query_filter: Optional[models.Filter]) -> models.Filter:
+    """Merge a mandatory tenant match into an optional caller filter.
+
+    The tenant condition is always present. Callers cannot pass a filter that
+    drops it: even if they include their own tenant clause, we still add ours
+    for the required ``tenant_id``.
+    """
+    tenant_id = _require_tenant_id(tenant_id)
+    tenant_cond = models.FieldCondition(
+        key=TENANT_FIELD, match=models.MatchValue(value=tenant_id)
+    )
+    if query_filter is None:
+        return models.Filter(must=[tenant_cond])
+
+    must = list(query_filter.must or [])
+    # Put tenant first so the isolation guarantee is obvious in serialized filters.
+    must.insert(0, tenant_cond)
+    return models.Filter(
+        must=must,
+        should=query_filter.should,
+        must_not=query_filter.must_not,
+        min_should=query_filter.min_should,
+    )
+
+
+def _scope_prefetch(tenant_id: str, prefetch: Any) -> Any:
+    """Ensure every Prefetch branch carries the tenant filter."""
+    if prefetch is None:
+        return None
+    if isinstance(prefetch, list):
+        return [_scope_prefetch(tenant_id, p) for p in prefetch]
+    if isinstance(prefetch, models.Prefetch):
+        merged = _with_tenant(tenant_id, prefetch.filter)
+        nested = _scope_prefetch(tenant_id, prefetch.prefetch) if prefetch.prefetch else None
+        # model_copy keeps forward-compat with new Prefetch fields
+        if hasattr(prefetch, "model_copy"):
+            return prefetch.model_copy(update={"filter": merged, "prefetch": nested})
+        return models.Prefetch(
+            query=prefetch.query,
+            using=prefetch.using,
+            filter=merged,
+            params=prefetch.params,
+            limit=prefetch.limit,
+            prefetch=nested,
+        )
+    return prefetch
+
+
+class TenantScopedClient:
+    """Thin wrapper: the only surface allowed to call read APIs on Qdrant.
+
+    Write/ingest paths still use the raw client (points already carry tenant_id
+    in payload). Isolation at read time is what this enforces.
+    """
+
+    def __init__(self, client: QdrantClient) -> None:
+        self._client = client
+
+    @property
+    def raw(self) -> QdrantClient:
+        """Escape hatch for ingest, collection admin, and tests — not for search."""
+        return self._client
+
+    def query_points(
+        self,
+        *,
+        tenant_id: str,
+        collection_name: str,
+        query_filter: Optional[models.Filter] = None,
+        prefetch: Any = None,
+        **kwargs: Any,
+    ):
+        qf = _with_tenant(tenant_id, query_filter)
+        pf = _scope_prefetch(tenant_id, prefetch)
+        return self._client.query_points(
+            collection_name=collection_name,
+            query_filter=qf,
+            prefetch=pf,
+            **kwargs,
+        )
+
+    def scroll(
+        self,
+        *,
+        tenant_id: str,
+        collection_name: str,
+        scroll_filter: Optional[models.Filter] = None,
+        **kwargs: Any,
+    ):
+        sf = _with_tenant(tenant_id, scroll_filter)
+        return self._client.scroll(
+            collection_name=collection_name,
+            scroll_filter=sf,
+            **kwargs,
+        )
+
+    def retrieve(
+        self,
+        *,
+        tenant_id: str,
+        collection_name: str,
+        ids: list,
+        **kwargs: Any,
+    ):
+        # retrieve has no filter param on all client versions; post-filter by tenant
+        points = self._client.retrieve(collection_name=collection_name, ids=ids, **kwargs)
+        tenant_id = _require_tenant_id(tenant_id)
+        return [
+            p
+            for p in points
+            if (p.payload or {}).get(TENANT_FIELD) == tenant_id
+        ]

src/tenantq/search.py

@@ -24,6 +24,7 @@
 )
 from .embeddings import Embedder
 from .metrics import QUERY_LATENCY
+from .scoped_client import TenantScopedClient
 
 Mode = Literal["dense", "sparse", "hybrid"]
 
@@ -90,14 +91,23 @@ def search(
     created_before: Optional[int] = None,
     prefetch_limit: int = 50,
 ) -> List[SearchHit]:
-    """Run a tenant-isolated search in the requested retrieval mode."""
+    """Run a tenant-isolated search in the requested retrieval mode.
+
+    All Qdrant reads go through :class:`TenantScopedClient`, which requires
+    ``tenant_id`` and injects the tenant filter so a missing filter cannot
+    silently cross tenants.
+    """
     qfilter = build_filter(tenant_id, category, created_after, created_before)
+    # build_filter already includes tenant_id; scoped client re-injects it as a
+    # second hard guarantee (duplicate must-conditions are fine for MatchValue).
+    scoped = TenantScopedClient(client)
     params = models.SearchParams(hnsw_ef=settings.hnsw.hnsw_ef)
 
     start = time.perf_counter()
     if mode == "dense":
         dense_vec = embedder.embed_dense([query])[0]
-        res = client.query_points(
+        res = scoped.query_points(
+            tenant_id=tenant_id,
             collection_name=settings.collection,
             query=dense_vec,
             using=DENSE_VECTOR_NAME,
@@ -108,7 +118,8 @@ def search(
         )
     elif mode == "sparse":
         sv = embedder.embed_sparse([query])[0]
-        res = client.query_points(
+        res = scoped.query_points(
+            tenant_id=tenant_id,
             collection_name=settings.collection,
             query=models.SparseVector(indices=sv.indices, values=sv.values),
             using=SPARSE_VECTOR_NAME,
@@ -119,7 +130,8 @@ def search(
     elif mode == "hybrid":
         dense_vec = embedder.embed_dense([query])[0]
         sv = embedder.embed_sparse([query])[0]
-        res = client.query_points(
+        res = scoped.query_points(
+            tenant_id=tenant_id,
             collection_name=settings.collection,
             prefetch=[
                 models.Prefetch(

tests/test_scoped_access.py

@@ -0,0 +1,94 @@
+"""Structural isolation: no raw query_points/scroll/retrieve outside the wrapper."""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import pytest
+
+from tenantq.scoped_client import TenantScopedClient, _with_tenant
+
+PKG = Path(__file__).resolve().parents[1] / "src" / "tenantq"
+
+# Modules allowed to call the underlying client read APIs.
+_ALLOWED_READ_MODULES = frozenset({"scoped_client.py"})
+
+_READ_ATTRS = ("query_points", "scroll", "retrieve")
+
+
+def _python_files():
+    return sorted(PKG.glob("*.py"))
+
+
+# Raw client receivers only. Wrapper calls (scoped.query_points) are the allowed path.
+_RAW_CLIENT_READ = re.compile(
+    r"(?<![\w.])(?:client|_client)\.(%s)\s*\(" % "|".join(_READ_ATTRS)
+)
+
+
+def test_no_raw_read_apis_outside_scoped_client():
+    """A deliberately leaky call site in package code would fail this test.
+
+    Direct ``client.query_points`` / ``_client.scroll`` / etc. are banned outside
+    scoped_client.py. Using TenantScopedClient (e.g. ``scoped.query_points``) is
+    required.
+    """
+    offenders: list[str] = []
+    for path in _python_files():
+        if path.name in _ALLOWED_READ_MODULES:
+            continue
+        text = path.read_text(encoding="utf-8")
+        for i, line in enumerate(text.splitlines(), 1):
+            stripped = line.lstrip()
+            if stripped.startswith("#"):
+                continue
+            if _RAW_CLIENT_READ.search(line):
+                offenders.append(f"{path.name}:{i}: {stripped}")
+    assert not offenders, (
+        "tenant-scoped reads must go through TenantScopedClient; "
+        "raw client reads found:\n  " + "\n  ".join(offenders)
+    )
+
+
+def test_leaky_query_pattern_is_detected_by_guard():
+    """Simulate a leaky code path: the guard regex must match it.
+
+    This is the acceptance case: if someone adds ``client.query_points(...)``
+    without the wrapper, CI fails.
+    """
+    leaky = "    res = client.query_points(collection_name=c, query=q)\n"
+    ok = "    res = scoped.query_points(tenant_id=t, collection_name=c, query=q)\n"
+    assert _RAW_CLIENT_READ.search(leaky)
+    assert not _RAW_CLIENT_READ.search(ok)
+
+
+def test_search_module_uses_scoped_client_not_raw():
+    text = (PKG / "search.py").read_text(encoding="utf-8")
+    assert "TenantScopedClient" in text
+    assert "scoped.query_points" in text
+    assert "client.query_points" not in text
+
+
+def test_with_tenant_requires_non_empty():
+    with pytest.raises(ValueError, match="tenant_id"):
+        _with_tenant("", None)
+    with pytest.raises(ValueError, match="tenant_id"):
+        _with_tenant("   ", None)
+
+
+def test_with_tenant_injects_condition():
+    from tenantq.config import TENANT_FIELD
+
+    f = _with_tenant("acme", None)
+    assert f.must
+    cond = f.must[0]
+    assert cond.key == TENANT_FIELD
+    assert cond.match.value == "acme"
+
+
+def test_scoped_client_requires_tenant_id_kwarg(ingested, settings):
+    scoped = TenantScopedClient(ingested)
+    with pytest.raises(TypeError):
+        # tenant_id is keyword-only required
+        scoped.query_points(collection_name=settings.collection, query=[0.0] * 8)  # type: ignore[call-arg]