diff --git a/.env.example b/.env.example index e6bbd3b..96f6e8b 100644 --- a/.env.example +++ b/.env.example @@ -1,16 +1,15 @@ # LeoCRM v1.0 - Environment Variables Template # === REQUIRED === -DATABASE_URL=postgresql+asyncpg://leocrm:leocrm@localhost:5432/leocrm +DATABASE_URL=postgresql+asyncpg://crm_api:your_password@localhost:5432/crm_db +AUTH_DATABASE_URL=postgresql+asyncpg://crm_auth:your_password@localhost:5432/crm_db +WORKER_DATABASE_URL=postgresql+asyncpg://crm_worker:your_password@localhost:5432/crm_db +MIGRATION_DATABASE_URL=postgresql+asyncpg://crm_migration:your_password@localhost:5432/crm_db REDIS_URL=redis://localhost:6379/0 # === REQUIRED for Docker/Production === -# Migration DB URL (owner user, can bypass RLS for DDL) -MIGRATION_DATABASE_URL=postgresql+asyncpg://crm_migration:your_password@localhost:5432/crm_db # Redis password (required in Docker) REDIS_PASSWORD=your_redis_password -# Runtime DB password (set crm_runtime role password on startup) -RUNTIME_DB_PASSWORD=your_runtime_password # === OPTIONAL (with defaults) === diff --git a/alembic/env.py b/alembic/env.py index 28da21f..36e4ee4 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -20,7 +20,8 @@ if config.config_file_name is not None: target_metadata = Base.metadata settings = get_settings() -config.set_main_option("sqlalchemy.url", settings.database_url) +# Use migration_database_url (crm_migration role, table owner) for Alembic +config.set_main_option("sqlalchemy.url", settings.migration_database_url or settings.database_url) def run_migrations_offline() -> None: diff --git a/alembic/versions/0085_restore_tenant_rls.py b/alembic/versions/0085_restore_tenant_rls.py new file mode 100644 index 0000000..5fad194 --- /dev/null +++ b/alembic/versions/0085_restore_tenant_rls.py @@ -0,0 +1,185 @@ +"""Restore tenant RLS, transfer ownership, fix roles and grants. + +This migration implements Phase 1 of the Sanierungsplan: + +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"], +} + +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: Fix crm_migration role — remove BYPASSRLS + _exec("ALTER ROLE crm_migration NOBYPASSRLS") + + # Step 3: Transfer ALL table ownership to crm_migration + for table in ALL_TABLES: + _exec(f"ALTER TABLE public.{table} OWNER TO crm_migration") + + # 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 + _exec("DROP ROLE IF EXISTS crm_runtime") + + # 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"GRANT {priv_str} ON public.{table} TO crm_auth") + + # Step 8: Grant CRUD on tenant tables to crm_api and crm_worker + for table in TENANT_TABLES: + _exec(f"GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_api") + _exec(f"GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_worker") + + # 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"GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_api") + + # Step 10: Grant worker global table access + for table, privs in WORKER_GLOBAL_TABLES.items(): + priv_str = ", ".join(privs) + _exec(f"GRANT {priv_str} ON public.{table} TO crm_worker") + + 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"GRANT SELECT, INSERT, UPDATE, DELETE ON public.{table} TO crm_worker") + + # 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"DROP POLICY IF EXISTS tenant_isolation ON public.{table}") + _exec(f"DROP POLICY IF EXISTS {table}_tenant_isolation ON public.{table}") + _exec(f"ALTER TABLE public.{table} ENABLE ROW LEVEL SECURITY") + _exec(f"ALTER TABLE public.{table} FORCE ROW LEVEL SECURITY") + _exec(policy_template.format(table=table)) + + # Step 12: Disable RLS on global tables + for table in GLOBAL_TABLES: + _exec(f"DROP POLICY IF EXISTS tenant_isolation ON public.{table}") + _exec(f"DROP POLICY IF EXISTS {table}_tenant_isolation ON public.{table}") + _exec(f"ALTER TABLE public.{table} DISABLE ROW LEVEL SECURITY") + + # 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 diff --git a/app/config.py b/app/config.py index 2f82494..120fe80 100644 --- a/app/config.py +++ b/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 diff --git a/app/core/db/__init__.py b/app/core/db/__init__.py index cea63b8..c51e677 100644 --- a/app/core/db/__init__.py +++ b/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 diff --git a/app/routes/auth.py b/app/routes/auth.py index 821d9bf..335257f 100644 --- a/app/routes/auth.py +++ b/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) diff --git a/app/services/auth_service.py b/app/services/auth_service.py index 7ff9668..330b0b8 100644 --- a/app/services/auth_service.py +++ b/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() diff --git a/docker-compose.yml b/docker-compose.yml index ed7efcd..72c61c7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -81,9 +81,11 @@ services: redis: condition: service_healthy environment: - # App/worker uses crm_api (NOSUPERUSER, NOBYPASSRLS) — RLS enforced + # API uses crm_api (NOSUPERUSER, NOBYPASSRLS) — RLS enforced DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required} - # Migration/DDL uses crm_migration (owner, can bypass RLS for DDL) + # Auth uses crm_auth (NOSUPERUSER, NOBYPASSRLS) — identity tables only + AUTH_DATABASE_URL: ${AUTH_DATABASE_URL:-postgresql+asyncpg://crm_auth:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}} + # Migration/DDL uses crm_migration (owner, NOBYPASSRLS) MIGRATION_DATABASE_URL: ${MIGRATION_DATABASE_URL:-postgresql+asyncpg://crm_migration:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}} REDIS_URL: ${REDIS_URL:-redis://:${REDIS_PASSWORD}@redis:6379/0} SECRET_KEY: ${SECRET_KEY:?SECRET_KEY is required (min 32 chars)} @@ -152,9 +154,10 @@ services: memory: 512M cpus: "1.0" environment: - # Worker uses crm_runtime too — RLS enforced - DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required} - MIGRATION_DATABASE_URL: ${MIGRATION_DATABASE_URL:-postgresql+asyncpg://${POSTGRES_USER:-crm_user}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}} + # Worker uses crm_worker (NOSUPERUSER, NOBYPASSRLS) — RLS enforced + DATABASE_URL: ${WORKER_DATABASE_URL:-${DATABASE_URL:?DATABASE_URL is required}} + WORKER_DATABASE_URL: ${WORKER_DATABASE_URL:-postgresql+asyncpg://crm_worker:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}} + MIGRATION_DATABASE_URL: ${MIGRATION_DATABASE_URL:-postgresql+asyncpg://crm_migration:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB:-crm_db}} REDIS_URL: ${REDIS_URL:-redis://:${REDIS_PASSWORD}@redis:6379/0} SECRET_KEY: ${SECRET_KEY:?SECRET_KEY is required (min 32 chars)} FRONTEND_URL: ${FRONTEND_URL:-http://localhost:8000} diff --git a/tests/test_rls_coverage.py b/tests/test_rls_coverage.py new file mode 100644 index 0000000..1a6b72a --- /dev/null +++ b/tests/test_rls_coverage.py @@ -0,0 +1,299 @@ +"""Automated RLS coverage check for all tenant tables. + +This test verifies that every table in the application schema that has a +``tenant_id`` column has: + +1. RLS enabled (relrowsecurity = true) +2. FORCE ROW LEVEL SECURITY enabled (relforcerowsecurity = true) +3. A tenant isolation policy using ``app.current_tenant_id`` +4. No fail-open/bootstrap policy (IS NULL, = '', COALESCE patterns) +5. Policy scoped to runtime roles (crm_api, crm_worker) — not PUBLIC +6. Policy has both USING and WITH CHECK clauses + +It also verifies that runtime roles are not superuser, do not bypass RLS, +and are not table owners. + +This test does NOT mock the RLS boundary — it queries pg_catalog directly. +""" + +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!!" + + +# Use admin connection to inspect catalog +_ADMIN_DB_URL = os.environ.get( + "RLS_TEST_ADMIN_DB_URL", + "postgresql+asyncpg://postgres@localhost:5432/leocrm_test", +) + + +def _skip_if_no_db(): + """Skip if the test database is not available.""" + 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 + + +_skip_reason = "Test database not available for RLS coverage check." + + +@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=_skip_reason) +async def test_all_tenant_tables_have_rls_enabled(admin_session: AsyncSession): + """Every table with tenant_id must have RLS enabled.""" + result = await admin_session.execute(text(""" + SELECT c.relname + FROM pg_class c + JOIN pg_attribute a ON a.attrelid = c.oid + WHERE c.relnamespace = 'public'::regnamespace + AND c.relkind = 'r' + AND a.attname = 'tenant_id' + AND c.relrowsecurity = false + ORDER BY c.relname + """)) + tables_without_rls = [row[0] for row in result.fetchall()] + assert len(tables_without_rls) == 0, \ + f"Tables with tenant_id but RLS DISABLED: {tables_without_rls}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_all_tenant_tables_have_force_rls(admin_session: AsyncSession): + """Every table with tenant_id must have FORCE ROW LEVEL SECURITY.""" + result = await admin_session.execute(text(""" + SELECT c.relname + FROM pg_class c + JOIN pg_attribute a ON a.attrelid = c.oid + WHERE c.relnamespace = 'public'::regnamespace + AND c.relkind = 'r' + AND a.attname = 'tenant_id' + AND c.relforcerowsecurity = false + ORDER BY c.relname + """)) + tables_without_force = [row[0] for row in result.fetchall()] + assert len(tables_without_force) == 0, \ + f"Tables with tenant_id but FORCE RLS DISABLED: {tables_without_force}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_all_tenant_tables_have_isolation_policy(admin_session: AsyncSession): + """Every tenant table must have a tenant isolation policy.""" + result = await admin_session.execute(text(""" + SELECT c.relname + FROM pg_class c + JOIN pg_attribute a ON a.attrelid = c.oid + WHERE c.relnamespace = 'public'::regnamespace + AND c.relkind = 'r' + AND a.attname = 'tenant_id' + AND NOT EXISTS ( + SELECT 1 FROM pg_policy p + WHERE p.polrelid = c.oid + AND p.polname LIKE '%tenant_isolation%' + ) + ORDER BY c.relname + """)) + tables_without_policy = [row[0] for row in result.fetchall()] + assert len(tables_without_policy) == 0, \ + f"Tables with tenant_id but no isolation policy: {tables_without_policy}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_no_fail_open_policies(admin_session: AsyncSession): + """No RLS policy should use fail-open patterns (IS NULL, = '', COALESCE).""" + result = await admin_session.execute(text(""" + SELECT tablename, policyname, qual, with_check + FROM pg_policies + WHERE schemaname = 'public' + AND ( + qual ILIKE '%current_setting%IS NULL%' + OR qual ILIKE '%current_setting%= ''%' + OR qual ILIKE '%COALESCE%current_setting%' + OR with_check ILIKE '%current_setting%IS NULL%' + OR with_check ILIKE '%current_setting%= ''%' + OR with_check ILIKE '%COALESCE%current_setting%' + ) + """)) + bad_policies = result.fetchall() + assert len(bad_policies) == 0, \ + f"Fail-open RLS policies found: {bad_policies}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_policies_scoped_to_runtime_roles(admin_session: AsyncSession): + """Tenant isolation policies must be scoped to crm_api/crm_worker, not PUBLIC.""" + result = await admin_session.execute(text(""" + SELECT tablename, policyname, roles + FROM pg_policies + WHERE schemaname = 'public' + AND policyname LIKE '%tenant_isolation%' + AND NOT ('{crm_api,crm_worker}' = roles) + AND NOT (roles @> '{crm_api}' AND roles @> '{crm_worker}') + """)) + wrong_scope = result.fetchall() + assert len(wrong_scope) == 0, \ + f"Policies not scoped to crm_api+crm_worker: {wrong_scope}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_policies_use_app_current_tenant_id(admin_session: AsyncSession): + """All tenant policies must use app.current_tenant_id, not app.tenant_id.""" + result = await admin_session.execute(text(""" + SELECT tablename, policyname, qual, with_check + FROM pg_policies + WHERE schemaname = 'public' + AND policyname LIKE '%tenant_isolation%' + AND ( + qual ILIKE '%app.tenant_id%' + OR with_check ILIKE '%app.tenant_id%' + ) + """)) + old_var = result.fetchall() + assert len(old_var) == 0, \ + f"Policies using legacy app.tenant_id: {old_var}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_policies_have_with_check(admin_session: AsyncSession): + """Tenant policies must have WITH CHECK for write protection.""" + result = await admin_session.execute(text(""" + SELECT tablename, policyname + FROM pg_policies + WHERE schemaname = 'public' + AND policyname LIKE '%tenant_isolation%' + AND with_check IS NULL + """)) + no_check = result.fetchall() + assert len(no_check) == 0, \ + f"Policies without WITH CHECK: {no_check}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_runtime_roles_not_superuser(admin_session: AsyncSession): + """crm_api and crm_worker must not be superuser or bypass RLS.""" + result = await admin_session.execute(text(""" + SELECT rolname, rolsuper, rolbypassrls + FROM pg_roles + WHERE rolname IN ('crm_api', 'crm_worker', 'crm_migration', 'crm_auth') + ORDER BY rolname + """)) + for row in result.fetchall(): + rolname, rolsuper, rolbypassrls = row + assert rolsuper is False, f"{rolname} is SUPERUSER!" + assert rolbypassrls is False, f"{rolname} has BYPASSRLS!" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_runtime_roles_not_table_owner(admin_session: AsyncSession): + """crm_api and crm_worker must not own any tables.""" + result = await admin_session.execute(text(""" + SELECT tablename, tableowner + FROM pg_tables + WHERE schemaname = 'public' + AND tableowner IN ('crm_api', 'crm_worker', 'crm_auth') + """)) + owned = result.fetchall() + assert len(owned) == 0, \ + f"Runtime roles own tables: {owned}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_crm_runtime_role_dropped(admin_session: AsyncSession): + """crm_runtime legacy role must not exist.""" + result = await admin_session.execute(text(""" + SELECT 1 FROM pg_roles WHERE rolname = 'crm_runtime' + """)) + exists = result.fetchone() + assert exists is None, "crm_runtime role still exists — should have been dropped" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_no_rls_on_global_tables(admin_session: AsyncSession): + """Global tables (no tenant_id) must NOT have RLS enabled.""" + # Tables without tenant_id should not have RLS + result = await admin_session.execute(text(""" + SELECT c.relname + FROM pg_class c + WHERE c.relnamespace = 'public'::regnamespace + AND c.relkind = 'r' + AND c.relrowsecurity = true + AND NOT EXISTS ( + SELECT 1 FROM pg_attribute a + WHERE a.attrelid = c.oid AND a.attname = 'tenant_id' + ) + ORDER BY c.relname + """)) + global_with_rls = [row[0] for row in result.fetchall()] + assert len(global_with_rls) == 0, \ + f"Global tables (no tenant_id) with RLS enabled: {global_with_rls}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_crm_auth_has_minimal_access(admin_session: AsyncSession): + """crm_auth should only have access to identity tables, not business data.""" + result = await admin_session.execute(text(""" + SELECT table_name + FROM information_schema.role_table_grants + WHERE table_schema = 'public' + AND grantee = 'crm_auth' + AND table_name NOT IN ('users', 'user_tenants', 'tenants', 'password_reset_tokens') + """)) + extra_access = [row[0] for row in result.fetchall()] + assert len(extra_access) == 0, \ + f"crm_auth has access to non-identity tables: {extra_access}" + + +@pytest.mark.asyncio +@pytest.mark.skipif(_skip_if_no_db(), reason=_skip_reason) +async def test_alembic_version_not_accessible_to_runtime(admin_session: AsyncSession): + """crm_api and crm_worker must not have access to alembic_version.""" + result = await admin_session.execute(text(""" + SELECT grantee + FROM information_schema.role_table_grants + WHERE table_schema = 'public' + AND table_name = 'alembic_version' + AND grantee IN ('crm_api', 'crm_worker', 'crm_auth') + """)) + accessors = [row[0] for row in result.fetchall()] + assert len(accessors) == 0, \ + f"Runtime roles have access to alembic_version: {accessors}"