phase1: separate DB roles, RLS restoration, login on crm_auth

- config.py: add auth_database_url, worker_database_url, migration_database_url
- db/__init__.py: separate engines for auth/worker/migration + get_auth_db/get_worker_db
- auth.py: all auth endpoints use get_auth_db (crm_auth role)
- auth_service.py: remove login fallback, require active membership, check status
- auth_service.py: switch_tenant checks active membership status
- alembic/env.py: use migration_database_url for Alembic
- docker-compose.yml: add AUTH_DATABASE_URL, WORKER_DATABASE_URL
- .env.example: add all 4 DB URLs with separate roles
- migration 0085: transfer ownership to crm_migration, fix BYPASSRLS,
  enable RLS+FORCE on all tenant tables, drop old policies, create new
  fail-closed policies scoped to crm_api+crm_worker, revoke excessive grants,
  grant minimal crm_auth access, drop crm_runtime, set default privileges
- tests/test_rls_coverage.py: automated RLS coverage check (13 tests)
- tests/test_cross_tenant_security_v2.py: RLS tests with unprivileged role

app/config.py

@@ -22,8 +22,11 @@ class Settings(BaseSettings):
     environment: Literal["development", "production", "testing"] = "development"
     log_level: str = "INFO"
 
-    # Database
+    # Database — separate connections for auth, API, worker, and migrations
     database_url: str = "postgresql+asyncpg://leocrm:leocrm@localhost:5432/leocrm_test"
+    auth_database_url: str = ""  # Falls back to database_url if empty
+    worker_database_url: str = ""  # Falls back to database_url if empty
+    migration_database_url: str = ""  # Falls back to database_url if empty
     db_pool_size: int = 10
     db_max_overflow: int = 20
     db_echo: bool = False

app/core/db/__init__.py

@@ -55,13 +55,22 @@ class TenantMixin(TimestampMixin, SoftDeleteMixin):
     tenant_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False, index=True)
 
 
-# Global engine and session factory
+# Global engines and session factories — separate per role
 _engine: AsyncEngine | None = None
 _session_factory: async_sessionmaker[AsyncSession] | None = None
 
+_auth_engine: AsyncEngine | None = None
+_auth_session_factory: async_sessionmaker[AsyncSession] | None = None
+
+_worker_engine: AsyncEngine | None = None
+_worker_session_factory: async_sessionmaker[AsyncSession] | None = None
+
+_migration_engine: AsyncEngine | None = None
+_migration_session_factory: async_sessionmaker[AsyncSession] | None = None
+
 
 def get_engine() -> AsyncEngine:
-    """Get or create the global async engine."""
+    """Get or create the global async engine (crm_api role)."""
     global _engine
     if _engine is None:
         settings = get_settings()
@@ -75,7 +84,7 @@ def get_engine() -> AsyncEngine:
 
 
 def get_session_factory() -> async_sessionmaker[AsyncSession]:
-    """Get or create the global session factory."""
+    """Get or create the global session factory (crm_api role)."""
     global _session_factory
     if _session_factory is None:
         _session_factory = async_sessionmaker(
@@ -86,6 +95,99 @@ def get_session_factory() -> async_sessionmaker[AsyncSession]:
     return _session_factory
 
 
+def get_auth_engine() -> AsyncEngine:
+    """Get or create the auth engine (crm_auth role).
+
+    Used for login, tenant resolution, password reset — no tenant context needed.
+    Falls back to the main engine if AUTH_DATABASE_URL is not set.
+    """
+    global _auth_engine
+    if _auth_engine is None:
+        settings = get_settings()
+        url = settings.auth_database_url or settings.database_url
+        _auth_engine = create_async_engine(
+            url,
+            pool_size=settings.db_pool_size,
+            max_overflow=settings.db_max_overflow,
+            echo=settings.db_echo,
+        )
+    return _auth_engine
+
+
+def get_auth_session_factory() -> async_sessionmaker[AsyncSession]:
+    """Get or create the auth session factory (crm_auth role)."""
+    global _auth_session_factory
+    if _auth_session_factory is None:
+        _auth_session_factory = async_sessionmaker(
+            bind=get_auth_engine(),
+            expire_on_commit=False,
+            class_=AsyncSession,
+        )
+    return _auth_session_factory
+
+
+def get_worker_engine() -> AsyncEngine:
+    """Get or create the worker engine (crm_worker role).
+
+    Used by ARQ worker for background job processing.
+    Falls back to the main engine if WORKER_DATABASE_URL is not set.
+    """
+    global _worker_engine
+    if _worker_engine is None:
+        settings = get_settings()
+        url = settings.worker_database_url or settings.database_url
+        _worker_engine = create_async_engine(
+            url,
+            pool_size=settings.db_pool_size,
+            max_overflow=settings.db_max_overflow,
+            echo=settings.db_echo,
+        )
+    return _worker_engine
+
+
+def get_worker_session_factory() -> async_sessionmaker[AsyncSession]:
+    """Get or create the worker session factory (crm_worker role)."""
+    global _worker_session_factory
+    if _worker_session_factory is None:
+        _worker_session_factory = async_sessionmaker(
+            bind=get_worker_engine(),
+            expire_on_commit=False,
+            class_=AsyncSession,
+        )
+    return _worker_session_factory
+
+
+def get_migration_engine() -> AsyncEngine:
+    """Get or create the migration engine (crm_migration role).
+
+    Used by Alembic for DDL operations. This engine connects as the table owner.
+    Falls back to the main engine if MIGRATION_DATABASE_URL is not set.
+    """
+    global _migration_engine
+    if _migration_engine is None:
+        settings = get_settings()
+        url = settings.migration_database_url or settings.database_url
+        _migration_engine = create_async_engine(
+            url,
+            pool_size=2,
+            max_overflow=0,
+            echo=settings.db_echo,
+        )
+    return _migration_engine
+
+
+def get_migration_session_factory() -> async_sessionmaker[AsyncSession]:
+    """Get or create the migration session factory (crm_migration role)."""
+    global _migration_session_factory
+    if _migration_session_factory is None:
+        _migration_session_factory = async_sessionmaker(
+            bind=get_migration_engine(),
+            expire_on_commit=False,
+            class_=AsyncSession,
+        )
+    return _migration_session_factory
+
+
 # Backward-compat alias: code imports `async_session_maker` from app.core.db.
 # Behaves like the session factory — calling it returns an AsyncSession.
 # We use a wrapper class so `async with async_session_maker() as db:` works.
@@ -102,7 +204,10 @@ async_session_maker = _AsyncSessionMakerWrapper()
 
 
 async def get_db() -> AsyncGenerator[AsyncSession, None]:
-    """FastAPI dependency: yield an async database session."""
+    """FastAPI dependency: yield an async database session (crm_api role).
+
+    Used for normal API requests with tenant context set via RLS.
+    """
     factory = get_session_factory()
     async with factory() as session:
         try:
@@ -113,6 +218,38 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]:
             raise
 
 
+async def get_auth_db() -> AsyncGenerator[AsyncSession, None]:
+    """FastAPI dependency: yield an auth database session (crm_auth role).
+
+    Used for login, tenant resolution, password reset — no tenant context needed.
+    The auth role has access only to identity tables (users, user_tenants, tenants,
+    password_reset_tokens), not to tenant-scoped business data.
+    """
+    factory = get_auth_session_factory()
+    async with factory() as session:
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
+
+
+async def get_worker_db() -> AsyncGenerator[AsyncSession, None]:
+    """FastAPI dependency: yield a worker database session (crm_worker role).
+
+    Used by ARQ worker for background job processing with tenant context.
+    """
+    factory = get_worker_session_factory()
+    async with factory() as session:
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
+
+
 async def set_tenant_context(session: AsyncSession, tenant_id: uuid.UUID | str) -> None:
     """Set PostgreSQL session variable for RLS tenant context.
 
@@ -175,17 +312,39 @@ async def create_db_session(
 
 
 async def close_engine() -> None:
-    """Dispose the global engine (for shutdown)."""
-    global _engine, _session_factory
-    if _engine is not None:
-        await _engine.dispose()
-        _engine = None
-        _session_factory = None
+    """Dispose all global engines (for shutdown)."""
+    global _engine, _session_factory, _auth_engine, _auth_session_factory
+    global _worker_engine, _worker_session_factory, _migration_engine, _migration_session_factory
+    for eng in (_engine, _auth_engine, _worker_engine, _migration_engine):
+        if eng is not None:
+            await eng.dispose()
+    _engine = None
+    _session_factory = None
+    _auth_engine = None
+    _auth_session_factory = None
+    _worker_engine = None
+    _worker_session_factory = None
+    _migration_engine = None
+    _migration_session_factory = None
 
 
 def reset_engine_for_testing(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
-    """Replace the global engine with a test engine. Returns a session factory."""
+    """Replace all global engines with a test engine. Returns a session factory.
+
+    In tests, all roles share the same test engine since RLS is not enforced
+    with the test database user (which is typically a superuser).
+    """
     global _engine, _session_factory
+    global _auth_engine, _auth_session_factory
+    global _worker_engine, _worker_session_factory
+    global _migration_engine, _migration_session_factory
     _engine = engine
     _session_factory = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
+    # In tests, all engines point to the same test DB
+    _auth_engine = engine
+    _auth_session_factory = _session_factory
+    _worker_engine = engine
+    _worker_session_factory = _session_factory
+    _migration_engine = engine
+    _migration_session_factory = _session_factory
     return _session_factory

app/routes/auth.py

@@ -9,7 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import get_settings
 from app.core.auth import get_redis
-from app.core.db import get_db
+from app.core.db import get_auth_db
 from app.core.rate_limit import check_rate_limit, get_client_ip, reset_rate_limit
 from app.schemas.auth import (
     AuthResponse,
@@ -30,7 +30,7 @@ settings = get_settings()
 async def login(
     request: Request,
     body: LoginRequest,
-    db: AsyncSession = Depends(get_db),
+    db: AsyncSession = Depends(get_auth_db),
 ):
     """Login with email+password. Sets session cookie."""
     ip = get_client_ip(request)
@@ -95,7 +95,7 @@ async def login(
 @router.post("/logout", response_model=MessageResponse)
 async def logout(
     request: Request,
-    db: AsyncSession = Depends(get_db),
+    db: AsyncSession = Depends(get_auth_db),
 ):
     """Logout — invalidate session, clear cookie."""
     session_id = request.cookies.get(settings.session_cookie_name)
@@ -116,7 +116,7 @@ async def logout(
 @router.get("/me", response_model=AuthResponse)
 async def me(
     request: Request,
-    db: AsyncSession = Depends(get_db),
+    db: AsyncSession = Depends(get_auth_db),
 ):
     """Get current user + active tenant."""
     session_id = request.cookies.get(settings.session_cookie_name)
@@ -151,7 +151,7 @@ async def me(
 @router.get("/me/permissions")
 async def me_permissions(
     request: Request,
-    db: AsyncSession = Depends(get_db),
+    db: AsyncSession = Depends(get_auth_db),
     current_user: dict = Depends(get_current_user),
 ):
     """Get resolved permissions for the current user.
@@ -171,7 +171,7 @@ async def me_permissions(
 async def switch_tenant(
     request: Request,
     body: SwitchTenantRequest,
-    db: AsyncSession = Depends(get_db),
+    db: AsyncSession = Depends(get_auth_db),
 ):
     """Switch active tenant for current session."""
     session_id = request.cookies.get(settings.session_cookie_name)
@@ -211,7 +211,7 @@ async def switch_tenant(
 async def password_reset_request(
     request: Request,
     body: PasswordResetRequest,
-    db: AsyncSession = Depends(get_db),
+    db: AsyncSession = Depends(get_auth_db),
 ):
     """Request password reset. Always returns 200 (no user enumeration)."""
     ip = get_client_ip(request)
@@ -231,7 +231,7 @@ async def password_reset_request(
 async def password_reset_confirm(
     request: Request,
     body: PasswordResetConfirm,
-    db: AsyncSession = Depends(get_db),
+    db: AsyncSession = Depends(get_auth_db),
 ):
     """Reset password with a valid token."""
     ip = get_client_ip(request)

app/services/auth_service.py

@@ -63,7 +63,10 @@ class AuthService:
             return None
 
         # Get user's default tenant or the one matching slug
-        ut_q = select(UserTenant).where(UserTenant.user_id == user.id)
+        ut_q = select(UserTenant).where(
+            UserTenant.user_id == user.id,
+            UserTenant.status == "active",
+        )
         if tenant_slug:
             ut_q = ut_q.join(Tenant, UserTenant.tenant_id == Tenant.id).where(
                 Tenant.slug == tenant_slug
@@ -73,12 +76,27 @@ class AuthService:
         ut_result = await db.execute(ut_q)
         user_tenant = ut_result.scalar_one_or_none()
 
-        # Fallback: just get first tenant membership
+        # No fallback — if tenant_slug was provided, the membership must exist
+        # and be active in exactly that tenant. If no slug, the default membership
+        # must exist and be active.
         if user_tenant is None:
-            ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
+            if tenant_slug:
+                # Specific tenant requested but no active membership — fail
+                return None
+            # No default membership — check if there are multiple active memberships
+            ut_q2 = select(UserTenant).where(
+                UserTenant.user_id == user.id,
+                UserTenant.status == "active",
+            )
             ut_result2 = await db.execute(ut_q2)
-            user_tenant = ut_result2.scalar_one_or_none()
-            if user_tenant is None:
+            active_memberships = ut_result2.scalars().all()
+            if len(active_memberships) == 1:
+                # Exactly one active membership — use it
+                user_tenant = active_memberships[0]
+            elif len(active_memberships) == 0:
+                return None
+            else:
+                # Multiple active memberships without a default — must specify tenant_slug
                 return None
 
         tenant_q = select(Tenant).where(Tenant.id == user_tenant.tenant_id)
@@ -152,10 +170,11 @@ class AuthService:
 
         user_id = uuid.UUID(session_data["user_id"])
 
-        # Verify user is member of target tenant
+        # Verify user has an ACTIVE membership in target tenant
         ut_q = select(UserTenant).where(
             UserTenant.user_id == user_id,
             UserTenant.tenant_id == new_tenant_id,
+            UserTenant.status == "active",
         )
         ut_result = await db.execute(ut_q)
         user_tenant = ut_result.scalar_one_or_none()