Why RAG Needs Document-Level Access Control at Query Time

Your RAG demo goes great until someone in sales asks the chatbot about the restructuring plan and it answers. The vector store did nothing wrong. It ranks chunks by similarity, and the HR folder was very similar to the question.

RAG access control has to run at query time, on every search, using the caller's identity. By the end you'll have a permission-aware retrieval query, a fair answer to "aren't per-user indexes simpler?", and a list of the places leaks hide after the filter is in.

Why RAG access control can't live in the prompt

If you index everything into one store, retrieval hands the model whatever matches best, and the model decides what to show. A system prompt line like "never reveal confidential documents" is a request, not a control. Prompt injection is the art of talking a model out of its instructions.

OWASP covers this under Vector and Embedding Weaknesses in its 2025 LLM Top 10. It warns about context leakage between users in shared stores and recommends permission-aware vector stores.

The rule is simple. A chunk the caller may not read should never leave the database, so the model has nothing to leak.

Add permission-aware retrieval with metadata filters

Store the groups allowed to read each chunk next to its embedding, and filter on them in the same query that runs the similarity search. This example uses Postgres with pgvector, but the idea is the same in any store with vector store metadata filtering.

CREATE TABLE chunks (
  id             bigserial PRIMARY KEY,
  doc_id         text NOT NULL,
  content        text NOT NULL,
  embedding      vector(1536) NOT NULL,
  allowed_groups text[] NOT NULL
);

CREATE INDEX ON chunks USING gin (allowed_groups);
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);

This gives every chunk an allowed_groups array, a GIN index for the permission filter and an HNSW index for the vector search. Match the vector(1536) size to your embedding model.

import psycopg

SEARCH_SQL = '''
SELECT doc_id, content
FROM chunks
WHERE allowed_groups && %(groups)s::text[]
ORDER BY embedding <=> %(query)s::vector
LIMIT 5
'''

def retrieve(conn: psycopg.Connection, query_embedding: list[float], caller_groups: list[str]):
    vector = '[' + ','.join(str(x) for x in query_embedding) + ']'
    with conn.transaction(), conn.cursor() as cur:
        cur.execute('SET LOCAL hnsw.iterative_scan = strict_order')
        cur.execute(SEARCH_SQL, {'groups': caller_groups, 'query': vector})
        return cur.fetchall()

Three details matter. The filter sits in the WHERE clause, not in application code after the top 5 come back, so a restricted chunk never takes a slot. caller_groups comes from your identity provider's verified token or a server-side directory lookup, never from the request body. And an empty list matches nothing, so a missing identity fails closed.

Azure AI Search documents the same pattern with a group_ids/any(g:search.in(g, '...')) filter. It is blunt that the principal is just a string, so authenticating the caller is your job.

One pgvector gotcha: with an HNSW index, the README says the filter is applied after the index scan. At the default hnsw.ef_search of 40, a filter matching 10% of rows returns about 4 rows on average, even with LIMIT 10. Since 0.8.0, the hnsw.iterative_scan setting keeps scanning until enough rows match, capped by hnsw.max_scan_tuples (default 20,000). Skip it and users with narrow permissions get thin answers and blame the model.

Are per-user indexes simpler than document permissions?

The strongest objection: skip the filter, build one index per user, and isolation is structural. It's a fair instinct. For a few hard-isolated tenants, like separate customers or legal entities, a collection per tenant is the right call. Use that as the outer wall and keep group filters inside it.

Per user, it falls apart. Say a handbook is readable by 400 employees. Per-user indexes embed and store it 400 times, and when HR changes who may read it, you rewrite 400 copies. Company permissions are group-based, nested and always changing, so you'd mirror every change into every index. That doesn't remove the stale-permission problem, it multiplies it. With a filter, the document is one row and the permission is one column.

Where RAG data leakage hides after the filter

Filtered search is necessary, not sufficient. Leaks hide wherever you store or reuse output derived from restricted content.

Shared caches

A semantic cache returns a stored answer when a new question resembles an old one. If the first asker was in HR, the sales user who asks next gets an HR-grade answer without touching your filter. Put the permission scope in the cache key:

import hashlib

def cache_key(query: str, caller_groups: list[str]) -> str:
    scope = hashlib.sha256('|'.join(sorted(caller_groups)).encode()).hexdigest()
    return scope + ':' + hashlib.sha256(query.encode()).hexdigest()

Two callers with different group sets now never share an entry. The cost is a lower hit rate, which is the price of correctness.

Summaries and derived data

Document summaries, context blurbs added at ingest, folder digests and conversation memory are all new text generated from source content. Each one inherits the access of its sources. A summary of documents A and B should be visible only to people who can read both, so store the intersection of their groups in allowed_groups and filter on it like any chunk.

Stale permission copies

Copying ACLs into metadata at ingest creates a second source of truth. Someone leaves a group or a file gets unshared, and the index still says yes.

Push ACL changes from the source system as events and update allowed_groups in place, with no re-embedding needed. Expire cached group memberships in minutes, not days. For highly sensitive sources, re-check the top few hits against the source system before they reach the prompt.

Start with a test that tries to leak

Before refactoring anything, write the test that fails today. Seed one restricted chunk, then query as someone who shouldn't see it:

def test_restricted_chunk_never_reaches_other_groups(conn, embed):
    # fixture seeded doc 'layoff-plan' with allowed_groups = ARRAY['hr']
    query = embed('Who is affected by the restructuring?')

    hr_docs = {doc_id for doc_id, _ in retrieve(conn, query, ['hr'])}
    sales_docs = {doc_id for doc_id, _ in retrieve(conn, query, ['sales'])}

    assert 'layoff-plan' in hr_docs
    assert 'layoff-plan' not in sales_docs
    assert retrieve(conn, query, []) == []

embed is whatever function produces your query vectors. The first assertion proves the data is really indexed, so the second one means something. Then repeat the same two-user check against your cache and your summaries, because that's where it will fail first.

In production, the only access control you can trust is the one a test can prove, and query-time filtering is the one you can test.