Stop Leaking Data in Multi-Tenant Apps

Source: Dev.to

Why Your application logic isn’t Enough: The Case for Database-Level Row-Level Security

You’ve built a robust multi‑tenant SaaS. You’ve implemented tenant_id filters in every query, added middleware checks, and written unit tests. But application‑layer security is fragile: a missed filter, a rogue script, or a direct database connection can bypass your entire safety net.
Move your security perimeter from the code to the database itself.

The Short Summary

In this deep dive we dismantle the myth that Row‑Level Security (RLS) is just a “Postgres feature” and show how to integrate it seamlessly into your SQLAlchemy ORM and Alembic migration workflow. We go beyond theory to cover production‑ready implementation details.

Why relying solely on WHERE tenant_id = ? in your code is a ticking time bomb

• Filters live in the application layer and can be omitted or overridden.
• Direct database access (e.g., admin tools, ad‑hoc queries) ignores those filters.
• A single oversight can expose all tenants’ data.

Step‑by‑step Alembic migration scripts to enable RLS and define policies without breaking your CI/CD pipeline

1. Create a migration that adds the tenant_id column (if not present) and enables RLS on the target tables.

   # alembic revision script
   from alembic import op
   import sqlalchemy as sa
   
   def upgrade():
       op.execute("ALTER TABLE orders ENABLE ROW LEVEL SECURITY;")
       op.execute("""
           CREATE POLICY tenant_isolation ON orders
           USING (tenant_id = current_setting('app.current_tenant')::uuid);
       """)

2. Add a reversible downgrade that drops the policy and disables RLS.

3. Run migrations as part of your normal CI/CD process; the statements are idempotent and safe to execute on production.

How to inject dynamic tenant context into SQLAlchemy sessions using contextvars and event listeners

import contextvars
from sqlalchemy import event
from sqlalchemy.orm import Session

# Context variable that holds the current tenant UUID
current_tenant = contextvars.ContextVar("current_tenant")

def set_tenant(tenant_id: str):
    current_tenant.set(tenant_id)

@event.listens_for(Session, "before_flush")
def apply_tenant_setting(session, flush_context, instances):
    tenant = current_tenant.get(None)
    if tenant:
        session.execute(
            f"SET LOCAL app.current_tenant = '{tenant}'"
        )

• Call set_tenant() at the beginning of each request (e.g., in FastAPI middleware).
• All subsequent ORM queries automatically inherit the tenant context, and the database enforces the RLS policy.

Critical pitfalls

• Admin bypass strategies: Ensure super‑user roles do not automatically disable RLS; create explicit policies for admin accounts if needed.
• Performance indexing: Index the tenant_id column (and any columns used in RLS predicates) to avoid full‑table scans.
• Policy types:
  • PERMISSIVE policies add to the allowed rows.
  • RESTRICTIVE policies define the exact set of rows that can be accessed. Choose the appropriate type based on your security model.

Stop hoping developers never forget a filter. Enforce security where it matters most: at the data layer.

Read the full technical guide with complete code examples, migration templates, and testing strategies on https://www.adrianovieira.eng.br/en/posts/architecture/row-level-security-sqlachemy-alembic-guide/.