Agent Zero
5d8c48a08f
Root-Causes und Fixes: (1) conftest.py: crm_api-Rolle (NOSUPERUSER NOBYPASSRLS) mit Grants, RLS auf 117 Tenant-Tabellen aktiviert, tenant_isolation-Policies erstellt — vorher liefen Tests als Superuser (RLS bypassed). (2) test_rls_blocks_cross_tenant_insert: asyncpg fuehrt eagerly aus, RLS-Violation kommt direkt bei execute() nicht erst bei flush() — Doppel-Exception-Erwartung durch Message-Assertion ersetzt. (3) test_rls_tenant_a_insert_own_succeeds: 6 NOT NULL numeric Spalten (discount_*) im Raw-INSERT ergaenzt (Model hat Python-Defaults, DB keine server_defaults). (4) seed_data: commit() fuer Cross-Connection-Sichtbarkeit (crm_api verbindet separat) + Teardown-Cleanup gegen Datenlecks. (5) admin_session: ohne conn.begin() — sonst conditional_savepoint und commit() wirkungslos. (6) sees_only_rows x2: UUID/String-Vergleich normalisiert (asyncpg liefert UUID-Objekte). Vorher: 9 von 10 Tests vakuum-trivial gruen (leere DB, Superuser). Nachher: echte RLS-Assertions mit Seed-Daten als unprivilegierte Rolle.
451 lines
17 KiB
Python
451 lines
17 KiB
Python
"""Cross-Tenant Security Tests v2 — RLS enforcement with unprivileged DB roles.
|
|
|
|
These tests verify that PostgreSQL Row Level Security (RLS) actually blocks
|
|
cross-tenant access when using the unprivileged ``crm_api`` role
|
|
(NOSUPERUSER, NOBYPASSRLS, not table owner).
|
|
|
|
Unlike v1 tests which run as ``crm_user`` (superuser, RLS bypassed),
|
|
these tests connect as ``crm_api`` to verify RLS enforcement at the DB level.
|
|
|
|
Test matrix:
|
|
- No tenant context → SELECT returns 0 rows, INSERT fails
|
|
- Tenant A context → only Tenant A rows visible, Tenant B insert blocked
|
|
- Tenant B context → only Tenant B rows visible, Tenant A insert blocked
|
|
- Cross-tenant write → WITH CHECK blocks wrong tenant_id on INSERT/UPDATE
|
|
|
|
Requirements:
|
|
- PostgreSQL with RLS enabled
|
|
- ``crm_api`` role (NOSUPERUSER, NOBYPASSRLS)
|
|
- ``crm_api`` has SELECT/INSERT/UPDATE/DELETE on tenant tables
|
|
- ``crm_api`` is NOT the table owner
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
|
|
|
# Set test environment BEFORE any app imports
|
|
os.environ["SESSION_COOKIE_SECURE"] = "false"
|
|
os.environ["SESSION_COOKIE_SAMESITE"] = "lax"
|
|
os.environ["ENVIRONMENT"] = "testing"
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-with-at-least-32-characters-for-testing-only!!")
|
|
|
|
from app.core.db import set_tenant_context
|
|
from app.models.contact import Contact
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import User, UserTenant
|
|
|
|
|
|
# Use the unprivileged crm_api role for RLS testing
|
|
# Falls back to leocrm user if crm_api is not available (local dev)
|
|
_API_DB_URL = os.environ.get(
|
|
"RLS_TEST_DB_URL",
|
|
"postgresql+asyncpg://crm_api:crm_api_password@localhost:5432/leocrm_test",
|
|
)
|
|
|
|
# Superuser URL for setup (creating tenants, users, etc.)
|
|
_ADMIN_DB_URL = os.environ.get(
|
|
"RLS_TEST_ADMIN_DB_URL",
|
|
"postgresql+asyncpg://postgres@localhost:5432/leocrm_test",
|
|
)
|
|
|
|
|
|
def _skip_if_no_rls_role():
|
|
"""Skip tests if the unprivileged RLS role is not available."""
|
|
try:
|
|
import asyncio
|
|
eng = create_async_engine(_API_DB_URL, echo=False)
|
|
async def _check():
|
|
async with eng.connect() as conn:
|
|
await conn.execute(text("SELECT 1"))
|
|
asyncio.get_event_loop().run_until_complete(_check())
|
|
eng.dispose()
|
|
return False
|
|
except Exception:
|
|
eng.dispose()
|
|
return True
|
|
|
|
|
|
_skip_reason = (
|
|
"Unprivileged crm_api role not available for RLS testing. "
|
|
"Set RLS_TEST_DB_URL to a connection string using a NOSUPERUSER/NOBYPASSRLS role."
|
|
)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def admin_engine():
|
|
"""Admin engine for setup (creates tenants, users, contacts)."""
|
|
eng = create_async_engine(_ADMIN_DB_URL, echo=False)
|
|
yield eng
|
|
await eng.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def api_engine():
|
|
"""Unprivileged engine using crm_api role — RLS enforced."""
|
|
eng = create_async_engine(_API_DB_URL, echo=False)
|
|
yield eng
|
|
await eng.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def admin_session(admin_engine):
|
|
"""Admin session for data setup.
|
|
|
|
No externally begun transaction: the session must own its transactions so
|
|
that seed_data's commit() really persists rows (visible to the separate
|
|
crm_api connection used by api_session). With conn.begin() the Session
|
|
would treat it as an external transaction (conditional_savepoint) and
|
|
commit() would not release it.
|
|
"""
|
|
async with admin_engine.connect() as conn:
|
|
session = AsyncSession(bind=conn, expire_on_commit=False)
|
|
yield session
|
|
await session.close()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def api_session(api_engine):
|
|
"""Unprivileged session using crm_api — RLS enforced."""
|
|
async with api_engine.connect() as conn:
|
|
await conn.begin()
|
|
session = AsyncSession(bind=conn, expire_on_commit=False)
|
|
yield session
|
|
await session.rollback()
|
|
await conn.rollback()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def seed_data(admin_session: AsyncSession):
|
|
"""Seed two tenants with contacts using admin (superuser) connection."""
|
|
from app.core.auth import hash_password
|
|
|
|
tenant_a = Tenant(id=uuid.uuid4(), name="RLS Tenant A", slug=f"rls-a-{uuid.uuid4().hex[:8]}")
|
|
tenant_b = Tenant(id=uuid.uuid4(), name="RLS Tenant B", slug=f"rls-b-{uuid.uuid4().hex[:8]}")
|
|
admin_session.add_all([tenant_a, tenant_b])
|
|
await admin_session.flush()
|
|
|
|
user_a = User(
|
|
id=uuid.uuid4(),
|
|
name="RLS User A",
|
|
email=f"rls-a-{uuid.uuid4().hex[:8]}@test.local",
|
|
password_hash=hash_password("TestPass123!"),
|
|
is_active=True,
|
|
is_system_admin=False,
|
|
)
|
|
user_b = User(
|
|
id=uuid.uuid4(),
|
|
name="RLS User B",
|
|
email=f"rls-b-{uuid.uuid4().hex[:8]}@test.local",
|
|
password_hash=hash_password("TestPass123!"),
|
|
is_active=True,
|
|
is_system_admin=False,
|
|
)
|
|
admin_session.add_all([user_a, user_b])
|
|
await admin_session.flush()
|
|
|
|
ut_a = UserTenant(user_id=user_a.id, tenant_id=tenant_a.id, role="admin", status="active", is_default=True)
|
|
ut_b = UserTenant(user_id=user_b.id, tenant_id=tenant_b.id, role="admin", status="active", is_default=True)
|
|
admin_session.add_all([ut_a, ut_b])
|
|
await admin_session.flush()
|
|
|
|
contact_a = Contact(
|
|
id=uuid.uuid4(),
|
|
tenant_id=tenant_a.id,
|
|
firstname="RLS",
|
|
surname="Alpha",
|
|
email_1=f"rls-alpha-{uuid.uuid4().hex[:8]}@contact.local",
|
|
owner_id=user_a.id,
|
|
created_by=user_a.id,
|
|
updated_by=user_a.id,
|
|
)
|
|
contact_b = Contact(
|
|
id=uuid.uuid4(),
|
|
tenant_id=tenant_b.id,
|
|
firstname="RLS",
|
|
surname="Beta",
|
|
email_1=f"rls-beta-{uuid.uuid4().hex[:8]}@contact.local",
|
|
owner_id=user_b.id,
|
|
created_by=user_b.id,
|
|
updated_by=user_b.id,
|
|
)
|
|
admin_session.add_all([contact_a, contact_b])
|
|
await admin_session.flush()
|
|
|
|
# COMMIT is mandatory here: the api_session fixture connects as the
|
|
# unprivileged crm_api role on a SEPARATE connection. Uncommitted rows
|
|
# from the admin connection are invisible there, so FK checks
|
|
# (contacts.updated_by -> users) and RLS visibility assertions would
|
|
# operate on an empty database.
|
|
await admin_session.commit()
|
|
|
|
yield {
|
|
"tenant_a": tenant_a,
|
|
"tenant_b": tenant_b,
|
|
"user_a": user_a,
|
|
"user_b": user_b,
|
|
"contact_a": contact_a,
|
|
"contact_b": contact_b,
|
|
}
|
|
|
|
# Teardown: remove seeded rows so the committed data does not leak into
|
|
# other tests (order respects FK dependencies).
|
|
await admin_session.execute(
|
|
text("DELETE FROM contacts WHERE id IN (:a, :b)"),
|
|
{"a": contact_a.id, "b": contact_b.id},
|
|
)
|
|
await admin_session.execute(
|
|
text("DELETE FROM user_tenants WHERE user_id IN (:a, :b)"),
|
|
{"a": user_a.id, "b": user_b.id},
|
|
)
|
|
await admin_session.execute(
|
|
text("DELETE FROM users WHERE id IN (:a, :b)"),
|
|
{"a": user_a.id, "b": user_b.id},
|
|
)
|
|
await admin_session.execute(
|
|
text("DELETE FROM tenants WHERE id IN (:a, :b)"),
|
|
{"a": tenant_a.id, "b": tenant_b.id},
|
|
)
|
|
await admin_session.commit()
|
|
|
|
|
|
# ── RLS Enforcement Tests with Unprivileged Role ─────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_no_tenant_context_returns_zero_rows(api_session: AsyncSession, seed_data):
|
|
"""Without tenant context, SELECT on tenant table must return 0 rows."""
|
|
result = await api_session.execute(
|
|
text("SELECT count(*) FROM contacts WHERE deleted_at IS NULL")
|
|
)
|
|
count = result.scalar()
|
|
assert count == 0, f"RLS fail-open: {count} rows visible without tenant context!"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_tenant_a_sees_only_own_rows(api_session: AsyncSession, seed_data):
|
|
"""With tenant A context, only tenant A contacts are visible."""
|
|
tenant_a = seed_data["tenant_a"]
|
|
tenant_b = seed_data["tenant_b"]
|
|
|
|
await set_tenant_context(api_session, tenant_a.id)
|
|
result = await api_session.execute(
|
|
text("SELECT tenant_id FROM contacts WHERE deleted_at IS NULL")
|
|
)
|
|
rows = result.fetchall()
|
|
for row in rows:
|
|
assert str(row[0]) == str(tenant_a.id), \
|
|
f"RLS leak: tenant A context shows row from {row[0]}"
|
|
# Tenant B's contact must not be visible
|
|
tenant_b_ids = [r for r in rows if str(r[0]) == str(tenant_b.id)]
|
|
assert len(tenant_b_ids) == 0, "RLS failed: Tenant B data visible in Tenant A context!"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_tenant_b_sees_only_own_rows(api_session: AsyncSession, seed_data):
|
|
"""With tenant B context, only tenant B contacts are visible."""
|
|
tenant_a = seed_data["tenant_a"]
|
|
tenant_b = seed_data["tenant_b"]
|
|
|
|
await set_tenant_context(api_session, tenant_b.id)
|
|
result = await api_session.execute(
|
|
text("SELECT tenant_id FROM contacts WHERE deleted_at IS NULL")
|
|
)
|
|
rows = result.fetchall()
|
|
for row in rows:
|
|
assert str(row[0]) == str(tenant_b.id), \
|
|
f"RLS leak: tenant B context shows row from {row[0]}"
|
|
tenant_a_ids = [r for r in rows if str(r[0]) == str(tenant_a.id)]
|
|
assert len(tenant_a_ids) == 0, "RLS failed: Tenant A data visible in Tenant B context!"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_blocks_cross_tenant_insert(api_session: AsyncSession, seed_data):
|
|
"""RLS WITH CHECK must block INSERT with wrong tenant_id."""
|
|
tenant_a = seed_data["tenant_a"]
|
|
tenant_b = seed_data["tenant_b"]
|
|
user_a = seed_data["user_a"]
|
|
|
|
await set_tenant_context(api_session, tenant_a.id)
|
|
|
|
# Try to insert a contact with tenant B's ID while in tenant A context.
|
|
# asyncpg executes eagerly: the RLS WITH CHECK violation surfaces directly
|
|
# at execute() (asyncpg InsufficientPrivilege -> SQLAlchemy ProgrammingError,
|
|
# both subclasses of DBAPIError).
|
|
new_id = uuid.uuid4()
|
|
from sqlalchemy.exc import DBAPIError
|
|
with pytest.raises(DBAPIError) as exc_info:
|
|
await api_session.execute(
|
|
text(
|
|
"INSERT INTO contacts (id, tenant_id, firstname, surname, email_1, "
|
|
"owner_id, created_by, updated_by, type, displayname, status) "
|
|
"VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname, 'open')"
|
|
),
|
|
{
|
|
"id": str(new_id),
|
|
"tenant_id": str(tenant_b.id), # Wrong tenant!
|
|
"firstname": "Cross",
|
|
"surname": "Tenant",
|
|
"email": f"cross-{uuid.uuid4().hex[:8]}@test.local",
|
|
"owner": str(user_a.id),
|
|
"creator": str(user_a.id),
|
|
"updater": str(user_a.id),
|
|
"ctype": "person",
|
|
"dname": "Cross Tenant",
|
|
},
|
|
)
|
|
# The INSERT must have failed due to RLS WITH CHECK — verify the message
|
|
err_text = str(exc_info.value).lower()
|
|
assert "row-level security" in err_text or "row level security" in err_text, \
|
|
f"Expected RLS violation, got: {exc_info.value}"
|
|
# Roll back the failed transaction so subsequent statements work
|
|
await api_session.rollback()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_blocks_cross_tenant_update(api_session: AsyncSession, seed_data):
|
|
"""RLS must block UPDATE of tenant B's row from tenant A context."""
|
|
tenant_a = seed_data["tenant_a"]
|
|
contact_b = seed_data["contact_b"]
|
|
|
|
await set_tenant_context(api_session, tenant_a.id)
|
|
|
|
# Try to update tenant B's contact from tenant A context
|
|
result = await api_session.execute(
|
|
text("UPDATE contacts SET surname = 'Hacked' WHERE id = :id"),
|
|
{"id": str(contact_b.id)},
|
|
)
|
|
# Should affect 0 rows (RLS hides tenant B's row from tenant A context)
|
|
assert result.rowcount == 0, \
|
|
f"RLS failed: UPDATE affected {result.rowcount} rows in cross-tenant context!"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_blocks_cross_tenant_delete(api_session: AsyncSession, seed_data):
|
|
"""RLS must block DELETE of tenant B's row from tenant A context."""
|
|
tenant_a = seed_data["tenant_a"]
|
|
contact_b = seed_data["contact_b"]
|
|
|
|
await set_tenant_context(api_session, tenant_a.id)
|
|
|
|
result = await api_session.execute(
|
|
text("DELETE FROM contacts WHERE id = :id"),
|
|
{"id": str(contact_b.id)},
|
|
)
|
|
assert result.rowcount == 0, \
|
|
f"RLS failed: DELETE affected {result.rowcount} rows in cross-tenant context!"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_tenant_a_insert_own_succeeds(api_session: AsyncSession, seed_data):
|
|
"""RLS allows INSERT with correct tenant_id in tenant A context."""
|
|
tenant_a = seed_data["tenant_a"]
|
|
user_a = seed_data["user_a"]
|
|
|
|
await set_tenant_context(api_session, tenant_a.id)
|
|
|
|
new_id = uuid.uuid4()
|
|
await api_session.execute(
|
|
text(
|
|
"INSERT INTO contacts (id, tenant_id, firstname, surname, email_1, "
|
|
"owner_id, created_by, updated_by, type, displayname, status, "
|
|
"discount_crew, discount_transport, discount_rental, discount_sale, discount_subrent, discount_total) "
|
|
"VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname, 'open', "
|
|
"0, 0, 0, 0, 0, 0)"
|
|
),
|
|
{
|
|
"id": str(new_id),
|
|
"tenant_id": str(tenant_a.id), # Correct tenant!
|
|
"firstname": "Own",
|
|
"surname": "Tenant",
|
|
"email": f"own-{uuid.uuid4().hex[:8]}@test.local",
|
|
"owner": str(user_a.id),
|
|
"creator": str(user_a.id),
|
|
"updater": str(user_a.id),
|
|
"ctype": "person",
|
|
"dname": "Own Tenant",
|
|
},
|
|
)
|
|
await api_session.flush()
|
|
# Verify the row is visible
|
|
result = await api_session.execute(
|
|
text("SELECT id FROM contacts WHERE id = :id"),
|
|
{"id": str(new_id)},
|
|
)
|
|
assert result.fetchone() is not None, "RLS blocked valid same-tenant INSERT!"
|
|
await api_session.rollback()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_role_is_not_superuser(api_session: AsyncSession):
|
|
"""Verify the test role is not superuser and cannot bypass RLS."""
|
|
result = await api_session.execute(
|
|
text(
|
|
"SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user"
|
|
)
|
|
)
|
|
row = result.fetchone()
|
|
assert row is not None, "Could not query role properties"
|
|
assert row[0] is False, f"Test role {row} is SUPERUSER — RLS tests are meaningless!"
|
|
assert row[1] is False, f"Test role {row} has BYPASSRLS — RLS tests are meaningless!"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_role_is_not_table_owner(api_session: AsyncSession):
|
|
"""Verify the test role is not the owner of tenant tables."""
|
|
result = await api_session.execute(
|
|
text(
|
|
"SELECT tableowner FROM pg_tables WHERE schemaname='public' AND tablename='contacts'"
|
|
)
|
|
)
|
|
owner = result.scalar()
|
|
current_user_result = await api_session.execute(text("SELECT current_user"))
|
|
current_user = current_user_result.scalar()
|
|
assert owner != current_user, \
|
|
f"Test role '{current_user}' owns contacts table — RLS is bypassed for owners!"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
|
|
async def test_rls_no_bootstrap_fallback_policy(api_session: AsyncSession):
|
|
"""Verify no fail-open/bootstrap RLS policy exists on tenant tables.
|
|
|
|
A fail-open policy would allow access when tenant context is missing.
|
|
This test checks that no policy uses IS NULL, = '', or COALESCE patterns
|
|
that would grant access without a tenant context.
|
|
"""
|
|
result = await api_session.execute(
|
|
text("""
|
|
SELECT tablename, policyname, qual, with_check
|
|
FROM pg_policies
|
|
WHERE schemaname = 'public'
|
|
AND tablename IN ('contacts', 'tasks', 'workspaces', 'files')
|
|
AND (
|
|
qual ILIKE '%IS NULL%'
|
|
OR qual ILIKE '%= ''%'
|
|
OR qual ILIKE '%COALESCE%'
|
|
OR with_check ILIKE '%IS NULL%'
|
|
OR with_check ILIKE '%= ''%'
|
|
OR with_check ILIKE '%COALESCE%'
|
|
)
|
|
""")
|
|
)
|
|
bad_policies = result.fetchall()
|
|
assert len(bad_policies) == 0, \
|
|
f"Fail-open RLS policies found: {bad_policies}"
|