Cross-tenant leak: anti-patterns

Capsule overview

You already have the three multi-tenancy models clear (capsule 02), a shared schema with its mitigations (capsule 03), RLS conceptually (capsule 04), RLS integrated with FastAPI + SQLAlchemy async (capsule 05), and schema-per-tenant for enterprise cases (capsule 06). This capsule brings together in one place the concrete anti-patterns that end in a cross-tenant leak in production — the bugs that have shown up most in public and private postmortems, written as real code so you recognize them in code review before they reach deploy.

Every anti-pattern in this capsule follows the same format: the buggy code, the documented case where it happened (public or representative), why the human mind falls into the trap, how RLS prevents it even when the code is badly written (the reason the module recommends RLS as the safety net), and the complementary mitigation at the code level so the guarantee doesn't depend on a single layer.

By the end you'll have a mental catalog of five anti-patterns you'll be able to smell in any multi-tenant PR. You're going to understand why the combination "shared schema + RLS + isolation tests + lint" is defense in depth and not redundancy. And you'll be able to explain to a teammate why each layer contributes — even when RLS is already enabled, the other layers are still necessary.

Mental model: defense in depth, not a single wall

A cross-tenant leak usually isn't the work of a single bad line of code. It's usually the combination of several layers failing at the same time: the code has a bug, the code review didn't catch it because the reviewer was in a hurry, the lint didn't flag it because the pattern is too subtle, the tests didn't catch it because there was only one tenant in the test environment, and the database didn't block it because RLS wasn't enabled or the connecting role was the owner.

Think of the protection as a series of stacked meshes, each with different holes. A single mesh lets a lot of errors through. Five stacked meshes, with holes in different places, catch almost everything. That's the idea of defense in depth:

Every anti-pattern in this capsule fails at least one of the layers above. The capsule's question isn't "how do I get one layer to catch everything?" — it's "how many layers do I have active so that no application bug becomes a public incident?".

Why does anticipating these anti-patterns matter?

1. They're the most expensive bugs in multi-tenant SaaS. A cross-tenant leak isn't a bug that breaks a feature — it's a security incident. You trigger the incident response plan, you communicate with affected customers, you possibly notify regulators (GDPR, CCPA), and the reputational damage exceeds the technical damage. Anticipating them in code review saves orders of magnitude more than fixing them later.

2. Almost all of them follow repeated patterns. Despite the severity, the publicly documented cross-tenant leaks fall into five or six categories. If you recognize the patterns, you detect them at the speed of reading the PR. This capsule trains you for that speed.

3. RLS doesn't excuse you from detecting them. RLS is the final safety net, but it isn't an excuse for writing careless code. If your code has systematic filtering bugs, it's also going to have them in other aspects (authorization, input validation, error handling). Well-written code + RLS is real defense in depth. Badly-written code + RLS is hoping a single layer catches everything that escapes.

4. It's the conversation you're going to have most in a SaaS code review. When you're a tech lead or senior, you're going to flag "I don't see tenant_id in this WHERE" on many PRs. Having clear vocabulary (the anti-patterns' names) makes that conversation reproducible and teachable to new devs.

Anti-pattern 1: a direct query with no tenant filter

The most classic, the most documented, the easiest to commit.

The buggy code

# app/api/projects.py
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import get_session
from app.db.models import Project
from app.auth import get_current_tenant, Tenant

router = APIRouter()


@router.get("/projects/recent")
async def list_recent_projects(
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    # BUG: forgot the .where(Project.tenant_id == current_tenant.id)
    result = await db.execute(
        select(Project)
        .order_by(Project.created_at.desc())
        .limit(10)
    )
    return result.scalars().all()

A documented case

GitHub published a postmortem on its engineering blog in 2021 where a new mobile app endpoint exposed private repositories from different organizations in the "recent activity" list. The cause: a new SQLAlchemy query that forgot the filter by organization in the "global view" branch. The bug lasted less than 24 hours in production thanks to a customer reporting it quickly, but it managed to expose metadata for hundreds of private repos to users who shouldn't have seen them.

At smaller companies the pattern is identical: a new endpoint, a query written in a hurry before the demo, a PR approved by a reviewer who trusted that "the other endpoints filter, so this one will too." The lapse is human and systematic.

Why the human mind falls into the trap

The buggy code looks correct. The query is valid, the ORM accepts it, the unit tests pass (because there's only one tenant in the test setup), and the endpoint works locally. The bug only appears when there's data from more than one tenant in the database. In development, that case almost never gets actively tested.

Another factor: "filtering" is something the dev knows they have to do but doesn't see as part of the endpoint's behavior. They see it as "boilerplate." And boilerplate is the first thing that gets forgotten under pressure.

How RLS prevents it even with the bug

With a shared schema + RLS + FORCE ROW LEVEL SECURITY enabled on projects, the same buggy query gets transformed at the PostgreSQL level into:

-- What the ORM sends:
SELECT * FROM projects ORDER BY created_at DESC LIMIT 10;

-- What PostgreSQL runs (RLS adds the invisible WHERE):
SELECT * FROM projects
WHERE tenant_id = current_setting('app.tenant_id')::BIGINT
ORDER BY created_at DESC
LIMIT 10;

The endpoint returns only the request's tenant's projects. The bug in the code doesn't become a cross-tenant leak because the database applied the filter on its own. The student who understood capsules 04 and 05 recognizes this immediately: the get_tenant_session dependency ran SET LOCAL app.tenant_id at the start of the transaction, the RLS policy read that setting, and it filtered the rows before returning them to the ORM.

Complementary mitigation

• The repository pattern (capsule 03): the query would never be written directly; it would call await project_repo.list_recent(), where the repo receives tenant_id in its constructor.
• A grep in CI: a rule that fails if it finds select(Project) outside app/db/repositories/.
• Isolation tests (capsule 05): the setup creates two tenants with data in both. The endpoint's test calls with tenant A and verifies no tenant B rows show up.
• A code review checklist: a "Multitenancy" section in the PR template that asks you to check "all the new queries filter by tenant_id or go through a repository that does."

Anti-pattern 2: .get(Model, id) in SQLAlchemy without checking the tenant

A particularly insidious variant because the buggy code is shorter than the correct code.

The buggy code

# app/api/projects.py
@router.get("/projects/{project_id}")
async def get_project(
    project_id: int,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    # BUG: .get() looks up by PK without filtering by tenant
    project = await db.get(Project, project_id)
    if project is None:
        raise HTTPException(status_code=404, detail="Project not found")
    return project

A representative case

In a private postmortem from a B2B scaleup (shared at an internal conference in 2024), an endpoint used db.get(Invoice, invoice_id) to serve an invoice's detail. A malicious user who was also a customer of another tenant discovered that by changing the ID in the URL they could read competitors' invoices. The bug went undetected for months because the IDs were consecutive and predictable. The leak ended in a legal notice, damage mitigation, and a rushed migration to RLS.

The "IDOR" pattern (Insecure Direct Object Reference) is one of OWASP's top items precisely because of this: APIs that trust the request's ID is legitimate without checking ownership.

Why the human mind falls into the trap

db.get(Project, project_id) is the "idiomatic" SQLAlchemy way to fetch by PK. It's what shows up in the tutorials. It's what the dev writes first when they think "I need project X." Adding a tenant filter afterward feels like "adding boilerplate to clean code."

Also, the endpoint works without the filter: the dev tries it with their own user from their own tenant, sees the right project, considers the feature done. The IDOR only appears when someone deliberately tries an ID belonging to another tenant.

How RLS prevents it even with the bug

With RLS enabled on projects, db.get(Project, project_id) translates internally into a SELECT by PK. That SELECT goes through the policy:

-- What SQLAlchemy sends internally:
SELECT * FROM projects WHERE id = $1;

-- What PostgreSQL runs:
SELECT * FROM projects
WHERE id = $1
  AND tenant_id = current_setting('app.tenant_id')::BIGINT;

If project_id belongs to another tenant, the query returns 0 rows. SQLAlchemy interprets that as "doesn't exist," db.get() returns None, and the endpoint returns a 404. The attacker gets the same 404 they'd get if the ID really didn't exist. Indistinguishable from the outside. That indistinguishability is a key property: you don't leak information about the existence of someone else's IDs.

Complementary mitigation

• A repository: await project_repo.get(project_id) where the repo always filters by tenant_id.
• Lint/grep: detect db.get(Model, ...) for multi-tenant models outside the allowed files.
• An isolation test: create a project with tenant B, try to read it with tenant A's session, expect a 404. This test is one of the ones that catches the most leaks.

Anti-pattern 3: the tenant filter in a JOIN that gets forgotten in one branch of the query builder

A subtle pattern that shows up in dynamically built queries.

The buggy code

# app/api/projects.py
from sqlalchemy import select, and_

@router.get("/projects/search")
async def search_projects(
    q: str | None = None,
    archived: bool = False,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    stmt = select(Project).where(Project.tenant_id == current_tenant.id)

    if q:
        # BUG: when there's a search, the dev "rebuilds" the query from scratch
        # and forgets to re-apply the tenant filter.
        stmt = (
            select(Project)
            .join(ProjectTag, ProjectTag.project_id == Project.id)
            .where(ProjectTag.name.ilike(f"%{q}%"))
        )

    if archived:
        stmt = stmt.where(Project.archived_at.is_not(None))
    else:
        stmt = stmt.where(Project.archived_at.is_(None))

    result = await db.execute(stmt.limit(50))
    return result.scalars().all()

The Project.tenant_id == current_tenant.id filter gets set on the builder's first line, but the if q: discards it by reassigning stmt = select(Project).... Only in the "no search" branch does the filter persist.

A representative case

A typical scaleup case where a search endpoint evolves over months with several devs adding branches. Each new branch gets added as if condition: stmt = ... without auditing whether the earlier branches correctly propagate the tenant filter. A new dev adds a "with join" branch, tries it with their own tenant, sees it works, merges. Three months later a customer reports the search returns results from another company with project names similar to the search term.

Why the human mind falls into the trap

SQLAlchemy's query builder is compositional, but the dev doesn't always reason compositionally. When they think "I need to add a join," their instinct is "I start from select(Project) and build it from scratch" instead of "I take the stmt that already has the filter and add the join to it." That difference between "rebuilding" and "extending" causes the bug.

Also, queries with branching if/elif are hard to audit mentally. The reviewer looks at the new branch (if q:), verifies the JOIN is correct, and doesn't notice the base tenant_id filter got lost in that branch.

How RLS prevents it even with the bug

Even though the query builder's code loses the tenant_id filter in the search branch, PostgreSQL applies the policy to the final query, no matter how the ORM built it. The policy filters by tenant at the database level, and the JOIN's result only contains rows from the setting's tenant.

An important detail: for this to work, the policy has to be enabled on ALL the tables involved in the JOIN. If projects has RLS but project_tags doesn't, an attacker could do a JOIN that crosses tenants. The operational rule: every table with tenant_id gets RLS enabled and FORCE ROW LEVEL SECURITY.

Complementary mitigation

• Refactor the builder: define the base query with the filter and only add where/join, never reassign stmt = select(Model). This gets tested with a regression test that compiles the final SQL and verifies tenant_id appears.
• A repository with a limited builder: the repository exposes methods like repo.search(q, archived) and builds the query internally. The endpoint never assembles queries.
• An isolation test per branch: parameterize the test to cover every combination (q=None/q="abc", archived=True/False) with data from two tenants. The test passes only if every branch filters correctly.

Anti-pattern 4: a cron job that forgets to set the tenant context

The bug that only appears at night, in the nightly query nobody audits.

The buggy code

# app/jobs/notify_overdue.py
import asyncio
from datetime import datetime, timedelta
from sqlalchemy import select

from app.db.session import SessionLocal
from app.db.models import Task
from app.notifications import send_email


async def notify_overdue_tasks():
    """A cron job that runs every night at 02:00 UTC."""
    async with SessionLocal() as db:
        # BUG: doesn't set SET LOCAL app.tenant_id because the job is "global"
        result = await db.execute(
            select(Task)
            .where(Task.due_date < datetime.utcnow())
            .where(Task.status == "open")
            .where(Task.notified_overdue.is_(False))
        )
        tasks = result.scalars().all()

        for task in tasks:
            # BUG #2: sends an email to the "owner" without checking which tenant they belong to
            await send_email(
                to=task.assignee_email,
                subject=f"Overdue task: {task.title}",
                body=f"Your task '{task.title}' has been overdue since {task.due_date}.",
            )
            task.notified_overdue = True

        await db.commit()


if __name__ == "__main__":
    asyncio.run(notify_overdue_tasks())

If the app is on a shared schema with no RLS, the job works "fine" (it touches every tenant's tasks). The bug is that the assignee_email can belong to a different tenant from the original task's (e.g. the task was reassigned, the assignee's email corresponds to their personal account, etc.) — and the email's content can reveal information from another tenant's task.

If the app has RLS enabled, the job fails because with no app.tenant_id set, the queries return no rows (or blow up, depending on how current_setting is configured). The dev "fixes" the job by setting an arbitrary tenant_id or by using the owner role — and there they fall into the next level of bug.

A representative case

At a project management scaleup (a case shared anonymously at a PostgreSQL Europe conference in 2023), a nightly "reminders" job sent emails with tasks' content. The job connected with a DB superuser role to "access all the tenants." After a refactor, the job's query started JOINing with the users table to resolve the assignee's email — but the JOIN had a bug and matched assignees with users from any tenant. Result: users from company A received emails with the content of company B's tasks for 11 days, until a user reported "I'm getting emails about tasks I never created."

Why the human mind falls into the trap

Cron jobs don't get thought of as "multi-tenant endpoints." They get thought of as "global scripts that maintain the system." The dev writes the job as if it were a DBA script: "I'm going to read all the overdue tasks and send emails." It doesn't occur to them that each task belongs to a tenant and that the tenant context is relevant.

Another factor: the jobs run at night, with no direct human observability. The errors show up in logs nobody reads if there are no alerts. A job sending crossed emails can run 30 consecutive nights before somebody notices something odd.

How RLS prevents it even with the bug

RLS explicitly blocks this anti-pattern. If the job connects with the app_user role (not superuser, not owner), the queries with no SET LOCAL app.tenant_id return 0 rows. The job "does nothing" — immediately visible in the logs ("0 overdue tasks today, is that right?"). That forces the dev to think through the correct model:

The correct pattern for cross-tenant jobs:

async def notify_overdue_tasks_correct():
    """Iterates per tenant and processes each one in isolation."""
    async with SessionLocal() as db:
        # 1. The list of tenants — use a BYPASSRLS role only for this query
        async with db.begin():
            await db.execute(text("SET LOCAL row_security = off"))
            tenant_ids = (await db.execute(
                select(Tenant.id).where(Tenant.active.is_(True))
            )).scalars().all()

    # 2. For each tenant, open a session with ITS context
    for tenant_id in tenant_ids:
        async with SessionLocal() as db:
            async with db.begin():
                await db.execute(
                    text("SET LOCAL app.tenant_id = :tid"),
                    {"tid": str(tenant_id)},
                )
                result = await db.execute(
                    select(Task)
                    .where(Task.due_date < datetime.utcnow())
                    .where(Task.status == "open")
                    .where(Task.notified_overdue.is_(False))
                )
                tasks = result.scalars().all()

                for task in tasks:
                    await send_email(...)
                    task.notified_overdue = True

The job processes each tenant in its own transaction with its own context. RLS guarantees that within each iteration, only that tenant's tasks get read. The emails never cross tenants.

Complementary mitigation

• A dedicated role for jobs with BYPASSRLS only for the "list active tenants" query, never for data queries.
• An operational alert: if the job processes "0 rows" for N consecutive days, alert. It reflects the context not being set.
• A test for the job: run the job in an environment with two tenants and verify each email sent contains only the recipient's tenant's data.
• A code review specific to jobs: every PR touching app/jobs/ requires a senior review with a check of "does this job operate per-tenant or globally? does the pattern match?".

Anti-pattern 5: a bulk operation that gets tenant_id from the client's payload

The "whatever the client says goes" bug.

The buggy code

# app/api/tasks.py
from pydantic import BaseModel

class TaskBulkCreate(BaseModel):
    tenant_id: int  # ⚠️ comes from the client's payload
    items: list[dict]


@router.post("/tasks/bulk")
async def bulk_create_tasks(
    payload: TaskBulkCreate,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    # BUG: uses payload.tenant_id instead of current_tenant.id
    new_tasks = [
        Task(
            tenant_id=payload.tenant_id,  # ⚠️ controlled by the client
            title=item["title"],
            status=item.get("status", "open"),
        )
        for item in payload.items
    ]
    db.add_all(new_tasks)
    await db.commit()
    return {"created": len(new_tasks)}

A malicious client sends {"tenant_id": 999, "items": [...]} and creates tasks attributed to someone else's tenant. If that tenant later has a GET /tasks endpoint filtered by its own tenant_id, it's going to see tasks an attacker injected into it. Worse: if the endpoint exposes those tasks in notifications, dashboards, or exports, the attacker got a primitive for inserting arbitrary content into the victim's app.

A representative case

In a breach reported by TechCrunch in 2022 about a marketing automation tool, a "bulk import" endpoint accepted an account_id field in the JSON. A security researcher discovered that by changing the account_id they could create contacts in someone else's accounts. The company patched it within hours but the incident was public. The postmortem (published under NDA) confirmed the bug had existed since day 1 of the endpoint, simply because the dev confused "the client sends data to import" with "the client sends the account to import into."

Why the human mind falls into the trap

The pattern shows up when the endpoint is "technical" (bulk import, data sync, a B2B integration). The dev thinks in terms of the payload: "the client sends me a JSON with everything I need to create the rows, including which tenant they belong to." It makes sense if you think of the tool as an ETL.

But the rule in multi-tenant SaaS is the opposite: the tenant_id NEVER comes from the payload. It comes from the request's authenticated context (JWT, session, an API key with a tenant scope). The client shouldn't be able to "ask" for their data to be created in another tenant — the question doesn't make sense in the correct model.

How RLS prevents it even with the bug

With the RLS policy configured with WITH CHECK (capsule 04), PostgreSQL validates that every INSERT has a tenant_id equal to the session setting's:

CREATE POLICY tenant_isolation ON tasks
    USING (tenant_id = current_setting('app.tenant_id')::BIGINT)
    WITH CHECK (tenant_id = current_setting('app.tenant_id')::BIGINT);

If the session has app.tenant_id = 1 and the code tries to insert a row with tenant_id = 999, PostgreSQL rejects the INSERT with an error:

ERROR:  new row violates row-level security policy for table "tasks"

The bulk insert fails completely. The client gets a 500 (or a 400 if you catch it and translate). The attacker can't create rows in someone else's tenant even though the application's code allows it in the ORM.

Complementary mitigation

• A Pydantic schema with no tenant_id: the TaskBulkCreate shouldn't have tenant_id as a field. The tenant gets obtained from the authenticated request, not from the payload. Making the field invalid at the model level is the strongest form of prevention.
• A lint to detect Model(tenant_id=payload...): an easy static pattern to detect.
• A security test: try sending tenant_id=999 with a session authenticated as tenant_id=1 and verify the endpoint rejects it (400 or 500, but rejects).
• Document the principle in the style guide: "the tenant_id gets obtained from the authenticated request, NEVER from the payload."

Bonus anti-pattern: the app's role is the tables' owner (RLS doesn't apply)

It isn't a bug in the application code — it's a PostgreSQL configuration bug that silently disables all your RLS protection.

The problem

PostgreSQL, by default, does NOT apply RLS to the table's owner. If your setup created the tables connected as postgres (the superuser role that's also the owner), and your app connects as that same role, the RLS policies get ignored. All the protection you built in capsules 04-05 is switched off.

-- As postgres (the table's owner):
SET LOCAL app.tenant_id = '1';
SELECT * FROM tasks;
-- Output: ALL the rows (RLS ignored for being the owner).

A representative case

It's one of the most common bugs in projects turning on RLS for the first time. The dev follows a tutorial, runs ALTER TABLE ... ENABLE ROW LEVEL SECURITY, writes the policy, tests that it works from a new role (SET ROLE app_user), confirms the isolation works. Then they connect the app to production using the credentials they already had configured — which turn out to be the postgres role's. RLS is enabled at the schema level but ignored at runtime. The team believes it has working RLS. It doesn't.

How to prevent it: three explicit actions

1. FORCE ROW LEVEL SECURITY on every table with RLS enabled. This forces the policies to be applied even to the owner.

ALTER TABLE tasks FORCE ROW LEVEL SECURITY;

2. The app connects with a non-owner role, dedicated, with minimal permissions. Typically app_user (capsule 05).

3. A regression test: a test that connects as the role the app uses in production, runs a query with no SET LOCAL app.tenant_id, and verifies it returns 0 rows or fails. If the test passes with rows returned, there's a configuration bug.

Process-level mitigation

• Mandatory code review on any change to pg_hba.conf, connection strings, or role configuration.
• The secret manager should have clear separation: DATABASE_URL (the app role, not the owner) vs DATABASE_ADMIN_URL (the postgres role, only for migrations and maintenance).
• The deploy pipelines never use DATABASE_ADMIN_URL to run the app — only migrations.

Traps and common mistakes when designing the defense

Mistake 1 (conceptual): thinking RLS is "magic that makes everything else unnecessary"

Symptom: the team turns on RLS and abandons the repository pattern, the isolation tests, and tenant-specific code review. "It's not needed anymore, RLS protects us."

Why it happens: RLS sometimes gets sold as "the definitive solution." And conceptually it protects against cross-tenant leaks in storage. But it doesn't protect against: authorization bugs at the feature level (a user who shouldn't be able to delete tasks even though they're their own tenant's), bugs in jobs that connect with BYPASSRLS from a configuration mistake, bugs in legitimate cross-tenant queries that write to another tenant's table. Each of those cases requires other layers.

How to tell: review the last PR merged to production. Is there a test that verifies the new endpoint doesn't expose another tenant's data? If the answer is "no, RLS protects it," the team fell into the trap.

How to fix it: treat RLS as one more layer in defense in depth. Keep isolation tests per endpoint, the repository pattern, explicit code review of tenant filtering.

Mistake 2 (operational): assuming a test that passes locally proves isolation in production

Symptom: the isolation tests run against a local Postgres that has RLS enabled, they pass with no problems, the team merges to production. In production the app connects with an owner role (the bonus anti-pattern above) and RLS doesn't apply. A hidden bug.

Why it happens: the test environment uses a different PostgreSQL setup from production's. Generally the test runs in a container with the postgres user, who is the owner — the same bug production has.

How to tell: run the isolation tests connecting with the same role the app uses in production. If the tests pass there, the isolation is real.

How to fix it: the test environment mirrors the production setup exactly: a non-owner role, FORCE ROW LEVEL SECURITY, the same policies. Ideally, the test creates the app_user role and connects as that role specifically.

Mistake 3 (conceptual): confusing "validating input" with "filtering by tenant"

Symptom: the dev adds strict Pydantic validation thinking "if the input is valid, there's no leak." But the validation doesn't include checking that the referenced IDs belong to the request's tenant.

Why it happens: Pydantic validates types and formats, not permissions. A project_id: int is valid to Pydantic as long as it's an integer, regardless of which tenant it belongs to.

How to tell: review the endpoints that receive IDs in the body or query params. Is there a check that the ID belongs to the request's tenant? If the check is only "it's an int," there's a potential IDOR.

How to fix it: ownership checking is the repository's responsibility (which filters by tenant_id) or RLS's. Input validation is complementary, not a replacement.

Exercises

Exercise 1: code review a PR with multiple anti-patterns

This PR comes to you for review. Identify ALL the cross-tenant leak risks (classify by the catalog's anti-pattern).

# app/api/notes.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import get_session
from app.db.models import Note, Folder
from app.auth import get_current_tenant, Tenant

router = APIRouter()


class NoteCreate(BaseModel):
    folder_id: int
    title: str
    content: str
    tenant_id: int  # the client sends it


@router.post("/notes")
async def create_note(
    payload: NoteCreate,
    db: AsyncSession = Depends(get_session),
    current_tenant: Tenant = Depends(get_current_tenant),
):
    folder = await db.get(Folder, payload.folder_id)
    if folder is None:
        raise HTTPException(status_code=404, detail="Folder not found")

    note = Note(
        tenant_id=payload.tenant_id,
        folder_id=payload.folder_id,
        title=payload.title,
        content=payload.content,
    )
    db.add(note)
    await db.commit()
    return {"id": note.id}


@router.get("/notes/recent")
async def list_recent_notes(
    limit: int = 20,
    db: AsyncSession = Depends(get_session),
):
    result = await db.execute(
        select(Note)
        .order_by(Note.created_at.desc())
        .limit(limit)
    )
    return result.scalars().all()

Exercise 2: predict behavior with and without RLS

For each query, predict what happens in two scenarios: (a) without RLS and (b) with RLS enabled + a standard tenant policy. Assume the session has app.tenant_id = 1 and that tasks exist with tenant_id = 1 and tenant_id = 2.

i. SELECT COUNT(*) FROM tasks;

ii. SELECT * FROM tasks WHERE id = (SELECT id FROM tasks WHERE tenant_id = 2 LIMIT 1); — the attacker tries to read a tenant 2 task by nesting a subquery.

iii. INSERT INTO tasks (tenant_id, title) VALUES (2, 'injected'); — the code has a bug and attributes the task to someone else's tenant.

iv. UPDATE tasks SET status = 'done' WHERE 1=1; — a catastrophic query that forgets the WHERE.

Exercise 3: write a test that catches five anti-patterns in a single run

Design a pytest test that covers anti-patterns 1, 2, 3, and 5 against an /api/notes endpoint. The test has to create two tenants (Acme and Globex) with data in both, authenticate as Acme, and try several actions that should fail or return empty.

# tests/security/test_notes_isolation.py
import pytest
from httpx import AsyncClient

from app.db.models import Tenant, Folder, Note


@pytest.fixture
async def two_tenants_with_data(db):
    acme = Tenant(slug="acme-test", name="Acme")
    globex = Tenant(slug="globex-test", name="Globex")
    db.add_all([acme, globex])
    await db.flush()

    acme_folder = Folder(tenant_id=acme.id, name="Acme Folder")
    globex_folder = Folder(tenant_id=globex.id, name="Globex Folder")
    db.add_all([acme_folder, globex_folder])
    await db.flush()

    acme_note = Note(
        tenant_id=acme.id, folder_id=acme_folder.id,
        title="Acme Note", content="hello",
    )
    globex_note = Note(
        tenant_id=globex.id, folder_id=globex_folder.id,
        title="Globex Note", content="secret",
    )
    db.add_all([acme_note, globex_note])
    await db.commit()

    return {
        "acme": acme, "globex": globex,
        "acme_folder": acme_folder, "globex_folder": globex_folder,
        "acme_note": acme_note, "globex_note": globex_note,
    }


@pytest.mark.asyncio
async def test_no_cross_tenant_in_recent_notes(
    client: AsyncClient, two_tenants_with_data
):
    """Anti-pattern 1: a query with no filter must NOT return other tenants' notes."""
    data = two_tenants_with_data
    headers = {"X-Tenant-ID": str(data["acme"].id)}

    response = await client.get("/api/notes/recent", headers=headers)
    assert response.status_code == 200

    titles = [n["title"] for n in response.json()]
    assert "Acme Note" in titles
    assert "Globex Note" not in titles, (
        "BREACH: the recent list includes another tenant's notes"
    )


@pytest.mark.asyncio
async def test_no_idor_via_note_id(client, two_tenants_with_data):
    """Anti-pattern 2: requesting another tenant's note_id must return a 404."""
    data = two_tenants_with_data
    headers = {"X-Tenant-ID": str(data["acme"].id)}

    response = await client.get(
        f"/api/notes/{data['globex_note'].id}",
        headers=headers,
    )
    assert response.status_code == 404, (
        f"BREACH: Globex's note (id={data['globex_note'].id}) "
        f"is accessible from Acme's session. Status={response.status_code}"
    )


@pytest.mark.asyncio
async def test_no_payload_tenant_id_injection(
    client, two_tenants_with_data
):
    """Anti-pattern 5: a tenant_id in the payload must be ignored or rejected."""
    data = two_tenants_with_data
    headers = {"X-Tenant-ID": str(data["acme"].id)}

    response = await client.post(
        "/api/notes",
        headers=headers,
        json={
            "folder_id": data["acme_folder"].id,
            "title": "Injected",
            "content": "leak attempt",
            "tenant_id": data["globex"].id,  # an injection attempt
        },
    )
    # Acceptable: 422 (Pydantic rejects the field) or 200 (it ignores it and uses current_tenant)
    if response.status_code == 200:
        # Verify the note got created in Acme, not in Globex
        note_id = response.json()["id"]
        check = await client.get(
            f"/api/notes/{note_id}",
            headers={"X-Tenant-ID": str(data["globex"].id)},
        )
        assert check.status_code == 404, (
            "BREACH: the payload's tenant_id was respected, the note got created in Globex"
        )


@pytest.mark.asyncio
async def test_no_idor_via_folder_id_in_create(
    client, two_tenants_with_data
):
    """Anti-pattern 2 + 3: using another tenant's folder_id when creating a note."""
    data = two_tenants_with_data
    headers = {"X-Tenant-ID": str(data["acme"].id)}

    response = await client.post(
        "/api/notes",
        headers=headers,
        json={
            "folder_id": data["globex_folder"].id,  # Globex's folder
            "title": "cross-folder attempt",
            "content": "...",
        },
    )
    assert response.status_code in (404, 422), (
        f"BREACH: I was able to create a note in Globex's folder from an Acme session. "
        f"Status={response.status_code}"
    )

Exercise 4: refactor a vulnerable cron job

Take this cron job and refactor it so it's multi-tenant safe. Assume the app has RLS enabled and an admin_app role with BYPASSRLS available only for listing tenants.

# app/jobs/weekly_summary.py
import asyncio
from sqlalchemy import select
from app.db.session import SessionLocal
from app.db.models import Project
from app.notifications import send_email


async def send_weekly_summaries():
    """Sends a weekly project summary to each PM."""
    async with SessionLocal() as db:
        # Reads ALL the projects from ALL the tenants
        result = await db.execute(
            select(Project).where(Project.archived_at.is_(None))
        )
        projects = result.scalars().all()

        for p in projects:
            await send_email(
                to=p.manager_email,
                subject=f"Weekly summary: {p.name}",
                body=f"Your project {p.name} has {p.task_count} tasks...",
            )


if __name__ == "__main__":
    asyncio.run(send_weekly_summaries())

# app/jobs/weekly_summary.py
import asyncio
import logging
from sqlalchemy import select, text

from app.db.session import SessionLocal, AdminSessionLocal
# AdminSessionLocal uses a role with BYPASSRLS, configured in app/db/session.py
from app.db.models import Project, Tenant
from app.notifications import send_email

logger = logging.getLogger(__name__)


async def list_active_tenant_ids() -> list[int]:
    """
    Uses the admin role with BYPASSRLS only to enumerate tenants.
    This is the job's only cross-tenant operation.
    """
    async with AdminSessionLocal() as db:
        async with db.begin():
            result = await db.execute(
                select(Tenant.id).where(Tenant.active.is_(True))
            )
            return list(result.scalars().all())


async def send_summary_for_tenant(tenant_id: int) -> int:
    """
    Processes a single tenant in its own transaction with its own context.
    RLS guarantees the queries only return this tenant's data.
    """
    async with SessionLocal() as db:
        async with db.begin():
            await db.execute(
                text("SET LOCAL app.tenant_id = :tid"),
                {"tid": str(tenant_id)},
            )

            result = await db.execute(
                select(Project).where(Project.archived_at.is_(None))
            )
            projects = list(result.scalars().all())

            for p in projects:
                await send_email(
                    to=p.manager_email,
                    subject=f"Weekly summary: {p.name}",
                    body=f"Your project {p.name} has {p.task_count} tasks...",
                )
            return len(projects)


async def send_weekly_summaries():
    """Orchestrates the job: lists tenants and processes each one in isolation."""
    tenant_ids = await list_active_tenant_ids()
    logger.info("Processing %d active tenants", len(tenant_ids))

    total_emails = 0
    for tenant_id in tenant_ids:
        try:
            count = await send_summary_for_tenant(tenant_id)
            total_emails += count
            logger.info("Tenant %d: %d emails sent", tenant_id, count)
        except Exception:
            logger.exception("Processing failed for tenant %d", tenant_id)
            # Continue with the next tenant — one failure doesn't abort the job

    logger.info("Job completed. Total emails: %d", total_emails)


if __name__ == "__main__":
    asyncio.run(send_weekly_summaries())

Exercise 5: explain why RLS isn't an excuse to ignore the other layers

Your tech lead says: "we already have RLS enabled. That guarantees isolation. We can remove the per-endpoint isolation tests and simplify the repository pattern to speed up dev." Articulate 3 arguments for why that's a bad idea.

Exercise 6: design a CI check that catches anti-pattern 5

Design a lint rule or a script that detects anti-pattern 5 (tenant_id in a Pydantic payload) in a codebase using FastAPI + Pydantic. The script has to run in CI and fail the build if it finds Pydantic models that have tenant_id as a field.

#!/bin/bash
# scripts/check_no_tenant_id_in_pydantic.sh
set -e

# Look for definitions like "tenant_id: int" or "tenant_id: str" in files
# that import pydantic.
violations=$(
    grep -rln "from pydantic" app/ tests/ \
        --include="*.py" \
    | xargs grep -nE "^\s+tenant_id:\s+(int|str)" 2>/dev/null \
    | grep -v "# allow-tenant-in-payload" \
    || true
)

if [ -n "$violations" ]; then
    echo "ERROR: Pydantic models cannot have tenant_id field."
    echo "tenant_id must come from authenticated context, never from client payload."
    echo ""
    echo "Violations:"
    echo "$violations"
    echo ""
    echo "If this is intentional (rare admin endpoint), add comment:"
    echo "    tenant_id: int  # allow-tenant-in-payload"
    exit 1
fi

echo "OK: no Pydantic models expose tenant_id field"

# .github/workflows/ci.yml
- name: Security checks
  run: |
    bash scripts/check_no_tenant_id_in_pydantic.sh

# scripts/check_no_tenant_id_in_pydantic.py
import ast
import sys
from pathlib import Path


def find_pydantic_models_with_tenant_id(file_path: Path) -> list[tuple[int, str]]:
    """Returns a list of (line, class_name) that have tenant_id."""
    tree = ast.parse(file_path.read_text())
    violations = []

    # Detect pydantic imports
    has_pydantic = any(
        isinstance(n, ast.ImportFrom) and n.module == "pydantic"
        for n in ast.walk(tree)
    )
    if not has_pydantic:
        return []

    for node in ast.walk(tree):
        if not isinstance(node, ast.ClassDef):
            continue

        # Does it inherit from BaseModel?
        is_pydantic = any(
            (isinstance(b, ast.Name) and b.id == "BaseModel")
            or (isinstance(b, ast.Attribute) and b.attr == "BaseModel")
            for b in node.bases
        )
        if not is_pydantic:
            continue

        for stmt in node.body:
            if isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name):
                if stmt.target.id == "tenant_id":
                    violations.append((stmt.lineno, node.name))

    return violations


def main(paths: list[str]) -> int:
    all_violations = []
    for path_str in paths:
        for py_file in Path(path_str).rglob("*.py"):
            for line, class_name in find_pydantic_models_with_tenant_id(py_file):
                all_violations.append((py_file, line, class_name))

    if not all_violations:
        print("OK: no Pydantic models expose tenant_id field")
        return 0

    print("ERROR: Pydantic models cannot have tenant_id field.")
    print("tenant_id must come from authenticated context, never from client.")
    print()
    for f, line, name in all_violations:
        print(f"  {f}:{line} — class {name}")
    return 1


if __name__ == "__main__":
    sys.exit(main(["app", "tests"]))

Summary and next step

In this capsule you learned:

• Five concrete anti-patterns that end in a cross-tenant leak: a direct query with no filter, db.get() without checking the tenant, a filter lost in a branching query builder, a cron job with no tenant context, and a tenant_id received from the client's payload.
• The bonus configuration anti-pattern: the app's role is the tables' owner, which silently disables RLS without anyone noticing until the incident.
• How RLS prevents each anti-pattern even when the code has a bug, demonstrating why the combination "defensive code + RLS" is the recommended architecture.
• The defense in depth principle: six stacked layers (discipline, repository, lint, code review, tests, RLS) catch more than any individual layer. Removing layers because "RLS covers it" is a false simplification.
• Concrete isolation tests covering the anti-patterns in the CI environment with two real tenants.
• How to detect anti-patterns in code review with reproducible vocabulary you can teach new devs.

Before moving on you should be able to:

• Recognize any of the five anti-patterns in a PR at reading speed.
• Explain to a teammate why RLS doesn't replace the other defense layers.
• Design a pytest test that reproduces a specific anti-pattern against two tenants.
• Configure a CI check (grep or AST) that detects anti-pattern 5 (tenant_id in a Pydantic payload).
• Refactor a cron job vulnerable to the "no tenant context" pattern using BYPASSRLS only to enumerate tenants and SET LOCAL per iteration.

Next capsule — Project: multi-tenant TaskFlow with RLS. You're going to put the whole module into practice by building a mini SaaS task management API with three fictional tenants (Acme, Globex, and Initech). You're going to implement a shared schema + RLS + an audit log + cursor pagination + soft deletes in a single codebase, and the project's "malicious" tests are going to verify that none of this capsule's anti-patterns slip through. The capstone project of module 8 (the complete TaskFlow) extends exactly this foundation.

Resources

1. PostgreSQL — Row Security Policies (especially WITH CHECK) — the official reference for policies, required reading for understanding the nuances of each operation.
2. Supabase — Row Level Security guide — a pragmatic guide from the provider that most popularized RLS for modern multi-tenancy; it includes examples of real anti-patterns.
3. OWASP — Insecure Direct Object References (IDOR) — OWASP's top ten, covering anti-pattern 2 from a generic security perspective, not just multi-tenant.
4. GitHub Engineering Blog — postmortem analysis — a collection of postmortems where several incidents cover filtering anti-patterns (search for "tenant" or "isolation").
5. Crunchy Data — Row Level Security for Tenants — a practical implementation with an emphasis on FORCE ROW LEVEL SECURITY and the owner problem.
6. Brandur — Postgres-only stacks at scale — a perspective on defense in depth and each layer's limits in real systems.
7. HashiCorp — Multi-tenancy security patterns whitepaper — a general reference on isolation patterns in multi-tenant systems (not just PostgreSQL).
8. pganalyze — Best practices for Postgres RLS — a detailed technical analysis with performance benchmarks for policies.

Module 4 — SQL Patterns for Production APIs Guide

Next capsule: The multi-tenant TaskFlow project with RLS — you put all the module's patterns into practice in a real mini-API with three tenants and isolation tests that catch this capsule's anti-patterns.