A retrieval endpoint can return the wrong collection even when its ranking is correct. Add a collection policy before selecting results, then apply it again when a caller reads a passage. A search handle must not become a way around the current policy.
When you need this. Use this boundary when callers can search different document collections and a later read must remain tied to the identity that issued its handle.
This lab uses original Python API summaries and an explicitly synthetic collection map. The public fixture Bearer tokens select seeded principals; they are not JWTs, secret credentials or real organizational permissions. No model or answer generation runs.
Ranking boundary. This lab authorizes eligibility before selecting results and again before a read, but computes native BM25 statistics over the shared fixture corpus. It is not an isolation test for ranking statistics. The later developer knowledge assistant instead computes statistics over the permitted version and collection; that is a different ranking contract, so its scores cannot be compared directly with these scores.
Resolve the fixture identity before creating tools
The server resolves the Authorization header to a fixture principal. The caller supplies a collection name, not a principal or permission assertion. serialization contains the JSON summary; files contains the file-reading and writing summaries, including their original documentation versions. The fixture policy grants different collection sets to the seeded reader and editor.
POST /search accepts a literal alphanumeric term, collection, exact version and bounded top-k count. The strict schema supplies defaults and rejects extra fields or booleans used as counts. Body admission and strict JSON parsing run before the endpoint.
Unknown identity. A missing or unknown fixture token is refused before tool construction. Denied collection. An authenticated fixture principal still needs a grant for the requested collection. Neither refusal runs native search.
Apply collection eligibility before selecting top-k
The search tool loads the pinned snapshot and executes native SQLite FTS5 ranking. The inherited constructor imposes a small corpus bound. Within that bound, the wrapper fetches every native match, filters the requested collection and documentation version, then selects top-k.
Filtering only after taking top-k can discard an ineligible match while hiding an eligible result below it. The recorded eligible_before_k case searches a term that also matches file summaries but requests only serialization. The regression checks that the allowed JSON summary still occupies the requested slot.
This filter controls returned content. Native BM25 statistics still use the shared fixture corpus; this does not provide tenant-isolated scoring or prevent statistical inference. A larger corpus would need an eligible-query design whose candidate bound remains valid.
Recheck the issued handle before returning source text
Search returns metadata handles without passage text. Each handle binds the fixture principal, collection, policy identity, corpus snapshot, search request and exact passage provenance. The server retains a bounded, expiring authoritative copy. The opaque issued ID locates that copy; it is not an independent authorization credential.
POST /read resolves the current token again. It checks the current grants and exact issued metadata before returning pinned source text. Policy identity is rederived from current trusted policy state. A restart, eviction or expiry can invalidate a handle even if the passage still exists.
Foreign handle. A different principal cannot read a handle issued to the editor, even when both principals can access that collection. Changed handle. Editing a span or other issued field is refused. Revoked grant. Changing the trusted policy cannot preserve a previously authorized read. A stale corpus binding also fails.
Tool failure. A trusted failing factory produces the recorded service error without exposing private exception text. This fixture is server code, not a public request switch.
| Executed case | HTTP status | Outcome |
|---|---|---|
| unknown identity | 401 | unauthenticated |
| collection denied | 403 | collection_denied |
| authorized search | 200 | 2 eligible passage handles |
| authorized read | 200 | source text returned |
| foreign handle | 403 | collection_denied |
| tampered handle | 403 | handle_binding_mismatch |
| eligible before k | 200 | 1 eligible passage handles |
| invalid boolean k | 422 | invalid_request |
| empty matches | 200 | 0 eligible passage handles |
| controlled tool failure | 503 | tool_failure |
Run and inspect the collection boundary
Download the complete authorization lab. It includes the pinned snapshot, synthetic policy, copied retrieval core, exact installed dependency lock and tests. Prepare the virtual environment using the README and requirements.lock.
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.lock"""Synthetic collection authorization, bounded issued handles, no identity claims."""
import copy
import hashlib
import json
import math
import secrets
import time
from collections import OrderedDict
from pathlib import Path
from tools import DocumentationTools, load_snapshot, ToolError
from version_documents import canonical_hash
class AccessError(ValueError):
def __init__(self, code, status=403):self.code=code;self.status=status;super().__init__(code)
class Authorization:
def __init__(self, snapshot=None, policy=None, tool_factory=DocumentationTools, *, clock=time.monotonic, max_handles=64, handle_ttl=60):
self.snapshot=copy.deepcopy(snapshot if snapshot is not None else load_snapshot())
self.policy=copy.deepcopy(policy if policy is not None else json.loads((Path(__file__).parent/'fixtures/policy.json').read_text()))
ids={d['source_id'] for d in self.snapshot['documents']}
if set(self.policy)!= {'schema_version','description','collections','tokens','grants'} or self.policy['schema_version']!=1:raise ValueError('invalid_policy')
collections=self.policy['collections'];grants=self.policy['grants'];tokens=self.policy['tokens']
if not collections or not grants or not tokens:raise ValueError('empty_policy')
for name,sources in collections.items():
if not isinstance(name,str) or not isinstance(sources,list) or not sources or len(set(sources))!=len(sources) or not set(sources)<=ids:raise ValueError('invalid_collection')
for principal,allowed in grants.items():
if not isinstance(principal,str) or not isinstance(allowed,list) or not set(allowed)<=set(collections):raise ValueError('invalid_grants')
if any(not isinstance(t,str) or not t or p not in grants for t,p in tokens.items()):raise ValueError('invalid_tokens')
if type(max_handles) is not int or not 1<=max_handles<=1024 or not isinstance(handle_ttl,(int,float)) or isinstance(handle_ttl,bool) or not math.isfinite(handle_ttl) or not 0<handle_ttl<=3600:raise ValueError('invalid_handle_limits')
self.corpus_id=self.snapshot['snapshot_id'];self.tool_factory=tool_factory
self.clock=clock;self.max_handles=max_handles;self.handle_ttl=handle_ttl;self.issued=OrderedDict();self.search_calls=0;self.read_calls=0
@property
def policy_id(self):return canonical_hash(self.policy)
def resolve(self, authorization):
if not isinstance(authorization,str) or not authorization.startswith('Bearer '):raise AccessError('unauthenticated',401)
token=authorization[7:]
if token not in self.policy['tokens']:raise AccessError('unauthenticated',401)
return self.policy['tokens'][token]
def authorize(self, principal, collection):
if principal not in self.policy['grants']:raise AccessError('unauthenticated',401)
if not isinstance(collection,str) or collection not in self.policy['collections'] or collection not in self.policy['grants'][principal]:raise AccessError('collection_denied')
def _prune(self):
now=self.clock()
for token,entry in list(self.issued.items()):
if entry[0]<=now:del self.issued[token]
def search(self, principal, request):
self.authorize(principal,request['collection'])
# Tool construction and invocation occur only after identity/grant checks.
tools=self.tool_factory(self.snapshot)
try:
self.search_calls+=1
ranked=tools.index.search(request['query'],limit=100)['results']
eligible=[r for r in ranked if r['version']==request['version'] and r['source_id'] in self.policy['collections'][request['collection']]][:request['k']]
request_id=canonical_hash({'principal':principal,'policy_id':self.policy_id,'corpus_id':self.corpus_id,'request':request})
handles=[];self._prune()
for rank,row in enumerate(eligible,1):
source=tools.rows[row['chunk_id']]
handle={'principal':principal,'collection':request['collection'],'policy_id':self.policy_id,'corpus_id':self.corpus_id,'search_request_id':request_id,'rank':rank,'native_score':row['native_score'],**{k:v for k,v in source.items() if k not in ('text','url')},'issued_id':secrets.token_hex(16)}
self.issued[handle['issued_id']]=(self.clock()+self.handle_ttl,copy.deepcopy(handle),copy.deepcopy(source))
while len(self.issued)>self.max_handles:self.issued.popitem(last=False)
handles.append(handle)
return {'principal':principal,'policy_id':self.policy_id,'corpus_id':self.corpus_id,'request':copy.deepcopy(request),'search_request_id':request_id,'results':copy.deepcopy(handles)}
finally:tools.close()
def read(self, principal, handle):
if not isinstance(handle,dict):raise AccessError('invalid_handle')
self.authorize(principal,handle.get('collection'))
self._prune();entry=self.issued.get(handle.get('issued_id')) if isinstance(handle.get('issued_id'),str) else None
if entry is None:raise AccessError('unissued_or_expired_handle')
expected,source=entry[1:]
try:equal=canonical_hash(handle)==canonical_hash(expected)
except (TypeError,ValueError):equal=False
if not equal or handle.get('principal')!=principal or handle.get('policy_id')!=self.policy_id or handle.get('corpus_id')!=self.corpus_id:raise AccessError('handle_binding_mismatch')
if source['source_id'] not in self.policy['collections'][handle['collection']]:raise AccessError('collection_denied')
self.read_calls+=1
body={**copy.deepcopy(handle),'text':source['text'],'url':source['url']}
return {**body,'result_sha256':canonical_hash(body)}python run.py --output runs/my-run/result.json
python -m unittest discover -vunknown_identity: HTTP 401
collection_denied: HTTP 403
authorized_search: HTTP 200
authorized_read: HTTP 200
foreign_handle: HTTP 403
tampered_handle: HTTP 403
eligible_before_k: HTTP 200
invalid_boolean_k: HTTP 422
empty_matches: HTTP 200
controlled_tool_failure: HTTP 503The driver starts actual Uvicorn children on loopback sockets and sends HTTPX requests. Each search creates, uses and closes its SQLite connection on the same event-loop thread. The issued-handle registry lives only in this process. This is not a production identity system, shared session store, CORS policy or deployment recipe.
Change the caller from editor to reader between search and read. Before running the case, predict whether a shared collection grant is enough to reuse the editor’s exact handle. Then compare the source text, handle principal and refusal code. The named tests also cover current-policy changes, the corpus bound and real HTTP body admission.
Inspect the complete saved result
{
"schema_version": 1,
"mode": "actual-loopback-http-public-fixture-authorization",
"model_called": false,
"observations": [
{
"name": "unknown_identity",
"method": "POST",
"path": "/search",
"status_code": 401,
"body": {
"error": {
"code": "unauthenticated"
}
}
},
{
"name": "collection_denied",
"method": "POST",
"path": "/search",
"status_code": 403,
"body": {
"error": {
"code": "collection_denied"
}
}
},
{
"name": "authorized_search",
"method": "POST",
"path": "/search",
"status_code": 200,
"body": {
"principal": "editor",
"policy_id": "0b9b91312bdcd4070944dbba0616396aa12e4d05b8893fc2d3dffdfcf85f2933",
"corpus_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"request": {
"query": "file",
"collection": "files",
"version": "3.12",
"k": 2
},
"search_request_id": "b98a6fd39b4469ca1e9d593ba26760eb9323ca7151a860624d18d77132ea4966",
"results": [
{
"principal": "editor",
"collection": "files",
"policy_id": "0b9b91312bdcd4070944dbba0616396aa12e4d05b8893fc2d3dffdfcf85f2933",
"corpus_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"search_request_id": "b98a6fd39b4469ca1e9d593ba26760eb9323ca7151a860624d18d77132ea4966",
"rank": 1,
"native_score": -1.6120689655172415e-06,
"passage_id": "c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"source_id": "path-read",
"version": "3.12",
"document_revision": "b58385b9bf597a951c16c035653d639929924e53eae3e5ada255afd6ee1592f2",
"source_sha256": "dc097ca63486ace45279f412ea266bf754fcd2f0b1153f5d763ce905dafbc2c0",
"start_line": 1,
"end_line": 1,
"text_sha256": "dc097ca63486ace45279f412ea266bf754fcd2f0b1153f5d763ce905dafbc2c0",
"byte_count": 154,
"issued_id": "1f3d30af4901e605229346c82d152e52"
},
{
"principal": "editor",
"collection": "files",
"policy_id": "0b9b91312bdcd4070944dbba0616396aa12e4d05b8893fc2d3dffdfcf85f2933",
"corpus_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"search_request_id": "b98a6fd39b4469ca1e9d593ba26760eb9323ca7151a860624d18d77132ea4966",
"rank": 2,
"native_score": -1.3203883495145633e-06,
"passage_id": "c66643a4609c6ee157458bf4353f7446b9c6a200a3ef453846d6a18d5b5069a7",
"source_id": "path-write",
"version": "3.12",
"document_revision": "28c6349eaf80b629322edb310bb871dc45a04af9fad3b752bc75d12f0d7314e7",
"source_sha256": "a48b4b9f763cd6309ebe32af19c5a814adcf810c2259683da9f328ccee46948e",
"start_line": 1,
"end_line": 1,
"text_sha256": "a48b4b9f763cd6309ebe32af19c5a814adcf810c2259683da9f328ccee46948e",
"byte_count": 224,
"issued_id": "de5f6d5a226ab7b569c6085d92b5f117"
}
]
}
},
{
"name": "authorized_read",
"method": "POST",
"path": "/read",
"status_code": 200,
"body": {
"principal": "editor",
"collection": "files",
"policy_id": "0b9b91312bdcd4070944dbba0616396aa12e4d05b8893fc2d3dffdfcf85f2933",
"corpus_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"search_request_id": "b98a6fd39b4469ca1e9d593ba26760eb9323ca7151a860624d18d77132ea4966",
"rank": 1,
"native_score": -1.6120689655172415e-06,
"passage_id": "c737db948e8111ae00e3442d8a22c80c876e43446dc1ee176ffe8ce142a41900",
"source_id": "path-read",
"version": "3.12",
"document_revision": "b58385b9bf597a951c16c035653d639929924e53eae3e5ada255afd6ee1592f2",
"source_sha256": "dc097ca63486ace45279f412ea266bf754fcd2f0b1153f5d763ce905dafbc2c0",
"start_line": 1,
"end_line": 1,
"text_sha256": "dc097ca63486ace45279f412ea266bf754fcd2f0b1153f5d763ce905dafbc2c0",
"byte_count": 154,
"issued_id": "1f3d30af4901e605229346c82d152e52",
"text": "Path.read_text(encoding=None, errors=None) opens a file, returns its text as a string and closes the file. Set encoding=\"utf-8\" when reading a UTF-8 file.",
"url": "https://docs.python.org/3.12/library/pathlib.html#pathlib.Path.read_text",
"result_sha256": "afa0d28da92290a84385faba1e50b71201cb1a4c90627afe97acd4556aa8e109"
}
},
{
"name": "foreign_handle",
"method": "POST",
"path": "/read",
"status_code": 403,
"body": {
"error": {
"code": "collection_denied"
}
}
},
{
"name": "tampered_handle",
"method": "POST",
"path": "/read",
"status_code": 403,
"body": {
"error": {
"code": "handle_binding_mismatch"
}
}
},
{
"name": "eligible_before_k",
"method": "POST",
"path": "/search",
"status_code": 200,
"body": {
"principal": "reader",
"policy_id": "0b9b91312bdcd4070944dbba0616396aa12e4d05b8893fc2d3dffdfcf85f2933",
"corpus_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"request": {
"query": "file",
"collection": "serialization",
"version": "3.12",
"k": 1
},
"search_request_id": "d45bff89d14d764520ae0c073a2f7d5e91b918322ff799e888286dfddb228a0b",
"results": [
{
"principal": "reader",
"collection": "serialization",
"policy_id": "0b9b91312bdcd4070944dbba0616396aa12e4d05b8893fc2d3dffdfcf85f2933",
"corpus_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"search_request_id": "d45bff89d14d764520ae0c073a2f7d5e91b918322ff799e888286dfddb228a0b",
"rank": 1,
"native_score": -1.352622061482821e-06,
"passage_id": "e5455737b11664449f653280d89dfc2faaa9a14de16e9c3eb24296ca2ceea51c",
"source_id": "json-text",
"version": "3.12",
"document_revision": "0c4b565b2b1900787d6da9f8539c471f8cb2245c00abf887a57d460364abc57a",
"source_sha256": "0f5efa9669a97f93d5e7e8fe91de80a0d954643947ac9619dba41d3904166288",
"start_line": 1,
"end_line": 1,
"text_sha256": "0f5efa9669a97f93d5e7e8fe91de80a0d954643947ac9619dba41d3904166288",
"byte_count": 199,
"issued_id": "6afdce4328b7a71396113c09bf7b22e6"
}
]
}
},
{
"name": "invalid_boolean_k",
"method": "POST",
"path": "/search",
"status_code": 422,
"body": {
"error": {
"code": "invalid_request"
}
}
},
{
"name": "empty_matches",
"method": "POST",
"path": "/search",
"status_code": 200,
"body": {
"principal": "editor",
"policy_id": "0b9b91312bdcd4070944dbba0616396aa12e4d05b8893fc2d3dffdfcf85f2933",
"corpus_id": "0615d22a3315e97c0a88c6c566c546877304f5a28a97e83f68dd36881d4bbd0e",
"request": {
"query": "zzzz",
"collection": "files",
"version": "3.12",
"k": 2
},
"search_request_id": "ea2156f3143c29769c55518c788f2b3fdaf6dc7ea4264b84db7a5fd4555eb873",
"results": []
}
},
{
"name": "controlled_tool_failure",
"method": "POST",
"path": "/search",
"status_code": 503,
"body": {
"error": {
"code": "tool_failure"
}
}
}
]
}Check the policy boundary
A handle binds a passage to the principal that searched for it. A shared collection grant does not let another principal reuse that handle. Run test_foreign_handle_refused and test_policy_revision_and_revoked_grant_refused to check that both a changed caller and a revoked grant stop the read before source text is returned.
Key takeaways
Authorize the resolved principal before selecting results and again before a read. Exact passage metadata is necessary for source consistency; it does not replace a current grant check.
Best practices
Keep fixture credentials visibly synthetic. Preserve the collection/version filter before top-k, bound issued state, and fail closed on changed identities or bindings. Treat shared ranking statistics and single-process state as explicit limits.
Read the explanation
Read-time authorization and exact issued-handle checks both run before source text is returned.
What’s next
Continue with Cache Retrieval Results.
References
The implementation uses the same pinned FastAPI request and lifespan boundary as Serve the RAG Application with FastAPI. Collection grants, issued-state lifetime and read-time checks are application policies demonstrated by this lab, not built-in production authorization guarantees.