phase0: fix cross-plugin import, remove app.tenant_id, create cross-tenant v2 tests
- Fix report_generator/jobs.py: use DmsContract instead of direct DMS import
- Remove app.tenant_id from set_tenant_context (only app.current_tenant_id)
- Create tests/test_cross_tenant_security_v2.py with real RLS tests using
unprivileged crm_api role (NOSUPERUSER, NOBYPASSRLS)
- Fix existing tests referencing app.tenant_id
- Git baseline tag v-phase0-baseline at 11d6faa
- Production DB backup at /tmp/crm_backup_20260731_015514.dump
This commit is contained in:
Agent Zero
@@ -441,9 +441,9 @@ async def test_tenant_context_variable_consistency(
|
||||
db_session: AsyncSession,
|
||||
tenant_a: Tenant,
|
||||
):
|
||||
"""Test that set_tenant_context sets both app.current_tenant_id and app.tenant_id."""
|
||||
"""Test that set_tenant_context sets app.current_tenant_id (the only standard)."""
|
||||
await set_tenant_context(db_session, tenant_a.id)
|
||||
|
||||
|
||||
# Check app.current_tenant_id
|
||||
result = await db_session.execute(
|
||||
text("SELECT current_setting('app.current_tenant_id', true)")
|
||||
@@ -451,11 +451,3 @@ async def test_tenant_context_variable_consistency(
|
||||
current_tid = result.scalar()
|
||||
assert current_tid == str(tenant_a.id), \
|
||||
f"app.current_tenant_id not set correctly: {current_tid}"
|
||||
|
||||
# Check app.tenant_id (legacy)
|
||||
result = await db_session.execute(
|
||||
text("SELECT current_setting('app.tenant_id', true)")
|
||||
)
|
||||
legacy_tid = result.scalar()
|
||||
assert legacy_tid == str(tenant_a.id), \
|
||||
f"app.tenant_id not set correctly: {legacy_tid}"
|
||||
|
||||
@@ -0,0 +1,410 @@
|
||||
"""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["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."""
|
||||
async with admin_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 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."""
|
||||
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="$2b$12$testhash",
|
||||
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="$2b$12$testhash",
|
||||
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()
|
||||
|
||||
return {
|
||||
"tenant_a": tenant_a,
|
||||
"tenant_b": tenant_b,
|
||||
"user_a": user_a,
|
||||
"user_b": user_b,
|
||||
"contact_a": contact_a,
|
||||
"contact_b": contact_b,
|
||||
}
|
||||
|
||||
|
||||
# ── 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 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[0] for r in rows if 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 row[0] == str(tenant_b.id), \
|
||||
f"RLS leak: tenant B context shows row from {row[0]}"
|
||||
tenant_a_ids = [r[0] for r in rows if 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
|
||||
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) "
|
||||
"VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname)"
|
||||
),
|
||||
{
|
||||
"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 should fail due to RLS WITH CHECK
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
await api_session.flush()
|
||||
assert "row level security" in str(exc_info.value).lower() or "rls" in str(exc_info.value).lower(), \
|
||||
f"Expected RLS error, got: {exc_info.value}"
|
||||
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) "
|
||||
"VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname)"
|
||||
),
|
||||
{
|
||||
"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}"
|
||||
@@ -209,11 +209,8 @@ async def test_rls_disabled_on_system_tables(db):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tenant_context_variable_consistency(db, tenant_a):
|
||||
"""set_tenant_context must set both app.current_tenant_id and app.tenant_id."""
|
||||
"""set_tenant_context must set app.current_tenant_id (the only standard)."""
|
||||
await set_tenant_context(db, tenant_a.id)
|
||||
|
||||
result = await db.execute(text("SELECT current_setting('app.current_tenant_id', true)"))
|
||||
assert result.scalar() == str(tenant_a.id), "app.current_tenant_id not set correctly"
|
||||
|
||||
result = await db.execute(text("SELECT current_setting('app.tenant_id', true)"))
|
||||
assert result.scalar() == str(tenant_a.id), "app.tenant_id not set correctly"
|
||||
|
||||
Reference in New Issue
Block a user