Cápsula 05: Multi-tenant isolation — convertir el aislamiento en regla de plataforma

Descripción de la cápsula

En sistemas RAG multi-tenant, tenant_id no es feature de organización — es obligación de seguridad y compliance. Una query que devuelva un documento del cliente equivocado es un incident reportable: GDPR, HIPAA, SOC2, contractual. No es algo que se puede "arreglar después" — es algo que NUNCA puede pasar.

Esta cápsula te enseña los patrones para garantizar aislamiento técnicamente, no por convención. La convención falla: cualquier dev nuevo puede olvidar el filter, cualquier refactor puede romper la línea que asegura tenant_id, cualquier endpoint nuevo puede saltarse el guardrail. La solución es diseñar el sistema para que sea imposible ejecutar una query sin tenant_id.

Al finalizar esta cápsula serás capaz de:

• ✅ Implementar TenantContext que encapsula identidad y permisos del request
• ✅ Diseñar secure_query como wrapper obligatorio (no se puede saltar)
• ✅ Implementar tests de aislamiento automatizados que corren en CI
• ✅ Configurar pre-commit hooks que detectan llamadas directas inseguras
• ✅ Auditar logs para detectar bypass attempts
• ✅ Anticipar y prevenir las cinco vulnerabilidades comunes de aislamiento

Tiempo estimado: 30-35 minutos

La realidad: convenciones fallan, code design no

Tres formas de "aislamiento" en sistemas multi-tenant, ordenadas de menos a más seguras:

Nivel 1: convención (el dev "se acuerda")

# Documentación dice: "siempre incluir workspace_id en filters"
results = collection.query(
    query_texts=[q],
    where={"workspace_id": user.workspace_id},
)

Falla porque:

• Devs olvidan en endpoints nuevos.
• Refactor accidental remueve la línea.
• Code review no siempre lo detecta.

Nivel 2: linter/test que verifica

# Pre-commit hook que detecta `collection.query(` sin where filter

Falla porque:

• Hooks pueden ignorarse con --no-verify.
• Tests pueden no cubrir todos los paths.
• Regex puede tener falsos negativos.

Nivel 3: imposible saltarlo por design (el approach correcto)

# La función pública requiere tenant_id en signature
def secure_query(query: str, tenant: TenantContext, ...):
    if not tenant.workspace_id:
        raise TenantIsolationError(...)
    where = {"workspace_id": tenant.workspace_id, ...}
    ...

# Y collection.query() NUNCA se llama directamente desde código de aplicación

Funciona porque:

• TypeError si llamas sin tenant_id (Python type checking).
• Excepción explícita si tenant_id es vacío.
• collection.query() directo está prohibido por convención + linter.

Vamos al nivel 3.

TenantContext: encapsular identidad y permisos

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


@dataclass(frozen=True)  # immutable: nadie puede modificar el contexto
class TenantContext:
    """Contexto de seguridad inmutable por 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 es obligatorio")
        if not self.user_id:
            raise TenantIsolationError("user_id es obligatorio")


class TenantIsolationError(Exception):
    """Error explícito para violaciones de aislamiento."""
    pass


# Construir desde request authenticated
def context_from_request(request) -> TenantContext:
    """Extrae TenantContext desde request autenticado."""
    if not request.authenticated:
        raise PermissionError("Request no autenticado")

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

Punto crítico: TenantContext se construye desde request autenticado, no desde inputs del usuario. El workspace_id viene del JWT validado en el middleware, NO de un parámetro de URL.

# ❌ Inseguro: workspace_id desde input
@app.post("/search")
def search(query: str, workspace_id: str):  # ← workspace_id puede ser controlado por atacante
    ctx = TenantContext(workspace_id=workspace_id, ...)

# ✅ Seguro: workspace_id desde token validado
@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: wrapper obligatorio

# 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,
):
    """
    Wrapper obligatorio para todas las queries al vector DB.

    Garantiza:
    - workspace_id siempre presente en filter
    - additional_filters NO puede sobreescribir workspace_id
    - Logging de cada query para auditoría
    - Raise explícito si algo está mal
    """
    # Validar TenantContext
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("TenantContext con workspace_id obligatorio")

    # Construir filter con workspace_id forzado
    where = _build_secure_where(tenant, additional_filters)

    # Logging para auditoría (sin contenido de la query — 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,
        }
    )

    # Ejecutar 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:
    """Construye where blindando workspace_id."""
    base = {"workspace_id": tenant.workspace_id}

    if not additional_filters:
        return base

    # Validar que additional_filters NO intenta sobreescribir workspace_id
    if "workspace_id" in additional_filters:
        raise TenantIsolationError(
            "additional_filters NO puede contener workspace_id. "
            "Es controlado por TenantContext."
        )

    # Si hay $or en additional_filters, asegurar que no relaja tenant
    _validate_no_tenant_bypass(additional_filters)

    # Combinar con $and explícito
    return {"$and": [base, additional_filters]}


def _validate_no_tenant_bypass(filters: dict) -> None:
    """Recursivamente verifica que no haya $or que escape del tenant."""
    if not isinstance(filters, dict):
        return

    for key, value in filters.items():
        if key == "$or" and isinstance(value, list):
            # Cada elemento del $or debe estar dentro del workspace
            # (esto se asegura por el $and externo, pero validamos contenido)
            for item in value:
                _validate_no_tenant_bypass(item)
        elif isinstance(value, dict):
            _validate_no_tenant_bypass(value)

Uso correcto

# 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"},  # opcional
        n_results=5,
    )
    return results

Tests de aislamiento automatizados

Los tests de aislamiento son non-negotiable. Deben correr en CI antes de cada deploy:

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


@pytest.fixture
def collection_with_two_tenants():
    """Setup: collection con docs de tenant A y B."""
    client = chromadb.Client()
    collection = client.create_collection("test_isolation")

    # 100 docs de 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 de 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 NUNCA debe ver docs de tenant B."""
    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 vio doc de {meta['workspace_id']}"
        )


def test_empty_workspace_id_raises_error():
    """Sin workspace_id no se puede crear 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 NO puede sobreescribir 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"},  # intent de bypass
        )


def test_or_filter_still_respects_tenant(collection_with_two_tenants):
    """$or filters no pueden escapar del 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,
    )

    # Aún con $or interno, todos deben ser tenant_a
    for meta in results["metadatas"][0]:
        assert meta["workspace_id"] == "tenant_a"


def test_concurrent_tenants_no_leakage(collection_with_two_tenants):
    """Queries concurrentes de tenants distintos no se cruzan."""
    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}"

Estos tests deben:

• Correr en cada PR (CI).
• Bloquear merges si fallan.
• Auditarse cuando se agregan tests nuevos.

Pre-commit hook anti-bypass

# scripts/check_no_direct_query.py
"""
Pre-commit hook: detecta llamadas directas a collection.query()
sin pasar por secure_query.
"""
import re
import sys
from pathlib import Path

PATTERN = re.compile(r'\.query\(')
ALLOWED_FILES = {"src/secure_query.py", "tests/"}  # solo aquí permite

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)

Configurar en .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$

Auditoría de logs

Si tu sistema sirve datos sensibles (médico, financiero, legal), considera audit logging:

# audit_log.py
import json
from datetime import datetime


def log_query_for_audit(tenant: TenantContext, query: str, n_results: int):
    """Logs estructurados para 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),  # NO el query mismo (privacy)
        "n_results": n_results,
    }
    audit_logger.info(json.dumps(audit_record))


# Integrar en secure_query
def secure_query(...):
    log_query_for_audit(tenant, query_text, n_results)
    # ... resto

Beneficios:

• Compliance trail (HIPAA, SOC2 lo requieren).
• Detección de comportamiento anómalo (un user_id consultando 1000 queries/seg = bot).
• Debugging post-incident.

Trampas y errores comunes

Trampa 1: workspace_id desde URL parameter

Cubierta arriba. Atacante puede modificar el parameter para acceder a otros tenants.

Trampa 2: dict.update() sobreescribe workspace_id

# ❌ Mal
where = {"workspace_id": "tenant_a"}
where.update(user_filters)  # si user_filters tiene workspace_id, lo sobreescribe

Síntoma: filter del usuario gana sobre el del sistema.

Cómo prevenir: validar antes (visto en _build_secure_where).

Trampa 3: bypass con $or al nivel raíz

# ❌ Mal
where = {
    "$or": [
        {"workspace_id": "tenant_a"},  # ← user puede meter otra
        {"workspace_id": "tenant_b"},
    ]
}

Síntoma: filter parece tener tenant_a, pero también incluye tenant_b.

Cómo prevenir: wrapping con $and. tenant_id va al nivel raíz, $or va dentro como condition adicional.

Trampa 4: cache cross-tenant

# ❌ Mal
@lru_cache(maxsize=1000)
def cached_query(query_text):
    # cache key es solo el query → resultados se comparten entre tenants
    return collection.query(...)

Cómo prevenir: incluir workspace_id en la 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})

Trampa 5: scripts admin sin contexto

# ❌ Mal: script de migración que olvida tenant
all_docs = collection.get()  # cross-tenant data

Cómo prevenir: scripts admin requieren AdminContext explícito + audit log que indique "intencional cross-tenant access para tarea X".

Ejercicio aplicado

Escenario: eres AI Engineer en una plataforma SaaS de RRHH. Datos sensibles: salarios, evaluaciones, datos personales.

Stakeholders piden:

• Aislamiento estricto entre 50 empresas (workspaces).
• Cada empresa tiene roles: admin, hr_manager, manager, employee.
• Algunos docs son confidential (solo HR), otros public_to_company, otros team_only.
• Compliance: GDPR + SOC2 audit en 3 meses.

Tu trabajo:

1. Diseña el sistema de aislamiento + permisos.
2. Define los tests de aislamiento automatizados.
3. Plan de auditoría para SOC2.

# rrhh_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 y user_id obligatorios")


def hr_secure_query(query: str, tenant: HRTenantContext, n_results: int = 5):
    """Query con visibility según rol."""

    visibility_clauses = []

    # 1. Docs públicos de la empresa - todos los roles los ven
    visibility_clauses.append({"visibility": "public_to_company"})

    # 2. Docs team_only - solo si el user es del team
    if tenant.team_ids:
        visibility_clauses.append({
            "$and": [
                {"visibility": "team_only"},
                {"team_id": {"$in": tenant.team_ids}},
            ]
        })

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

    # 4. Datos personales del propio empleado
    visibility_clauses.append({
        "$and": [
            {"visibility": "personal"},
            {"owner_id": tenant.user_id},
        ]
    })

    where = {
        "$and": [
            {"workspace_id": tenant.workspace_id},  # tenant isolation
            {"$or": visibility_clauses},             # visibility según rol
        ]
    }

    # 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():
    """Employee NO ve docs confidential."""
    ctx = HRTenantContext(
        workspace_id="acme",
        user_id="emp_1",
        user_role="employee",
    )
    # Setup: insertar un doc confidential en tenant acme
    results = hr_secure_query("salary", ctx)
    for meta in results["metadatas"][0]:
        assert meta["visibility"] != "confidential"


def test_employee_sees_only_own_personal():
    """Employee NO ve datos personales de otros."""
    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 ve confidential pero solo de su workspace."""
    ctx = HRTenantContext(workspace_id="acme", user_id="hr_1", user_role="hr_manager")
    # Insertar confidential en acme y en otro 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 está OK porque es HR


def test_role_escalation_via_input_blocked():
    """No se puede pasar role como input para escalar."""
    # Endpoint recibe role del JWT validado, no de input
    # Verificar que no hay endpoint que permita pasar role manualmente


def test_audit_log_records_query():
    """Cada query queda en audit log."""
    ctx = HRTenantContext(workspace_id="acme", user_id="emp_1", user_role="employee")
    hr_secure_query("benefits", ctx)
    # Verificar que el log tiene un record con workspace_id, user_id, role, timestamp


def test_cross_tenant_concurrent():
    """50 tenants haciendo queries concurrentes - cero leakage."""
    # Setup: 50 tenants × 100 docs cada uno
    # Ejecutar 200 queries concurrentes (4 por tenant)
    # Verificar que ningún resultado está en workspace incorrecto
    ...

Resumen y siguiente paso

Lo que aprendiste:

• Aislamiento por código > convención. Imposible saltarlo por design.
• TenantContext inmutable construido desde request autenticado, NO desde inputs del usuario.
• secure_query wrapper obligatorio que SIEMPRE incluye workspace_id.
• Tests automatizados en CI que validan isolation antes de cada merge.
• Pre-commit hooks que detectan llamadas directas inseguras.
• Audit logging para compliance (HIPAA, SOC2, GDPR).
• Trampas: workspace_id desde URL, dict.update bypass, $or al nivel raíz, cache cross-tenant.

Checkpoint: antes de avanzar, deberías poder:

• Implementar TenantContext y secure_query con guardrails.
• Diseñar suite de tests de isolation que cubran cross-tenant, role escalation, concurrencia.
• Configurar pre-commit hook que prohíbe collection.query() directo.

Siguiente cápsula: 06 — Time-based + tag filtering.

Cubrimos isolation. La cápsula 06 cubre los filtros operacionales más comunes: por fecha (recencia, retention) y por tags (sub-segmentación fina). Aplicaciones prácticas con casos típicos.

Recursos

1. OWASP Multi-Tenancy — Riesgos
2. Pinecone — Multi-tenancy Patterns — Comparación
3. SOC2 Trust Services Criteria — Requisitos compliance
4. GDPR Article 32 — Security — Legal context
5. Pydantic for Validation — Para tenant context
6. FastAPI Security — JWT y auth

Tiempo estimado: 30-35 minutos Siguiente: 06-time-based-and-tag-filtering.md