Build Tenant-Isolated Vector Search

Build a Qdrant gateway that prevents cross-tenant vector search leaks.

Introduction

30 Second Summary

A search result can look perfectly relevant while quietly exposing information that belongs to someone else. In a shared system, one missing access rule can turn a useful feature into a data leak.

In this project, you will build a local two-tenant vector search lab with Qdrant plus a FastAPI gateway. You will reproduce a cross-tenant leak before enforcing the tenant boundary with automated tests.

What You'll Build

Picture sending the same search through your local API with two demo keys and seeing each response return only the documents assigned to that tenant.

By the end of this project, you'll have:

  • A visible cross-tenant data leak where one unfiltered search returns documents belonging to Acme plus Globex.
  • A protected /search endpoint that derives tenant identity from a demo API key before returning matching documents.
  • An automated security test suite that reports five passing checks for tenant access, rejected credentials, plus override resistance.
  • Secret Mission: Add a third tenant to the shared collection without creating another search route or weakening isolation.

Are there any prerequisites?

You need a supported 64-bit Windows 10 or Windows 11 machine with hardware virtualization enabled.

Your machine needs WSL 2.1.5 or later plus at least 8 GB of RAM.

You also need permission to install Docker Desktop 4.93.0.

Before We Start

This checkpoint sets the security boundary before any setup begins. Acme responses must exclude Globex and every other tenant because the gateway enforces a tenant filter from trusted API-key-derived context.

Set Up Qdrant and the Python Environment

Your security rule becomes testable once the lab has a real Qdrant server to query. Running it through Docker Desktop gives you a visible database boundary that later searches can cross or protect.

The gateway also needs a Python virtual environment with exactly pinned packages. This step makes both sides reproducible before you create any tenant data.

In this step, get ready to:
  • Install Docker Desktop 4.93.0 with its WSL 2 backend.
  • Start Qdrant 1.19.1 through Docker Compose with persistent local storage.
  • Prepare the isolated Python environment with every pinned dependency.
Install Docker Desktop and verify its engine

Docker Desktop provides the engine that runs Qdrant as a local server. The WSL 2 backend supplies the Linux environment needed by the Qdrant container.

Check Docker Desktop licensing

Docker Desktop is free for personal use and education. It is also free for non-commercial open source work.

Larger-enterprise commercial use requires a paid subscription. Government use also requires a paid subscription.

Choose the tab that matches the Docker Desktop state on your Windows machine.

✔️ I have Docker Desktop 4.93.0

Your installed version already matches the lab. Confirm its installation settings before checking the engine.

  • Confirm Docker Desktop is installed for your current Windows account.
  • Confirm the WSL 2 backend is enabled in Docker Desktop.
  • Press the Windows key to open Windows Search.
  • Type Docker Desktop into Windows Search.
  • Press Enter to start Docker Desktop.
  • Keep Docker Desktop running while you complete the lab.

ⓧ I have an older Docker Desktop

The installer may request Windows approval. It may also request a restart.

  • Open the official Docker Desktop installation guide in your browser.
  • Download the Windows installer for Docker Desktop 4.93.0.
  • Launch the downloaded installer for your current Windows account.
  • Keep the WSL 2 backend enabled during installation.
  • Complete the installer prompts.
  • Press the Windows key to open Windows Search.
  • Type Docker Desktop into Windows Search.
  • Press Enter to start the updated Docker Desktop installation.

ⓧ Docker Desktop is not installed

The installer may request Windows approval. It may also request a restart.

  • Open the official Docker Desktop installation guide in your browser.
  • Download the Windows installer for Docker Desktop 4.93.0.
  • Launch the downloaded installer for your current Windows account.
  • Keep the WSL 2 backend enabled during installation.
  • Complete the installer prompts.
  • Press the Windows key to open Windows Search.
  • Type Docker Desktop into Windows Search.
  • Press Enter to start Docker Desktop.
  • Press the Windows key to open Windows Search.
  • Type PowerShell into Windows Search.
  • Press Enter to open PowerShell.
  • Verify that the Docker client can reach the running engine by running this command:
docker version

What does this check prove?

The docker version command asks both the local Docker client and the Docker engine for version details. Seeing both sections proves that the engine is available.

You should see separate Client and Server details. Your Docker engine is now ready to run Qdrant containers.

Missing the Server details?

  • Confirm Docker Desktop is still running.
  • Wait for the Docker engine to finish starting.
  • Run docker version again from the same PowerShell window.

Ask for help with the engine connection: Help me diagnose why docker version does not show Server details on Windows.

Start Qdrant with Docker Compose

The Compose file describes the Qdrant container as reproducible configuration. It also limits the database port to your Windows machine.

  • Press the Windows key to open Windows Search.
  • Type VS Code into Windows Search.
  • Press Enter to open VS Code.
  • Click File in the top menu bar.
  • Choose Open Folder from the File menu.
  • Select your Desktop in the folder dialog.
  • Create a folder named tenant-search-lab on your Desktop.
  • Select tenant-search-lab as the folder to open.

Your project workspace

The Explorer sidebar should now show tenant-search-lab as the open folder. Every lab file stays inside this workspace.

  • Click the New File control in the Explorer sidebar.
  • Enter docker-compose.yml as the file name.

You should see docker-compose.yml in the Explorer sidebar.

  • Add the Qdrant service definition to docker-compose.yml by copying this configuration:
services:
  qdrant:
    image: qdrant/qdrant:v1.19.1
    ports:
      - "127.0.0.1:6333:6333"
    volumes:
      - qdrant_storage:/qdrant/storage

volumes:
  qdrant_storage:

How does this configuration protect the lab?

The image pin keeps Qdrant on version 1.19.1. The loopback address exposes port 6333 only through your Windows machine.

The Docker named volume stores Qdrant data under /qdrant/storage. That data remains available when Compose removes the container without deleting volumes.

  • Save docker-compose.yml.
  • Click Terminal in the VS Code top menu.
  • Click New Terminal to open the integrated PowerShell terminal.
  • Start Qdrant in detached mode by running this command:
docker compose up -d

What does this command do?

The docker compose up -d command creates the Qdrant container and its network. It also creates the qdrant_storage volume.

Detached mode leaves Qdrant running in the background. Your terminal remains available for the Python setup.

You should see Qdrant's welcome response. That response confirms the pinned container is running through the loopback-only port.

Qdrant welcome response missing?

  • Confirm Docker Desktop is still running.
  • Confirm the VS Code terminal is inside tenant-search-lab.
  • Confirm docker-compose.yml is saved before running the Compose command again.

Ask for help with the container: Help me diagnose why Qdrant is not reachable on localhost port 6333.

Create the isolated Python environment

The virtual environment keeps the gateway packages separate from other Python projects. Pinning each package makes the lab reproducible on Python 3.13.1.

  • Create .venv inside tenant-search-lab by running these commands in the VS Code PowerShell terminal:
python -m venv .venv
.venv\Scripts\Activate.ps1

What do these commands do?

The first command creates an isolated environment in .venv. The second command activates its PowerShell script.

Activation directs later Python and pip commands into this environment. Your project dependencies stay isolated.

You should see the environment name at the start of your PowerShell prompt. This confirms that .venv is active.

Virtual environment not active?

  • Confirm the VS Code terminal profile is PowerShell.
  • Confirm the terminal is inside tenant-search-lab.
  • Run .venv\Scripts\Activate.ps1 again after checking the path.

Ask for help with activation: Help me activate my .venv environment in PowerShell.

  • Click the New File control in the Explorer sidebar.
  • Enter requirements.txt as the file name.

You should see requirements.txt beside docker-compose.yml in the Explorer sidebar.

  • Add every pinned dependency to requirements.txt by copying these lines:
fastapi==0.141.1
httpx==0.28.1
pytest==9.1.1
qdrant-client==1.19.1
uvicorn==0.54.0

What will these packages provide?

FastAPI provides the gateway framework. Uvicorn serves that gateway locally.

HTTPX supports test requests. pytest runs the automated checks. The Qdrant Python client sends collection and search operations to the local server.

  • Save requirements.txt.

Before you run the installation, will pip accept every pinned version without a dependency conflict?

  • Install the pinned packages into the active environment by running this command:
python -m pip install -r requirements.txt

What does this command do?

The python -m pip install -r requirements.txt command uses the active environment's Python interpreter. Pip reads each exact package version from requirements.txt.

The installation should complete without a dependency conflict. Your isolated gateway environment is ready.

Package installation failed?

  • Confirm the environment name still appears at the start of the PowerShell prompt.
  • Confirm every line in requirements.txt matches the reference below.
  • Confirm your machine is using Python 3.13.1 for this environment.

Ask for help with the dependency conflict: Help me diagnose my pinned Python package installation failure.

✔️ Awesome, I've got everything!

Great work. Qdrant is responding through Docker while your pinned Python packages are installed inside the active virtual environment.

ⓧ I'd like to double check the full code

Compare your complete docker-compose.yml file with this reference:

services:
  qdrant:
    image: qdrant/qdrant:v1.19.1
    ports:
      - "127.0.0.1:6333:6333"
    volumes:
      - qdrant_storage:/qdrant/storage

volumes:
  qdrant_storage:

What to compare

Check the image pin and loopback port mapping. Check that the service mount uses the same qdrant_storage name declared under volumes.

Compare your complete requirements.txt file with this reference:

fastapi==0.141.1
httpx==0.28.1
pytest==9.1.1
qdrant-client==1.19.1
uvicorn==0.54.0

What to compare

Check every package name and version pin. Each line must match before you run the installation command again.

Your database server and Python toolchain are ready. Next, you will seed two tenants into one shared collection and expose the missing-filter leak.

Seed Two Tenants and Trigger a Data Leak

Your local Qdrant server is running with persistent storage. The lab now needs realistic tenant data before you can examine its security boundary.

A shared collection keeps small tenants in one place. Its query boundary determines which payloads a caller can retrieve.

In this step, you will seed deterministic Acme and Globex documents. You will then inspect what a plain vector search can retrieve from the shared collection.

In this step, get ready to:
  • Define reusable settings for the shared collection.
  • Seed deterministic Acme and Globex documents with tenant payloads.
  • Run a shared vector search to inspect every returned tenant.
Define the shared lab settings

Deterministic vectors give this lab repeatable rankings. The same query produces comparable results on every run.

The demo API key map links each credential to one tenant identity. Later steps use this mapping as trusted authorization context.

  • In the Explorer sidebar from earlier, right-click the lab project folder.
  • Select New File.
  • Enter settings.py as the file name.
  • Define the shared connection values by pasting this code into settings.py:
QDRANT_URL = "http://localhost:6333"
COLLECTION_NAME = "tenant_documents"
QUERY_VECTOR = [1.0, 0.0, 0.0, 0.0]

API_KEY_TO_TENANT = {
    "acme-demo-key": "tenant-a",
    "globex-demo-key": "tenant-b",
}

What Do These Settings Control?

  • The QDRANT_URL value points the Python client at the local Qdrant server.
  • The COLLECTION_NAME value gives every script one shared collection name.
  • The QUERY_VECTOR value makes every search use the same four-dimensional input.
  • The API_KEY_TO_TENANT mapping assigns the Acme key to tenant-a.
  • The same mapping assigns the Globex key to tenant-b.
  • Save settings.py.
  • Confirm that settings.py remains listed in the Explorer sidebar.

Can't See the Settings File?

Check that you created settings.py inside the lab project folder shown in VS Code. A file created outside this folder cannot be imported by the scripts you build next.

Confirm that the file name ends with .py.

Help me check why my settings file is missing from the VS Code project.

✔️ Awesome, I've got everything!

Your shared connection values are saved in settings.py.

ⓧ I'd like to double check the full code

  • Compare your complete settings.py file with this reference:
QDRANT_URL = "http://localhost:6333"
COLLECTION_NAME = "tenant_documents"
QUERY_VECTOR = [1.0, 0.0, 0.0, 0.0]

API_KEY_TO_TENANT = {
    "acme-demo-key": "tenant-a",
    "globex-demo-key": "tenant-b",
}

How to Compare This File

Check each constant name against the reference. Pay close attention to the commas inside API_KEY_TO_TENANT.

Seed the shared tenant collection

A Qdrant collection stores points. Each point carries a vector plus a payload.

A payload index helps Qdrant organize the tenant_id field for tenant-scoped retrieval. The is_tenant=True setting identifies this field as the tenant key.

  • Right-click the lab project folder in the Explorer sidebar.
  • Select New File.
  • Enter seed.py as the file name.
  • Build the collection schema by pasting this code into seed.py:
from qdrant_client import QdrantClient, models

from settings import COLLECTION_NAME, QDRANT_URL

client = QdrantClient(url=QDRANT_URL)

if client.collection_exists(COLLECTION_NAME):
    client.delete_collection(COLLECTION_NAME)

client.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config=models.VectorParams(size=4, distance=models.Distance.COSINE),
)

client.create_payload_index(
    collection_name=COLLECTION_NAME,
    field_name="tenant_id",
    field_schema=models.KeywordIndexParams(
        type=models.KeywordIndexType.KEYWORD,
        is_tenant=True,
    ),
)

What Does This Code Do?

  • The Qdrant client connects to the server through QDRANT_URL.
  • The existence check removes an older copy of tenant_documents.
  • The collection configuration accepts four-dimensional vectors.
  • The cosine distance setting controls how Qdrant compares vectors.
  • The keyword payload index identifies tenant_id as the tenant field.
  • Save seed.py.
  • Create the empty collection schema by running this command in the active PowerShell terminal:
python seed.py

What Does This Command Do?

This command runs the current seed.py file through the active virtual environment. The script recreates the collection schema in the running Qdrant container.

  • Navigate to http://localhost:6333/dashboard in your browser.
  • Select the tenant_documents collection.
  • Confirm that its vector configuration uses four dimensions.
  • Confirm that tenant_id appears as an indexed payload field.

You should see the new tenant_documents collection with its four-dimensional cosine configuration. This proves the schema reached the running server.

Collection Missing From Qdrant?

Return to the active PowerShell terminal. Confirm that its prompt still shows (.venv).

Check that the Qdrant container from the previous step is still running. Refresh the Qdrant dashboard after rerunning the script.

Help me diagnose why seed.py did not create tenant_documents in Qdrant.

  • Return to seed.py in the editor.
  • Locate the closing parenthesis beneath the client.create_payload_index( call.
  • Add the four tenant documents below that call by pasting this code:
client.upsert(
    collection_name=COLLECTION_NAME,
    points=[
        models.PointStruct(
            id=1,
            vector=[1.0, 0.0, 0.0, 0.0],
            payload={"tenant_id": "tenant-a", "document": "Acme launch plan"},
        ),
        models.PointStruct(
            id=2,
            vector=[0.8, 0.2, 0.0, 0.0],
            payload={"tenant_id": "tenant-a", "document": "Acme support runbook"},
        ),
        models.PointStruct(
            id=3,
            vector=[1.0, 0.0, 0.0, 0.0],
            payload={"tenant_id": "tenant-b", "document": "Globex acquisition memo"},
        ),
        models.PointStruct(
            id=4,
            vector=[0.8, 0.2, 0.0, 0.0],
            payload={"tenant_id": "tenant-b", "document": "Globex password rotation"},
        ),
    ],
)

print("Seeded 4 documents across tenant-a and tenant-b.")

How Are the Points Organized?

  • Points 1 through 2 carry the tenant-a payload.
  • Points 3 through 4 carry the tenant-b payload.
  • The fixed vectors keep the ranking reproducible.
  • The document strings make every result recognizable in the terminal.
  • The final message confirms that the complete seed operation finished.
  • Save seed.py.
  • Recreate the collection with all four points by running:
python seed.py

What Should I See?

You should see Seeded 4 documents across tenant-a and tenant-b. in PowerShell. That message confirms that Qdrant now holds both tenants in one shared collection.

That gives you a reproducible data layer. Acme and Globex now share the same collection while retaining visible tenant labels in every payload.

Seed Script Did Not Finish?

Check that settings.py sits beside seed.py in the lab project folder. The import fails when the files are in different folders.

Compare every parenthesis inside the client.upsert( call. Each models.PointStruct( entry must close before the next one begins.

Help me troubleshoot why seed.py cannot insert the four tenant documents.

✔️ Awesome, I've got everything!

Your shared collection now contains two Acme points plus two Globex points.

ⓧ I'd like to double check the full code

  • Compare your complete seed.py file with this reference:
from qdrant_client import QdrantClient, models

from settings import COLLECTION_NAME, QDRANT_URL

client = QdrantClient(url=QDRANT_URL)

if client.collection_exists(COLLECTION_NAME):
    client.delete_collection(COLLECTION_NAME)

client.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config=models.VectorParams(size=4, distance=models.Distance.COSINE),
)

client.create_payload_index(
    collection_name=COLLECTION_NAME,
    field_name="tenant_id",
    field_schema=models.KeywordIndexParams(
        type=models.KeywordIndexType.KEYWORD,
        is_tenant=True,
    ),
)

client.upsert(
    collection_name=COLLECTION_NAME,
    points=[
        models.PointStruct(
            id=1,
            vector=[1.0, 0.0, 0.0, 0.0],
            payload={"tenant_id": "tenant-a", "document": "Acme launch plan"},
        ),
        models.PointStruct(
            id=2,
            vector=[0.8, 0.2, 0.0, 0.0],
            payload={"tenant_id": "tenant-a", "document": "Acme support runbook"},
        ),
        models.PointStruct(
            id=3,
            vector=[1.0, 0.0, 0.0, 0.0],
            payload={"tenant_id": "tenant-b", "document": "Globex acquisition memo"},
        ),
        models.PointStruct(
            id=4,
            vector=[0.8, 0.2, 0.0, 0.0],
            payload={"tenant_id": "tenant-b", "document": "Globex password rotation"},
        ),
    ],
)

print("Seeded 4 documents across tenant-a and tenant-b.")

How to Compare This File

Check the collection schema first. Next, compare the four point IDs with their tenant payloads.

Trigger the cross-tenant leak

A vector query ranks every eligible point by similarity. The next script asks the shared collection for its four closest points.

The script prints each returned tenant_id beside its document. Those labels make the query boundary visible.

  • Right-click the lab project folder in the Explorer sidebar.
  • Select New File.
  • Enter unsafe_search.py as the file name.
  • Build the search script by pasting this code into unsafe_search.py:
from qdrant_client import QdrantClient

from settings import COLLECTION_NAME, QDRANT_URL, QUERY_VECTOR

client = QdrantClient(url=QDRANT_URL)

points = client.query_points(
    collection_name=COLLECTION_NAME,
    query=QUERY_VECTOR,
    with_payload=True,
    limit=4,
).points

print("Unsafe results:")
for point in points:
    payload = point.payload or {}
    print(f"- {payload.get('tenant_id')}: {payload.get('document')} ({point.score:.3f})")

What Does This Code Do?

  • The client connects to the same local Qdrant server used by seed.py.
  • The query_points( call searches with the deterministic QUERY_VECTOR.
  • The with_payload=True argument returns each document payload.
  • The limit=4 argument allows all four seeded points into the result set.
  • The loop prints each tenant label with its document text.
  • Save unsafe_search.py.

Before you run this, which tenant IDs do you think a shared collection query will return?

  • Run the search in the active PowerShell terminal with this command:
python unsafe_search.py

The Search Crosses the Tenant Boundary

You should see results containing both tenant-a and tenant-b. Acme documents appear beside Globex documents in one result set.

The query_points( call received no query_filter. Qdrant therefore searched across every vector in the collection.

You have reproduced the intended security failure. The tenant label exists in every payload while the query ignores it.

Don't See Both Tenant IDs?

Rerun seed.py to restore all four points. Confirm that its completion message names both tenant IDs.

Check that unsafe_search.py uses limit=4.

Help me diagnose why unsafe_search.py does not return both seeded tenants.

✔️ Awesome, I've got everything!

Your terminal shows Acme and Globex documents mixed in the same search results.

ⓧ I'd like to double check the full code

  • Compare your complete unsafe_search.py file with this reference:
from qdrant_client import QdrantClient

from settings import COLLECTION_NAME, QDRANT_URL, QUERY_VECTOR

client = QdrantClient(url=QDRANT_URL)

points = client.query_points(
    collection_name=COLLECTION_NAME,
    query=QUERY_VECTOR,
    with_payload=True,
    limit=4,
).points

print("Unsafe results:")
for point in points:
    payload = point.payload or {}
    print(f"- {payload.get('tenant_id')}: {payload.get('document')} ({point.score:.3f})")

How to Compare This File

Check that the query uses QUERY_VECTOR with a limit of four. Confirm that the output loop reads both payload fields.

Your shared collection now exposes a reproducible cross-tenant leak. Next, you will put the search behind a gateway to test whether authentication alone closes the gap.

Put a Naive Search Gateway in Front

Your unsafe script already showed that one shared Qdrant collection can return Acme and Globex records together. An API gateway can identify a caller while still leaving the database query unconstrained.

In this step, you'll build that gateway with FastAPI. You will test whether an identity check alone protects the stored data.

In this step, get ready to:
  • Create a FastAPI app for the existing Qdrant server.
  • Map each demo API key to its assigned tenant.
  • Run the /search route with the Acme key to inspect its response.
Create the gateway foundation

The app's header parameter gives tenant_from_api_key() the caller's demo key. The helper returns the matching tenant_id or raises an HTTP error.

  • Create main.py beside settings.py by using the new-file control in the VS Code Explorer sidebar.
  • Paste this gateway foundation into main.py:
from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from qdrant_client import QdrantClient

from settings import API_KEY_TO_TENANT, COLLECTION_NAME, QDRANT_URL, QUERY_VECTOR

app = FastAPI(title="Tenant-Isolated Vector Search Gateway")
qdrant = QdrantClient(url=QDRANT_URL)


def tenant_from_api_key(x_api_key: str | None) -> str:
    if x_api_key is None:
        raise HTTPException(status_code=401, detail="Missing X-API-Key header")

    tenant_id = API_KEY_TO_TENANT.get(x_api_key)
    if tenant_id is None:
        raise HTTPException(status_code=403, detail="Invalid API key")

    return tenant_id

What does this code do?

  • The FastAPI(...) call creates the gateway application.
  • The QdrantClient(...) call points the gateway at the existing local database.
  • The tenant_from_api_key() helper looks up the caller in API_KEY_TO_TENANT.
  • The helper rejects missing or unknown keys before a search begins.
  • Save main.py.
  • Confirm main.py appears beside settings.py in the Explorer sidebar.

Are the imports underlined?

  • Check that VS Code is still using the Python interpreter inside the active .venv.

Ask for help with the interpreter if the imports remain unresolved. Help me connect VS Code to the active virtual environment for this FastAPI project.

Add the naive search route

The search helper uses the deterministic vector from settings.py to query the shared collection. It converts each returned point into data that FastAPI can send as JSON.

  • Add tenant_search() below tenant_from_api_key() by pasting this code:
def tenant_search(tenant_id: str) -> list[dict[str, object]]:
    points = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        query=QUERY_VECTOR,
        with_payload=True,
        limit=4,
    ).points

    results = []
    for point in points:
        payload = point.payload or {}
        results.append({"id": point.id, "score": point.score, "tenant_id": payload.get("tenant_id"), "document": payload.get("document")})
    return results

How does the search helper work?

  • The query_points() call searches tenant_documents with the shared query vector.
  • The with_payload=True setting includes each document's stored metadata.
  • The loop formats every matching point into the gateway's response structure.
  • Save main.py.
  • Confirm tenant_search() appears below tenant_from_api_key() in the editor.

Is the helper flagged in VS Code?

  • Check that the def tenant_search line starts at the far-left edge of the editor.
  • Check that every line inside the function uses consistent indentation.

Use the guided prompt if the function still shows a syntax warning. Help me check the indentation and syntax of tenant_search in main.py.

The route connects the trusted header lookup to the search helper. Its response identifies the assigned tenant alongside the returned results.

  • Add the /search route below tenant_search() by pasting this code:
@app.get("/search")
def search(x_api_key: Annotated[str | None, Header()] = None):
    tenant_id = tenant_from_api_key(x_api_key)
    return {"tenant_id": tenant_id, "results": tenant_search(tenant_id)}

How does the route handle identity?

  • The Header() annotation maps x_api_key to the X-API-Key request header.
  • The route derives tenant_id before passing it to tenant_search().
  • The returned dictionary becomes the JSON response for /search.
  • Save main.py.
  • Confirm the search() route appears below tenant_search() in the editor.

✔️ Awesome, I've got everything!

Your gateway code is saved. The file now contains the app, tenant lookup, search helper, and /search route.

ⓧ I'd like to double check the full code

Compare your saved main.py with this complete version.

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from qdrant_client import QdrantClient

from settings import API_KEY_TO_TENANT, COLLECTION_NAME, QDRANT_URL, QUERY_VECTOR

app = FastAPI(title="Tenant-Isolated Vector Search Gateway")
qdrant = QdrantClient(url=QDRANT_URL)


def tenant_from_api_key(x_api_key: str | None) -> str:
    if x_api_key is None:
        raise HTTPException(status_code=401, detail="Missing X-API-Key header")

    tenant_id = API_KEY_TO_TENANT.get(x_api_key)
    if tenant_id is None:
        raise HTTPException(status_code=403, detail="Invalid API key")

    return tenant_id


def tenant_search(tenant_id: str) -> list[dict[str, object]]:
    points = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        query=QUERY_VECTOR,
        with_payload=True,
        limit=4,
    ).points

    results = []
    for point in points:
        payload = point.payload or {}
        results.append({"id": point.id, "score": point.score, "tenant_id": payload.get("tenant_id"), "document": payload.get("document")})
    return results


@app.get("/search")
def search(x_api_key: Annotated[str | None, Header()] = None):
    tenant_id = tenant_from_api_key(x_api_key)
    return {"tenant_id": tenant_id, "results": tenant_search(tenant_id)}
Trigger the authenticated leak

Uvicorn loads the application from main.py and serves it locally. The running server keeps its terminal session occupied while you test the gateway.

  • Switch back to the active PowerShell terminal from earlier.
  • Start the gateway by running:
uvicorn main:app

What does this command do?

Uvicorn imports the app object from main.py. It serves the gateway at http://127.0.0.1:8000.

The terminal remains occupied while the gateway is running.

  • Leave the PowerShell terminal running.
  • Navigate to http://127.0.0.1:8000/docs in your browser.
  • Confirm the interactive documentation lists the /search operation.

Can't open the API documentation?

  • Check that the PowerShell terminal still shows the running Uvicorn process.
  • Confirm the active terminal is inside the project folder containing main.py.

Use the guided prompt if the documentation still does not load. Help me diagnose why my local FastAPI documentation is not loading while Uvicorn is running.

The gateway now identifies Acme before searching. The next request tests whether that identity check changes which Qdrant points are eligible for retrieval.

  • Expand the /search operation in the interactive documentation.
  • Click Try it out.
  • Enter acme-demo-key in the X-API-Key header field.

Before you send the request, do you think identifying Acme is enough to limit every result to tenant-a?

  • Click Execute.

You'll see the top-level tenant_id set to tenant-a. The results array still includes tenant-b with a Globex document.

Why did the leak survive authentication?

The tenant_from_api_key() helper correctly identifies Acme. Authentication is working.

The tenant_search() helper calls query_points() without a query_filter. Qdrant therefore searches every vector in the shared collection.

Don't see the leaked Globex result?

  • Confirm the response body contains a populated results array.
  • Confirm the Qdrant container from the previous step is still running.
  • Confirm the seeded tenant_documents collection is still available.

Use the guided prompt if the response is empty or the request fails. Help me diagnose why the naive FastAPI search is not returning the seeded Qdrant documents.

That is the experiment complete: your gateway authenticates Acme, yet the search still crosses the tenant boundary. Next, you'll make the trusted tenant context control every Qdrant query.

Centralize and Enforce the Tenant Filter

Your FastAPI gateway already derives a trusted tenant identity from each demo API key. The database query still searches every vector in the shared collection.

The leak exists because authorization context and database filtering follow separate paths. This step joins them inside tenant_search() so every request carries its assigned tenant into Qdrant.

In this step, get ready to:
  • Give the gateway access to the Qdrant filter models.
  • Inject the trusted tenant filter into every vector search.
  • Compare isolated Acme and Globex responses through the single search route.
Give main.py access to filter models

The QdrantClient class sends requests to the database. The filter classes live in the models namespace.

  • In main.py, find the Qdrant client import near the top of the file.
  • Replace that import with the line below:
from qdrant_client import QdrantClient, models

Why import models?

This import keeps QdrantClient available. It also exposes the filter classes through models.

Enforce the tenant boundary in tenant_search

The tenant_search() function is the single enforcement point for the payload filter. It receives the tenant identity created by tenant_from_api_key() before it constructs the database query.

  • In main.py, locate the complete tenant_search() function.
  • Replace that function with the code below:
def tenant_search(tenant_id: str) -> list[dict[str, object]]:
    points = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        query=QUERY_VECTOR,
        query_filter=models.Filter(
            must=[
                models.FieldCondition(
                    key="tenant_id",
                    match=models.MatchValue(value=tenant_id),
                )
            ]
        ),
        with_payload=True,
        limit=4,
    ).points

    results = []
    for point in points:
        payload = point.payload or {}
        results.append(
            {
                "id": point.id,
                "score": point.score,
                "tenant_id": payload.get("tenant_id"),
                "document": payload.get("document"),
            }
        )
    return results

What does this filter enforce?

  • The models.Filter object groups the conditions that every matched point must satisfy.
  • The models.FieldCondition object targets the tenant_id payload field.
  • The models.MatchValue object uses the trusted tenant_id supplied by the route.
  • The query_filter argument passes that condition with the vector query.
  • The result loop keeps the response shape consistent for both tenants.
  • Save main.py.

The running Uvicorn process loaded the previous version of the gateway. Restarting it loads the tenant filter you just added.

  • Switch back to the terminal running Uvicorn.
  • Press Ctrl+C to stop the current gateway.
  • Restart the gateway by running this command:
uvicorn main:app

What does this command do?

Uvicorn imports the app object from main.py. The restarted process now uses the filtered version of tenant_search().

You'll see startup logs indicating that the gateway is ready at http://127.0.0.1:8000.

Gateway not restarting?

  • Confirm that the project virtual environment is still active.
  • Check that the Qdrant import includes both QdrantClient and models.
  • Compare the brackets inside models.Filter with the full code below.

Help me diagnose why the restarted gateway is failing.

✔️ Awesome, I've got everything!

Great. Your saved main.py now derives the tenant from the API key before applying the matching Qdrant filter.

ⓧ I'd like to double check the full code

Compare your complete main.py with this reference:

from typing import Annotated

from fastapi import FastAPI, Header, HTTPException
from qdrant_client import QdrantClient, models

from settings import API_KEY_TO_TENANT, COLLECTION_NAME, QDRANT_URL, QUERY_VECTOR

app = FastAPI(title="Tenant-Isolated Vector Search Gateway")
qdrant = QdrantClient(url=QDRANT_URL)


def tenant_from_api_key(x_api_key: str | None) -> str:
    if x_api_key is None:
        raise HTTPException(status_code=401, detail="Missing X-API-Key header")

    tenant_id = API_KEY_TO_TENANT.get(x_api_key)
    if tenant_id is None:
        raise HTTPException(status_code=403, detail="Invalid API key")

    return tenant_id


def tenant_search(tenant_id: str) -> list[dict[str, object]]:
    points = qdrant.query_points(
        collection_name=COLLECTION_NAME,
        query=QUERY_VECTOR,
        query_filter=models.Filter(
            must=[
                models.FieldCondition(
                    key="tenant_id",
                    match=models.MatchValue(value=tenant_id),
                )
            ]
        ),
        with_payload=True,
        limit=4,
    ).points

    results = []
    for point in points:
        payload = point.payload or {}
        results.append(
            {
                "id": point.id,
                "score": point.score,
                "tenant_id": payload.get("tenant_id"),
                "document": payload.get("document"),
            }
        )
    return results


@app.get("/search")
def search(x_api_key: Annotated[str | None, Header()] = None):
    tenant_id = tenant_from_api_key(x_api_key)
    return {"tenant_id": tenant_id, "results": tenant_search(tenant_id)}

This file keeps tenant derivation in the route path. It applies the matching filter inside every call to tenant_search().

Compare isolated Acme and Globex responses

The route now has one authorization input. Comparing both demo identities shows that the filter value changes while the endpoint stays the same.

  • Return to the API documentation from earlier.
  • Refresh the page to load the restarted gateway.
  • Expand the /search route.
  • Confirm that the request form provides an X-API-Key header input.
  • Confirm that the request form has no tenant_id selector.

Before you send the first request, predict which tenant IDs the Acme key can return.

  • Enable the interactive request form for /search.
  • Enter acme-demo-key as the X-API-Key header value.
  • Send the request.

You'll see tenant-a at the top of the response. Every item inside results also has tenant-a as its tenant ID.

That closes the Acme side of the leak. Globex documents no longer appear in Acme's search results.

Before you send the request again, predict how changing only the API key changes the response.

  • Replace the current header value with globex-demo-key.
  • Send the request again.

You'll see tenant-b at the top of the response. Every item inside results now has tenant-b as its tenant ID.

Both callers use the same /search route. The trusted API key mapping controls the filter without accepting a caller-supplied tenant selector.

You have closed the cross-tenant leak through one centralized enforcement point. Next, you'll turn these manual checks into automated tests that catch future isolation regressions.

Prove Isolation with Negative Tests

Your filtered gateway now keeps Acme data separate from Globex data during manual requests. The next goal is to turn that tenant isolation into evidence that runs on demand.

Manual checks can miss regressions in trusted API keys or database filters. Automated negative tests with pytest expose those failures before they reach a real tenant.

In this step, get ready to:
  • Create positive tests for the Acme and Globex access paths.
  • Add negative tests for missing credentials, invalid credentials, and tenant override attempts.
  • Remove the tenant filter temporarily to prove the isolation tests detect the regression.
Build the positive isolation tests

The FastAPI test client calls the gateway inside the test process. Each positive test checks the response status before comparing the tenant IDs returned by the gateway.

  • Create test_gateway.py beside main.py in the VS Code Explorer from earlier.

You should see test_gateway.py listed beside the existing project files.

  • Add the test client, tenant helper, and Acme test by pasting this code into test_gateway.py:
from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def result_tenants(response) -> set[str]:
    return {item["tenant_id"] for item in response.json()["results"]}


def test_acme_receives_only_tenant_a():
    response = client.get("/search", headers={"X-API-Key": "acme-demo-key"})
    assert response.status_code == 200
    assert result_tenants(response) == {"tenant-a"}

What does this code test?

  • TestClient(app) creates a client for requests to the gateway.
  • result_tenants() reduces the response to the tenant IDs that matter for isolation.
  • test_acme_receives_only_tenant_a() proves that the Acme key produces a successful response containing only tenant-a data.
  • Save test_gateway.py.
  • Verify the first access rule by running this command in the active project terminal:
python -m pytest -q

What does this check?

Pytest discovers the Acme test in test_gateway.py. The test passes only when every returned result belongs to tenant-a.

You should see pytest report one passing test. This confirms the first positive access path is protected.

Is the Acme test failing?

Confirm the Qdrant container is still running. Make sure the terminal prompt still shows the active .venv environment.

Check that main.py still passes query_filter to qdrant.query_points.

Help me diagnose the failing Acme tenant-isolation test.

A passing Acme test covers only one side of the boundary. The Globex path needs its own assertion so either tenant mapping can fail independently.

  • Add the Globex test below test_acme_receives_only_tenant_a() in test_gateway.py:
def test_globex_receives_only_tenant_b():
    response = client.get("/search", headers={"X-API-Key": "globex-demo-key"})
    assert response.status_code == 200
    assert result_tenants(response) == {"tenant-b"}

What does this code test?

test_globex_receives_only_tenant_b() sends the Globex key through the same route. Its final assertion rejects any result outside tenant-b.

  • Save test_gateway.py.
  • Verify both positive access paths by running this command:
python -m pytest -q

What does this check?

This run executes the Acme test plus the new Globex test. Both tenant mappings must return isolated result sets.

You should see pytest report two passing tests. Both tenants now have executable positive evidence.

Is only one test running?

Check that both function names start with test_. Make sure the Globex function sits outside the Acme function.

Help me find why pytest discovers only one gateway test.

Add credential and override tests

Positive tests prove valid callers reach their own data. Negative tests prove the gateway rejects missing credentials while ignoring caller input that tries to widen access.

  • Add the missing-key test below the Globex test in test_gateway.py:
def test_missing_api_key_is_rejected():
    response = client.get("/search")
    assert response.status_code == 401

What does this code test?

test_missing_api_key_is_rejected() sends a request without X-API-Key. The test requires the gateway to return 401.

  • Save test_gateway.py.
  • Verify the missing-credential rule by running this command:
python -m pytest -q

What does this check?

This run adds the missing-key path to the two tenant access checks. A request without authorization context must stop before any search result is returned.

You should see pytest report three passing tests. The gateway now has evidence for rejecting a request with no API key.

Is the missing-key test failing?

Confirm the request contains no headers argument. Check that tenant_from_api_key() raises the 401 response when x_api_key is None.

Help me debug the missing API key test.

A supplied key still needs validation against the trusted mapping. The next test separates an unknown credential from a missing credential.

  • Add the invalid-key test below test_missing_api_key_is_rejected() in test_gateway.py:
def test_invalid_api_key_is_rejected():
    response = client.get("/search", headers={"X-API-Key": "invalid-demo-key"})
    assert response.status_code == 403

What does this code test?

test_invalid_api_key_is_rejected() supplies a key absent from API_KEY_TO_TENANT. The required 403 response proves that an unknown key cannot become a tenant identity.

  • Save test_gateway.py.
  • Verify the invalid-credential rule by running this command:
python -m pytest -q

What does this check?

This run exercises both credential rejection paths plus the two valid tenant paths. Each path must produce its own expected status or tenant set.

You should see pytest report four passing tests. Unknown credentials can no longer slip through unnoticed.

Is the invalid-key test failing?

Check that the test sends invalid-demo-key. Confirm that the key is absent from API_KEY_TO_TENANT in settings.py.

Help me debug the invalid API key test.

The final negative test simulates a caller who adds a tenant selector to the URL. The Acme key must remain the only trusted source of tenant identity.

  • Add the override test below test_invalid_api_key_is_rejected() in test_gateway.py:
def test_caller_cannot_override_tenant():
    response = client.get(
        "/search?tenant_id=tenant-b",
        headers={"X-API-Key": "acme-demo-key"},
    )
    assert response.status_code == 200
    assert result_tenants(response) == {"tenant-a"}

What does this code test?

test_caller_cannot_override_tenant() combines the Acme key with ?tenant_id=tenant-b in the URL. The response must still contain only tenant-a results.

  • Save test_gateway.py.
  • Verify all five access rules by running this command:
python -m pytest -q

What does this check?

This run covers both valid tenants, both credential failures, and the attempted tenant override. The suite now represents the complete authorization boundary.

You should see 5 passed in the terminal. Your gateway now has automated evidence for each required access path.

Is the override test failing?

Confirm the request uses acme-demo-key while the URL contains tenant_id=tenant-b. Check that the route derives tenant_id only through tenant_from_api_key().

Help me debug the tenant override test.

✔️ Awesome, I've got everything!

Great. Double check that test_gateway.py is saved with all five tests.

ⓧ I'd like to double check the full code

Compare your complete test_gateway.py file with this reference:

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def result_tenants(response) -> set[str]:
    return {item["tenant_id"] for item in response.json()["results"]}


def test_acme_receives_only_tenant_a():
    response = client.get("/search", headers={"X-API-Key": "acme-demo-key"})
    assert response.status_code == 200
    assert result_tenants(response) == {"tenant-a"}


def test_globex_receives_only_tenant_b():
    response = client.get("/search", headers={"X-API-Key": "globex-demo-key"})
    assert response.status_code == 200
    assert result_tenants(response) == {"tenant-b"}


def test_missing_api_key_is_rejected():
    response = client.get("/search")
    assert response.status_code == 401


def test_invalid_api_key_is_rejected():
    response = client.get("/search", headers={"X-API-Key": "invalid-demo-key"})
    assert response.status_code == 403


def test_caller_cannot_override_tenant():
    response = client.get(
        "/search?tenant_id=tenant-b",
        headers={"X-API-Key": "acme-demo-key"},
    )
    assert response.status_code == 200
    assert result_tenants(response) == {"tenant-a"}
Break the filter and prove the tests catch it

A regression test earns trust by failing when its protection disappears. You will temporarily remove the database filter so the suite can expose the same cross-tenant leak from earlier.

  • Switch back to main.py in VS Code.
  • Find this block inside tenant_search():
query_filter=models.Filter(
            must=[
                models.FieldCondition(
                    key="tenant_id",
                    match=models.MatchValue(value=tenant_id),
                )
            ]
        ),

Why is this block the boundary?

query_filter limits the vector search to the tenant derived from the API key. Removing this block makes every stored vector eligible again.

  • Delete only the displayed query_filter block from tenant_search().
  • Save main.py.

Before you run the suite, which tests do you think will detect the missing filter?

  • Run the complete suite against the unfiltered gateway with this command:
python -m pytest -q

What does this run prove?

The tenant-set assertions now compare isolated expectations with mixed search results. The credential-rejection paths still exercise authentication before a database query runs.

You should see the isolation tests fail while the credential rejection tests still pass. That failure is intentional because the suite caught the missing tenant filter.

Did every test still pass?

Confirm that you removed the entire query_filter=models.Filter(...) argument from tenant_search(). Make sure you saved main.py before running the suite.

Help me work out why removing query_filter did not make the isolation tests fail.

The failing suite has proved that the tests detect the regression. Restore the trusted filter so the project finishes in its protected state.

  • Restore the deleted block between query=QUERY_VECTOR and with_payload=True by pasting this code:
query_filter=models.Filter(
            must=[
                models.FieldCondition(
                    key="tenant_id",
                    match=models.MatchValue(value=tenant_id),
                )
            ]
        ),

What does the restored code do?

The restored filter binds the database query to the trusted tenant_id. Every call through tenant_search() regains the same isolation rule.

  • Save main.py.

Before the final run, do you expect every access rule to pass again?

  • Confirm the restored gateway by running the full suite one final time:
python -m pytest -q

What does the final run prove?

The final run checks the restored filter against all five authorization scenarios. A passing suite proves that tenant identity comes from the API key while database retrieval remains constrained to that tenant.

You should see 5 passed in the terminal. That is the hard proof that your gateway rejects bad credentials and resists tenant override attempts.

Are tests still failing after restoration?

Check that the restored block sits inside the qdrant.query_points() call. Confirm the block ends with a comma before with_payload=True.

Compare main.py with the completed filtered version from the previous step.

Help me restore the tenant filter so all five tests pass.

You have moved beyond a manual demo. Your tenant boundary now comes with a repeatable test suite that fails when isolation disappears.

Secret mission

Add a Third Tenant Without Weakening Isolation

Your gateway isolates two tenants through one trusted mapping and one filtered route. Extend the same design to Initech, then prove that adding tenant-c does not weaken Acme's boundary.

Clean Up Your Resources

Clean Up Your Resources

This local lab creates no metered cloud usage charges for Qdrant. You can keep it running, pause its processes while retaining data, or remove it entirely.

Cost warning

The open-source packages and Qdrant image are free. Docker Desktop is free for personal use, education, non-commercial open source, and qualifying small businesses.

Larger-enterprise commercial use and government use require a paid subscription. Pausing the container does not change that licensing requirement.

Resources you used:

  • A local Qdrant container plus its Docker Compose network.
  • The qdrant_storage named volume containing tenant_documents, its tenant-aware keyword index, and six deterministic points.
  • The .venv Python virtual environment with the pinned gateway and test dependencies.
  • The local project folder containing docker-compose.yml, requirements.txt, settings.py, seed.py, unsafe_search.py, main.py, and test_gateway.py.

Keep the gateway boundary intact

The gateway demonstrates application-layer isolation. Qdrant stays on Windows loopback so unsafe_search.py can bypass the gateway for this lab.

Production deployments need a private database boundary.

  • Place Qdrant on a private network.
  • Prevent clients from connecting around the gateway.

Keep everything running

No action is needed. Choose this if you are still testing the gateway or extending the tenant model.

  • Leave the Qdrant container running so direct searches remain available.
  • Keep the Uvicorn gateway running so /search remains available at http://127.0.0.1:8000.
  • Retain qdrant_storage so all six tenant documents persist.
  • Retain .venv plus the project files so you can rerun the tests.

Pause - I'll come back to this later

Pausing frees local memory while keeping every file and seeded point ready for your return.

  • Switch back to the terminal running Uvicorn.
  • Press Ctrl+C to stop the gateway.
  • Switch back to the project terminal from earlier.
  • Stop the Qdrant container while preserving qdrant_storage by running this command:
docker compose down

What stays after pausing?

The docker compose down command removes the Qdrant container. It also removes the Compose network.

The named qdrant_storage volume remains. Your tenant_documents collection returns when you start Qdrant again.

Having trouble pausing Qdrant?

  • Confirm the terminal is in the folder containing docker-compose.yml.
  • Check that Docker Desktop is running before retrying the command.
  • Help me troubleshoot this cleanup command.

Your lab is safely paused. All six tenant documents remain in qdrant_storage for the next startup.

Delete - I don't want to use this again

Deletion gives you a clean local reset.

This cleanup is intentionally thorough, but it targets only this Compose lab and the project folder you select.

  • Stop the gateway with Ctrl+C if its Uvicorn terminal is still active.
  • Switch back to the project terminal from earlier.
  • Remove the Qdrant container, Compose network, and named volume by running this command:
docker compose down -v

What does this delete?

The docker compose down -v command removes the Qdrant container. It also removes the Compose network.

The volume removal permanently deletes qdrant_storage plus its tenant_documents collection.

Still seeing the Docker resources?

  • Confirm the terminal is in the folder containing docker-compose.yml.
  • Check that Docker Desktop is running before retrying the command.
  • Help me finish removing the Docker resources.

The Docker resources are gone. The project folder still contains .venv plus your lab files.

  • Close VS Code so Windows can release the project folder.
  • Press the Windows key to open search.
  • Type File Explorer into search.
  • Press Enter to open it.

File Explorer lets you remove the complete project folder without assuming a folder name or path.

  • Return to the parent folder where you created the lab.
  • Select the project folder containing docker-compose.yml.
  • Press Shift+Delete to remove the selected folder permanently.
  • Confirm the Windows deletion prompt.

That is a clean reset. Your virtual environment and every local lab file are now removed.

Docker Desktop remains installed for other local projects.

Nice Work!

Nice Work!

You did it: your local Qdrant lab now demonstrates a reproducible cross-tenant vector-search leak. Your FastAPI gateway closes that gap through centralized tenant filter injection.

You've learned how to:

  • Reproduced a cross-tenant vector-search leak by running an unfiltered query against the shared tenant_documents collection.
  • Centralized tenant authorization in one /search route. The gateway derives identity from X-API-Key before tenant_search() injects the matching Qdrant payload filter.
  • Built six authorization tests with pytest. Positive checks prove tenant-specific access. Negative checks prove credential rejection. The override check proves caller input cannot widen tenant scope.
  • Secret Mission: Added Initech as tenant-c inside the shared collection. The unchanged /search route now isolates all three tenants without a tenant-specific branch.

Ready to quiz yourself?