Learning Hub

How to Stop Cross-Tenant Data Leaks in a FastAPI App With PostgreSQL Row-Level Security

Learn to stop cross-tenant data leaks in a FastAPI and PostgreSQL app with row-level security, including the owner, connection-pool, unique-key, foreign-key and view traps, backed by 14 tests and...

October 16, 2026 38 Min Read
41

What you will build, and why one forgotten WHERE can leak a whole database

Many web products serve many customers from one program and one database. Each customer is called a tenant. Acme Plumbing’s invoices and Birch Bakery’s invoices sit in the same invoices table, and the only thing keeping them apart is a tenant_id column. The usual protection is a habit: every query must end with WHERE tenant_id = %s. The habit works until one query, written in a hurry, leaves the filter out. Nothing crashes and no error is raised. Acme simply starts seeing Birch’s data. Our tutorial on a multi-tenant SaaS API with Node.js builds that habit into a repository layer, and OWASP’s Multi-Tenant Security Cheat Sheet adds a second line of defense: “Add data-layer checks as defense in depth where the architecture supports them.”

Table Of Content

PostgreSQL has such a data-layer check built in. It is called row-level security, or RLS. You attach a rule to a table, and from then on the database applies the rule to every query that an ordinary role runs against that table, whatever SQL the application sends. The PostgreSQL manual describes the effect this way: “all normal access to the table for selecting rows or modifying rows must be allowed by a row security policy.” With a rule in place, the forgotten WHERE stops being a data leak, because the database adds the missing filter for you.

In this tutorial you will reproduce the forgotten-filter bug, then fix it with RLS in small steps, and at most steps you will break something on purpose so you can see why the step is needed. You will meet the table-owner loophole that makes a fresh policy do nothing, the connection-pool trap that carries one customer’s identity into the next customer’s request, three places where tenants can still learn about each other (a unique constraint, a foreign key and a view), and the limit of the whole approach. Then you will wire it into a FastAPI app that contains two deliberate bugs, add a test suite that fails if the protection is ever removed, and measure what the policy costs on a table with two million rows.

Everything below was run, not copied: PostgreSQL 18.6 on Windows 11, Python 3.13.14, psycopg 3.3.6 and FastAPI 0.143.0. Process ids and timings will differ on your machine; the patterns are what to look for. One warning before you start. Row-level security protects you from mistakes in your application code. It does not protect you from someone who can run any SQL they like as your application’s database role, and Step 9 shows why. Think of it as a seat belt, not a vault.

Terms to know before we start

Tenant. One customer organization that uses your product. Its data must never be visible to another tenant.

Role. PostgreSQL’s word for a database login (or a group of logins). Roles own tables and receive permissions.

Table owner. The role that created a table. Owners get special treatment in row-level security, which is the subject of Step 5.

Policy. A rule attached to a table that says which rows a role may see or change. A policy has a USING expression for rows that already exist and, optionally, a WITH CHECK expression for rows being written.

Connection pool. A small set of database connections that a program keeps open and lends to requests, because opening a connection is slow. One connection serves many requests, and therefore many tenants, over its lifetime.

Session setting. A named value, such as app.tenant_id, that a database connection remembers and that SQL can read with current_setting(). PostgreSQL calls these configuration parameters. We use one to tell the database which tenant a request belongs to.

Prerequisites

  • Python 3.13 (I used 3.13.14) and pip.
  • A throwaway PostgreSQL 18 server on your own machine, where you can connect as the postgres superuser. I used 18.6. The lab connects to 127.0.0.1 on port 55432 with trust authentication, which means no passwords. That is acceptable only for a disposable local server. Step 2 drops and recreates a database called tenantlab and two roles, so never aim it at a shared or production server. If your server listens elsewhere, set the environment variables LAB_PGHOST and LAB_PGPORT.
  • Basic SQL and Python. You should know what SELECT, INSERT and a primary key are. You do not need to know FastAPI or row-level security; both are explained as they appear.
  • About an hour. Each step runs in seconds; the performance test in Step 12 takes about ten.

Step 1: Create the project and two small helpers

Create an empty folder, a virtual environment inside it, and install the packages. All the files in this tutorial live in that one folder and import each other, so keep them together. The httpx2 package is there because Starlette’s test client, which Steps 10 and 11 use, requires it.

python -m venv .venv
# activate it: .venv\Scripts\activate on Windows, source .venv/bin/activate on Linux and macOS
pip install "psycopg[binary,pool]==3.3.6" fastapi==0.143.0 uvicorn==0.54.0 httpx2==2.13.1 pytest==9.1.1

Now create lab.py. It holds two ordinary helpers, dsn() and connect(), that build a connection to a given role, and one helper that you will use all the way through, attempt().

"""lab.py: connection helpers shared by every step."""
import os

import psycopg

HOST = os.environ.get("LAB_PGHOST", "127.0.0.1")
PORT = os.environ.get("LAB_PGPORT", "55432")
DBNAME = "tenantlab"


def dsn(user, dbname=DBNAME):
    return f"host={HOST} port={PORT} user={user} dbname={dbname}"


def connect(user, dbname=DBNAME, **kwargs):
    return psycopg.connect(dsn(user, dbname), **kwargs)


def attempt(tenant_id, sql, params=(), user="app_user"):
    """Run one statement as `tenant_id` and always roll it back; return a one-line result."""
    with connect(user) as conn:
        if tenant_id is not None:
            conn.execute("SELECT set_config('app.tenant_id', %s, true)", [str(tenant_id)])
        try:
            cur = conn.execute(sql, params)
            if cur.description:  # a SELECT or RETURNING: show the rows
                return f"ok, rows={cur.fetchall()}"
            return f"ok, {cur.rowcount} row(s) changed"
        except psycopg.Error as e:
            detail = f" [{e.diag.message_detail}]" if e.diag.message_detail else ""
            return f"{type(e).__name__}: {e.diag.message_primary}{detail}"
        finally:
            conn.rollback()

attempt() runs one SQL statement as a chosen role, optionally after telling the database which tenant the request belongs to, and returns a one-line summary: the rows, the number of rows changed, or the error. It always rolls back, so you can try destructive statements without changing a single row. The call you see inside it, set_config('app.tenant_id', ..., true), is the heart of the technique. Step 6 explains the third argument; for now read it as “remember this tenant id until the end of this transaction.”

Step 2: Create the lab database, two roles and some data

The next script builds everything the rest of the tutorial uses: a database, two roles, three tables and a few rows for two tenants. It is safe to run again whenever you want a clean start.

"""step02_setup.py: create the lab database, two roles, three tables and a little data."""
from lab import connect

STATEMENTS = [
    "DROP DATABASE IF EXISTS tenantlab WITH (FORCE)",
    "DROP ROLE IF EXISTS app_user",
    "DROP ROLE IF EXISTS app_owner",
    "CREATE ROLE app_owner LOGIN",  # owns the tables; used for migrations only
    "CREATE ROLE app_user LOGIN",   # what the web application connects as
    "CREATE DATABASE tenantlab OWNER app_owner",
]

SCHEMA = """
CREATE TABLE tenants (
    id   integer PRIMARY KEY,
    name text NOT NULL
);
CREATE TABLE customers (
    id        integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id integer NOT NULL REFERENCES tenants (id),
    email     text NOT NULL UNIQUE,
    name      text NOT NULL
);
CREATE TABLE invoices (
    id           integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id    integer NOT NULL REFERENCES tenants (id),
    customer_id  integer NOT NULL REFERENCES customers (id),
    amount_cents integer NOT NULL
);
GRANT SELECT, INSERT, UPDATE, DELETE ON customers, invoices TO app_user;
"""

SEED = """
INSERT INTO tenants VALUES (1, 'Acme Plumbing'), (2, 'Birch Bakery');
INSERT INTO customers (tenant_id, email, name) VALUES
    (1, '[email protected]', 'Alice Ng'),
    (1, '[email protected]',  'Omar Reyes'),
    (2, '[email protected]',   'Bob Katz');
INSERT INTO invoices (tenant_id, customer_id, amount_cents) VALUES
    (1, 1, 12000), (1, 1, 4550), (1, 2, 88000),
    (2, 3, 2300),  (2, 3, 9900);
"""

with connect("postgres", dbname="postgres", autocommit=True) as admin:
    for statement in STATEMENTS:
        admin.execute(statement)

with connect("app_owner", autocommit=True) as owner:
    owner.execute(SCHEMA)
    owner.execute(SEED)

with connect("app_user") as conn:
    print("role         superuser  bypassrls")
    for name, superuser, bypass in conn.execute(
        "SELECT rolname, rolsuper, rolbypassrls FROM pg_roles "
        "WHERE rolname IN ('postgres', 'app_owner', 'app_user') ORDER BY rolname"
    ):
        print(f"{name:<12} {superuser!s:<10} {bypass}")
    print()
    for table, owner_name in conn.execute(
        "SELECT tablename, tableowner FROM pg_tables WHERE schemaname = 'public' ORDER BY tablename"
    ):
        print(f"table {table:<10} owned by {owner_name}")

Run it with python step02_setup.py. You should see:

role         superuser  bypassrls
app_owner    False      False
app_user     False      False
postgres     True       True

table customers  owned by app_owner
table invoices   owned by app_owner
table tenants    owned by app_owner

What just happened

Two roles matter. app_owner creates and owns the tables; in a real project it is the role your migrations use. app_user is the role your web application connects as. It may read and write the two tenant tables and nothing else. It does not own them, so it cannot change their policies, and it has no permission at all on the tenants table. It also has no TRUNCATE permission, which matters because the manual says that “Operations that apply to the whole table, such as TRUNCATE and REFERENCES, are not subject to row security.” That separation is not decoration. PostgreSQL treats table owners differently from other roles, and a web app that connects as its own table owner can end up with a policy that does nothing. Step 5 proves it.

The output also shows that the postgres role, the superuser of this server, has its bypassrls column set to True, so it ignores row security completely. Remember that: never point your application at a superuser, not even in development, or the policies you write will never run.

The tables are deliberately naive in two places. customers.email is unique across the whole table, and invoices.customer_id references customers (id) on its own. Both look harmless. Step 8 shows how each one lets tenants learn about each other, and fixes them.

Step 3: Reproduce the bug

Before fixing anything, watch the failure happen. The script below plays an Acme employee using an application whose developer wrote two data-access functions. list_invoices remembers the tenant filter. get_invoice does not.

"""step03_forgotten_where.py: reproduce the bug this tutorial exists to contain."""
from lab import connect


def list_invoices(conn, tenant_id):
    return conn.execute(
        "SELECT id, tenant_id, amount_cents FROM invoices WHERE tenant_id = %s ORDER BY id",
        [tenant_id],
    ).fetchall()


def get_invoice(conn, invoice_id):
    # The bug: whoever wrote this query forgot "AND tenant_id = ...".
    return conn.execute(
        "SELECT id, tenant_id, amount_cents FROM invoices WHERE id = %s", [invoice_id]
    ).fetchone()


with connect("app_user") as conn:
    print("Acme (tenant 1) lists its invoices :", list_invoices(conn, 1))
    print("Acme asks for invoice 2 (its own)  :", get_invoice(conn, 2))
    print("Acme asks for invoice 4 (Birch's)  :", get_invoice(conn, 4))

Run python step03_forgotten_where.py:

Acme (tenant 1) lists its invoices : [(1, 1, 12000), (2, 1, 4550), (3, 1, 88000)]
Acme asks for invoice 2 (its own)  : (2, 1, 4550)
Acme asks for invoice 4 (Birch's)  : (4, 2, 2300)

The last line is the leak. Acme asked for invoice 4 and received a row whose tenant_id is 2: Birch Bakery’s invoice, with its amount. No error, no warning. This is not an exotic mistake. It is what you get when someone adds a “get one invoice” endpoint by copying a query, or uses an ORM call that fetches a row by primary key. For the rest of the tutorial we leave that function exactly as it is and make the database refuse to hand out the wrong row.

Step 4: Turn on row-level security and write the first policy

Row-level security is switched on per table with ALTER TABLE invoices ENABLE ROW LEVEL SECURITY and then shaped by one or more policies. Before writing a policy it is worth seeing what the switch does on its own, because that shows what PostgreSQL does when you forget the policy.

First, the application’s queries, in a module of their own. Notice that no query filters by tenant:

"""repo.py: the application's queries. None of them filters by tenant, on purpose."""


def list_invoices(conn):
    return conn.execute(
        "SELECT id, tenant_id, customer_id, amount_cents FROM invoices ORDER BY id"
    ).fetchall()


def get_invoice(conn, invoice_id):
    return conn.execute(
        "SELECT id, tenant_id, customer_id, amount_cents FROM invoices WHERE id = %s", [invoice_id]
    ).fetchone()


def create_invoice(conn, tenant_id, customer_id, amount_cents):
    return conn.execute(
        "INSERT INTO invoices (tenant_id, customer_id, amount_cents) VALUES (%s, %s, %s) RETURNING id",
        [tenant_id, customer_id, amount_cents],
    ).fetchone()

Now the step itself. The script enables RLS on both tenant tables, counts invoices as the owner and as the application role, creates a policy on each table, and repeats the checks.

"""step04_policy.py: turn on row-level security, see the default-deny rule, then add a policy."""
from lab import attempt, connect
from repo import get_invoice

COUNT = "SELECT count(*) FROM invoices"
IDS = "SELECT id FROM invoices ORDER BY id"

with connect("app_owner", autocommit=True) as owner:
    owner.execute("ALTER TABLE invoices ENABLE ROW LEVEL SECURITY")
    owner.execute("ALTER TABLE customers ENABLE ROW LEVEL SECURITY")

print("RLS enabled, no policy yet")
print("  app_owner counts invoices :", attempt(None, COUNT, user="app_owner"))
print("  app_user  counts invoices :", attempt(None, COUNT))

POLICIES = """
CREATE POLICY tenant_isolation ON invoices
    USING (tenant_id = current_setting('app.tenant_id')::integer);
CREATE POLICY tenant_isolation ON customers
    USING (tenant_id = current_setting('app.tenant_id')::integer);
"""
with connect("app_owner", autocommit=True) as owner:
    owner.execute(POLICIES)

print("\nafter CREATE POLICY")
print("  app_user, no tenant set   :", attempt(None, COUNT))
print("  app_user as tenant 1      :", attempt(1, IDS))
print("  app_user as tenant 2      :", attempt(2, IDS))
print("  app_owner, no tenant set  :", attempt(None, COUNT, user="app_owner"))

with connect("app_user") as conn:
    conn.execute("SELECT set_config('app.tenant_id', '1', true)")
    print("\nAcme asks for invoice 2 (its own)  :", get_invoice(conn, 2))
    print("Acme asks for invoice 4 (Birch's)  :", get_invoice(conn, 4))

Run python step04_policy.py:

RLS enabled, no policy yet
  app_owner counts invoices : ok, rows=[(5,)]
  app_user  counts invoices : ok, rows=[(0,)]

after CREATE POLICY
  app_user, no tenant set   : UndefinedObject: unrecognized configuration parameter "app.tenant_id"
  app_user as tenant 1      : ok, rows=[(1,), (2,), (3,)]
  app_user as tenant 2      : ok, rows=[(4,), (5,)]
  app_owner, no tenant set  : ok, rows=[(5,)]

Acme asks for invoice 2 (its own)  : (2, 1, 1, 4550)
Acme asks for invoice 4 (Birch's)  : None

Reading the output

Default deny. Right after ENABLE ROW LEVEL SECURITY and before any policy, app_user counts 0 invoices. The manual says: “If no policy exists for the table, a default-deny policy is used, meaning that no rows are visible or can be modified.” Forgetting a policy therefore makes the table look empty, which fails safe.

The policy. USING (tenant_id = current_setting('app.tenant_id')::integer) says that a row is visible only when its tenant_id equals the number stored in the session setting app.tenant_id. The manual explains what current_setting does when nobody has set that value: “If there is no such setting, current_setting throws an error unless missing_ok is supplied and is true (in which case NULL is returned).” That is the error on the first line after the policy, and it also fails safe.

The result. As tenant 1, the application sees invoices 1 to 3. As tenant 2, it sees 4 and 5. And the buggy lookup from Step 3 now returns None for invoice 4. The query is unchanged; the database filtered the row out before the application could see it.

The loophole. The owner, app_owner, still counts 5 invoices, both before and after the policy exists. The manual warns about this: “the table’s owner is typically not subject to row security policies.” Step 5 closes it.

Step 5: Close the owner loophole and refuse privileged roles

The manual lists the exemptions in one place. First: “Superusers and roles with the BYPASSRLS attribute always bypass the row security system when accessing a table.” Second: “Table owners normally bypass row security as well”, though an owner can opt in with ALTER TABLE invoices FORCE ROW LEVEL SECURITY. That gives us two jobs.

The first job is to run FORCE ROW LEVEL SECURITY on both tables, so that even app_owner is filtered. This protects you on the day someone points the application at the owner role by mistake. The second job is a guard that refuses to start the application as a superuser or as a role with the BYPASSRLS attribute, because FORCE does not constrain those. OWASP’s cheat sheet says so directly: “It does not constrain a superuser or a role with the BYPASSRLS attribute”. Here is the guard. It starts the file tenancy.py, which the next step completes:

"""tenancy.py: the two helpers every request goes through."""
from contextlib import contextmanager

from psycopg.rows import tuple_row


def assert_unprivileged(conn):
    """Refuse to run as a role that skips row-level security."""
    cur = conn.cursor(row_factory=tuple_row)  # the pool may hand out dict rows; unpack plain tuples
    superuser, bypass = cur.execute(
        "SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user"
    ).fetchone()
    if superuser or bypass:
        raise RuntimeError("this role bypasses row-level security; connect as a plain application role")

The guard asks PostgreSQL about the connection itself. The pg_roles view lists every role with two flags, rolsuper and rolbypassrls, and current_user is the role this connection is using. OWASP recommends exactly this kind of check: “Check the deployed request role, not only the role declared in source-controlled configuration.” One detail needs explaining: the first line of the function asks for a plain tuple row. The FastAPI app in Step 10 hands out dictionary rows, and my first version unpacked such a row into two variables, received the two column names, which are non-empty strings, and refused to start as if every role were privileged.

Now apply FORCE and test the guard:

"""step05_force.py: close the owner loophole and refuse privileged roles."""
from lab import attempt, connect
from tenancy import assert_unprivileged

COUNT = "SELECT count(*) FROM invoices"
NEW = "INSERT INTO invoices (tenant_id, customer_id, amount_cents) VALUES (1, 1, 5000)"


def show(label, result):
    print(f"{label:<38}: {result}")


show("before FORCE, owner, no tenant set", attempt(None, COUNT, user="app_owner"))

with connect("app_owner", autocommit=True) as owner:
    owner.execute("ALTER TABLE invoices FORCE ROW LEVEL SECURITY")
    owner.execute("ALTER TABLE customers FORCE ROW LEVEL SECURITY")

show("after FORCE, owner, no tenant set", attempt(None, COUNT, user="app_owner"))
show("after FORCE, owner as tenant 1", attempt(1, COUNT, user="app_owner"))
show("after FORCE, owner inserts, no tenant", attempt(None, NEW, user="app_owner"))
show("superuser, no tenant set", attempt(None, COUNT, user="postgres"))

for role in ("app_user", "postgres"):
    with connect(role) as conn:
        try:
            assert_unprivileged(conn)
            show(f"assert_unprivileged as {role}", "passes")
        except RuntimeError as exc:
            show(f"assert_unprivileged as {role}", f"RuntimeError: {exc}")

Run python step05_force.py:

before FORCE, owner, no tenant set    : ok, rows=[(5,)]
after FORCE, owner, no tenant set     : UndefinedObject: unrecognized configuration parameter "app.tenant_id"
after FORCE, owner as tenant 1        : ok, rows=[(3,)]
after FORCE, owner inserts, no tenant : UndefinedObject: unrecognized configuration parameter "app.tenant_id"
superuser, no tenant set              : ok, rows=[(5,)]
assert_unprivileged as app_user       : passes
assert_unprivileged as postgres       : RuntimeError: this role bypasses row-level security; connect as a plain application role

Reading the output

Before FORCE, the owner counts all 5 invoices. After it, the owner gets the same error as any other role until a tenant is set, and then sees only that tenant’s 3 invoices. A data migration that runs as the owner and inserts rows now needs a tenant too, as the “owner inserts” line shows; schema changes such as ALTER TABLE are not affected, and Step 8 runs several of them as the owner after FORCE. The superuser still counts 5, exactly as OWASP warned. The guard passes for app_user and raises for postgres, so an application that starts with the wrong credentials will stop instead of silently running without protection.

Step 6: Set the tenant per transaction, not per connection

So far we set the tenant by hand. In a real application every request has to do it, and how it does it matters more than it looks, because of connection pools. A pool keeps a few connections open and lends one to each request. When a request finishes, the pool takes the connection back and lends it to the next request, which may belong to a different tenant.

Session or transaction: the third argument of set_config

PostgreSQL’s set_config(name, value, is_local) can store a setting in two ways. The manual says: “If is_local is true, the new value will only apply during the current transaction. If you want the new value to apply for the rest of the current session, use false instead.” For the SET command the matching statement is: “The effects of SET LOCAL last only till the end of the current transaction, whether committed or not.” A session here means the life of one database connection, and a pool keeps connections alive across many requests. psycopg’s connection pool documents what it does when a request gives a connection back: the open transaction is committed or rolled back, and “if a reset callback was provided, the connection is passed to it, to allow application-specific cleanup if needed.” Nothing in that list clears a session setting on its own. OWASP draws the conclusion for tenant data: “A normal SET persists until the session ends after its transaction commits, while SET LOCAL lasts only for the current transaction.” Its rule for the application is: “Never assume a newly borrowed connection has the correct setting.”

So the helper every request will use opens a transaction, sets the tenant with is_local true, and hands the connection to the caller. Add it to tenancy.py:

# tenancy.py (continued)
@contextmanager
def tenant_transaction(pool, tenant_id):
    """Borrow a connection and open a transaction in which app.tenant_id is set.

    set_config(..., true) means "local to this transaction": the value disappears at
    COMMIT or ROLLBACK, so it cannot leak to the next request that gets this connection.
    """
    with pool.connection() as conn:
        with conn.transaction():
            conn.execute("SELECT set_config('app.tenant_id', %s, true)", [str(int(tenant_id))])
            yield conn

Three contexts are nested on purpose. pool.connection() borrows a connection and gives it back; conn.transaction() opens a transaction and commits it at the end; and the set_config call inside the transaction runs before the caller’s queries. When the block ends, the tenant value ends with it.

Watch the pool leak, then fix it

The script below creates a pool with exactly one connection, so every request reuses it. It runs two requests each time: request A names tenant 1, and request B forgets to name any tenant, the way a background job or a new code path might. The first number in each result is the PostgreSQL backend process id; seeing the same number for A and B proves they used the same connection. The script does this three times: with a session-level setting, with the transaction-level helper, and finally after the fix that is explained below.

"""step06_pool.py: set the tenant per transaction, not per connection; and the empty-string trap."""
import psycopg
from psycopg_pool import ConnectionPool

from lab import connect, dsn
from repo import list_invoices
from tenancy import tenant_transaction


def new_pool():
    pool = ConnectionPool(dsn("app_user"), min_size=1, max_size=1, open=True)  # one connection: every request reuses it
    pool.wait()
    return pool


def request_with_session_setting(pool, tenant_id):
    with pool.connection() as conn:
        conn.execute("SELECT set_config('app.tenant_id', %s, false)", [str(tenant_id)])  # false = whole session
        return conn.info.backend_pid, [(r[0], r[1]) for r in list_invoices(conn)]


def request_with_transaction_setting(pool, tenant_id):
    with tenant_transaction(pool, tenant_id) as conn:
        return conn.info.backend_pid, [(r[0], r[1]) for r in list_invoices(conn)]


def request_without_context(pool):
    with pool.connection() as conn:
        setting = conn.execute("SELECT current_setting('app.tenant_id', true)").fetchone()[0]
        try:
            rows = [(r[0], r[1]) for r in list_invoices(conn)]
        except psycopg.Error as exc:
            rows = f"{type(exc).__name__}: {exc.diag.message_primary}"
        conn.rollback()
        return conn.info.backend_pid, setting, rows


print("1. tenant stored with set_config(..., false), the whole session")
pool = new_pool()
print("   request A (tenant 1):", request_with_session_setting(pool, 1))
print("   request B (no tenant):", request_without_context(pool))
pool.close()

print("\n2. tenant stored with set_config(..., true), this transaction only")
pool = new_pool()
print("   request A (tenant 1):", request_with_transaction_setting(pool, 1))
print("   request B (no tenant):", request_without_context(pool))
pool.close()

Run python step06_pool.py. The first two demonstrations print:

1. tenant stored with set_config(..., false), the whole session
   request A (tenant 1): (7644, [(1, 1), (2, 1), (3, 1)])
   request B (no tenant): (7644, '1', [(1, 1), (2, 1), (3, 1)])

2. tenant stored with set_config(..., true), this transaction only
   request A (tenant 1): (19520, [(1, 1), (2, 1), (3, 1)])
   request B (no tenant): (19520, '', 'InvalidTextRepresentation: invalid input syntax for type integer: ""')

Demonstration 1 is a data leak. Request B named no tenant, yet it received Acme’s three invoices, and the second value in its result shows why: the connection still held '1' from request A. In production, B would be another customer’s request.

Demonstration 2 is safe, but strange. With the transaction-level setting, B receives no data. It receives an error instead, and the setting it reads is an empty string. The empty string is the interesting part. In my test, once a connection had used the setting name, the value read back as an empty string, not as missing, after the transaction ended. The policy’s cast of that empty string to an integer then fails with invalid input syntax for type integer. That is a different error from the one a brand-new connection gave in Step 4, so the same bug reports itself two ways depending on whether the connection is fresh or reused. OWASP leaves out the missing_ok argument in its example policy on purpose, calling the resulting error “a deliberate fail-closed choice”, and that is a fair design. The catch is only that the error is not the same every time.

Make a missing tenant mean “no rows”

The fix is a small function that turns both situations, never set and set to an empty string, into NULL, and a policy that compares against it. The manual says that for a policy expression “if false or null is returned then the row is not visible”, so NULL means no row matches. The function is marked STABLE because it reads a setting but changes nothing, and ALTER POLICY swaps the expression without dropping the policy.

# step06_pool.py (continued)
FIX = """
CREATE FUNCTION app_tenant_id() RETURNS integer
    LANGUAGE sql STABLE
    AS $$ SELECT NULLIF(current_setting('app.tenant_id', true), '')::integer $$;

ALTER POLICY tenant_isolation ON invoices  USING (tenant_id = app_tenant_id());
ALTER POLICY tenant_isolation ON customers USING (tenant_id = app_tenant_id());
"""
with connect("app_owner", autocommit=True) as owner:
    owner.execute(FIX)

print("\n3. after the fix: a policy that treats a missing tenant as no tenant")
pool = new_pool()
print("   request A (tenant 1):", request_with_transaction_setting(pool, 1))
print("   request B (no tenant):", request_without_context(pool))
pool.close()

The third demonstration prints:

3. after the fix: a policy that treats a missing tenant as no tenant
   request A (tenant 1): (13712, [(1, 1), (2, 1), (3, 1)])
   request B (no tenant): (13712, '', [])

Now request B gets an empty list on the reused connection. Either design fails closed; I prefer this one because a forgotten tenant behaves the same on a fresh and a reused connection. If you would rather have a loud error for a missing tenant, keep the plain current_setting form and test both cases. What you must not do is store the tenant for the whole session. The rule to remember is short: call set_config(..., true) inside the same transaction that runs the queries, on every request.

Step 7: Protect writes, not just reads

A policy has two halves. The manual says: “Existing table rows are checked against the expression specified in USING, while new rows that would be created via INSERT or UPDATE are checked against the expression specified in WITH CHECK.” We wrote only a USING expression, and the manual covers that case too: “If only a USING clause is specified, then that clause will be used for both USING and WITH CHECK cases.” So the same rule that hides other tenants’ rows should also stop a tenant from writing rows into another tenant. The script below tries six writes. Every attempt is rolled back, so your data stays as it was.

"""step07_writes.py: policies protect writes too. Every attempt is rolled back."""
from lab import attempt

NEW = "INSERT INTO invoices (tenant_id, customer_id, amount_cents) VALUES (%s, %s, 5000)"
MOVE = "UPDATE invoices SET tenant_id = 2 WHERE id = 1"
TOUCH = "UPDATE invoices SET amount_cents = 0 WHERE id = 4"
WIPE = "DELETE FROM invoices WHERE id = 4"

print("tenant 1 inserts its own invoice      :", attempt(1, NEW, [1, 1]))
print("tenant 1 inserts one for tenant 2     :", attempt(1, NEW, [2, 3]))
print("tenant 1 moves its invoice to tenant 2:", attempt(1, MOVE))
print("tenant 1 updates invoice 4 (Birch's)  :", attempt(1, TOUCH))
print("tenant 1 deletes invoice 4 (Birch's)  :", attempt(1, WIPE))
print("tenant 2 deletes invoice 4 (its own)  :", attempt(2, WIPE))

Run python step07_writes.py:

tenant 1 inserts its own invoice      : ok, 1 row(s) changed
tenant 1 inserts one for tenant 2     : InsufficientPrivilege: new row violates row-level security policy for table "invoices"
tenant 1 moves its invoice to tenant 2: InsufficientPrivilege: new row violates row-level security policy for table "invoices"
tenant 1 updates invoice 4 (Birch's)  : ok, 0 row(s) changed
tenant 1 deletes invoice 4 (Birch's)  : ok, 0 row(s) changed
tenant 2 deletes invoice 4 (its own)  : ok, 1 row(s) changed

Reading the output

Tenant 1 can insert an invoice for itself. It cannot insert one for tenant 2, and it cannot move its own invoice to tenant 2 with an UPDATE; both fail with new row violates row-level security policy. That error is the WITH CHECK half at work.

The next two lines are different. Tenant 1 tried to update and delete invoice 4, which belongs to Birch, and PostgreSQL reported success with 0 rows changed. No error was raised because, as the manual puts it, “Typically, no error occurs when a row is not visible.” To the policy, Birch’s invoice does not exist for tenant 1, so there was nothing to update. This has a practical consequence for your code: when an update or delete reports 0 rows, answer “not found” (HTTP 404), not “done.” The last line shows that tenant 2 can delete its own invoice 4, so the policy blocks only the wrong tenant.

Step 8: Close three side doors that RLS leaves open

A policy filters the rows a query reads and writes. It does not filter everything a database does with those rows. The manual is direct about one family of exceptions: “Referential integrity checks, such as unique or primary key constraints and foreign key references, always bypass row security to ensure that data integrity is maintained.” It tells schema designers to avoid information leaks through these checks, which it calls covert channels. Views are a second door. The script below walks through all three, then applies the fixes and repeats the checks.

"""step08_side_doors.py: three leaks that row-level security does not close by itself."""
from lab import attempt, connect

DUP = "INSERT INTO customers (tenant_id, email, name) VALUES (2, '[email protected]', 'Alice B')"
REF = "INSERT INTO invoices (tenant_id, customer_id, amount_cents) VALUES (2, %s, 100)"
TOTALS = "SELECT tenant_id, invoices, total_cents FROM invoice_totals ORDER BY 1"

print("A. unique constraint")
print("   Birch adds Acme's customer email :", attempt(2, DUP))

print("\nB. foreign key")
print("   Birch invoices Acme's customer 1 :", attempt(2, REF, [1]))
print("   Birch invoices customer 999      :", attempt(2, REF, [999]))

with connect("postgres", autocommit=True) as admin:  # a migration that ran as the superuser
    admin.execute(
        "CREATE VIEW invoice_totals AS "
        "SELECT tenant_id, count(*) AS invoices, sum(amount_cents) AS total_cents "
        "FROM invoices GROUP BY tenant_id"
    )
    admin.execute("GRANT SELECT ON invoice_totals TO app_user")

with connect("app_owner", autocommit=True) as owner:  # the same view, created by the table owner
    owner.execute(
        "CREATE VIEW owner_totals AS "
        "SELECT tenant_id, count(*) AS invoices, sum(amount_cents) AS total_cents "
        "FROM invoices GROUP BY tenant_id"
    )
    owner.execute("GRANT SELECT ON owner_totals TO app_user")

print("\nC. views")
print("   created by the superuser         :", attempt(1, TOTALS))
print("   created by app_owner (FORCE on)  :", attempt(1, TOTALS.replace("invoice_totals", "owner_totals")))

FIXES = """
ALTER TABLE customers DROP CONSTRAINT customers_email_key;
ALTER TABLE customers ADD CONSTRAINT customers_tenant_email_key UNIQUE (tenant_id, email);

ALTER TABLE customers ADD CONSTRAINT customers_tenant_id_id_key UNIQUE (tenant_id, id);
ALTER TABLE invoices DROP CONSTRAINT invoices_customer_id_fkey;
ALTER TABLE invoices ADD CONSTRAINT invoices_customer_fk
    FOREIGN KEY (tenant_id, customer_id) REFERENCES customers (tenant_id, id);
"""
with connect("app_owner", autocommit=True) as owner:
    owner.execute(FIXES)
    owner.execute("DROP VIEW owner_totals")
with connect("postgres", autocommit=True) as admin:
    admin.execute("ALTER VIEW invoice_totals SET (security_invoker = true)")

print("\nafter the fixes")
print("   A. Birch adds the same email     :", attempt(2, DUP))
print("   B. Birch invoices Acme's customer:", attempt(2, REF, [1]))
print("   B. Birch invoices customer 999   :", attempt(2, REF, [999]))
print("   C. tenant 1 reads invoice_totals :", attempt(1, TOTALS))

Run python step08_side_doors.py:

A. unique constraint
   Birch adds Acme's customer email : UniqueViolation: duplicate key value violates unique constraint "customers_email_key"

B. foreign key
   Birch invoices Acme's customer 1 : ok, 1 row(s) changed
   Birch invoices customer 999      : ForeignKeyViolation: insert or update on table "invoices" violates foreign key constraint "invoices_customer_id_fkey" [Key is not present in table "customers".]

C. views
   created by the superuser         : ok, rows=[(1, 3, 104550), (2, 2, 12200)]
   created by app_owner (FORCE on)  : ok, rows=[(1, 3, 104550)]

after the fixes
   A. Birch adds the same email     : ok, 1 row(s) changed
   B. Birch invoices Acme's customer: ForeignKeyViolation: insert or update on table "invoices" violates foreign key constraint "invoices_customer_fk" [Key is not present in table "customers".]
   B. Birch invoices customer 999   : ForeignKeyViolation: insert or update on table "invoices" violates foreign key constraint "invoices_customer_fk" [Key is not present in table "customers".]
   C. tenant 1 reads invoice_totals : ok, rows=[(1, 3, 104550)]

A. The unique constraint answers questions it should not

Birch tried to add a customer with Acme’s email address and got a UniqueViolation naming the constraint. PostgreSQL left the key value out of this message, but the failure alone answers a question Birch was not allowed to ask: does this address already exist somewhere in the system? Repeated for many addresses, that is a way to find out which people are customers of other tenants.

The fix for A

Make uniqueness a per-tenant property: UNIQUE (tenant_id, email). After the fix the same insert succeeds, and two tenants can legitimately share an email address.

B. The foreign key accepts a reference across tenants

This one is worse than a leak, because it corrupts the data model. Birch’s insert pointing at Acme’s customer 1 succeeded: the database accepted an invoice that belongs to one tenant and points at a customer of another. Any report or export that runs with a privileged role and joins the two tables will now mix tenants. Meanwhile a made-up id such as 999 was rejected, so Birch could also tell real ids from fake ones.

The fix for B

Use a composite foreign key, FOREIGN KEY (tenant_id, customer_id) REFERENCES customers (tenant_id, id), which needs a matching UNIQUE (tenant_id, id) on the customers table. After the fix, both attempts fail with the same message, so the error no longer tells Birch anything.

C. A view can read with someone else’s rights

The last block creates the same totals view twice, once as the superuser and once as the table owner. Tenant 1 reading the superuser’s view sees both tenants’ totals, including Birch’s 2 invoices and 12,200 cents. The owner’s view is filtered. The manual explains why: “If any of the underlying base relations has row-level security enabled, then by default, the row-level security policies of the view owner are applied.” The owner is subject to the policy because of FORCE; the superuser never is.

The fix for C

Use the view option security_invoker: “if the view has security_invoker set to true, then the policies and permissions of the invoking user are used instead.” With ALTER VIEW invoice_totals SET (security_invoker = true) the view returns only tenant 1’s row. Views created by migrations running as a superuser are an easy way to reintroduce this leak.

Step 9: Know what this approach cannot do

The policy trusts whatever value the session setting holds. So the last demonstration is deliberately short: inside a transaction that belongs to tenant 1, one extra statement changes the tenant.

"""step09_limits.py: what row-level security keyed on a session setting cannot do."""
from psycopg_pool import ConnectionPool

from lab import dsn
from tenancy import tenant_transaction

pool = ConnectionPool(dsn("app_user"), min_size=1, max_size=1, open=True)
pool.wait()
with tenant_transaction(pool, 1) as conn:
    ids = lambda: [r[0] for r in conn.execute("SELECT id FROM invoices ORDER BY id")]
    print("inside tenant 1's transaction        :", ids())
    conn.execute("SELECT set_config('app.tenant_id', '2', true)")  # any SQL the app role can run can do this
    print("after one set_config call            :", ids())
pool.close()

Run python step09_limits.py:

inside tenant 1's transaction        : [1, 2, 3]
after one set_config call            : [4, 5]

Any code that can run SQL as app_user can call set_config, and from then on it is somebody else’s tenant. In a correct application nobody sends that statement. An attacker who finds a SQL injection hole, however, gets exactly this power. Row-level security therefore belongs next to the other defenses, not in place of them: keep every query parameterized (our SQL injection tutorial shows how), authorize every object as in our IDOR and BOLA tutorial, and keep the application role’s permissions small. If a customer needs a different kind of boundary than rows can give, OWASP’s comparison lists separate schemas, a “Namespace and database-role boundary”, and separate databases, a “Database and credential boundary” that it calls strong when credentials, network access, administrative paths and backups are isolated, at a “higher operational cost.” For shared tables it states what you must bring: “Requires enforceable tenant ownership, policy coverage for classified tenant-owned tables, constrained request roles, and negative-path tests.” Steps 2 to 8 gave you tenant ownership, policies and a constrained request role; Step 11 adds the coverage check and the negative-path tests.

Step 10: Wire it into a FastAPI app

Now put the pieces behind HTTP. The app below has two deliberate bugs, and we want to see them contained. Its list endpoint and its single-invoice endpoint never mention a tenant, and its create endpoint trusts a tenant_id sent by the client. Do not copy those habits. The point is that with RLS in place they stop being leaks.

The routes are plain def functions. FastAPI’s documentation says what that means: “it is run in an external threadpool that is then awaited, instead of being called directly (as it would block the server).” That suits psycopg’s blocking pool, which its documentation says “can be used to request connections from multiple concurrent threads.” The tenant comes from the API key, never from the request body. A real application would take it from a verified login token; the small dictionary of keys is only a stand-in.

"""app.py: a small invoice API. It contains two deliberate bugs that row-level security must contain."""
from contextlib import asynccontextmanager

import psycopg
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from pydantic import BaseModel

from lab import dsn
from repo import create_invoice, get_invoice, list_invoices
from tenancy import assert_unprivileged, tenant_transaction

API_KEYS = {"acme-key": 1, "birch-key": 2}  # stand-in for real authentication


@asynccontextmanager
async def lifespan(app):
    pool = ConnectionPool(
        dsn("app_user"), min_size=1, max_size=4, kwargs={"row_factory": dict_row}, open=True
    )
    try:
        with pool.connection() as conn:
            assert_unprivileged(conn)  # refuse to start as a role that skips row-level security
        app.state.pool = pool
        yield
    finally:
        pool.close()


app = FastAPI(lifespan=lifespan)


def current_tenant(authorization: str | None = Header(default=None)) -> int:
    key = (authorization or "").removeprefix("Bearer ").strip()
    if key not in API_KEYS:
        raise HTTPException(status_code=401, detail="unknown API key")
    return API_KEYS[key]


class InvoiceIn(BaseModel):
    tenant_id: int  # deliberate bug 2: the client is allowed to name the tenant
    customer_id: int
    amount_cents: int


@app.get("/invoices")
def read_invoices(request: Request, tenant_id: int = Depends(current_tenant)):
    with tenant_transaction(request.app.state.pool, tenant_id) as conn:
        return list_invoices(conn)  # deliberate bug 1: no tenant filter anywhere


@app.get("/invoices/{invoice_id}")
def read_invoice(invoice_id: int, request: Request, tenant_id: int = Depends(current_tenant)):
    with tenant_transaction(request.app.state.pool, tenant_id) as conn:
        invoice = get_invoice(conn, invoice_id)
    if invoice is None:
        raise HTTPException(status_code=404, detail="invoice not found")
    return invoice


@app.post("/invoices", status_code=201)
def add_invoice(body: InvoiceIn, request: Request, tenant_id: int = Depends(current_tenant)):
    try:
        with tenant_transaction(request.app.state.pool, tenant_id) as conn:
            return create_invoice(conn, body.tenant_id, body.customer_id, body.amount_cents)
    except psycopg.errors.InsufficientPrivilege:
        raise HTTPException(status_code=403, detail="that row does not belong to your tenant")
    except psycopg.errors.ForeignKeyViolation:
        raise HTTPException(status_code=422, detail="no such customer")

Call it in-process with Starlette’s test client:

"""step10_api.py: call the API in-process with Starlette's test client."""
from fastapi.testclient import TestClient

from app import app

ACME = {"Authorization": "Bearer acme-key"}
BIRCH = {"Authorization": "Bearer birch-key"}


def show(label, response):
    print(f"{label:<42}{response.status_code} {response.json()}")


with TestClient(app) as client:
    show("no API key, GET /invoices/2", client.get("/invoices/2"))
    show("Acme, GET /invoices/2 (its own)", client.get("/invoices/2", headers=ACME))
    show("Acme, GET /invoices/4 (Birch's)", client.get("/invoices/4", headers=ACME))
    show("Birch, GET /invoices/4 (its own)", client.get("/invoices/4", headers=BIRCH))
    ids = lambda r: [row["id"] for row in r.json()]
    print(f"{'Acme, GET /invoices, ids':<42}{ids(client.get('/invoices', headers=ACME))}")
    print(f"{'Birch, GET /invoices, ids':<42}{ids(client.get('/invoices', headers=BIRCH))}")
    body = {"tenant_id": 2, "customer_id": 3, "amount_cents": 100}
    show("Acme, POST an invoice for tenant 2", client.post("/invoices", headers=ACME, json=body))
    body = {"tenant_id": 1, "customer_id": 999, "amount_cents": 100}
    show("Acme, POST an invoice for customer 999", client.post("/invoices", headers=ACME, json=body))

Run python step10_api.py:

no API key, GET /invoices/2               401 {'detail': 'unknown API key'}
Acme, GET /invoices/2 (its own)           200 {'id': 2, 'tenant_id': 1, 'customer_id': 1, 'amount_cents': 4550}
Acme, GET /invoices/4 (Birch's)           404 {'detail': 'invoice not found'}
Birch, GET /invoices/4 (its own)          200 {'id': 4, 'tenant_id': 2, 'customer_id': 3, 'amount_cents': 2300}
Acme, GET /invoices, ids                  [1, 2, 3]
Birch, GET /invoices, ids                 [4, 5]
Acme, POST an invoice for tenant 2        403 {'detail': 'that row does not belong to your tenant'}
Acme, POST an invoice for customer 999    422 {'detail': 'no such customer'}

Reading the output

A request without a key gets 401. Acme reads its own invoice 2 and gets 200, but when it asks for Birch’s invoice 4 the route returns 404: the buggy query found no row, so the app reported “not found.” Birch reads the same invoice successfully. Each list contains only its own tenant’s invoices even though the route has no filter. The two refused writes are the deliberate create bug being contained: an invoice for tenant 2 sent by Acme gets 403, and an invoice for a customer that does not exist gets 422 from the foreign key. The app also calls the guard from Step 5 at start-up, so it would refuse to run as a privileged role.

Try it with a real server

The same checks over real HTTP, with uvicorn and curl. Start the server in one terminal:

uvicorn app:app --port 8772

Then run the four requests from a second terminal. This is what the script step10b_live.py does for you, and what it printed:

"""step10b_live.py: start the real server, call it with curl, stop it."""
import os
import subprocess
import sys
import time
import urllib.request

PORT = 8772
BASE = f"http://127.0.0.1:{PORT}"
server = subprocess.Popen([sys.executable, "-m", "uvicorn", "app:app", "--port", str(PORT),
                           "--log-level", "warning"])


def curl(shown, args):
    print(f"$ {shown}")
    out = subprocess.run(["curl", "-s", "-w", "\nHTTP %{http_code}", *args], capture_output=True, text=True).stdout
    print(out + "\n")


try:
    for _ in range(100):
        try:
            urllib.request.urlopen(f"{BASE}/openapi.json", timeout=1)
            break
        except OSError:
            time.sleep(0.2)
    acme = ["-H", "Authorization: Bearer acme-key"]
    curl(f"curl -s -H 'Authorization: Bearer acme-key' {BASE}/invoices/2", [*acme, f"{BASE}/invoices/2"])
    curl(f"curl -s -H 'Authorization: Bearer acme-key' {BASE}/invoices/4", [*acme, f"{BASE}/invoices/4"])
    curl(f"curl -s -H 'Authorization: Bearer birch-key' {BASE}/invoices",
         ["-H", "Authorization: Bearer birch-key", f"{BASE}/invoices"])
    body = '{"tenant_id": 2, "customer_id": 3, "amount_cents": 100}'
    curl(f"curl -s -X POST -H 'Authorization: Bearer acme-key' -H 'Content-Type: application/json' -d '{body}' {BASE}/invoices",
         [*acme, "-X", "POST", "-H", "Content-Type: application/json", "-d", body, f"{BASE}/invoices"])
finally:
    if os.name == "nt":  # the venv's python.exe is a launcher on Windows: stop its whole process tree
        subprocess.run(["taskkill", "/PID", str(server.pid), "/T", "/F"], capture_output=True)
    else:
        server.terminate()
$ curl -s -H 'Authorization: Bearer acme-key' http://127.0.0.1:8772/invoices/2
{"id":2,"tenant_id":1,"customer_id":1,"amount_cents":4550}
HTTP 200

$ curl -s -H 'Authorization: Bearer acme-key' http://127.0.0.1:8772/invoices/4
{"detail":"invoice not found"}
HTTP 404

$ curl -s -H 'Authorization: Bearer birch-key' http://127.0.0.1:8772/invoices
[{"id":4,"tenant_id":2,"customer_id":3,"amount_cents":2300},{"id":5,"tenant_id":2,"customer_id":3,"amount_cents":9900}]
HTTP 200

$ curl -s -X POST -H 'Authorization: Bearer acme-key' -H 'Content-Type: application/json' -d '{"tenant_id": 2, "customer_id": 3, "amount_cents": 100}' http://127.0.0.1:8772/invoices
{"detail":"that row does not belong to your tenant"}
HTTP 403

Step 11: Test it, and fail the build when a table is unprotected

A policy that is never tested is a policy you hope works. OWASP’s guidance for verifying tenant isolation begins with a rule: “Test isolation through the same role, connection path, and pooling mode used by the application.” It adds a reason: “A privileged test connection can make a correct policy appear broken, while a test that never exercises the deployed request role can miss a bypass.” Its checklist maps directly onto the tests below: “Test the authorization matrix.” “Discover coverage.” “Assert the request role.” “Exercise connection reuse.”

The test file has 14 tests. The second one is the test that pays for itself: it asks PostgreSQL itself for every table with a tenant_id column and fails unless that table has RLS enabled, forced, and at least one policy. When a teammate adds a new tenant table and forgets RLS, this test fails in continuous integration before the table reaches production.

"""test_tenancy.py: the safety net. Run it with: python -m pytest -q"""
import pytest
from fastapi.testclient import TestClient
from psycopg_pool import ConnectionPool

from app import app
from lab import attempt, connect, dsn
from tenancy import assert_unprivileged, tenant_transaction

IDS = "SELECT id FROM invoices ORDER BY id"
ACME = {"Authorization": "Bearer acme-key"}
BIRCH = {"Authorization": "Bearer birch-key"}


def test_application_role_is_unprivileged():
    with connect("app_user") as conn:
        assert_unprivileged(conn)
    with connect("postgres") as conn, pytest.raises(RuntimeError):
        assert_unprivileged(conn)


def test_every_table_with_a_tenant_id_column_is_protected():
    with connect("postgres") as conn:
        rows = conn.execute(
            """
            SELECT c.relname, c.relrowsecurity, c.relforcerowsecurity,
                   (SELECT count(*) FROM pg_policies p
                     WHERE p.schemaname = n.nspname AND p.tablename = c.relname) AS policies
            FROM pg_class c
            JOIN pg_namespace n ON n.oid = c.relnamespace
            JOIN pg_attribute a ON a.attrelid = c.oid AND a.attname = 'tenant_id' AND NOT a.attisdropped
            WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p')
            ORDER BY c.relname
            """
        ).fetchall()
    assert [r[0] for r in rows] == ["customers", "invoices"]
    unprotected = [r[0] for r in rows if not (r[1] and r[2] and r[3] >= 1)]
    assert unprotected == [], f"enable RLS, force it and add a policy on: {unprotected}"


@pytest.mark.parametrize("tenant_id, expected", [(1, [(1,), (2,), (3,)]), (2, [(4,), (5,)])])
def test_a_tenant_sees_only_its_own_rows(tenant_id, expected):
    assert attempt(tenant_id, IDS) == f"ok, rows={expected}"


def test_no_tenant_means_no_rows():
    assert attempt(None, IDS) == "ok, rows=[]"


def test_tenant_does_not_outlive_its_transaction():
    pool = ConnectionPool(dsn("app_user"), min_size=1, max_size=1, open=True)
    pool.wait()
    with tenant_transaction(pool, 1) as conn:
        assert len(conn.execute(IDS).fetchall()) == 3
    with pool.connection() as conn:
        assert conn.execute(IDS).fetchall() == []
    pool.close()


def test_owner_is_subject_to_the_policy():
    assert attempt(None, IDS, user="app_owner") == "ok, rows=[]"


def test_insert_for_another_tenant_is_refused():
    sql = "INSERT INTO invoices (tenant_id, customer_id, amount_cents) VALUES (2, 3, 1)"
    assert attempt(1, sql).startswith("InsufficientPrivilege")


def test_row_cannot_be_moved_to_another_tenant():
    assert attempt(1, "UPDATE invoices SET tenant_id = 2 WHERE id = 1").startswith("InsufficientPrivilege")


def test_other_tenants_rows_cannot_be_changed_or_deleted():
    assert attempt(1, "UPDATE invoices SET amount_cents = 0 WHERE id = 4") == "ok, 0 row(s) changed"
    assert attempt(1, "DELETE FROM invoices WHERE id = 4") == "ok, 0 row(s) changed"


def test_two_tenants_may_share_an_email_address():
    sql = "INSERT INTO customers (tenant_id, email, name) VALUES (2, '[email protected]', 'Alice B')"
    assert attempt(2, sql) == "ok, 1 row(s) changed"


def test_cross_tenant_customer_reference_is_refused():
    sql = "INSERT INTO invoices (tenant_id, customer_id, amount_cents) VALUES (2, 1, 1)"
    assert attempt(2, sql).startswith("ForeignKeyViolation")


def test_view_respects_the_policies():
    sql = "SELECT tenant_id FROM invoice_totals ORDER BY 1"
    assert attempt(1, sql) == "ok, rows=[(1,)]"


def test_a_forgotten_filter_is_contained_in_the_api():
    with TestClient(app) as client:
        assert client.get("/invoices/4", headers=ACME).status_code == 404
        assert client.get("/invoices/4", headers=BIRCH).status_code == 200
        assert [r["id"] for r in client.get("/invoices", headers=ACME).json()] == [1, 2, 3]
        body = {"tenant_id": 2, "customer_id": 3, "amount_cents": 100}
        assert client.post("/invoices", headers=ACME, json=body).status_code == 403
        assert client.get("/invoices").status_code == 401

Run python -m pytest -q:

.............. [100%]
14 passed in 0.77s

Prove the tests can fail

A test that cannot fail proves nothing. This script removes the protection on purpose with ALTER TABLE invoices NO FORCE ROW LEVEL SECURITY, runs two of the tests, and restores it.

"""step11_break_it.py: prove the tests notice when someone removes the protection."""
import os
import subprocess
import sys

from lab import connect

SELECT = "protected or owner_is_subject"


def run_tests():
    env = dict(os.environ, COLUMNS="170")
    r = subprocess.run(
        [sys.executable, "-m", "pytest", "-q", "-k", SELECT, "--tb=no", "-p", "no:cacheprovider"],
        capture_output=True, text=True, env=env,
    )
    for line in r.stdout.splitlines():
        if line.startswith("FAILED") or " passed" in line or " failed" in line:
            print("  ", line)


print("tests with the protection in place")
run_tests()
with connect("app_owner", autocommit=True) as owner:
    owner.execute("ALTER TABLE invoices NO FORCE ROW LEVEL SECURITY")
print("\nafter ALTER TABLE invoices NO FORCE ROW LEVEL SECURITY")
run_tests()
with connect("app_owner", autocommit=True) as owner:
    owner.execute("ALTER TABLE invoices FORCE ROW LEVEL SECURITY")
print("\nafter FORCE ROW LEVEL SECURITY again")
run_tests()

Run python step11_break_it.py:

tests with the protection in place
   2 passed, 12 deselected in 0.31s

after ALTER TABLE invoices NO FORCE ROW LEVEL SECURITY
   FAILED test_tenancy.py::test_every_table_with_a_tenant_id_column_is_protected - AssertionError: enable RLS, force it and add a policy on: ['invoices']
   FAILED test_tenancy.py::test_owner_is_subject_to_the_policy - AssertionError: assert 'ok, rows=[(1..., (4,), (5,)]' == 'ok, rows=[]'
   2 failed, 12 deselected in 0.30s

after FORCE ROW LEVEL SECURITY again
   2 passed, 12 deselected in 0.31s

With the protection in place, both tests pass. Without FORCE, the lint test names the table, and the owner test shows that the owner sees every tenant’s rows again. After the restore, both pass.

Step 12: Check what it costs

Every query now carries the policy expression. The manual says: “This expression will be evaluated for each row prior to any conditions or functions coming from the user’s query”, and it also says which policies are cheap: “This is the simplest and best-performing case.” So how much does a simple policy cost? The script builds a table of 2,000,000 invoices spread over 20 tenants, so tenant 7 owns 100,000 rows, or 5 percent. It times SELECT count(*), sum(amount_cents) as tenant 7 in three ways: with RLS off and the filter written into the query, with the plain policy from Step 6, and with the same policy written as tenant_id = (SELECT app_tenant_id()). It runs each way first without an index on tenant_id and then with one, plus 400 primary-key lookups. Each aggregate timing is the median of seven runs, and each lookup timing is the median of the 400. The script drops its table when it finishes.

"""step12_perf.py: what a policy costs on a table with 2,000,000 rows (20 tenants)."""
import random
import statistics
import time

from lab import connect

ROWS, TENANTS, ME = 2_000_000, 20, 7
SETUP = f"""
DROP TABLE IF EXISTS perf_invoices;
CREATE TABLE perf_invoices (
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id    integer NOT NULL,
    amount_cents integer NOT NULL
);
INSERT INTO perf_invoices (tenant_id, amount_cents)
SELECT 1 + g % {TENANTS}, (random() * 100000)::integer FROM generate_series(1, {ROWS}) g;
GRANT SELECT ON perf_invoices TO app_user;
ANALYZE perf_invoices;
"""
AGG = "SELECT count(*), sum(amount_cents) FROM perf_invoices"
VARIANTS = [  # label, policy expression (None = no RLS, filter written in the query)
    ("explicit WHERE tenant_id = 7, no RLS", None),
    ("policy: tenant_id = app_tenant_id()", "tenant_id = app_tenant_id()"),
    ("policy: tenant_id = (SELECT app_tenant_id())", "tenant_id = (SELECT app_tenant_id())"),
]


def configure(owner, expression):
    if expression is None:
        owner.execute("ALTER TABLE perf_invoices DISABLE ROW LEVEL SECURITY")
        return
    owner.execute("ALTER TABLE perf_invoices ENABLE ROW LEVEL SECURITY")
    owner.execute("ALTER TABLE perf_invoices FORCE ROW LEVEL SECURITY")
    owner.execute("DROP POLICY IF EXISTS tenant_isolation ON perf_invoices")
    owner.execute(f"CREATE POLICY tenant_isolation ON perf_invoices USING ({expression})")


def query_for(expression):
    return AGG + (f" WHERE tenant_id = {ME}" if expression is None else "")


def median_ms(sql, runs=7):
    with connect("app_user") as conn:
        conn.execute("SELECT set_config('app.tenant_id', %s, false)", [str(ME)])  # session level is fine for a benchmark
        times = []
        for _ in range(runs + 1):
            t0 = time.perf_counter()
            conn.execute(sql).fetchall()
            times.append((time.perf_counter() - t0) * 1000)
        return statistics.median(times[1:])  # the first run warms the cache


def lookup_us(n=400):
    rnd = random.Random(7)
    ids = [rnd.randrange(0, ROWS // TENANTS) * TENANTS + (ME - 1) for _ in range(n)]  # ids owned by tenant 7
    with connect("app_user") as conn:
        conn.execute("SELECT set_config('app.tenant_id', %s, false)", [str(ME)])
        times = []
        for i in ids:
            t0 = time.perf_counter()
            conn.execute("SELECT amount_cents FROM perf_invoices WHERE id = %s", [i]).fetchall()
            times.append((time.perf_counter() - t0) * 1e6)
        return statistics.median(times)


def plan_lines(sql):
    with connect("app_user") as conn:
        conn.execute("SELECT set_config('app.tenant_id', %s, false)", [str(ME)])
        rows = conn.execute("EXPLAIN (COSTS OFF) " + sql).fetchall()
    return [r[0].strip() for r in rows if any(k in r[0] for k in ("Scan", "Filter", "Index Cond", "InitPlan"))]


with connect("app_owner", autocommit=True) as owner:
    t0 = time.perf_counter()
    owner.execute(SETUP)
    print(f"loaded {ROWS:,} rows in {time.perf_counter() - t0:.1f} s; tenant {ME} owns {ROWS // TENANTS:,}")

    results, plans = {}, {}
    for indexed in (False, True):
        if indexed:
            owner.execute("CREATE INDEX perf_invoices_tenant_idx ON perf_invoices (tenant_id)")
            owner.execute("ANALYZE perf_invoices")
        for label, expression in VARIANTS:
            configure(owner, expression)
            sql = query_for(expression)
            results[(label, indexed)] = median_ms(sql)
            plans[(label, indexed)] = plan_lines(sql)
            if indexed:
                results[(label, "point")] = lookup_us()

print(f"\n{'':<46}{'no index':>10}{'index':>9}{'point lookup':>14}")
print(f"{'median of 7 runs':<46}{'ms':>10}{'ms':>9}{'microseconds':>14}")
for label, _ in VARIANTS:
    print(f"{label:<46}{results[(label, False)]:>10.1f}{results[(label, True)]:>9.1f}{results[(label, 'point')]:>14.0f}")

for key in [(VARIANTS[1][0], False), (VARIANTS[2][0], False), (VARIANTS[1][0], True)]:
    print(f"\nplan, {key[0]}, {'with' if key[1] else 'without'} index:")
    for line in plans[key]:
        print("   ", line)

with connect("app_owner", autocommit=True) as owner:
    owner.execute("DROP TABLE perf_invoices")

Run python step12_perf.py. The timing table is:

loaded 2,000,000 rows in 3.9 s; tenant 7 owns 100,000

                                                no index    index  point lookup
median of 7 runs                                      ms       ms  microseconds
explicit WHERE tenant_id = 7, no RLS                38.2     12.3            87
policy: tenant_id = app_tenant_id()                164.3     12.6            61
policy: tenant_id = (SELECT app_tenant_id())        44.2     12.4            76

And the plan lines that explain it:

plan, policy: tenant_id = app_tenant_id(), without index:
    ->  Seq Scan on perf_invoices
    Filter: (tenant_id = (NULLIF(current_setting('app.tenant_id'::text, true), ''::text))::integer)

plan, policy: tenant_id = (SELECT app_tenant_id()), without index:
    InitPlan 1
    ->  Seq Scan on perf_invoices
    Filter: (tenant_id = (InitPlan 1).col1)

plan, policy: tenant_id = app_tenant_id(), with index:
    ->  Bitmap Heap Scan on perf_invoices
    ->  Bitmap Index Scan on perf_invoices_tenant_idx
    Index Cond: (tenant_id = (NULLIF(current_setting('app.tenant_id'::text, true), ''::text))::integer)

What the numbers say

Without an index, the plain policy is the slow one. The hand-written filter took about 38 ms and the plain policy about 164 ms, roughly four times as long (my arithmetic from the table above). The first plan shows why: the whole expression sits in the sequential scan’s Filter, so PostgreSQL evaluates it row by row. In the second plan the sub-select appears as InitPlan 1, PostgreSQL’s name for a sub-query that is computed once and reused, and the time drops to 44 ms in the run shown.

With an index, there is no visible cost. All three variants took about 12 ms, and the last plan shows the policy expression sitting inside the Index Cond of the bitmap index scan, so the index does the filtering. Index tenant_id; our keyset pagination tutorial, for example, starts its composite index with it.

Point lookups are noise at this scale. Looking up single rows by primary key took between 44 and 87 microseconds across every variant and run, and the differences between variants were smaller than the differences between runs.

I ran the script five times and kept every output. Without an index, the hand-written filter took 38.1 to 39.5 ms, the plain policy 162.4 to 164.3 ms and the wrapped policy 44.2 to 51.8 ms; with an index, every variant took 11.9 to 12.7 ms. These numbers come from one machine and one table shape, so treat them as a guide to what to measure, not as promises. The advice that follows from them is cheap to adopt: index the tenant column, and write the call as (SELECT app_tenant_id()) so PostgreSQL computes it once.

Check that everything works end to end

From the project folder, with the server running, run the scripts in order and then the tests. The scripts build on each other (Step 5 forces RLS, Step 6 replaces the policy expression, Step 8 changes constraints), so keep this order, and run step02_setup.py again whenever you want a clean start:

python step02_setup.py
python step03_forgotten_where.py
python step04_policy.py
python step05_force.py
python step06_pool.py
python step07_writes.py
python step08_side_doors.py
python step09_limits.py
python step10_api.py
python -m pytest -q
python step11_break_it.py
python step12_perf.py

You are done when all of these are true: no script prints a traceback; Step 3 leaks invoice 4 while Step 4 returns None for it; Step 6 shows an empty list for request B in its last demonstration; Step 8 ends with one row in the view and the same foreign-key message twice; Step 10 prints 401, 200, 404, 200, the two id lists, 403 and 422; pytest prints 14 passed; and Step 11 shows two FAILED lines between its two green runs.

Common mistakes and how to spot them

  • The application connects as a superuser, a BYPASSRLS role or the table owner. Policies appear to do nothing. Run the guard from Step 5 at start-up and in tests, and use FORCE ROW LEVEL SECURITY.
  • The tenant is stored for the session. Pooled connections carry it into the next request. Always use set_config(..., true) inside the transaction that runs the queries.
  • A new table has a tenant_id column and no policy. Keep the lint test from Step 11 in your build.
  • Unique keys and foreign keys ignore the tenant. Make them composite with tenant_id so one tenant cannot probe or reference another.
  • A view or a migration runs as a privileged role. Set security_invoker on views and check who owns what.
  • Zero rows changed is treated as success. A hidden row is not an error. Return 404 when an update or delete touches nothing.
  • Tests run as a superuser. They will pass or fail for the wrong reasons. Run them as the application role through the same pool.

Where to go next

  • Read the two manual pages this tutorial leans on, Row Security Policies and CREATE POLICY. They cover per-command policies, policies for specific roles and restrictive policies that must all pass. When you split policies by command, give INSERT and UPDATE policies a WITH CHECK expression, because that is the one PostgreSQL applies to rows being written.
  • Plan your backups. The manual notes that when taking a backup, “it could be disastrous if row security silently caused some rows to be omitted”, and suggests setting row_security to off, which “does not in itself bypass row security; what it does is throw an error if any query’s results would get filtered by a policy.” Take backups with a role that bypasses row security, such as a superuser or a role with BYPASSRLS, and set row_security to off for the backup session: if that role ever turns out to be subject to a policy, the backup stops with an error instead of silently leaving rows out.
  • Strengthen the rest of the stack: preventing broken object level authorization in FastAPI, stopping mass assignment (the client-chosen tenant_id in Step 10 is a small case of it), and keyset pagination, whose composite index starts with the tenant column, like the index recommended in Step 12.
  • Replace the API-key stand-in with real authentication, and keep the rule from Step 10: the tenant comes from the verified identity, never from the request body.

Sources

Tags:

Application SecurityFastAPIMulti-TenancyPostgreSQLPythonRow-Level Security

Share

No Comment! Be the first one.

Leave a Reply

Your email address will not be published. Required fields are marked *

Related Posts