Phase 4: RAG & Agent Security · 55 min · Python · Vector databases · Row-level security
The Problem
Most RAG systems that serve multiple customers from shared infrastructure store every tenant's documents in the same vector index, tagged with a tenant_id field. That's a reasonable cost and operational decision — running a separate vector database per customer doesn't scale. But it means the isolation between Customer A's confidential contract and Customer B's support ticket is enforced entirely in application code, not by the storage layer. If that application-level check has a gap, cosine similarity does not know or care whose document it's ranking.
This is structurally the same bug class as SQL injection or missing row-level security in a traditional multi-tenant database — a query that should be scoped to one tenant's rows instead runs against the whole table. The difference in RAG systems is that the "table scan" is a similarity search, so the bug doesn't look like an obvious SQL statement missing a WHERE tenant_id = ? clause. It looks like a search(query, top_k=5) function that seems to work correctly in every test that happens to only use one tenant's data.
The most common real-world version of this bug: a tenant filter exists somewhere in the codebase — maybe on the primary user-facing search endpoint — but a second code path (an internal debug tool, a new feature, an admin panel, a background job) calls the underlying ranking function directly and nobody re-adds the filter. The vulnerability isn't "we forgot access control exists." It's "access control was bolted onto one entry point instead of built into the retrieval primitive itself."
Where the filter belongs
There are two ways to add a tenant filter to a vector search, and only one of them is actually secure:
Filter after ranking (post-filter). Run the similarity search across all documents, get back the global top-k, then discard results that don't belong to the requesting tenant. This is the pattern that produces both a security bug and a correctness bug: if the top-k happens to be dominated by another tenant's high-scoring documents, you can end up with zero or too few results after filtering — and worse, any code path that skips the filter step gets full cross-tenant access with no error, no exception, nothing that looks broken.
Filter before ranking (pre-filter / row-level security). Restrict the candidate set to only documents the requester is authorized to see, and only then compute similarity scores and take the top-k. A document from another tenant is never scored, never ranked, never a candidate — it structurally cannot appear in the output, regardless of how well it would have matched the query. This should be enforced inside the store's search() method itself, taking the requesting identity as a required argument, not as an optional parameter a caller can omit.
Isolation bugs are boring until they're catastrophic
Nobody designs a system intending to leak Customer A's contract to Customer B. It happens because a tenant_id filter that lived in one place (the main API handler) didn't travel with the retrieval function when it got reused elsewhere (an internal tool, a new agent, a batch export job). The fix isn't "remember to filter" — it's making the retrieval function impossible to call without specifying who's asking and enforcing eligibility as part of the function, not as a convention every caller has to remember.
Role-based document permissions within a tenant
Tenant isolation is the outer boundary; most real systems also need finer-grained permissions inside a single tenant. Not every document a company owns should be visible to every employee of that company — an internal product roadmap might be admin-only while a getting-started guide is visible to every member. This is the same principle as tenant isolation applied one level down: allowed_roles on the document, checked as part of the same pre-ranking filter, not as a separate pass.
A support engineer at Company A queries the shared knowledge base and gets back a document belonging to Company B that scored highly for their query. Investigation shows the vector store has a tenant_id field on every document, and the main customer-facing search endpoint does filter by it correctly. What's the most likely root cause?
This is the single most common real-world tenant isolation bug: the filter exists, and it works, on the entry point everyone tests — but it's implemented as a wrapper or a post-processing step rather than being inside the retrieval primitive itself. Any new code path that calls the underlying search function directly inherits zero isolation, with no error or warning. A broken embedding model or storage failure wouldn't selectively leak one tenant's data to another in a way that correlates with query relevance; malicious poisoning is a real attack but doesn't explain why the leaked document is a real, unmodified Company B document rather than an attacker-crafted one.
Detecting and auditing isolation failures
Fixing the bug isn't the end of the job — a security review needs evidence the fix holds, and production needs a way to detect if it regresses. Two concrete controls close this loop:
- Log every access decision, not just denials. For each retrieval, record the requesting tenant/role, the document's tenant/role requirement, and whether access was granted — before ranking runs. This produces the audit trail an AI security review actually asks for: "show me every case where a cross-tenant document was excluded from candidates in the last 30 days," not "trust that the code is correct."
- Test with adversarial tenants, not just happy-path tenants. A test suite that only ever queries within a single tenant's data will never exercise the isolation boundary at all. Isolation tests need to deliberately query as Tenant B for content that scores well against Tenant A's documents, and assert zero cross-tenant results — this is the same idea as a negative test case in traditional AppSec.
Your isolation test suite queries only within each tenant's own documents and always passes. A production incident later reveals Tenant B could retrieve Tenant A's documents. What was structurally wrong with the test suite, independent of the code bug itself?
A test suite that only ever queries within a single tenant's own data cannot distinguish "isolation works" from "isolation code doesn't exist" — both produce the same passing result, because no cross-tenant query is ever attempted. This is a structural blind spot, not a flakiness or performance issue. The fix is adding adversarial isolation tests: deliberately query as one tenant using terms crafted to score well against another tenant's real documents, and assert the result set is empty of cross-tenant content. This is the retrieval-layer equivalent of a negative security test case.
Build
Build it
The code artifact lesson_04_02_tenant_isolation.py builds directly on lesson 1's vector store and adds document-level ACLs:
Document gains tenant_id and allowed_roles fieldsAccessAttempt: an audit record logged for every retrieval decision, granted or deniedNaiveMultiTenantStore: has a tenant_id field on every document but the ranking path (search_unsafe) doesn't filter by it, and search_forgot_to_filter simulates a caller that skips enforcement — the exact bug pattern described aboveSecureMultiTenantStore: enforces _is_authorized() (tenant + role) as part of search() itself, before candidates are ranked, and logs every decisiondemo_isolation_bypass(): a Globex user retrieves Acme's confidential contract terms through the naive storedemo_fixed_isolation(): the same query against the secure store returns zero cross-tenant results, with denied attempts logged, plus a role-based check showing an admin-only roadmap document is hidden from a member
"""
lesson_04_02_tenant_isolation.py
Extends the lesson 1 vector store with multi-tenant document-level access
control. Demonstrates a tenant isolation bypass in a naive implementation
(filter applied after the fact, or not at all) and the fix (filter enforced
as part of the search itself, with denied-access attempts logged).
No API keys, no network calls. Uses the same deterministic hash embedding
as lesson 1 so the lab runs instantly and standalone.
Usage:
python lesson_04_02_tenant_isolation.py
"""
import hashlib
import math
import re
from dataclasses import dataclass, field
VECTOR_DIM = 64
# ── Embeddings (same deterministic stand-in as lesson 1) ───────────────────
def embed(text: str) -> list[float]:
vec = [0.0] * VECTOR_DIM
tokens = re.findall(r"[a-z0-9]+", text.lower())
for tok in tokens:
h = int(hashlib.sha256(tok.encode()).hexdigest(), 16)
idx = h % VECTOR_DIM
sign = 1.0 if (h // VECTOR_DIM) % 2 == 0 else -1.0
vec[idx] += sign
norm = math.sqrt(sum(v * v for v in vec)) or 1.0
return [v / norm for v in vec]
def cosine(a: list[float], b: list[float]) -> float:
return sum(x * y for x, y in zip(a, b))
# ── Document with tenant / ACL metadata ─────────────────────────────────────
@dataclass
class Document:
doc_id: str
text: str
tenant_id: str
allowed_roles: set[str] = field(default_factory=lambda: {"member"})
embedding: list[float] = field(default_factory=list)
@dataclass
class AccessAttempt:
"""Audit record for every search — this is the log a security review
needs to prove isolation actually holds, not just that it's intended."""
query: str
requesting_tenant: str
requesting_role: str
doc_id: str
doc_tenant: str
allowed: bool
reason: str
# ── Naive store: filters AFTER computing global top-k (the vulnerable path) ─
class NaiveMultiTenantStore:
"""Looks correct at a glance — it does check `tenant_id` — but the check
happens in the wrong place: after global ranking, and only as a client-side
filter the caller can choose to skip. This mirrors a real bug pattern: an
application-layer 'if doc.tenant_id == current_tenant' filter added late,
bolted onto a search function that was never designed to be tenant-aware."""
def __init__(self):
self.documents: dict[str, Document] = {}
self.access_log: list[AccessAttempt] = []
def add_document(self, doc_id, text, tenant_id, allowed_roles=None):
doc = Document(doc_id, text, tenant_id, allowed_roles or {"member"}, embed(text))
self.documents[doc_id] = doc
def search_unsafe(self, query: str, top_k: int = 3) -> list[tuple[Document, float]]:
"""The bug: ranks across ALL tenants' documents with no filter at all.
A caller has to remember to filter afterward — and in this store,
nothing enforces that they do."""
q_vec = embed(query)
scored = [(doc, cosine(q_vec, doc.embedding)) for doc in self.documents.values()]
scored.sort(key=lambda pair: pair[1], reverse=True)
return scored[:top_k]
def search_forgot_to_filter(self, query: str, requesting_tenant: str, requesting_role: str,
top_k: int = 3) -> list[tuple[Document, float]]:
"""Simulates the common real-world mistake: a developer calls the
underlying ranking function directly (for a new feature, a debug
endpoint, an internal admin tool) and never applies the tenant filter
that exists elsewhere in the codebase."""
results = self.search_unsafe(query, top_k=top_k)
for doc, score in results:
allowed = doc.tenant_id == requesting_tenant
self.access_log.append(AccessAttempt(
query, requesting_tenant, requesting_role, doc.doc_id, doc.tenant_id,
allowed, "returned without enforcement" if not allowed else "same tenant",
))
return results # returned unfiltered regardless of tenant match
# ── Fixed store: tenant + role filter is inside the search path itself ──────
class SecureMultiTenantStore:
"""The filter is not a helper you can forget to call — it's inside
`search()` itself. Every retrieval requires a requesting tenant and role,
and access decisions are logged whether they succeed or are denied."""
def __init__(self):
self.documents: dict[str, Document] = {}
self.access_log: list[AccessAttempt] = []
def add_document(self, doc_id, text, tenant_id, allowed_roles=None):
doc = Document(doc_id, text, tenant_id, allowed_roles or {"member"}, embed(text))
self.documents[doc_id] = doc
def _is_authorized(self, doc: Document, tenant_id: str, role: str) -> bool:
return doc.tenant_id == tenant_id and role in doc.allowed_roles
def search(self, query: str, requesting_tenant: str, requesting_role: str,
top_k: int = 3) -> list[tuple[Document, float]]:
"""Row-level security applied BEFORE ranking: only documents the
caller is authorized to see are ever scored and considered for top_k.
A document from another tenant literally never enters the candidate
set — it can't leak through a ranking edge case because it was never
a ranking candidate."""
q_vec = embed(query)
candidates = []
for doc in self.documents.values():
authorized = self._is_authorized(doc, requesting_tenant, requesting_role)
self.access_log.append(AccessAttempt(
query, requesting_tenant, requesting_role, doc.doc_id, doc.tenant_id,
authorized, "authorized" if authorized else "tenant/role mismatch — excluded pre-ranking",
))
if authorized:
candidates.append((doc, cosine(q_vec, doc.embedding)))
candidates.sort(key=lambda pair: pair[1], reverse=True)
return candidates[:top_k]
# ── Demo data: two tenants sharing infrastructure ────────────────────────────
TENANT_A_DOCS = [
("acme-contract-001", "Acme Corp master services agreement: total contract value $2.4M annually, "
"renewal date March 2027, includes SLA penalty clauses for downtime over 99.9%.",
"tenant-acme", {"member", "admin"}),
("acme-roadmap-002", "Acme Corp Q3 product roadmap: launching the enterprise SSO feature in August, "
"internal target date, not yet communicated to customers.", "tenant-acme", {"admin"}),
]
TENANT_B_DOCS = [
("globex-faq-001", "Globex Inc getting started guide: connect your account via the integrations "
"page and invite team members from settings.", "tenant-globex", {"member", "admin"}),
]
def demo_isolation_bypass() -> None:
print("=" * 70)
print("ATTACK: tenant isolation bypass — filter forgotten in the search path")
print("=" * 70)
store = NaiveMultiTenantStore()
for doc_id, text, tenant, roles in TENANT_A_DOCS + TENANT_B_DOCS:
store.add_document(doc_id, text, tenant, roles)
# A Globex user (tenant-globex, role=member) asks a question that happens
# to rank well against Acme's contract document, because the shared index
# has no tenant boundary in the ranking path.
query = "What is our contract renewal date and penalty terms?"
results = store.search_forgot_to_filter(query, requesting_tenant="tenant-globex",
requesting_role="member", top_k=3)
print(f"\nQuery from tenant-globex (role=member): {query!r}")
print("\nResults returned:")
leaked = False
for doc, score in results:
cross_tenant = doc.tenant_id != "tenant-globex"
if cross_tenant:
leaked = True
flag = " <-- CROSS-TENANT LEAK" if cross_tenant else ""
print(f" [{score:.3f}] {doc.doc_id} (tenant={doc.tenant_id}){flag}")
print(f"\nCross-tenant document leaked to tenant-globex: {leaked}")
if leaked:
print(" -> A Globex user just received Acme's contract terms. The 'tenant_id' field\n"
" existed on the document the whole time — nothing in the search path checked it.")
def demo_fixed_isolation() -> None:
print("\n" + "=" * 70)
print("DEFENSE: row-level filter enforced inside search(), before ranking")
print("=" * 70)
store = SecureMultiTenantStore()
for doc_id, text, tenant, roles in TENANT_A_DOCS + TENANT_B_DOCS:
store.add_document(doc_id, text, tenant, roles)
query = "What is our contract renewal date and penalty terms?"
results = store.search(query, requesting_tenant="tenant-globex", requesting_role="member", top_k=3)
print(f"\nQuery from tenant-globex (role=member): {query!r}")
print(f"\nResults returned: {len(results)}")
for doc, score in results:
print(f" [{score:.3f}] {doc.doc_id} (tenant={doc.tenant_id})")
if not results:
print(" (no results — tenant-globex has no documents relevant to this query, "
"and Acme's documents were never candidates)")
denied = [a for a in store.access_log if not a.allowed]
print(f"\nDenied access attempts logged: {len(denied)}")
for a in denied:
print(f" DENIED query={a.query!r} doc={a.doc_id} (tenant={a.doc_tenant}) reason={a.reason}")
# Role check: an Acme member (not admin) should not see the internal roadmap.
print("\n--- role-level check within the same tenant ---")
member_results = store.search("what's shipping next quarter", requesting_tenant="tenant-acme",
requesting_role="member", top_k=3)
admin_results = store.search("what's shipping next quarter", requesting_tenant="tenant-acme",
requesting_role="admin", top_k=3)
print(f"tenant-acme member sees roadmap doc: {any(d.doc_id == 'acme-roadmap-002' for d, _ in member_results)}")
print(f"tenant-acme admin sees roadmap doc: {any(d.doc_id == 'acme-roadmap-002' for d, _ in admin_results)}")
if __name__ == "__main__":
demo_isolation_bypass()
demo_fixed_isolation()