fix: RLS fail-closed migration + per-tenant startup code

alembic/versions/0084_rls_fail_closed_reactivate.py

@@ -17,7 +17,7 @@ Bootstrap and startup must use:
 
 Revision ID: 0084
 Revises: 0083
-""
+"""
 
 from alembic import op
 from sqlalchemy import text
@@ -75,21 +75,22 @@ def upgrade() -> None:
         conn.execute(text(f"ALTER TABLE {table} FORCE ROW LEVEL SECURITY"))
 
         # Fail-closed tenant isolation policy
-        # NULLIF converts empty string to NULL → comparison yields NULL → no rows returned
+        # NULLIF converts empty string to NULL -> comparison yields NULL -> no rows returned
         # This is fail-closed: missing tenant context = no access
-        conn.execute(text(f"""
+        policy_sql = (
-            CREATE POLICY {table}_tenant_isolation
+            "CREATE POLICY " + table + "_tenant_isolation "
-            ON {table}
+            "ON " + table + " "
-            AS PERMISSIVE
+            "AS PERMISSIVE "
-            FOR ALL
+            "FOR ALL "
-            TO crm_api
+            "TO crm_api "
-            USING (
+            "USING ("
-                tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid
+            "tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid"
+            ") "
+            "WITH CHECK ("
+            "tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid"
+            ")"
         )
-            WITH CHECK (
+        conn.execute(text(policy_sql))
-                tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid
-            )
-        """))
 
 
 def downgrade() -> None:

app/main.py

@@ -166,6 +166,16 @@ async def lifespan(app: FastAPI):
     event_bus = get_event_bus()
     async_session = async_sessionmaker(get_engine(), expire_on_commit=False)
 
+    # Load all tenant IDs for per-tenant plugin activation (RLS fail-closed requires tenant context)
+    from app.models.tenant import Tenant as TenantModel
+    from app.core.db import set_tenant_context
+
+    async with async_session() as db:
+        tenant_result = await db.execute(sa_select(TenantModel.id))
+        all_tenant_ids = [row[0] for row in tenant_result]
+        logger.info(f"Loaded {len(all_tenant_ids)} tenants for plugin activation")
+
+    # Install plugin records and run migrations (global, no tenant context needed)
     async with async_session() as db:
         for name in registry.resolve_load_order():
             plugin = registry.get_plugin(name)
@@ -179,9 +189,6 @@ async def lifespan(app: FastAPI):
             plugin_record = result.scalar_one_or_none()
 
             if plugin_record is None:
-                # Create DB record for this builtin plugin — inactive by default (except core)
-                # Only core plugins auto-activate on first install
-                # Existing plugins that are marked active in DB will be activated below
                 plugin_record = PluginModel(
                     name=name,
                     display_name=plugin.manifest.display_name,
@@ -206,22 +213,29 @@ async def lifespan(app: FastAPI):
                         logger.error(f"Deactivating plugin {name} due to migration failure")
                         plugin_record.active = False
                         plugin_record.status = "migration_failed"
-                    continue  # Skip activation if migration fails
+                    continue
 
             # Only activate plugins that are marked active in DB
             if not plugin_record.active:
                 logger.info(f"Plugin {name} is inactive — skipping activation")
                 continue
 
-            # Activate plugin (routes are already registered in create_app)
+            # Activate plugin with tenant context set for each tenant
+            # (RLS fail-closed requires app.current_tenant_id to be set for tenant-table writes)
+            activation_failed = False
+            for tenant_id in all_tenant_ids:
                 try:
+                    await set_tenant_context(db, tenant_id)
                     await plugin.on_activate(db, container, event_bus)
+                except Exception as exc:
+                    logger.error(f"[STARTUP] Failed to activate plugin {name} for tenant {tenant_id}: {exc}")
+                    activation_failed = True
+                    break
+
+            if not activation_failed:
                 plugin_record.status = "active"
                 logger.info(f"[STARTUP] Activated plugin: {name}")
-                logger.info(f"Activated plugin: {name}")
+            else:
-            except Exception as exc:
-                logger.error(f"[STARTUP] Failed to activate plugin {name}: {exc}")
-                logger.error(f"Failed to activate plugin {name}: {exc}")
                 plugin_record.active = False
                 plugin_record.status = "activation_failed"