leocrm/tests/test_cross_tenant_standalone.py

"""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 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"