0012. RLS: aislamiento por empresa y cómo se usa en código¶

Estado: Aceptado Fecha: 2026-09-03 Autor: Duval Alcivar

Contexto¶

Con 3 bases de datos compartidas por 14 microservicios, el aislamiento entre empresas se resuelve por Row-Level Security (RLS) de PostgreSQL usando empresa_id (ver ADR 0002).

Este ADR define cómo se implementa RLS en código para que ningún Repository tenga que escribir WHERE empresa_id a mano. La idea central: el SET app.current_empresa_id ocurre una sola vez por request, en un middleware, y RLS filtra automáticamente el resto.

1. ¿Qué es RLS?¶

RLS es una característica de PostgreSQL que filtra automáticamente las filas de una tabla según una condición. En nuestro caso: filtra por empresa_id.

Sin RLS: El programador debe acordarse de poner WHERE empresa_id = 5 en CADA query. Si se olvida una, se filtra datos de otra empresa.

Con RLS: PostgreSQL lo hace automáticamente. No puedes ver datos de otra empresa aunque te olvides del filtro.

Ejemplo sin RLS (MAL)¶

-- El programador debe recordar poner el filtro en CADA query
SELECT * FROM erp.trx_pedidos WHERE empresa_id = 5;
SELECT * FROM erp.trx_facturas WHERE empresa_id = 5;
SELECT * FROM erp.trx_movimientos WHERE empresa_id = 5;

-- Si se olvida UNA vez...
SELECT * FROM erp.trx_pedidos;  -- ¡ERROR! Ve pedidos de TODAS las empresas

Ejemplo con RLS (BIEN)¶

-- 1. Crear la política de aislamiento
CREATE POLICY aislamiento_empresa ON erp.trx_pedidos
    USING (empresa_id = NULLIF(current_setting('app.current_empresa_id', true), '')::BIGINT);

-- 2. Activar RLS en la tabla
ALTER TABLE erp.trx_pedidos ENABLE ROW LEVEL SECURITY;

-- 3. Ahora PostgreSQL filtra automáticamente
SET app.current_empresa_id = '5';

SELECT * FROM erp.trx_pedidos;
-- Solo devuelve pedidos WHERE empresa_id = 5
-- No necesitas escribir el filtro, PostgreSQL lo hace

2. Cómo se implementa en código (Python/FastAPI)¶

Paso 1: Crear la política SQL (una vez, al crear la tabla)¶

El esquema (tablas + política RLS) no se crea a mano. Se versiona con Alembic (de SQLAlchemy), cada cambio es un archivo versionado con rollback posible.

a) Crear el archivo de migración¶

cd svc-pedidos
alembic revision -m "crear tabla trx_pedidos"

b) Escribir la migración (crea la tabla + la política RLS)¶

# svc-pedidos/alembic/versions/20260903_001_crear_trx_pedidos.py
from alembic import op
import sqlalchemy as sa

revision = "20260903_001"
down_revision = None

def upgrade():
    op.create_table(
        "trx_pedidos",
        sa.Column("id", sa.UUID(), primary_key=True),
        sa.Column("empresa_id", sa.BigInteger(), nullable=False),
        sa.Column("cliente_id", sa.UUID(), nullable=False),
        sa.Column("sucursal_id", sa.UUID(), nullable=False),
        sa.Column("total", sa.Numeric(12, 2), nullable=False),
        sa.Column("fecha_pedido", sa.Date(), nullable=False),
        sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now()),
    )

    # ── Política RLS ─────────────────────────────────────────────
    # RLS no tiene helper de SQLAlchemy, se escribe SQL directo
    op.execute("""
        CREATE POLICY aislamiento_empresa ON trx_pedidos
        USING (empresa_id = NULLIF(current_setting('app.current_empresa_id', true), '')::BIGINT);
    """)
    op.execute("ALTER TABLE trx_pedidos ENABLE ROW LEVEL SECURITY;")

def downgrade():
    # al borrar la tabla se borra también su política
    op.drop_table("trx_pedidos")

c) Aplicar la migración¶

alembic upgrade head

Nota: la política RLS siempre se crea junto con su tabla en la misma migración, usando op.execute(...) con SQL directo. No se hace a mano ni en un script suelto.

Paso 2: Middleware — setear `empresa_id` UNA vez por request¶

Clave: el SET NO va en el Repository. Va en un middleware que se ejecuta una sola vez por request, ANTES de que entre a cualquier Router. Así ningún Repository tiene que escribir WHERE empresa_id ni SET.

# svc-pedidos/main.py
from fastapi import FastAPI, Request
from sqlalchemy import text

app = FastAPI()

# ══════════════════════════════════════════════════════════
#  MIDDLEWARE: se ejecuta 1 vez por request, ANTES del Router
#  Ningún Repository necesita saber de RLS
# ══════════════════════════════════════════════════════════
@app.middleware("http")
async def set_empresa_actual(request: Request, call_next):
    # 1. Extrae JWT del header
    token = request.headers.get("Authorization", "").replace("Bearer ", "")

    # 2. Decodifica y saca empresa_id (solo valida firma, sin tocar DB)
    payload = decode_jwt(token)
    empresa_id = payload.get("empresa_id")

    # 3. Setea la variable de sesión UNA vez por request
    if empresa_id:
        request.state.empresa_id = empresa_id   # Disponible en TODO el request
        db = request.app.state.db
        db.execute(text(f"SET app.current_empresa_id = '{empresa_id}'"))

    # 4. Sigue con el request normal
    response = await call_next(request)
    return response

Resultado: los Repositories quedan limpios. No escriben SET ni WHERE empresa_id:

# svc-pedidos/repositories/pedido_repository.py
class PedidoRepository:
    def __init__(self, db: Session):
        self.db = db

    def obtener_pedidos(self) -> List[TrxPedido]:
        # ✅ SIN SET, SIN WHERE empresa_id
        # RLS ya filtró a la empresa correcta (lo hizo el middleware)
        return self.db.query(TrxPedido).all()

Paso 3: Dependencia para extraer `empresa_id` del JWT¶

Para las rutas que ESCRIBEN (crear/actualizar/borrar), se usa una dependencia que lee request.state.empresa_id (que el middleware ya dejó):

# svc-pedidos/dependencies.py
from fastapi import Depends, HTTPException, Request

def get_empresa_actual(request: Request) -> int:
    """
    Devuelve el empresa_id que el middleware dejó en request.state.
    Nunca viene del request del cliente.
    """
    empresa_id = request.state.empresa_id
    if not empresa_id:
        raise HTTPException(status_code=400, detail="Token sin empresa_id")
    return empresa_id  # BIGINT del JWT

Paso 4: Router — usar la dependencia SOLO en rutas que escriben¶

# svc-pedidos/routers/pedido_router.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from ..dependencies import get_db, get_empresa_actual
from ..services.pedido_service import PedidoService

router = APIRouter()

# ── LEER lista: NO necesita empresa_id ✔
#    RLS ya filtra solo (middleware lo setteó)
@router.get("/api/v1/pedidos")
def listar_pedidos(db: Session = Depends(get_db)):
    return PedidoService(db).listar()

# ── LEER uno: NO necesita empresa_id ✔
@router.get("/api/v1/pedidos/{pedido_id}")
def obtener_pedido(pedido_id: str, db: Session = Depends(get_db)):
    return PedidoService(db).obtener(pedido_id)

# ── CREAR: SÍ necesita empresa_id ✏ (hay que guardarlo en la fila)
@router.post("/api/v1/pedidos")
def crear_pedido(
    data: PedidoCreate,
    empresa_id: int = Depends(get_empresa_actual),  # del JWT
    db: Session = Depends(get_db)
):
    return PedidoService(db).crear(data, empresa_id)

# ── ACTUALIZAR: SÍ necesita empresa_id ✏ (seguridad: no tocar otra empresa)
@router.put("/api/v1/pedidos/{pedido_id}")
def actualizar_pedido(
    pedido_id: str,
    data: PedidoUpdate,
    empresa_id: int = Depends(get_empresa_actual),
    db: Session = Depends(get_db)
):
    return PedidoService(db).actualizar(pedido_id, data, empresa_id)

# ── BORRAR: SÍ necesita empresa_id ✏ (seguridad)
@router.delete("/api/v1/pedidos/{pedido_id}")
def eliminar_pedido(
    pedido_id: str,
    empresa_id: int = Depends(get_empresa_actual),
    db: Session = Depends(get_db)
):
    return PedidoService(db).eliminar(pedido_id, empresa_id)

Cuándo usar Depends(get_empresa_actual):

Tipo de ruta	Ejemplo	¿Necesitas `get_empresa_actual`?	¿Por qué?
CREAR	`POST /api/v1/pedidos`	SÍ	Hay que guardar `empresa_id` en la columna del INSERT
ACTUALIZAR	`PUT /api/v1/pedidos/{id}`	SÍ	Seguridad: no modificar fila de otra empresa
BORRAR	`DELETE /api/v1/pedidos/{id}`	SÍ	Seguridad: no borrar fila de otra empresa
LEER (lista)	`GET /api/v1/pedidos`	NO	RLS ya filtra solo
LEER (uno)	`GET /api/v1/pedidos/{id}`	NO	RLS ya filtra solo (si no existe, 404)

3. Flujo completo de un request con RLS¶

USUARIO                  svc-pedidos                 PostgreSQL
   │                         │                            │
   │  GET /api/v1/pedidos       │                            │
   │  Authorization:         │                            │
   │  Bearer <JWT>           │                            │
   │  ──────────────────────>│                            │
   │                         │                            │
   │                         │  1. Extrae empresa_id = 5  │
   │                         │     del JWT                │
   │                         │                            │
   │                         │  2. Ejecuta:               │
   │                         │     SET app.empresa_id     │
   │                         │     _actual = '5'          │
   │                         │  ─────────────────────────>│
   │                         │                            │
   │                         │  3. Query:                 │
   │                         │     SELECT * FROM          │
   │                         │     trx_pedidos            │
   │                         │  ─────────────────────────>│
   │                         │                            │
   │                         │                    ┌───────┴───────┐
   │                         │                    │ RLS filtra:   │
   │                         │                    │ WHERE         │
   │                         │                    │ empresa_id=5  │
   │                         │                    └───────┬───────┘
   │                         │                            │
   │                         │  Solo devuelve pedidos     │
   │                         │  de empresa 5              │
   │                         │  <─────────────────────────│
   │                         │                            │
   │  [{pedidos...}]         │                            │
   │<────────────────────────│                            │

¿Qué pasa si intento ver datos de otra empresa?¶

# Usuario de empresa 5 intenta ver pedidos de empresa 2

# 1. JWT dice: empresa_id = 5
SET app.current_empresa_id = '5';

# 2. Query normal
SELECT * FROM erp.trx_pedidos;

# 3. PostgreSQL aplica RLS automáticamente:
WHERE empresa_id = 5  -- Solo ve pedidos de empresa 5

# Resultado: No ve NADA de empresa 2

4. RLS en cada servicio¶

Servicio	Tablas con RLS	Política
svc-identidad	`mae_sucursales`, `rel_usuario_sucursal`, `trx_refresh_tokens`, `log_sesiones`	`empresa_id`
svc-identidad (navegación)	`rel_empresa_aplicacion` (por `empresa_id`), `cfg_modulos`, `cfg_menu`, `cfg_menu_acciones`	`empresa_id`
svc-clientes	`mae_clientes`, `mae_contactos`	`empresa_id`
svc-pedidos	`trx_pedidos`	`empresa_id`
svc-contable	`trx_facturas`, `trx_asientos`	`empresa_id`
svc-operaciones	`trx_movimientos`, `mae_productos`	`empresa_id`
cada producto (auditoría)	`audit_log`	`empresa_id`

Nota: mae_usuarios, mae_empresas y cat_aplicaciones no tienen RLS porque son globales: usuarios y aplicaciones cruzan empresas (catálogo de productos), y mae_empresas es la propia fuente de empresa_id (no tiene empresa_id sobre sí misma). Ver ADRs 0004 y 0005 de svc-identidad para el detalle completo.

Cada tabla nueva debe tener RLS activado desde el primer commit.

5. Comunicación servicio a servicio (todo en tiempo real)¶

Service Account (cada microservicio tiene su propio JWT)¶

Principio: Cada servicio es dueño de sus datos y expone endpoints. Los demás solo consultan esos endpoints. No se duplican tablas, no se crean endpoints nuevos en otros servicios.

┌─────────────────────────────────────────────────────────────────────────┐
│  SERVICE-TO-SERVICE COMMUNICATION                                        │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  1. Cada servicio tiene una cuenta de servicio en svc-identidad          │
│                                                                         │
│  2. Al arrancar, obtiene un JWT con scope limitado:                     │
│     {                                                                   │
│       "sub": "svc-clientes-service-account",                            │
│       "scope": ["read:usuarios", "read:sucursales"],                    │
│       "iss": "svc-identidad",                                          │
│       "aud": "svc-identidad"                                            │
│     }                                                                   │
│                                                                         │
│  3. Cuando necesita datos de otro servicio:                             │
│     - Usa su JWT de servicio (NO el del usuario)                        │
│     - Llama al endpoint EXISTENTE del otro servicio                     │
│     - No crea endpoints nuevos, no duplica tablas                       │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Ejemplo: svc-clientes obtiene datos de usuario¶

# svc-clientes/repositories/usuario_repository.py
class UsuarioRepository:
    def __init__(self, http: HTTPClient):
        self.http = http
        self.service_token = os.getenv("SERVICE_TOKEN")  # JWT de servicio

    def obtener_usuario(self, usuario_id: UUID) -> dict:
        """
        Consulta svc-identidad para obtener datos del usuario.
        USA el token de servicio, NO el del usuario.
        Llama al endpoint EXISTENTE, no crea uno nuevo.
        """
        respuesta = self.http.get(
            f"http://svc-identidad/api/v1/usuarios/{usuario_id}",  # Endpoint ya existe
            headers={
                "Authorization": f"Bearer {self.service_token}",
                "X-Request-ID": str(uuid.uuid4())
            }
        )

        if respuesta.status_code == 200:
            return respuesta.json()

        raise Exception("Usuario no encontrado")

Endpoints centralizados (NO se duplican)¶

Servicio	Dueño de	Endpoints que expone
svc-identidad	usuarios, empresas, sucursales	`GET /api/v1/usuarios/{id}`, `GET /api/v1/empresas/{id}`, `GET /api/v1/sucursales/{id}`
svc-clientes	clientes, contactos	`GET /api/v1/clientes/{id}`
svc-pedidos	pedidos	`GET /api/v1/pedidos/{id}`
svc-contable	facturas, CxC	`GET /api/v1/facturas/{id}`
svc-operaciones	inventario, productos	`GET /api/v1/productos/{id}`

Qué consulta cada servicio¶

Servicio	Necesita datos de	Llama a	Endpoint
svc-clientes	usuario	svc-identidad	`GET /api/v1/usuarios/{id}`
svc-pedidos	usuario	svc-identidad	`GET /api/v1/usuarios/{id}`
svc-pedidos	cliente	svc-clientes	`GET /api/v1/clientes/{id}`
svc-operaciones	proveedor	svc-compras	`GET /api/v1/proveedores/{id}`
svc-documents	factura	svc-contable	`GET /api/v1/facturas/{id}`
svc-documents	cliente	svc-clientes	`GET /api/v1/clientes/{id}`

No se crean endpoints nuevos. No se duplican tablas. Solo se consultan los endpoints que ya existen.

Reglas no negociables¶

RLS activado en cada tabla desde el primer commit
El middleware hace SET app.current_empresa_id UNA vez por request — los Repositories no lo repiten
Depends(get_empresa_actual) solo en rutas que ESCRIBEN (crear/actualizar/borrar), no en las que leen
empresa_id viene del JWT, nunca del request del cliente
No hacer SQL cruzado entre bases (dblink/FDW prohibidos)
No duplicar tablas ni endpoints entre servicios — solo consultar los endpoints existentes

Consecuencias¶

RLS debe estar activado en cada tabla desde el primer commit
El SET app.current_empresa_id ocurre en el middleware, una sola vez por request
Depends(get_empresa_actual) solo en rutas que escriben (crear/actualizar/borrar)
empresa_id siempre viene del JWT, nunca del request del cliente
Los servicios se comunican con service account JWT, consultando endpoints existentes