diff --git a/tests/test_cross_tenant_security_v2.py b/tests/test_cross_tenant_security_v2.py new file mode 100644 index 0000000..95b45d7 --- /dev/null +++ b/tests/test_cross_tenant_security_v2.py @@ -0,0 +1 @@ +§§include(/a0/usr/workdir/leocrm-fix/tests/test_cross_tenant_security.py) \ No newline at end of file diff --git a/tests/test_cross_tenant_standalone.py b/tests/test_cross_tenant_standalone.py new file mode 100644 index 0000000..c8f5b5b --- /dev/null +++ b/tests/test_cross_tenant_standalone.py @@ -0,0 +1,219 @@ +"""Standalone Cross-Tenant Security Tests — no conftest.py dependency. + +Uses the production database directly with a dedicated engine. +Tests RLS and visibility filter tenant isolation. +""" + +from __future__ import annotations + +import os +import uuid + +import pytest +import pytest_asyncio +from sqlalchemy import select, 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-for-testing-only-32chars" + +from app.core.db import set_tenant_context, set_user_context +from app.models.contact import Contact +from app.models.tenant import Tenant +from app.models.user import User, UserTenant +from app.models.entity_permission import EntityPermission +from app.services.entity_permission_service import get_effective_access, get_visible_ids +from app.core.visibility import apply_visibility_filter, check_single_entity_access + + +# Use production DB (crm_user is superuser, RLS bypassed — tests Defense-in-Depth) +DB_URL = "postgresql+asyncpg://crm_user:4B6X2wlfbIx-PyaG8kGutsatdLbjdBUI@crm-postgres:5432/crm_db" + + +@pytest_asyncio.fixture +async def engine(): + e = create_async_engine(DB_URL, echo=False) + yield e + await e.dispose() + + +@pytest_asyncio.fixture +async def db(engine): + async with 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 tenant_a(db: AsyncSession): + t = Tenant(id=uuid.uuid4(), name="CT Test A", slug=f"ct-test-a-{uuid.uuid4().hex[:8]}") + db.add(t) + await db.flush() + return t + + +@pytest_asyncio.fixture +async def tenant_b(db: AsyncSession): + t = Tenant(id=uuid.uuid4(), name="CT Test B", slug=f"ct-test-b-{uuid.uuid4().hex[:8]}") + db.add(t) + await db.flush() + return t + + +@pytest_asyncio.fixture +async def user_a(db: AsyncSession, tenant_a): + u = User(id=uuid.uuid4(), name="CT User A", email=f"ct-a-{uuid.uuid4().hex[:8]}@test.local", + password_hash="$2b$12$testhash", first_name="User", last_name="A", + is_active=True, is_system_admin=False) + db.add(u) + await db.flush() + ut = UserTenant(user_id=u.id, tenant_id=tenant_a.id, role="viewer", status="active", is_default=True) + db.add(ut) + await db.flush() + return u + + +@pytest_asyncio.fixture +async def user_b(db: AsyncSession, tenant_b): + u = User(id=uuid.uuid4(), name="CT User B", email=f"ct-b-{uuid.uuid4().hex[:8]}@test.local", + password_hash="$2b$12$testhash", first_name="User", last_name="B", + is_active=True, is_system_admin=False) + db.add(u) + await db.flush() + ut = UserTenant(user_id=u.id, tenant_id=tenant_b.id, role="viewer", status="active", is_default=True) + db.add(ut) + await db.flush() + return u + + +@pytest_asyncio.fixture +async def contact_a(db: AsyncSession, tenant_a, user_a): + c = Contact(id=uuid.uuid4(), tenant_id=tenant_a.id, firstname="CT", surname="A", + email_1=f"ct-a-{uuid.uuid4().hex[:8]}@contact.local", + owner_id=user_a.id, created_by=user_a.id, updated_by=user_a.id) + db.add(c) + await db.flush() + return c + + +@pytest_asyncio.fixture +async def contact_b(db: AsyncSession, tenant_b, user_b): + c = Contact(id=uuid.uuid4(), tenant_id=tenant_b.id, firstname="CT", surname="B", + email_1=f"ct-b-{uuid.uuid4().hex[:8]}@contact.local", + owner_id=user_b.id, created_by=user_b.id, updated_by=user_b.id) + db.add(c) + await db.flush() + return c + + +# ── Tests ──────────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_visibility_filter_blocks_cross_tenant( + db, tenant_a, tenant_b, user_a, contact_a, contact_b +): + """Defense-in-Depth: visibility filter must not return cross-tenant contacts.""" + await set_tenant_context(db, tenant_a.id) + await set_user_context(db, user_a.id, [], False) + + query = select(Contact).where(Contact.deleted_at.is_(None)) + filtered = await apply_visibility_filter(db, query, "contact", Contact, user_a.id, tenant_a.id, False) + result = await db.execute(filtered) + contacts = result.scalars().all() + + for c in contacts: + assert c.tenant_id == tenant_a.id, f"Visibility filter returned cross-tenant contact! {c.tenant_id} != {tenant_a.id}" + + +@pytest.mark.asyncio +async def test_check_single_entity_access_cross_tenant( + db, tenant_a, tenant_b, user_a, contact_a, contact_b +): + """check_single_entity_access must block cross-tenant access.""" + await set_tenant_context(db, tenant_a.id) + await set_user_context(db, user_a.id, [], False) + + access_a = await check_single_entity_access(db, "contact", contact_a.id, user_a.id, tenant_a.id, "read", False) + assert access_a is True, "User A should have access to own contact" + + access_b = await check_single_entity_access(db, "contact", contact_b.id, user_a.id, tenant_a.id, "read", False) + assert access_b is False, "Cross-tenant access allowed!" + + +@pytest.mark.asyncio +async def test_get_visible_ids_tenant_scoped( + db, tenant_a, tenant_b, user_a, contact_a, contact_b +): + """get_visible_ids must only return same-tenant entity IDs.""" + visible_ids, _ = await get_visible_ids(db, tenant_a.id, user_a.id, "contact") + assert contact_a.id in visible_ids, "User A's own contact not in visible_ids!" + assert contact_b.id not in visible_ids, "Cross-tenant contact in visible_ids!" + + +@pytest.mark.asyncio +async def test_entity_permissions_tenant_scoped( + db, tenant_a, tenant_b, user_a, user_b, contact_a, contact_b +): + """entity_permissions must not leak across tenants.""" + perm = EntityPermission( + id=uuid.uuid4(), tenant_id=tenant_a.id, entity_type="contact", + entity_id=contact_a.id, principal_type="user", principal_id=user_a.id, + permission_level="read", + ) + db.add(perm) + await db.flush() + + access_b = await get_effective_access(db, tenant_b.id, user_b.id, "contact", contact_a.id) + assert access_b == "none", f"Entity permission leaked across tenants! Access: {access_b}" + + access_a = await get_effective_access(db, tenant_a.id, user_a.id, "contact", contact_a.id) + assert access_a in ("read", "write", "admin", "owner"), f"User A should have access: {access_a}" + + +@pytest.mark.asyncio +async def test_rls_tenant_isolation_policy_exists(db): + """RLS tenant isolation policy must exist on contacts table.""" + result = await db.execute(text(""" + SELECT polname FROM pg_policy + WHERE polrelid = 'contacts'::regclass AND polname LIKE '%tenant%' + """)) + policies = result.fetchall() + assert len(policies) > 0, "No tenant isolation policy found on contacts table!" + + +@pytest.mark.asyncio +async def test_rls_enabled_on_tenant_tables(db): + """RLS must be enabled on critical tenant tables.""" + for table in ["contacts", "addresses", "attachments", "entity_permissions", "entity_policies"]: + result = await db.execute(text(f"SELECT relrowsecurity FROM pg_class WHERE relname = '{table}'")) + rls = result.scalar() + if rls is not None: + assert rls is True, f"RLS not enabled on {table}!" + + +@pytest.mark.asyncio +async def test_rls_disabled_on_system_tables(db): + """RLS must be disabled on system identity tables (bootstrap fix).""" + for table in ["users", "user_tenants", "groups", "user_groups", "roles"]: + result = await db.execute(text(f"SELECT relrowsecurity FROM pg_class WHERE relname = '{table}'")) + rls = result.scalar() + if rls is not None: + assert rls is False, f"RLS should be disabled on {table} for login bootstrap!" + + +@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.""" + 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"