gate: migration 0086, crm_migration BYPASSRLS, audit_log fix, CI test for app.tenant_id

- Migration 0086: Remove FORCE RLS from 5 global tables
- Migration 0085: crm_migration keeps BYPASSRLS for data migrations
- Migration 0085: Remove audit_log from crm_auth grants
- auth_service.py: Audit log via separate API session (crm_api with tenant context)
- tests/test_no_legacy_tenant_var.py: CI test for app.tenant_id in policies

alembic/versions/0085_restore_tenant_rls.py

@@ -78,7 +78,6 @@ AUTH_TABLES = {
     "tenants": ["SELECT"],
     "password_reset_tokens": ["SELECT", "INSERT", "UPDATE", "DELETE"],
     "sessions": ["SELECT", "INSERT", "UPDATE", "DELETE"],
-    "audit_log": ["SELECT", "INSERT"],
 }
 
 WORKER_GLOBAL_TABLES = {
@@ -98,8 +97,9 @@ def upgrade() -> None:
     # Step 1: Create crm_platform_admin role
     _exec("DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_platform_admin') THEN CREATE ROLE crm_platform_admin NOSUPERUSER NOBYPASSRLS NOLOGIN; END IF; END $$;")
 
-    # Step 2: Fix crm_migration role — remove BYPASSRLS
+    # Step 2: crm_migration keeps BYPASSRLS for data migrations (NOSUPERUSER)
-    _exec("ALTER ROLE crm_migration NOBYPASSRLS")
+    # crm_migration is the table owner and needs to run tenant-wide data migrations
+    _exec("ALTER ROLE crm_migration NOSUPERUSER BYPASSRLS")
 
     # Step 3: Transfer ALL table ownership to crm_migration
     for table in ALL_TABLES:

alembic/versions/0086_fix_global_tables_force_rls.py

@@ -0,0 +1,38 @@
+"""Fix FORCE RLS on global tables.
+
+Migration 0085 disabled RLS on global tables but did not remove
+FORCE ROW LEVEL SECURITY from 5 tables that had it enabled from
+older migrations. This migration removes FORCE RLS from all
+global tables (tables without tenant_id).
+
+Revision ID: 0086
+Revises: 0085
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0086"
+down_revision = "0085"
+branch_labels = None
+depends_on = None
+
+
+GLOBAL_TABLES_WITH_FORCE_RLS = [
+    "api_tokens",
+    "sequences",
+    "sessions",
+    "tenant_plugin_activation",
+    "user_tenants",
+]
+
+
+def upgrade() -> None:
+    for table in GLOBAL_TABLES_WITH_FORCE_RLS:
+        op.execute(f"ALTER TABLE public.{table} NO FORCE ROW LEVEL SECURITY")
+
+
+def downgrade() -> None:
+    for table in GLOBAL_TABLES_WITH_FORCE_RLS:
+        op.execute(f"ALTER TABLE public.{table} FORCE ROW LEVEL SECURITY")

app/services/auth_service.py

@@ -109,13 +109,15 @@ class AuthService:
             db, redis, user, tenant.id, role=user_tenant.role
         )
 
-        # Set tenant context for audit log write (auth session uses crm_auth role)
+        # Log the login in audit trail via separate API session (crm_api with tenant context)
-        from app.core.db import set_tenant_context
+        # crm_auth must not write to tenant tables — audit_log is a tenant table
-        await set_tenant_context(db, tenant.id)
+        try:
-
+            from app.core.db import get_session_factory, set_tenant_context
-        # Log the login in audit trail
+            api_factory = get_session_factory()
+            async with api_factory() as audit_db:
+                await set_tenant_context(audit_db, tenant.id)
                 await log_audit(
-            db,
+                    audit_db,
                     tenant.id,
                     user.id,
                     "login",
@@ -123,6 +125,9 @@ class AuthService:
                     user.id,
                     changes={"email": email},
                 )
+                await audit_db.commit()
+        except Exception:
+            logger.warning("Failed to write login audit log via API session", exc_info=True)
 
         # Hook: auth.after_login
         await do_action("auth.after_login", db=db, user=user, tenant=tenant, role=user_tenant.role, session_id=session_id)

tests/test_no_legacy_tenant_var.py

@@ -0,0 +1,93 @@
+"""CI test: verify no RLS policy uses legacy app.tenant_id variable.
+
+After alembic upgrade head, all RLS policies must use app.current_tenant_id
+exclusively. This test fails if any policy in the database still references
+the old app.tenant_id variable.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+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!!"
+
+
+_ADMIN_DB_URL = os.environ.get(
+    "RLS_TEST_ADMIN_DB_URL",
+    "postgresql+asyncpg://postgres@localhost:5432/leocrm_test",
+)
+
+
+def _skip_if_no_db():
+    try:
+        import asyncio
+        eng = create_async_engine(_ADMIN_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
+
+
+@pytest_asyncio.fixture
+async def admin_session():
+    eng = create_async_engine(_ADMIN_DB_URL, echo=False)
+    async with eng.connect() as conn:
+        session = AsyncSession(bind=conn, expire_on_commit=False)
+        yield session
+        await session.rollback()
+        await conn.rollback()
+    await eng.dispose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(_skip_if_no_db(), reason="Test database not available")
+async def test_no_policy_uses_legacy_app_tenant_id(admin_session: AsyncSession):
+    """No RLS policy should reference the legacy app.tenant_id variable.
+
+    All policies must use app.current_tenant_id exclusively.
+    This test runs after alembic upgrade head to verify the final state.
+    """
+    result = await admin_session.execute(text("""
+        SELECT tablename, policyname, qual, with_check
+        FROM pg_policies
+        WHERE schemaname = 'public'
+        AND (
+            qual ILIKE '%app.tenant_id%'
+            OR with_check ILIKE '%app.tenant_id%'
+        )
+    """))
+    legacy_policies = result.fetchall()
+    assert len(legacy_policies) == 0, \
+        f"RLS policies still using legacy app.tenant_id: {legacy_policies}"
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(_skip_if_no_db(), reason="Test database not available")
+async def test_all_tenant_policies_use_current_tenant_id(admin_session: AsyncSession):
+    """All tenant isolation policies must use app.current_tenant_id."""
+    result = await admin_session.execute(text("""
+        SELECT tablename, policyname
+        FROM pg_policies
+        WHERE schemaname = 'public'
+        AND policyname LIKE '%tenant_isolation%'
+        AND (
+            qual NOT ILIKE '%app.current_tenant_id%'
+            AND with_check NOT ILIKE '%app.current_tenant_id%'
+        )
+    """))
+    wrong_policies = result.fetchall()
+    assert len(wrong_policies) == 0, \
+        f"Tenant policies not using app.current_tenant_id: {wrong_policies}"