203 lines
7.1 KiB
Python
203 lines
7.1 KiB
Python
"""Pruebas de aislamiento Row-Level Security por ``tenant_id`` y ``company_id``.
|
|
|
|
Las políticas se crean en la migración
|
|
``d1a2b3c4e5f6_enable_rls_tenant_company`` y dependen de las GUCs
|
|
``app.tenant_id`` / ``app.company_id`` fijadas por la aplicación.
|
|
|
|
Como ``postgres`` hace BYPASSRLS por defecto, los tests cambian de rol
|
|
dentro de la transacción a un usuario sin ese atributo (``anexo76_rls_test``)
|
|
antes de contar filas. Si la conexión de pruebas no tiene privilegios para
|
|
crear el rol, el set completo se salta con ``pytest.skip``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Iterator
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import DBAPIError, ProgrammingError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from tests.conftest import TestingSessionLocal, engine
|
|
from tests.fixtures.builders import ensure_tenant_company
|
|
|
|
|
|
TEST_ROLE = "anexo76_rls_test"
|
|
RLS_SCHEMAS = ("core", "a76", "a24", "public")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def rls_role() -> str:
|
|
"""Crea (idempotente) un rol no-superusuario con los privilegios mínimos
|
|
necesarios para ejercitar las políticas RLS en los tests."""
|
|
try:
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
f"""
|
|
DO $$
|
|
BEGIN
|
|
IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{TEST_ROLE}') THEN
|
|
CREATE ROLE {TEST_ROLE} NOLOGIN NOBYPASSRLS;
|
|
END IF;
|
|
END $$;
|
|
"""
|
|
)
|
|
)
|
|
for schema in RLS_SCHEMAS:
|
|
conn.execute(text(f'GRANT USAGE ON SCHEMA "{schema}" TO {TEST_ROLE}'))
|
|
conn.execute(
|
|
text(
|
|
f'GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA "{schema}" TO {TEST_ROLE}'
|
|
)
|
|
)
|
|
conn.execute(
|
|
text(
|
|
f'GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA "{schema}" TO {TEST_ROLE}'
|
|
)
|
|
)
|
|
except ProgrammingError as exc:
|
|
pytest.skip(f"No hay privilegio para preparar rol de pruebas RLS: {exc}")
|
|
return TEST_ROLE
|
|
|
|
|
|
@pytest.fixture
|
|
def rls_session() -> Iterator[Session]:
|
|
"""Sesión con transacción externa + rollback (no contamina BD real)."""
|
|
connection = engine.connect()
|
|
transaction = connection.begin()
|
|
session = TestingSessionLocal(bind=connection)
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
transaction.rollback()
|
|
connection.close()
|
|
|
|
|
|
def _allocate_id() -> int:
|
|
return 1_700_000_000 + (uuid.uuid4().int % 90_000_000)
|
|
|
|
|
|
def _bootstrap_two_tenants(session: Session) -> tuple[tuple[int, int], tuple[int, int]]:
|
|
tid_a, tid_b = _allocate_id(), _allocate_id()
|
|
cid_a, cid_b = _allocate_id(), _allocate_id()
|
|
ensure_tenant_company(session, tenant_id=tid_a, company_id=cid_a)
|
|
ensure_tenant_company(session, tenant_id=tid_b, company_id=cid_b)
|
|
return (tid_a, cid_a), (tid_b, cid_b)
|
|
|
|
|
|
def _seed_clients(session: Session, tenant_id: int, company_id: int, *, count: int) -> None:
|
|
"""Inserta filas en ``a76.clients_and_providers`` con SQL crudo para
|
|
aislarnos del ORM y medir filtrado puro a nivel de BD."""
|
|
for i in range(count):
|
|
session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO a76.clients_and_providers
|
|
(tenant_id, company_id, name, client_or_provider, is_active)
|
|
VALUES
|
|
(:tid, :cid, :name, 'BOTH', true)
|
|
"""
|
|
),
|
|
{"tid": tenant_id, "cid": company_id, "name": f"Seed {tenant_id}/{company_id}/{i}"},
|
|
)
|
|
session.flush()
|
|
|
|
|
|
def _count_under_role(
|
|
session: Session,
|
|
tenant_id: int | None,
|
|
company_id: int | None,
|
|
*,
|
|
role: str,
|
|
table: str = "a76.clients_and_providers",
|
|
) -> int:
|
|
"""Cuenta filas tras cambiar a un rol sin BYPASSRLS y fijar el contexto."""
|
|
session.execute(text(f"SET LOCAL ROLE {role}"))
|
|
session.execute(
|
|
text("SELECT set_config('app.tenant_id', :t, true), set_config('app.company_id', :c, true)"),
|
|
{
|
|
"t": "" if tenant_id is None else str(tenant_id),
|
|
"c": "" if company_id is None else str(company_id),
|
|
},
|
|
)
|
|
try:
|
|
return int(session.execute(text(f"SELECT count(*) FROM {table}")).scalar() or 0)
|
|
finally:
|
|
session.execute(text("RESET ROLE"))
|
|
|
|
|
|
def test_tenant_isolation_hides_rows_from_other_tenant(
|
|
rls_session: Session, rls_role: str
|
|
) -> None:
|
|
(tid_a, cid_a), (tid_b, cid_b) = _bootstrap_two_tenants(rls_session)
|
|
_seed_clients(rls_session, tid_a, cid_a, count=3)
|
|
_seed_clients(rls_session, tid_b, cid_b, count=2)
|
|
|
|
visible_a = _count_under_role(rls_session, tid_a, None, role=rls_role)
|
|
visible_b = _count_under_role(rls_session, tid_b, None, role=rls_role)
|
|
|
|
assert visible_a == 3, "Tenant A debe ver solo sus 3 filas"
|
|
assert visible_b == 2, "Tenant B debe ver solo sus 2 filas"
|
|
|
|
|
|
def test_company_scope_narrows_within_tenant(
|
|
rls_session: Session, rls_role: str
|
|
) -> None:
|
|
(tid_a, cid_a), _ = _bootstrap_two_tenants(rls_session)
|
|
|
|
second_company_id = _allocate_id()
|
|
ensure_tenant_company(rls_session, tenant_id=tid_a, company_id=second_company_id)
|
|
|
|
_seed_clients(rls_session, tid_a, cid_a, count=4)
|
|
_seed_clients(rls_session, tid_a, second_company_id, count=7)
|
|
|
|
without_company = _count_under_role(rls_session, tid_a, None, role=rls_role)
|
|
with_company_a = _count_under_role(rls_session, tid_a, cid_a, role=rls_role)
|
|
with_second = _count_under_role(rls_session, tid_a, second_company_id, role=rls_role)
|
|
|
|
assert without_company == 11, "Sin company context el tenant ve ambas compañías"
|
|
assert with_company_a == 4
|
|
assert with_second == 7
|
|
|
|
|
|
def test_missing_tenant_context_returns_zero_rows(
|
|
rls_session: Session, rls_role: str
|
|
) -> None:
|
|
(tid_a, cid_a), _ = _bootstrap_two_tenants(rls_session)
|
|
_seed_clients(rls_session, tid_a, cid_a, count=5)
|
|
|
|
visible = _count_under_role(rls_session, None, None, role=rls_role)
|
|
assert visible == 0, "Sin app.tenant_id la política debe devolver 0 filas"
|
|
|
|
|
|
def test_insert_violation_respects_tenant_policy(
|
|
rls_session: Session, rls_role: str
|
|
) -> None:
|
|
"""La cláusula ``WITH CHECK`` debe rechazar inserts fuera de contexto."""
|
|
(tid_a, cid_a), (tid_b, _) = _bootstrap_two_tenants(rls_session)
|
|
|
|
rls_session.execute(text(f"SET LOCAL ROLE {rls_role}"))
|
|
rls_session.execute(
|
|
text("SELECT set_config('app.tenant_id', :t, true), set_config('app.company_id', :c, true)"),
|
|
{"t": str(tid_a), "c": str(cid_a)},
|
|
)
|
|
try:
|
|
with pytest.raises(DBAPIError):
|
|
rls_session.execute(
|
|
text(
|
|
"""
|
|
INSERT INTO a76.clients_and_providers
|
|
(tenant_id, company_id, name, client_or_provider, is_active)
|
|
VALUES
|
|
(:tid, :cid, 'Cross-tenant attempt', 'BOTH', true)
|
|
"""
|
|
),
|
|
{"tid": tid_b, "cid": cid_a},
|
|
)
|
|
finally:
|
|
rls_session.rollback()
|