diff --git a/tests/conftest.py b/tests/conftest.py index 42965fc..6dadbbd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -236,6 +236,94 @@ def db_setup(): except Exception as e: print(f"[CONFTEST] Trigger fix FAILED: {e}") + # Grant crm_api role access + enable RLS + create tenant isolation + # policies (Block E / I-C): The Cross-Tenant Security tests connect as + # crm_api (NOSUPERUSER, NOBYPASSRLS) to verify RLS enforcement. + print("[CONFTEST] Setting up RLS grants and policies...") + try: + sync_eng3 = _get_sync_engine() + with sync_eng3.connect() as conn: + # 1. Ensure crm_api role exists + conn.execute(text( + "DO $$ BEGIN " + "IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_api') THEN " + "CREATE ROLE crm_api LOGIN PASSWORD 'crm_api_password' NOSUPERUSER NOBYPASSRLS; " + "END IF; END $$;" + )) + + # 2. Enable RLS on all tenant tables EXCEPT auth-critical ones + # (user_tenants must be readable without tenant context for login) + _no_rls_tables = "('user_tenants','tenants','users','audit_log','alembic_version','groups','user_groups','roles')" + conn.execute(text(f""" + DO $$ DECLARE r RECORD; + BEGIN + FOR r IN ( + SELECT c.relname AS tablename + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' + AND c.relkind = 'r' + AND c.relrowsecurity = false + AND c.relname NOT IN {_no_rls_tables} + AND EXISTS ( + SELECT 1 FROM information_schema.columns ic + WHERE ic.table_schema = 'public' + AND ic.table_name = c.relname + AND ic.column_name = 'tenant_id' + ) + ) LOOP + EXECUTE format('ALTER TABLE public.%I ENABLE ROW LEVEL SECURITY', r.tablename); + END LOOP; + END $$; + """)) + + # 3. Create standard tenant-isolation policy per table that has + # RLS enabled but no policy yet + conn.execute(text(""" + DO $$ DECLARE r RECORD; + BEGIN + FOR r IN ( + SELECT c.relname AS tablename + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' + AND c.relkind = 'r' + AND c.relrowsecurity = true + AND NOT EXISTS ( + SELECT 1 FROM pg_policies p + WHERE p.schemaname = 'public' + AND p.tablename = c.relname + ) + ) LOOP + EXECUTE format( + 'CREATE POLICY %I_tenant_isolation ON public.%I ' + 'FOR ALL USING (tenant_id = current_setting(''app.current_tenant_id'', true)::uuid) ' + 'WITH CHECK (tenant_id = current_setting(''app.current_tenant_id'', true)::uuid)', + r.tablename, r.tablename + ); + END LOOP; + END $$; + """)) + + # 4. Grant crm_api access to all tables + conn.execute(text("GRANT USAGE ON SCHEMA public TO crm_api")) + conn.execute(text("GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO crm_api")) + conn.execute(text("GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO crm_api")) + conn.commit() + + # Count results + with sync_eng3.connect() as verify_conn: + rls_count = verify_conn.execute(text( + "SELECT count(*) FROM pg_tables WHERE schemaname='public' AND rowsecurity=true" + )).scalar() + pol_count = verify_conn.execute(text( + "SELECT count(*) FROM pg_policies WHERE schemaname='public'" + )).scalar() + print(f"[CONFTEST] RLS setup complete: {rls_count} RLS tables, {pol_count} policies") + sync_eng3.dispose() + except Exception as e: + print(f"[CONFTEST] RLS setup FAILED: {e}") + yield diff --git a/tests/test_cross_tenant_security_v2.py b/tests/test_cross_tenant_security_v2.py index c141fe8..14d8699 100644 --- a/tests/test_cross_tenant_security_v2.py +++ b/tests/test_cross_tenant_security_v2.py @@ -96,13 +96,18 @@ async def api_engine(): @pytest_asyncio.fixture async def admin_session(admin_engine): - """Admin session for data setup.""" + """Admin session for data setup. + + No externally begun transaction: the session must own its transactions so + that seed_data's commit() really persists rows (visible to the separate + crm_api connection used by api_session). With conn.begin() the Session + would treat it as an external transaction (conditional_savepoint) and + commit() would not release it. + """ async with admin_engine.connect() as conn: - await conn.begin() session = AsyncSession(bind=conn, expire_on_commit=False) yield session - await session.rollback() - await conn.rollback() + await session.close() @pytest_asyncio.fixture @@ -173,7 +178,14 @@ async def seed_data(admin_session: AsyncSession): admin_session.add_all([contact_a, contact_b]) await admin_session.flush() - return { + # COMMIT is mandatory here: the api_session fixture connects as the + # unprivileged crm_api role on a SEPARATE connection. Uncommitted rows + # from the admin connection are invisible there, so FK checks + # (contacts.updated_by -> users) and RLS visibility assertions would + # operate on an empty database. + await admin_session.commit() + + yield { "tenant_a": tenant_a, "tenant_b": tenant_b, "user_a": user_a, @@ -182,6 +194,26 @@ async def seed_data(admin_session: AsyncSession): "contact_b": contact_b, } + # Teardown: remove seeded rows so the committed data does not leak into + # other tests (order respects FK dependencies). + await admin_session.execute( + text("DELETE FROM contacts WHERE id IN (:a, :b)"), + {"a": contact_a.id, "b": contact_b.id}, + ) + await admin_session.execute( + text("DELETE FROM user_tenants WHERE user_id IN (:a, :b)"), + {"a": user_a.id, "b": user_b.id}, + ) + await admin_session.execute( + text("DELETE FROM users WHERE id IN (:a, :b)"), + {"a": user_a.id, "b": user_b.id}, + ) + await admin_session.execute( + text("DELETE FROM tenants WHERE id IN (:a, :b)"), + {"a": tenant_a.id, "b": tenant_b.id}, + ) + await admin_session.commit() + # ── RLS Enforcement Tests with Unprivileged Role ───────────────────────────── @@ -210,10 +242,10 @@ async def test_rls_tenant_a_sees_only_own_rows(api_session: AsyncSession, seed_d ) rows = result.fetchall() for row in rows: - assert row[0] == str(tenant_a.id), \ + assert str(row[0]) == str(tenant_a.id), \ f"RLS leak: tenant A context shows row from {row[0]}" # Tenant B's contact must not be visible - tenant_b_ids = [r[0] for r in rows if r[0] == str(tenant_b.id)] + tenant_b_ids = [r for r in rows if str(r[0]) == str(tenant_b.id)] assert len(tenant_b_ids) == 0, "RLS failed: Tenant B data visible in Tenant A context!" @@ -230,9 +262,9 @@ async def test_rls_tenant_b_sees_only_own_rows(api_session: AsyncSession, seed_d ) rows = result.fetchall() for row in rows: - assert row[0] == str(tenant_b.id), \ + assert str(row[0]) == str(tenant_b.id), \ f"RLS leak: tenant B context shows row from {row[0]}" - tenant_a_ids = [r[0] for r in rows if r[0] == str(tenant_a.id)] + tenant_a_ids = [r for r in rows if str(r[0]) == str(tenant_a.id)] assert len(tenant_a_ids) == 0, "RLS failed: Tenant A data visible in Tenant B context!" @@ -246,33 +278,37 @@ async def test_rls_blocks_cross_tenant_insert(api_session: AsyncSession, seed_da await set_tenant_context(api_session, tenant_a.id) - # Try to insert a contact with tenant B's ID while in tenant A context + # Try to insert a contact with tenant B's ID while in tenant A context. + # asyncpg executes eagerly: the RLS WITH CHECK violation surfaces directly + # at execute() (asyncpg InsufficientPrivilege -> SQLAlchemy ProgrammingError, + # both subclasses of DBAPIError). new_id = uuid.uuid4() - await api_session.execute( - text( - "INSERT INTO contacts (id, tenant_id, firstname, surname, email_1, " - "owner_id, created_by, updated_by, type, displayname) " - "VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname)" - ), - { - "id": str(new_id), - "tenant_id": str(tenant_b.id), # Wrong tenant! - "firstname": "Cross", - "surname": "Tenant", - "email": f"cross-{uuid.uuid4().hex[:8]}@test.local", - "owner": str(user_a.id), - "creator": str(user_a.id), - "updater": str(user_a.id), - "ctype": "person", - "dname": "Cross Tenant", - }, - ) - - # The INSERT should fail due to RLS WITH CHECK - with pytest.raises(Exception) as exc_info: - await api_session.flush() - assert "row level security" in str(exc_info.value).lower() or "rls" in str(exc_info.value).lower(), \ - f"Expected RLS error, got: {exc_info.value}" + from sqlalchemy.exc import DBAPIError + with pytest.raises(DBAPIError) as exc_info: + await api_session.execute( + text( + "INSERT INTO contacts (id, tenant_id, firstname, surname, email_1, " + "owner_id, created_by, updated_by, type, displayname, status) " + "VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname, 'open')" + ), + { + "id": str(new_id), + "tenant_id": str(tenant_b.id), # Wrong tenant! + "firstname": "Cross", + "surname": "Tenant", + "email": f"cross-{uuid.uuid4().hex[:8]}@test.local", + "owner": str(user_a.id), + "creator": str(user_a.id), + "updater": str(user_a.id), + "ctype": "person", + "dname": "Cross Tenant", + }, + ) + # The INSERT must have failed due to RLS WITH CHECK — verify the message + err_text = str(exc_info.value).lower() + assert "row-level security" in err_text or "row level security" in err_text, \ + f"Expected RLS violation, got: {exc_info.value}" + # Roll back the failed transaction so subsequent statements work await api_session.rollback() @@ -325,8 +361,10 @@ async def test_rls_tenant_a_insert_own_succeeds(api_session: AsyncSession, seed_ await api_session.execute( text( "INSERT INTO contacts (id, tenant_id, firstname, surname, email_1, " - "owner_id, created_by, updated_by, type, displayname) " - "VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname)" + "owner_id, created_by, updated_by, type, displayname, status, " + "discount_crew, discount_transport, discount_rental, discount_sale, discount_subrent, discount_total) " + "VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname, 'open', " + "0, 0, 0, 0, 0, 0)" ), { "id": str(new_id),