fix: legacy app.tenant_id policies on _old tables + seed_admin.py rewrite

- Migration 0090: Drop legacy tenant_isolation policies on companies_old, company_contacts_old, contacts_old that used app.tenant_id variable. Create new policies using app.current_tenant_id for crm_api/crm_worker. - seed_admin.py: Rewrite to use migration engine (crm_migration) for bootstrap, set tenant context, create Tenant + Role + User + UserTenant. No longer passes tenant_id as User parameter. Fixes: 3 legacy app.tenant_id policies found in Gate 2 verification. Fixes: seed_admin.py incompatible with current User model.
2026-07-31 22:23:38 +02:00
parent b5191f0d11
commit 89b775b9ef
2 changed files with 98 additions and 18 deletions
@@ -0,0 +1,56 @@
+"""Fix legacy app.tenant_id policies on _old tables.
+
+Migration 0021 renamed old tables (contacts, companies, company_contacts) to *_old
+but their RLS policies still reference the old app.tenant_id variable.
+This migration drops those legacy policies and creates new ones using
+app.current_tenant_id to maintain consistency.
+
+Revision ID: 0090
+Revises: 0089
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0090"
+down_revision = "0089"
+branch_labels = None
+depends_on = None
+
+LEGACY_TABLES = ["companies_old", "company_contacts_old", "contacts_old"]
+
+
+def upgrade() -> None:
+    for table in LEGACY_TABLES:
+        op.execute(f"""
+            DO $$ BEGIN
+                IF EXISTS (
+                    SELECT 1 FROM information_schema.tables
+                    WHERE table_schema = 'public' AND table_name = '{table}'
+                ) THEN
+                    DROP POLICY IF EXISTS tenant_isolation ON public.{table};
+                    CREATE POLICY {table}_tenant_isolation
+                    ON public.{table}
+                    FOR ALL
+                    TO crm_api, crm_worker
+                    USING (tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid)
+                    WITH CHECK (tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid);
+                END IF;
+            END $$;
+        """)
+
+
+def downgrade() -> None:
+    for table in LEGACY_TABLES:
+        op.execute(f"""
+            DO $$ BEGIN
+                IF EXISTS (
+                    SELECT 1 FROM information_schema.tables
+                    WHERE table_schema = 'public' AND table_name = '{table}'
+                ) THEN
+                    DROP POLICY IF EXISTS {table}_tenant_isolation ON public.{table};
+                END IF;
+            END $$;
+        """)