leocrm/alembic/versions/0085_restore_tenant_rls.py

"""Restore tenant RLS, transfer ownership, fix roles and grants.

This migration implements Phase 1 of the security hardening:

1. Transfer ALL table ownership from crm_user (SUPERUSER) to crm_migration (NOSUPERUSER, NOBYPASSRLS)
2. ALTER ROLE crm_migration NOBYPASSRLS
3. Enable RLS + FORCE on ALL tenant tables (tables with tenant_id column)
4. Drop ALL old policies (scoped to {public} or using non-NULLIF patterns)
5. Create new fail-closed policies scoped to {crm_api, crm_worker}
6. Revoke excessive grants from crm_runtime, crm_worker, crm_api, crm_auth
7. Grant proper minimal permissions to crm_auth (identity tables only)
8. Grant CRUD to crm_api and crm_worker on tenant tables
9. Revoke alembic_version access from crm_api and crm_worker
10. Set default privileges for crm_migration owner
11. Drop crm_runtime legacy role
12. Create crm_platform_admin role (for one-time infrastructure only)

Revision ID: 0085
Revises: 0084
"""

from __future__ import annotations

from alembic import op

revision = "0085"
down_revision = "0084"
branch_labels = None
depends_on = None


TENANT_TABLES = [
    "addresses", "ai_agents", "ai_chat_attachments", "ai_chat_folders",
    "ai_chat_messages", "ai_chat_sessions", "ai_conversations", "ai_messages",
    "ai_models", "ai_presets", "ai_proactive_context_log", "ai_proactive_settings",
    "ai_proactive_suggestions", "ai_providers", "attachments", "audit_log",
    "automation_agent_definitions", "automation_agent_runs",
    "automation_agent_versions", "automation_cron_jobs",
    "automation_definitions", "automation_runs", "automation_versions",
    "backups", "bank_accounts", "calendar_entries", "calendar_entry_links",
    "calendar_shares", "calendars", "comm_conversation_mutes",
    "comm_conversation_pins", "comm_conversations", "comm_message_attachments",
    "comm_message_blocks", "comm_message_edits", "comm_message_reactions",
    "comm_message_reads", "comm_messages", "comm_participants",
    "contact_folder_permissions", "contact_folders", "contact_merge_history",
    "contact_pgp_keys", "contactpersons", "contacts", "currencies",
    "custom_field_definitions", "deletion_log", "entity_attachments",
    "entity_history", "entity_links", "entity_permissions", "entity_policies",
    "event_outbox", "files", "folders", "groups", "guest_users",
    "mail_account_delegates", "mail_account_send_permissions",
    "mail_accounts", "mail_attachments", "mail_folders",
    "mail_label_assignments", "mail_labels", "mail_rules", "mail_seen_by",
    "mail_signatures", "mail_sync_queue", "mail_templates", "mails",
    "mcp_server_configs", "notification_preferences", "notifications",
    "password_reset_tokens", "permission_delegations", "permission_templates",
    "permissions", "pgp_keys", "plugin_test_data", "report_instances",
    "report_templates", "resource_bookings", "resources", "roles",
    "saved_filters", "saved_views", "share_links", "subtasks",
    "system_settings", "tag_assignments", "tags", "tasks", "tax_rates",
    "unified_search_index_log", "unified_search_providers",
    "user_calendar_visibility", "user_groups", "user_preferences",
    "vacation_sent_log", "webhooks", "workflow_instances",
    "workflow_step_history", "workflows", "workspace_modules",
    "workspace_users", "workspace_widgets", "workspaces",
]

GLOBAL_TABLES = [
    "users", "tenants", "user_tenants", "sessions", "plugins",
    "plugin_allowlist", "plugin_migrations", "tenant_plugin_activation",
    "alembic_version", "notification_types", "api_tokens",
    "consumer_inbox", "outbox_deliveries", "guest_invitations",
    "sequences",
]

AUTH_TABLES = {
    "users": ["SELECT"],
    "user_tenants": ["SELECT"],
    "tenants": ["SELECT"],
    "password_reset_tokens": ["SELECT", "INSERT", "UPDATE", "DELETE"],
    "sessions": ["SELECT", "INSERT", "UPDATE", "DELETE"],
}

WORKER_GLOBAL_TABLES = {
    "event_outbox": ["SELECT", "INSERT", "UPDATE"],
    "outbox_deliveries": ["SELECT", "INSERT", "UPDATE"],
    "consumer_inbox": ["SELECT", "INSERT", "UPDATE", "DELETE"],
}

ALL_TABLES = TENANT_TABLES + GLOBAL_TABLES


def _exec(sql: str) -> None:
    op.execute(sql)


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: crm_migration keeps BYPASSRLS for data migrations (NOSUPERUSER)
    # 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 (only for tables that exist)
    for table in ALL_TABLES:
        _exec(f"DO $$ BEGIN IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = '{table}') THEN ALTER TABLE public.{table} OWNER TO crm_migration; END IF; END $$")

    # Transfer sequence ownership
    _exec("DO $$ DECLARE r RECORD; BEGIN FOR r IN SELECT sequence_name FROM information_schema.sequences WHERE sequence_schema = 'public' LOOP EXECUTE format('ALTER SEQUENCE public.%I OWNER TO crm_migration', r.sequence_name); END LOOP; END $$;")

    # Step 4: Revoke ALL grants from runtime roles
    for role in ("crm_runtime", "crm_worker", "crm_api", "crm_auth"):
        _exec(f"REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA public FROM {role}")
        _exec(f"REVOKE ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public FROM {role}")
        _exec(f"REVOKE ALL PRIVILEGES ON SCHEMA public FROM {role}")

    # Step 5: Drop crm_runtime role — revoke default privileges first, then drop
    _exec("ALTER DEFAULT PRIVILEGES IN SCHEMA public REVOKE SELECT, INSERT, UPDATE, DELETE ON TABLES FROM crm_runtime")
    _exec("ALTER DEFAULT PRIVILEGES IN SCHEMA public REVOKE USAGE, SELECT ON SEQUENCES FROM crm_runtime")
    _exec("DO $$ BEGIN DROP ROLE IF EXISTS crm_runtime; EXCEPTION WHEN insufficient_privilege THEN NULL; WHEN dependent_objects_still_exist THEN NULL; END $$")

    # Step 6: Grant schema USAGE to runtime roles
    _exec("GRANT USAGE ON SCHEMA public TO crm_api")
    _exec("GRANT USAGE ON SCHEMA public TO crm_worker")
    _exec("GRANT USAGE ON SCHEMA public TO crm_auth")

    # Step 7: Grant permissions to crm_auth (identity tables only)
    for table, privs in AUTH_TABLES.items():
        priv_str = ", ".join(privs)
        _exec(f"DO $$ BEGIN IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = '{table}') THEN GRANT {priv_str} ON public.{table} TO crm_auth; END IF; END $$")

    # Step 8: Grant CRUD on tenant tables to crm_api and crm_worker (only for tables that exist)
    for table in TENANT_TABLES:
        _exec(f"DO $$ BEGIN IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = '{table}') THEN GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_api; GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_worker; END IF; END $$")

    # Grant sequence USAGE to crm_api and crm_worker
    _exec("GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO crm_api")
    _exec("GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO crm_worker")

    # Step 9: Grant global table access to crm_api (except alembic_version)
    api_global_tables = [t for t in GLOBAL_TABLES if t != "alembic_version"]
    for table in api_global_tables:
        _exec(f"DO $$ BEGIN IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = '{table}') THEN GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_api; END IF; END $$")

    # Step 10: Grant worker global table access
    for table, privs in WORKER_GLOBAL_TABLES.items():
        priv_str = ", ".join(privs)
        _exec(f"DO $$ BEGIN IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = '{table}') THEN GRANT {priv_str} ON public.{table} TO crm_worker; END IF; END $$")

    worker_global_tables = [
        t for t in GLOBAL_TABLES
        if t != "alembic_version" and t not in WORKER_GLOBAL_TABLES
    ]
    for table in worker_global_tables:
        _exec(f"DO $$ BEGIN IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = '{table}') THEN GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_worker; END IF; END $$")

    # Step 11: Drop ALL old RLS policies and create new fail-closed ones
    policy_template = (
        "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)"
    )

    for table in TENANT_TABLES:
        _exec(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}; DROP POLICY IF EXISTS {table}_tenant_isolation ON public.{table}; ALTER TABLE public.{table} ENABLE ROW LEVEL SECURITY; ALTER TABLE public.{table} FORCE ROW LEVEL SECURITY; {policy_template.format(table=table)}; END IF; END $$")

    # Step 12: Disable RLS on global tables
    for table in GLOBAL_TABLES:
        _exec(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}; DROP POLICY IF EXISTS {table}_tenant_isolation ON public.{table}; ALTER TABLE public.{table} DISABLE ROW LEVEL SECURITY; END IF; END $$")

    # Step 13: Set default privileges for crm_migration owner
    _exec("ALTER DEFAULT PRIVILEGES FOR ROLE crm_migration IN SCHEMA public GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO crm_api")
    _exec("ALTER DEFAULT PRIVILEGES FOR ROLE crm_migration IN SCHEMA public GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO crm_worker")
    _exec("ALTER DEFAULT PRIVILEGES FOR ROLE crm_migration IN SCHEMA public GRANT USAGE, SELECT ON SEQUENCES TO crm_api")
    _exec("ALTER DEFAULT PRIVILEGES FOR ROLE crm_migration IN SCHEMA public GRANT USAGE, SELECT ON SEQUENCES TO crm_worker")


def downgrade() -> None:
    pass