Row-Level Security com SQLAlchemy e Alembic: Um Guia Completo

Este guia aborda a implementação de Row-Level Security (RLS) a partir do PostgreSQL 15+. Versões anteriores podem ter requisitos de sintaxe diferentes e estão fora do escopo aqui.

Introdução

Segurança em Nível de Linha (RLS) é um recurso do PostgreSQL (eles popularizaram o termo e têm suporte nativo robusto) que restringe o acesso a linhas específicas em uma tabela com base em funções de usuário ou contexto de sessão. Quando combinado com o ORM SQLAlchemy e migrações Alembic, o RLS fornece uma solução robusta para aplicações multi-tenant onde o isolamento de dados é crítico.

Descrevo a implementação desde migrações de banco de dados até integração no nível da aplicação, abordando armadilhas comuns e considerações de segurança.

Padrões de Design Arquitetural

A implementação incorpora vários padrões conhecidos, como:

Variável de Contexto – Usando contextvars para transmitir o contexto do tenant através do ciclo de vida da requisição
Event Listener – Eventos SQLAlchemy para configuração automática de sessão
Migration – Tratando políticas RLS como mudanças de esquema versionadas via Alembic
Defense-in-Depth – Camadas de segurança de banco de dados ao lado de autorização da aplicação

Métodos de Implementação

Estas são as técnicas táticas usadas para realizar a arquitetura:

Variáveis de sessão PostgreSQL (SET app.current_tenant)
Alembic op.execute() para SQL bruto em migrações
Hooks de eventos SQLAlchemy (before_cursor_execute)
Definições de políticas baseadas em função no PostgreSQL

Fundamentos do RLS do PostgreSQL

O RLS opera no nível do banco de dados, o que significa que as políticas são aplicadas independentemente de como as consultas chegam ao banco de dados. Isso fornece segurança de defesa em profundidade que complementa a autorização no nível da aplicação.

Conceitos Chave:

Policy: Define condições para acesso a linhas usando USING (para SELECT/UPDATE/DELETE) e WITH CHECK (para INSERT/UPDATE)
Session Variables: Configurações personalizadas como app.current_tenant que as políticas referenciam
Permissive vs. Restrictive: Políticas permissivas permitem acesso se qualquer política corresponder; políticas restritivas exigem que todas as políticas correspondam

Habilitando RLS via Migrações Alembic

Como a configuração do RLS é no nível do banco de dados, ela pertence às migrações Alembic e não ao código Python.

Modelo de Migração

"""habilitar_politicas_rls

ID da Revisão: habilitar_rls_v1
Revisa: revisão_anterior
Data de Criação: 2024-01-15 10:00:00.000000

"""
from alembic import op

def upgrade():
    # Habilitar RLS na tabela
    op.execute("ALTER TABLE seu_nome_de_tabela ENABLE ROW LEVEL SECURITY")

    # Remover políticas existentes se existirem (para idempotência)
    op.execute("DROP POLICY IF EXISTS tenant_isolation ON seu_nome_de_tabela")
    op.execute("DROP POLICY IF EXISTS tenant_isolation_insert ON seu_nome_de_tabela")
    op.execute("DROP POLICY IF EXISTS tenant_isolation_update ON seu_nome_de_tabela")

    # Criar política SELECT
    op.execute("""
        CREATE POLICY tenant_isolation ON seu_nome_de_tabela
        FOR SELECT
        USING (tenant_id = current_setting('app.current_tenant')::integer)
    """)

    # Criar política INSERT
    op.execute("""
        CREATE POLICY tenant_isolation_insert ON seu_nome_de_tabela
        FOR INSERT
        WITH CHECK (tenant_id = current_setting('app.current_tenant')::integer)
    """)

    # Criar política UPDATE
    op.execute("""
        CREATE POLICY tenant_isolation_update ON seu_nome_de_tabela
        FOR UPDATE
        USING (tenant_id = current_setting('app.current_tenant')::integer)
        WITH CHECK (tenant_id = current_setting('app.current_tenant')::integer)
    """)

    # Criar política DELETE
    op.execute("""
        CREATE POLICY tenant_isolation_delete ON seu_nome_de_tabela
        FOR DELETE
        USING (tenant_id = current_setting('app.current_tenant')::integer)
    """)

def downgrade():
    op.execute("DROP POLICY IF EXISTS tenant_isolation ON seu_nome_de_tabela")
    op.execute("DROP POLICY IF EXISTS tenant_isolation_insert ON seu_nome_de_tabela")
    op.execute("DROP POLICY IF EXISTS tenant_isolation_update ON seu_nome_de_tabela")
    op.execute("DROP POLICY IF EXISTS tenant_isolation_delete ON seu_nome_de_tabela")
    op.execute("ALTER TABLE seu_nome_de_tabela DISABLE ROW LEVEL SECURITY")

Definições de Modelos SQLAlchemy

Aqui está a estrutura de modelo SQLAlchemy corrigida com sintaxe adequada:

from sqlalchemy import Column, Integer, String, ForeignKey, event
from sqlalchemy.orm import declarative_base, relationship, Session
from contextvars import ContextVar

Base = declarative_base()

# Contexto do tenant thread-safe usando contextvars
current_tenant_id: ContextVar[int | None] = ContextVar("current_tenant_id", default=None)

class Tenant(Base):
    __tablename__ = 'tenants'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    users = relationship('User', back_populates='tenant')

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    tenant_id = Column(Integer, ForeignKey('tenants.id'), nullable=False)
    tenant = relationship('Tenant', back_populates='users')

class Document(Base):
    __tablename__ = 'documents'
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    content = Column(String)
    tenant_id = Column(Integer, ForeignKey('tenants.id'), nullable=False)
    created_by = Column(Integer, ForeignKey('users.id'))

Contexto do Tenant no SQLAlchemy

O contexto do tenant deve ser definido por sessão de banco de dados, idealmente no início de cada requisição.

Opção 1: Usando Event Listeners SQLAlchemy

from sqlalchemy import event

@event.listens_for(Session, "before_cursor_execute")
def receive_before_cursor_execute(session, cursor, context, statement, parameters,
                                   executemany, context):
    tenant_id = current_tenant_id.get()
    if tenant_id is not None:
        cursor.execute(f"SET app.current_tenant = {tenant_id}")

Opção 2: Usando Middleware & FastAPI

Exemplo com FastAPI.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from contextlib import asynccontextmanager

app = FastAPI()

@asynccontextmanager
async def tenant_context(tenant_id: int):
    token = current_tenant_id.set(tenant_id)
    try:
        yield
    finally:
        current_tenant_id.reset(token)

@app.get("/documents")
async def list_documents(db: Session = Depends(get_db)):
    tenant_id = current_tenant_id.get()
    if tenant_id is None:
        raise HTTPException(status_code=401, detail="Contexto do tenant não definido")

    # RLS filtrará automaticamente os resultados
    documents = db.query(Document).all()
    return documents

Opção 3: Usando SQLAlchemy 2.0

from sqlalchemy import text

def set_tenant_context(session: Session, tenant_id: int):
    # com Opções de Execução
    session.execute(text(f"SET app.current_tenant = {tenant_id}"))
    session.commit()  # Requerido para comandos SET

Considerações de Segurança

Bypass de Administrador/Superusuário

Administradores podem precisar contornar o RLS para manutenção. O PostgreSQL suporta isso através de políticas baseadas em função:

CREATE POLICY admin_bypass ON seu_nome_de_tabela
    FOR ALL
    TO admin_role
    USING (true)
    WITH CHECK (true);

Ou desabilitar RLS temporariamente (arriscado):

ALTER TABLE seu_nome_de_tabela DISABLE ROW LEVEL SECURITY;

-- Realize operações de administrador...

ALTER TABLE seu_nome_de_tabela ENABLE ROW LEVEL SECURITY;

Forçar Segurança em Nível de Linha

Por padrão, o RLS não é aplicado aos proprietários da tabela. Para forçar RLS mesmo para proprietários:

ALTER TABLE seu_nome_de_tabela FORCE ROW LEVEL SECURITY;

Isso é recomendado para sistemas de produção com múltiplos inquilinos (tenants).

Políticas Permissivas vs. Restritivas

-- Permissivo (padrão): Qualquer política correspondente concede acesso
CREATE POLICY policy_a ON table_name FOR SELECT USING (...);
CREATE POLICY policy_b ON table_name FOR SELECT USING (...);
-- Acesso concedido se policy_a OU policy_b corresponder

-- Restritivo: Todas as políticas devem corresponder
-- Requer definição explícita de política com palavra-chave RESTRICTIVE

Testando Implementação RLS

Verifique se o RLS funciona corretamente antes de implantar:

-- Teste 1: Verificar se RLS está habilitado
\d+ seu_nome_de_tabela
-- Procure por "Row Security: yes"

-- Teste 2: Alternar para função de usuário tenant
SET ROLE tenant_user;
SELECT * FROM seu_nome_de_tabela;
-- Deve retornar apenas linhas do tenant

-- Teste 3: Tentar acesso entre tenants
SET app.current_tenant = 999;  -- Tenant diferente
SELECT * FROM seu_nome_de_tabela;
-- Deve retornar resultado vazio

-- Teste 4: Verificar se bypass de administrador funciona
SET ROLE admin_role;
SELECT * FROM seu_nome_de_tabela;
-- Deve retornar todas as linhas

Considerações de Performance

RLS adiciona sobrecarga a cada consulta. Otimize com:

Indexar colunas tenant_id:

CREATE INDEX idx_documents_tenant_id ON documents(tenant_id);

Usar pool de conexões: Cada conexão mantém suas próprias variáveis de sessão

Monitorar planos de consulta:

EXPLAIN ANALYZE SELECT * FROM seu_nome_de_tabela;

Considerar desnormalização para dados frequentemente acessados com escopo de tenant.

Armadilhas Comuns

Problema	Solução
RLS não filtrando resultados	Garantir que `ENABLE ROW LEVEL SECURITY` esteja definido
Usuários administradores vendo todos os dados	Usar `FORCE ROW LEVEL SECURITY`
Variável de sessão não persistindo	Definir por conexão, não por consulta
Migração falha em tabelas existentes	Remover políticas primeiro com `IF EXISTS`
Degradação de performance	Adicionar índices nas colunas `tenant_id`

Principais Descobertas e Conclusões

Se você está construindo arquiteturas multi-tenant modernas, estas são as percepções não negociáveis da nossa análise:

“Defesa em Profundidade (Defense-in-Depth)” por definição: RLS não é um substituto para lógica de aplicação; é a última linha de defesa. Mesmo que sua camada de API tenha um bug, o banco de dados silenciosamente descartará linhas não autorizadas. Isso centraliza a lógica de segurança, reduzindo a superfície de erro humano no código Python.
Alembic é seu Amigo, abrace-o: Muitas equipes temem gerenciar políticas RLS via migrações. Nós provamos que tratar políticas RLS como objetos de esquema de primeira classe (removendo/recriando em upgrade() e downgrade()) garante que sua postura de segurança evolua exatamente como seu esquema. Sem mais scripts SQL manuais em servidores de produção.
Contexto é imperativo: A mágica do RLS depende de variáveis de sessão (ex: app.current_tenant). Um padrão thread-safe usando Python’s contextvars e event listeners SQLAlchemy para definir automaticamente este contexto no início de cada requisição, garantindo que não haja código repetitivo na sua lógica de negócios.
Paradoxo do Bypass de Administrador: Um ponto de falha comum é negar acesso aos seus próprios administradores. Descrevi as configurações específicas de função PostgreSQL e configurações FORCE ROW LEVEL SECURITY necessárias para permitir que superusuários contornem políticas para manutenção sem desabilitar RLS completamente.
Performance: Verifique. O RLS está lento? Sim, ligeiramente. Com indexação adequada nas colunas tenant_id e o planejamento de queries certo, a sobrecarga é baixa se comparada ao custo de uma violação de dados.

Referências

Imagens geradas por IA

Adriano Vieira