diff --git a/alembic/versions/0085_restore_tenant_rls.py b/alembic/versions/0085_restore_tenant_rls.py index 5c6cd27..e9944bd 100644 --- a/alembic/versions/0085_restore_tenant_rls.py +++ b/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 - _exec("ALTER ROLE crm_migration NOBYPASSRLS") + # 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 for table in ALL_TABLES: diff --git a/alembic/versions/0086_fix_global_tables_force_rls.py b/alembic/versions/0086_fix_global_tables_force_rls.py new file mode 100644 index 0000000..b80de22 --- /dev/null +++ b/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") diff --git a/app/services/auth_service.py b/app/services/auth_service.py index be5c19e..53257b2 100644 --- a/app/services/auth_service.py +++ b/app/services/auth_service.py @@ -109,20 +109,25 @@ class AuthService: db, redis, user, tenant.id, role=user_tenant.role ) - # Set tenant context for audit log write (auth session uses crm_auth role) - from app.core.db import set_tenant_context - await set_tenant_context(db, tenant.id) - - # Log the login in audit trail - await log_audit( - db, - tenant.id, - user.id, - "login", - "user", - user.id, - changes={"email": email}, - ) + # Log the login in audit trail via separate API session (crm_api with tenant context) + # crm_auth must not write to tenant tables — audit_log is a tenant table + try: + from app.core.db import get_session_factory, set_tenant_context + api_factory = get_session_factory() + async with api_factory() as audit_db: + await set_tenant_context(audit_db, tenant.id) + await log_audit( + audit_db, + tenant.id, + user.id, + "login", + "user", + 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) diff --git a/tests/test_no_legacy_tenant_var.py b/tests/test_no_legacy_tenant_var.py new file mode 100644 index 0000000..61c3142 --- /dev/null +++ b/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}"