Capsule 05: Multi-tenant isolation — turning isolation into a platform rule

Capsule description

In multi-tenant RAG systems, tenant_id isn't an organizational feature — it's a security and compliance obligation. A query that returns a document belonging to the wrong customer is a reportable incident: GDPR, HIPAA, SOC2, contractual. It isn't something you can "fix later" — it's something that must NEVER happen.

This capsule teaches you the patterns for guaranteeing isolation technically, not by convention. Conventions fail: any new dev can forget the filter, any refactor can drop the line that enforces tenant_id, any new endpoint can skip the guardrail. The solution is to design the system so that it's impossible to run a query without a tenant_id.

By the end of this capsule you'll be able to:

• ✅ Implement a TenantContext that encapsulates the request's identity and permissions
• ✅ Design secure_query as a mandatory wrapper (one you can't skip)
• ✅ Implement automated isolation tests that run in CI
• ✅ Configure pre-commit hooks that catch unsafe direct calls
• ✅ Audit the logs to detect bypass attempts
• ✅ Anticipate and prevent the five common isolation vulnerabilities

Estimated time: 30-35 minutes

The reality: conventions fail, code design doesn't

Three forms of "isolation" in multi-tenant systems, ordered from least to most secure:

Level 1: convention (the dev "remembers")

# The documentation says: "always include workspace_id in the filters"
results = collection.query(
    query_texts=[q],
    where={"workspace_id": user.workspace_id},
)

It fails because:

• Devs forget it in new endpoints.
• An accidental refactor removes the line.
• Code review doesn't always catch it.

Level 2: a linter/test that verifies it

# A pre-commit hook that catches `collection.query(` with no where filter

It fails because:

• Hooks can be skipped with --no-verify.
• Tests may not cover every path.
• The regex can have false negatives.

Level 3: impossible to skip by design (the right approach)

# The public function requires tenant_id in its signature
def secure_query(query: str, tenant: TenantContext, ...):
    if not tenant.workspace_id:
        raise TenantIsolationError(...)
    where = {"workspace_id": tenant.workspace_id, ...}
    ...

# And collection.query() is NEVER called directly from application code

It works because:

• A TypeError if you call it without a tenant_id (Python type checking).
• An explicit exception if the tenant_id is empty.
• A direct collection.query() is forbidden by convention + linter.

Let's go to level 3.

TenantContext: encapsulating identity and permissions

# tenant_context.py
from dataclasses import dataclass
from typing import Optional


@dataclass(frozen=True)  # immutable: nobody can modify the context
class TenantContext:
    """An immutable security context, per request."""
    workspace_id: str
    user_id: str
    user_role: str = "member"
    project_ids: list[str] = None

    def __post_init__(self):
        if not self.workspace_id:
            raise TenantIsolationError("workspace_id is mandatory")
        if not self.user_id:
            raise TenantIsolationError("user_id is mandatory")


class TenantIsolationError(Exception):
    """An explicit error for isolation violations."""
    pass


# Build it from the authenticated request
def context_from_request(request) -> TenantContext:
    """Extracts the TenantContext from an authenticated request."""
    if not request.authenticated:
        raise PermissionError("Request is not authenticated")

    return TenantContext(
        workspace_id=request.user.workspace_id,  # from the JWT token
        user_id=request.user.id,
        user_role=request.user.role,
        project_ids=request.user.project_ids,
    )

The critical point: TenantContext is built from the authenticated request, not from user input. The workspace_id comes from the JWT validated in the middleware, NOT from a URL parameter.

# ❌ Insecure: workspace_id from input
@app.post("/search")
def search(query: str, workspace_id: str):  # ← workspace_id can be controlled by an attacker
    ctx = TenantContext(workspace_id=workspace_id, ...)

# ✅ Secure: workspace_id from the validated token
@app.post("/search")
def search(query: str, user: AuthenticatedUser = Depends(get_current_user)):
    ctx = TenantContext(workspace_id=user.workspace_id, user_id=user.id, ...)

secure_query: the mandatory wrapper

# secure_query.py
import logging

logger = logging.getLogger(__name__)


def secure_query(
    collection,
    query_text: str,
    tenant: TenantContext,
    additional_filters: dict = None,
    n_results: int = 5,
):
    """
    The mandatory wrapper for every query to the vector DB.

    It guarantees:
    - workspace_id is always present in the filter
    - additional_filters can NOT overwrite workspace_id
    - Every query is logged for auditing
    - An explicit raise if anything is wrong
    """
    # Validate the TenantContext
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("A TenantContext with a workspace_id is mandatory")

    # Build the filter with workspace_id forced in
    where = _build_secure_where(tenant, additional_filters)

    # Logging for auditing (without the query's content — privacy)
    logger.info(
        "vector_query_executed",
        extra={
            "workspace_id": tenant.workspace_id,
            "user_id": tenant.user_id,
            "filter_keys": list(where.keys()) if isinstance(where, dict) else None,
            "n_results": n_results,
        }
    )

    # Run the query
    return collection.query(
        query_texts=[query_text],
        where=where,
        n_results=n_results,
    )


def _build_secure_where(tenant: TenantContext, additional_filters: dict | None) -> dict:
    """Builds the where clause with workspace_id armored in."""
    base = {"workspace_id": tenant.workspace_id}

    if not additional_filters:
        return base

    # Validate that additional_filters does NOT try to overwrite workspace_id
    if "workspace_id" in additional_filters:
        raise TenantIsolationError(
            "additional_filters may NOT contain workspace_id. "
            "It is controlled by the TenantContext."
        )

    # If there's an $or in additional_filters, make sure it doesn't relax the tenant
    _validate_no_tenant_bypass(additional_filters)

    # Combine with an explicit $and
    return {"$and": [base, additional_filters]}


def _validate_no_tenant_bypass(filters: dict) -> None:
    """Recursively verifies there's no $or that escapes the tenant."""
    if not isinstance(filters, dict):
        return

    for key, value in filters.items():
        if key == "$or" and isinstance(value, list):
            # Every element of the $or has to stay inside the workspace
            # (the outer $and enforces this, but we validate the contents too)
            for item in value:
                _validate_no_tenant_bypass(item)
        elif isinstance(value, dict):
            _validate_no_tenant_bypass(value)

Correct usage

# The endpoint
@app.post("/api/search")
async def search(query: str, user: AuthenticatedUser = Depends(get_current_user)):
    tenant = TenantContext(
        workspace_id=user.workspace_id,
        user_id=user.id,
        user_role=user.role,
    )
    results = secure_query(
        collection=vector_collection,
        query_text=query,
        tenant=tenant,
        additional_filters={"category": "auth"},  # optional
        n_results=5,
    )
    return results

Automated isolation tests

Isolation tests are non-negotiable. They must run in CI before every deploy:

# tests/test_tenant_isolation.py
import pytest
from secure_query import secure_query, TenantContext, TenantIsolationError


@pytest.fixture
def collection_with_two_tenants():
    """Setup: a collection with docs from tenant A and B."""
    client = chromadb.Client()
    collection = client.create_collection("test_isolation")

    # 100 docs from tenant_a
    for i in range(100):
        collection.add(
            documents=[f"Document {i} for tenant A"],
            metadatas=[{"workspace_id": "tenant_a"}],
            ids=[f"a_doc_{i}"],
        )

    # 100 docs from tenant_b
    for i in range(100):
        collection.add(
            documents=[f"Document {i} for tenant B"],
            metadatas=[{"workspace_id": "tenant_b"}],
            ids=[f"b_doc_{i}"],
        )
    return collection


def test_query_returns_only_tenant_a(collection_with_two_tenants):
    """Tenant A must NEVER see tenant B's docs."""
    ctx = TenantContext(workspace_id="tenant_a", user_id="user_1")
    results = secure_query(collection_with_two_tenants, "Document", ctx, n_results=20)

    for meta in results["metadatas"][0]:
        assert meta["workspace_id"] == "tenant_a", (
            f"LEAK: tenant_a saw a doc from {meta['workspace_id']}"
        )


def test_empty_workspace_id_raises_error():
    """Without a workspace_id you can't create a TenantContext."""
    with pytest.raises(TenantIsolationError):
        TenantContext(workspace_id="", user_id="user_1")


def test_additional_filters_cannot_override_workspace(collection_with_two_tenants):
    """additional_filters may NOT overwrite workspace_id."""
    ctx = TenantContext(workspace_id="tenant_a", user_id="user_1")

    with pytest.raises(TenantIsolationError):
        secure_query(
            collection_with_two_tenants,
            "Document",
            ctx,
            additional_filters={"workspace_id": "tenant_b"},  # a bypass attempt
        )


def test_or_filter_still_respects_tenant(collection_with_two_tenants):
    """$or filters cannot escape the tenant scope."""
    ctx = TenantContext(workspace_id="tenant_a", user_id="user_1")
    results = secure_query(
        collection_with_two_tenants,
        "Document",
        ctx,
        additional_filters={"$or": [{"category": "X"}, {"category": "Y"}]},
        n_results=20,
    )

    # Even with the inner $or, every result must be tenant_a's
    for meta in results["metadatas"][0]:
        assert meta["workspace_id"] == "tenant_a"


def test_concurrent_tenants_no_leakage(collection_with_two_tenants):
    """Concurrent queries from different tenants don't cross over."""
    import threading

    leaks = []

    def query_as_tenant(workspace_id):
        ctx = TenantContext(workspace_id=workspace_id, user_id=f"user_{workspace_id}")
        results = secure_query(collection_with_two_tenants, "Document", ctx, n_results=20)
        for meta in results["metadatas"][0]:
            if meta["workspace_id"] != workspace_id:
                leaks.append((workspace_id, meta["workspace_id"]))

    threads = [
        threading.Thread(target=query_as_tenant, args=("tenant_a",))
        for _ in range(10)
    ] + [
        threading.Thread(target=query_as_tenant, args=("tenant_b",))
        for _ in range(10)
    ]

    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not leaks, f"Concurrent leakage detected: {leaks}"

These tests must:

• Run on every PR (CI).
• Block merges if they fail.
• Be audited whenever new tests are added.

The anti-bypass pre-commit hook

# scripts/check_no_direct_query.py
"""
Pre-commit hook: catches direct calls to collection.query()
that don't go through secure_query.
"""
import re
import sys
from pathlib import Path

PATTERN = re.compile(r'\.query\(')
ALLOWED_FILES = {"src/secure_query.py", "tests/"}  # only allowed here

errors = []
for py_file in Path("src").rglob("*.py"):
    if any(allowed in str(py_file) for allowed in ALLOWED_FILES):
        continue

    content = py_file.read_text()
    for line_num, line in enumerate(content.split("\n"), 1):
        if PATTERN.search(line) and "collection.query" in line:
            if "secure_query" not in line:
                errors.append(f"{py_file}:{line_num}: direct collection.query() detected")

if errors:
    print("\n".join(errors))
    print("\nUse secure_query() instead of collection.query() directly.")
    sys.exit(1)

Configure it in .pre-commit-config.yaml:

repos:
  - repo: local
    hooks:
      - id: no-direct-query
        name: No direct collection.query()
        entry: python scripts/check_no_direct_query.py
        language: system
        files: \.py$

Log auditing

If your system serves sensitive data (medical, financial, legal), consider audit logging:

# audit_log.py
import json
from datetime import datetime


def log_query_for_audit(tenant: TenantContext, query: str, n_results: int):
    """Structured logs for the audit trail."""
    audit_record = {
        "timestamp": datetime.utcnow().isoformat(),
        "event": "vector_query",
        "workspace_id": tenant.workspace_id,
        "user_id": tenant.user_id,
        "user_role": tenant.user_role,
        "query_hash": hash(query),  # NOT the query itself (privacy)
        "n_results": n_results,
    }
    audit_logger.info(json.dumps(audit_record))


# Wire it into secure_query
def secure_query(...):
    log_query_for_audit(tenant, query_text, n_results)
    # ... the rest

The benefits:

• A compliance trail (HIPAA and SOC2 require it).
• Detection of anomalous behavior (one user_id issuing 1000 queries/sec = a bot).
• Post-incident debugging.

Traps and common mistakes

Trap 1: workspace_id from a URL parameter

Covered above. An attacker can modify the parameter to access other tenants.

Trap 2: dict.update() overwrites workspace_id

# ❌ Bad
where = {"workspace_id": "tenant_a"}
where.update(user_filters)  # if user_filters has a workspace_id, it overwrites it

The symptom: the user's filter wins over the system's.

How to prevent it: validate beforehand (as seen in _build_secure_where).

Trap 3: a bypass with $or at the root level

# ❌ Bad
where = {
    "$or": [
        {"workspace_id": "tenant_a"},  # ← the user can slip another one in
        {"workspace_id": "tenant_b"},
    ]
}

The symptom: the filter looks like it has tenant_a, but it also includes tenant_b.

How to prevent it: wrap it with $and. tenant_id goes at the root level, and the $or goes inside as an additional condition.

Trap 4: a cross-tenant cache

# ❌ Bad
@lru_cache(maxsize=1000)
def cached_query(query_text):
    # the cache key is only the query → results get shared across tenants
    return collection.query(...)

How to prevent it: include the workspace_id in the cache key:

@lru_cache(maxsize=1000)
def cached_query(query_text, workspace_id):
    return collection.query(query_texts=[query_text], where={"workspace_id": workspace_id})

Trap 5: admin scripts with no context

# ❌ Bad: a migration script that forgets the tenant
all_docs = collection.get()  # cross-tenant data

How to prevent it: admin scripts require an explicit AdminContext + an audit log noting "intentional cross-tenant access for task X".

Applied exercise

Scenario: you're an AI Engineer at an HR SaaS platform. Sensitive data: salaries, performance reviews, personal information.

The stakeholders are asking for:

• Strict isolation between 50 companies (workspaces).
• Each company has roles: admin, hr_manager, manager, employee.
• Some docs are confidential (HR only), others public_to_company, others team_only.
• Compliance: GDPR + a SOC2 audit in 3 months.

Your job:

1. Design the isolation + permissions system.
2. Define the automated isolation tests.
3. An audit plan for SOC2.

# hr_secure_query.py
from dataclasses import dataclass, field
from typing import Literal


@dataclass(frozen=True)
class HRTenantContext:
    workspace_id: str
    user_id: str
    user_role: Literal["admin", "hr_manager", "manager", "employee"]
    team_ids: list[str] = field(default_factory=list)

    def __post_init__(self):
        if not self.workspace_id or not self.user_id:
            raise TenantIsolationError("workspace_id and user_id are mandatory")


def hr_secure_query(query: str, tenant: HRTenantContext, n_results: int = 5):
    """A query with visibility based on role."""

    visibility_clauses = []

    # 1. The company's public docs - every role sees them
    visibility_clauses.append({"visibility": "public_to_company"})

    # 2. team_only docs - only if the user is on the team
    if tenant.team_ids:
        visibility_clauses.append({
            "$and": [
                {"visibility": "team_only"},
                {"team_id": {"$in": tenant.team_ids}},
            ]
        })

    # 3. confidential docs - HR roles only
    if tenant.user_role in ("admin", "hr_manager"):
        visibility_clauses.append({"visibility": "confidential"})

    # 4. The employee's own personal data
    visibility_clauses.append({
        "$and": [
            {"visibility": "personal"},
            {"owner_id": tenant.user_id},
        ]
    })

    where = {
        "$and": [
            {"workspace_id": tenant.workspace_id},  # tenant isolation
            {"$or": visibility_clauses},             # visibility based on role
        ]
    }

    # Audit log
    audit_log({
        "event": "hr_query",
        "workspace_id": tenant.workspace_id,
        "user_id": tenant.user_id,
        "user_role": tenant.user_role,
        "n_results": n_results,
    })

    return collection.query(query_texts=[query], where=where, n_results=n_results)

def test_employee_cannot_see_confidential():
    """An employee does NOT see confidential docs."""
    ctx = HRTenantContext(
        workspace_id="acme",
        user_id="emp_1",
        user_role="employee",
    )
    # Setup: insert a confidential doc into the acme tenant
    results = hr_secure_query("salary", ctx)
    for meta in results["metadatas"][0]:
        assert meta["visibility"] != "confidential"


def test_employee_sees_only_own_personal():
    """An employee does NOT see other people's personal data."""
    ctx = HRTenantContext(workspace_id="acme", user_id="emp_1", user_role="employee")
    results = hr_secure_query("evaluation", ctx)
    for meta in results["metadatas"][0]:
        if meta["visibility"] == "personal":
            assert meta["owner_id"] == "emp_1"


def test_hr_manager_sees_confidential_only_in_workspace():
    """HR sees confidential docs, but only from their own workspace."""
    ctx = HRTenantContext(workspace_id="acme", user_id="hr_1", user_role="hr_manager")
    # Insert confidential docs in acme and in another workspace
    results = hr_secure_query("evaluation", ctx, n_results=20)
    for meta in results["metadatas"][0]:
        assert meta["workspace_id"] == "acme"  # cross-tenant block
        # confidential is OK here because they're HR


def test_role_escalation_via_input_blocked():
    """You can't pass a role as input to escalate."""
    # The endpoint takes the role from the validated JWT, not from input
    # Verify there's no endpoint that lets you pass a role manually


def test_audit_log_records_query():
    """Every query lands in the audit log."""
    ctx = HRTenantContext(workspace_id="acme", user_id="emp_1", user_role="employee")
    hr_secure_query("benefits", ctx)
    # Verify the log has a record with workspace_id, user_id, role, timestamp


def test_cross_tenant_concurrent():
    """50 tenants running concurrent queries - zero leakage."""
    # Setup: 50 tenants × 100 docs each
    # Run 200 concurrent queries (4 per tenant)
    # Verify no result belongs to the wrong workspace
    ...

Summary and next step

What you learned:

• Isolation by code > by convention. Impossible to skip by design.
• An immutable TenantContext built from the authenticated request, NOT from user input.
• A mandatory secure_query wrapper that ALWAYS includes the workspace_id.
• Automated tests in CI that validate isolation before every merge.
• Pre-commit hooks that catch unsafe direct calls.
• Audit logging for compliance (HIPAA, SOC2, GDPR).
• The traps: workspace_id from a URL, a dict.update bypass, an $or at the root level, a cross-tenant cache.

Checkpoint: before moving on, you should be able to:

• Implement TenantContext and secure_query with guardrails.
• Design an isolation test suite covering cross-tenant, role escalation, concurrency.
• Configure a pre-commit hook that forbids a direct collection.query().

Next capsule: 06 — Time-based + tag filtering.

We covered isolation. Capsule 06 covers the most common operational filters: by date (recency, retention) and by tags (fine-grained sub-segmentation). Practical applications with typical cases.

Resources

1. OWASP Multi-Tenancy — The risks
2. Pinecone — Multi-tenancy Patterns — A comparison
3. SOC2 Trust Services Criteria — Compliance requirements
4. GDPR Article 32 — Security — The legal context
5. Pydantic for Validation — For the tenant context
6. FastAPI Security — JWT and auth

Estimated time: 30-35 minutes Next: 06-time-based-and-tag-filtering.md