Preventing SQL Injection in Multi-Tenant Apps

A single injected predicate in a multi-tenant query can leak every tenant's rows at once, so the usual SQLi defenses have to be paired with hard tenant scoping. This page sits under connection pooling in multi-tenant systems, because pooled connections and shared query builders are where tenant-scoped injection most often slips through.

Problem Framing

In a single-tenant app, SQL injection is bad because an attacker reads or writes data they should not. In a multi-tenant app the blast radius is the whole customer base: one unparameterized WHERE clause, one interpolated schema name, or one connection handed back to the pool with a stale search_path, and a query meant for tenant A returns rows for tenants B through Z. The tenant identifier becomes a security boundary, and any string that reaches the SQL text without being bound as a parameter is a way to move that boundary.

Three things make this harder than ordinary injection defense. First, tenant routing is dynamic — the tenant ID arrives on every request from a header, subdomain, or JWT claim, so it is attacker-influenced input that touches query construction by design. Second, many teams switch tenants by changing an identifier (SET search_path, a schema prefix, a database name), and identifiers cannot be parameterized the way values can, so a careless "SET search_path = " + tenant is a direct injection point that no prepared statement protects. Third, connection pooling means the session that ran tenant A's query is reused for tenant B; if SET ROLE or search_path is not reset, even perfectly parameterized SQL executes against the wrong tenant's data.

The flow below shows where a tenant identifier must be validated and bound on its way to the database, and the two points where it commonly escapes the safe path.

Step-by-Step Guide

1. Validate the tenant identifier at the edge

Reject anything that is not a well-formed tenant ID before it reaches a query builder. Use a strict format (UUID v4 or ULID) and an allowlist of known tenants, not a denylist of bad characters — denylists miss encodings.

// Express middleware: validate tenant ID format before it touches SQL
const TENANT_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;

export function requireTenant(req, res, next) {
  const tenantId = req.headers['x-tenant-id'];
  if (typeof tenantId !== 'string' || !TENANT_ID.test(tenantId)) {
    return res.status(400).json({ error: 'invalid tenant id' });
  }
  req.tenantId = tenantId;
  next();
}

2. Bind tenant IDs as values, never build them into SQL text

A tenant ID that filters rows is a value, so it binds as $1. The only time it cannot be a parameter is when it selects a schema or table name — handle that separately in step 3.

-- Safe: tenant_id is a bound value
SELECT id, total, status
FROM orders
WHERE tenant_id = $1 AND status = $2;

3. Map identifiers through an allowlist, never interpolate them

You cannot parameterize a schema or table name, so resolve it from a server-controlled map keyed by the validated tenant ID. The attacker-supplied string is used only as a lookup key, never as SQL.

// Resolve a schema name from a controlled map, then quote it safely
import { Client } from 'pg';

const SCHEMA_BY_TENANT: Record<string, string> = {
  '550e8400-e29b-41d4-a716-446655440000': 'tenant_acme',
};

export async function setTenantSchema(client: Client, tenantId: string) {
  const schema = SCHEMA_BY_TENANT[tenantId];
  if (!schema) throw new Error('unknown tenant');
  // pg has no bind params for identifiers; quote_ident-style escaping instead
  await client.query(`SET search_path TO "${schema.replace(/"/g, '""')}"`);
}

4. Enforce the tenant filter in ORM middleware

Hand-written WHERE tenant_id = ? clauses get forgotten. Attach the filter at the ORM execution hook so every SELECT is scoped automatically, the way Prisma client extensions for tenant scoping and similar interceptors do.

# SQLAlchemy: auto-scope every SELECT to the current tenant
from sqlalchemy import event
from sqlalchemy.orm import Session, with_loader_criteria
from myapp.context import get_current_tenant
from myapp.models import TenantScoped

@event.listens_for(Session, "do_orm_execute")
def apply_tenant_filter(state):
    tenant_id = get_current_tenant()
    if tenant_id is None:
        raise PermissionError("no tenant context")
    if state.is_select:
        state.statement = state.statement.options(
            with_loader_criteria(TenantScoped,
                lambda cls: cls.tenant_id == tenant_id,
                include_aliases=True)
        )

5. Reset session state on every pool checkout

A connection carries search_path, SET ROLE, and session GUCs across requests. Reset them before reuse, or pair pooling with the patterns in PgBouncer transaction pooling for multi-tenant SaaS so server connections never carry tenant state at all.

// Reset session state before the connection is reused by another tenant
func resetSession(ctx context.Context, conn *sql.Conn) error {
    _, err := conn.ExecContext(ctx, "RESET ALL; SET search_path = public")
    return err
}

6. Use a low-privilege role and row-level security as a backstop

Application-layer filters are the first line, not the last. A tenant-scoped database role plus RLS means an injected query still cannot read another tenant's rows even if the app's filter is bypassed.

-- RLS makes the database refuse cross-tenant rows regardless of app code
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON orders
  USING (tenant_id = current_setting('app.tenant_id')::uuid);

Verification

Prove that injection and missing-context both fail. The first test asserts a malicious tenant header is rejected; the second asserts an unscoped query throws instead of returning all tenants.

import pytest

def test_rejects_sql_in_tenant_header(client):
    resp = client.get("/orders", headers={"X-Tenant-Id": "1' OR '1'='1"})
    assert resp.status_code == 400

def test_unscoped_query_raises_without_context(db_session):
    clear_current_tenant()
    with pytest.raises(PermissionError):
        db_session.query(Order).all()

def test_filter_is_bound_not_interpolated(db_session, caplog):
    set_current_tenant("550e8400-e29b-41d4-a716-446655440000")
    db_session.query(Order).all()
    # the emitted SQL must contain a bind placeholder, never the literal id
    assert "550e8400" not in caplog.text
    assert "tenant_id" in caplog.text

A passing log line for a parameterized query shows the placeholder and the bound value separately, never the value spliced into the statement:

SELECT orders.id, orders.total FROM orders WHERE orders.tenant_id = %(tenant_id_1)s
[parameters: {'tenant_id_1': '550e8400-e29b-41d4-a716-446655440000'}]

Failure Modes & Gotchas

FAQ

Do parameterized queries alone stop multi-tenant SQL injection? For values, yes — but tenant routing often relies on schema or table identifiers, which cannot be parameterized. Those must go through a server-controlled allowlist, and you should add row-level security so the database enforces isolation even if an app-layer filter is missed.

Is an ORM enough to prevent injection across tenants? No. ORMs parameterize generated SQL, but raw query methods, missing global tenant filters, and unvalidated context all reintroduce injection or cross-tenant leakage. Enforce the tenant filter in an execution hook and forbid unaudited raw SQL.

Why does connection pooling matter for SQL injection? A pooled connection retains session state like search_path and SET ROLE. If it is reused without a reset, even fully parameterized queries run against the previous tenant's context, producing a cross-tenant read with no injection string involved.