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

app/core/db/__init__.py

@@ -116,18 +116,15 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]:
 async def set_tenant_context(session: AsyncSession, tenant_id: uuid.UUID | str) -> None:
     """Set PostgreSQL session variable for RLS tenant context.
 
-    Sets both app.current_tenant_id (new standard) and app.tenant_id
+    Sets app.current_tenant_id (the only standard tenant context variable).
-    (legacy, used by migration 0044 policies) for backward compatibility.
+    The legacy app.tenant_id has been removed — all RLS policies now use
+    app.current_tenant_id exclusively.
     """
     tid = str(tenant_id)
     await session.execute(
         text("SELECT set_config('app.current_tenant_id', :tid, true)"),
         {"tid": tid},
     )
-    await session.execute(
-        text("SELECT set_config('app.tenant_id', :tid, true)"),
-        {"tid": tid},
-    )
 
 
 async def set_user_context(

app/plugins/builtins/report_generator/jobs.py

@@ -76,7 +76,9 @@ async def generate_report_job(
         {"dms_file_id": ..., "filename": ..., "format": ..., "size": ...}
     """
     import hashlib
-    from app.plugins.builtins.dms.models import File as DmsFile
+    from app.plugins.builtins.contracts import get_contract_registry
+    _dms_contract = get_contract_registry().get("dms")
+    DmsFile = _dms_contract.DmsFile
 
     async with create_db_session() as db:
         # 1. Fetch template

tests/test_cross_tenant_security.py

@@ -441,7 +441,7 @@ 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
@@ -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}"

tests/test_cross_tenant_security_v2.py

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

tests/test_cross_tenant_standalone.py

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