"""Executable PostgreSQL RLS verification for the production tenant boundary. Run only against an isolated disposable database: SOCIAL_TEST_ADMIN_DATABASE_URL=postgresql://... # BYPASSRLS migration/seeding role SOCIAL_TEST_TENANT_DATABASE_URL=postgresql://... # non-owner, non-BYPASSRLS role pytest tests/test_postgres_rls.py The test never falls back to SQLite because SQLite cannot validate PostgreSQL policies. It intentionally does not run in normal CI without those dedicated credentials. """ from __future__ import annotations import os from pathlib import Path import pytest admin_url = os.getenv("SOCIAL_TEST_ADMIN_DATABASE_URL", "").strip() tenant_url = os.getenv("SOCIAL_TEST_TENANT_DATABASE_URL", "").strip() pytestmark = pytest.mark.skipif( not (admin_url and tenant_url), reason="PostgreSQL RLS integration credentials are not configured.", ) def _asyncpg_url(value: str) -> str: return value.replace("postgresql+asyncpg://", "postgresql://", 1) async def _apply_migrations(connection: object, directory: Path) -> None: for migration in sorted(directory.glob("*.sql")): await connection.execute(migration.read_text(encoding="utf-8")) # type: ignore[attr-defined] async def _tenant_context(connection: object, workspace_id: str, user_id: str) -> None: await connection.execute("select set_config('app.workspace_id', $1, false)", workspace_id) # type: ignore[attr-defined] await connection.execute("select set_config('app.user_id', $1, false)", user_id) # type: ignore[attr-defined] @pytest.mark.asyncio async def test_postgres_rls_rejects_cross_workspace_reads_and_writes() -> None: asyncpg = pytest.importorskip("asyncpg") root = Path(__file__).resolve().parents[1] admin = await asyncpg.connect(_asyncpg_url(admin_url)) tenant = await asyncpg.connect(_asyncpg_url(tenant_url)) try: # The configured database must be disposable and owned by the # migration role. Never point these variables at a customer database. await _apply_migrations(admin, root / "app" / "security" / "migrations") await _apply_migrations(admin, root / "app" / "projects" / "migrations") await _apply_migrations(admin, root / "app" / "social" / "migrations") role = await tenant.fetchrow( "select r.rolsuper, r.rolbypassrls from pg_roles r where r.rolname = current_user" ) assert role is not None assert not role["rolsuper"] and not role["rolbypassrls"] # Seed two fully independent tenants as the dedicated privileged role. await admin.execute( """ insert into api_keys (id,name,key_prefix,key_hash,environment,status,scopes,created_at,requests_per_minute,concurrent_jobs,uploads_per_hour,processing_bytes_per_day) values ('key-a','A','mp_test_aaaaaaaa','aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa','test','active','[]'::jsonb,now(),1,1,1,1048576), ('key-b','B','mp_test_bbbbbbbb','bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb','test','active','[]'::jsonb,now(),1,1,1,1048576); insert into users (id,subject,display_name) values ('user-a','test:user-a','A'),('user-b','test:user-b','B'); insert into workspaces (id,slug,name) values ('workspace-a','test-a','A'),('workspace-b','test-b','B'); insert into workspace_memberships (id,workspace_id,user_id,role) values ('membership-a','workspace-a','user-a','owner'),('membership-b','workspace-b','user-b','owner'); insert into api_key_principals (id,api_key_id,workspace_id,user_id,membership_id) values ('principal-a','key-a','workspace-a','user-a','membership-a'),('principal-b','key-b','workspace-b','user-b','membership-b'); insert into media_assets (id,workspace_id,request_id,filename,mime_type,file_size,sha256) values ('asset-a','workspace-a','00000000-0000-0000-0000-000000000001','a.mp4','video/mp4',1,repeat('a',64)), ('asset-b','workspace-b','00000000-0000-0000-0000-000000000002','b.mp4','video/mp4',1,repeat('b',64)); insert into media_asset_variants (id,workspace_id,asset_id,request_id,filename,mime_type,file_size,sha256) values ('asset-variant-a','workspace-a','asset-a','00000000-0000-0000-0000-000000000003','a-variant.mp4','video/mp4',1,repeat('c',64)), ('asset-variant-b','workspace-b','asset-b','00000000-0000-0000-0000-000000000004','b-variant.mp4','video/mp4',1,repeat('d',64)); insert into generation_requests (id,workspace_id,created_by_user_id,provider,model_id,modality,spec,request_fingerprint,idempotency_key) values ('generation-request-a','workspace-a','user-a','test-provider','test-image','image','{"prompt":"A"}'::jsonb,repeat('1',64),'generation-key-a'), ('generation-request-b','workspace-b','user-b','test-provider','test-image','image','{"prompt":"B"}'::jsonb,repeat('2',64),'generation-key-b'); insert into generation_jobs (id,generation_request_id,workspace_id,provider,status,output_asset_id) values ('generation-job-a','generation-request-a','workspace-a','test-provider','queued','asset-a'), ('generation-job-b','generation-request-b','workspace-b','test-provider','queued','asset-b'); insert into generation_job_attempts (id,generation_job_id,attempt_number,status) values ('generation-attempt-a','generation-job-a',1,'started'),('generation-attempt-b','generation-job-b',1,'started'); insert into projects (id,workspace_id,created_by,name) values ('project-a','workspace-a','user-a','A'),('project-b','workspace-b','user-b','B'); update media_assets set project_id = 'project-a' where id = 'asset-a'; update media_assets set project_id = 'project-b' where id = 'asset-b'; insert into project_generation_jobs (id,workspace_id,project_id,generation_job_id,attached_by) values ('project-generation-a','workspace-a','project-a','generation-job-a','user-a'), ('project-generation-b','workspace-b','project-b','generation-job-b','user-b'); insert into project_editor_states (id,workspace_id,project_id,revision,schema_version,state,updated_by) values ('editor-a','workspace-a','project-a',1,1,'{"schemaVersion":1,"projectId":"project-a","timeline":{"timeUnit":"milliseconds","tracks":[],"transitions":[],"markers":[]},"renderSettings":{"format":"mp4","width":1920,"height":1080,"frameRate":30}}'::jsonb,'user-a'), ('editor-b','workspace-b','project-b',1,1,'{"schemaVersion":1,"projectId":"project-b","timeline":{"timeUnit":"milliseconds","tracks":[],"transitions":[],"markers":[]},"renderSettings":{"format":"mp4","width":1920,"height":1080,"frameRate":30}}'::jsonb,'user-b'); insert into project_render_jobs (id,workspace_id,project_id,editor_revision,editor_schema_version,editor_state,render_settings,request_fingerprint,idempotency_key,requested_by) values ('render-a','workspace-a','project-a',1,1,'{"schemaVersion":1,"projectId":"project-a"}'::jsonb,'{"format":"mp4"}'::jsonb,repeat('a',64),'render-key-a','user-a'), ('render-b','workspace-b','project-b',1,1,'{"schemaVersion":1,"projectId":"project-b"}'::jsonb,'{"format":"mp4"}'::jsonb,repeat('b',64),'render-key-b','user-b'); insert into audit_events (id,workspace_id,actor_user_id,event_type,entity_type,entity_id) values ('project-audit-a','workspace-a','user-a','project.created','project','project-a'), ('project-audit-b','workspace-b','user-b','project.created','project','project-b'); insert into social_accounts (id,workspace_id,provider,account_type,external_account_id,status) values ('account-a','workspace-a','youtube','channel','a','connected'),('account-b','workspace-b','youtube','channel','b','connected'); insert into social_account_tokens (id,social_account_id,encrypted_payload) values ('token-a','account-a','opaque'),('token-b','account-b','opaque'); insert into social_account_capabilities (id,social_account_id,capability,enabled) values ('cap-a','account-a','publish',true),('cap-b','account-b','publish',true); insert into media_variants (id,workspace_id,source_asset_id) values ('variant-a','workspace-a','asset-a'),('variant-b','workspace-b','asset-b'); insert into social_media_assets (id,workspace_id,canonical_asset_id,request_id,filename,mime_type,file_size) values ('social-asset-a','workspace-a','asset-a','00000000-0000-0000-0000-000000000001','a.mp4','video/mp4',1), ('social-asset-b','workspace-b','asset-b','00000000-0000-0000-0000-000000000002','b.mp4','video/mp4',1); insert into social_campaigns (id,workspace_id,name) values ('campaign-a','workspace-a','A'),('campaign-b','workspace-b','B'); insert into social_webhook_events (id,provider,event_type,external_event_id,workspace_id) values ('webhook-a','youtube','TEST','webhook-a','workspace-a'),('webhook-b','youtube','TEST','webhook-b','workspace-b'); insert into social_posts (id,workspace_id,campaign_id,media_asset_id,source_variant_id,status,publish_mode) values ('post-a','workspace-a','campaign-a','social-asset-a','variant-a','draft','draft'), ('post-b','workspace-b','campaign-b','social-asset-b','variant-b','draft','draft'); insert into social_post_targets (id,social_post_id,social_account_id,provider) values ('target-a','post-a','account-a','youtube'),('target-b','post-b','account-b','youtube'); insert into social_post_media (id,social_post_id,media_variant_id,media_asset_id) values ('post-media-a','post-a','variant-a','social-asset-a'),('post-media-b','post-b','variant-b','social-asset-b'); insert into social_schedules (id,social_post_id,scheduled_at,timezone) values ('schedule-a','post-a',now(),'UTC'),('schedule-b','post-b',now(),'UTC'); insert into social_jobs (id,workspace_id,social_post_id,social_post_target_id,provider,status) values ('job-a','workspace-a','post-a','target-a','youtube','queued'),('job-b','workspace-b','post-b','target-b','youtube','queued'); insert into social_job_attempts (id,social_job_id,attempt_number,status) values ('attempt-a','job-a',1,'started'),('attempt-b','job-b',1,'started'); insert into social_post_metrics (id,social_post_id,social_post_target_id,provider) values ('metric-a','post-a','target-a','youtube'),('metric-b','post-b','target-b','youtube'); insert into social_audit_events (id,workspace_id,event_type) values ('audit-a','workspace-a','TEST'),('audit-b','workspace-b','TEST'); """ ) await _tenant_context(tenant, "workspace-a", "user-a") for table in ( "users", "workspaces", "workspace_memberships", "api_key_principals", "media_assets", "media_asset_variants", "generation_requests", "generation_jobs", "generation_job_attempts", "social_accounts", "projects", "project_generation_jobs", "project_editor_states", "project_render_jobs", "audit_events", "social_account_tokens", "social_account_capabilities", "media_variants", "social_media_assets", "social_posts", "social_post_targets", "social_post_media", "social_schedules", "social_jobs", "social_job_attempts", "social_post_metrics", "social_audit_events", "social_campaigns", "social_webhook_events", ): assert await tenant.fetchval(f"select count(*) from {table}") == 1, table # RLS WITH CHECK rejects direct reassignment to tenant B. The relation # integrity triggers independently reject cross-tenant child links. with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into social_accounts (id,workspace_id,provider,account_type,external_account_id,status) values ('blocked-account','workspace-b','youtube','channel','blocked','connected')" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into workspace_memberships (id,workspace_id,user_id,role) values ('blocked-membership','workspace-b','user-a','member')" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into media_assets (id,workspace_id,request_id,filename,mime_type,file_size,sha256) values ('blocked-asset','workspace-b','00000000-0000-0000-0000-000000000006','blocked.mp4','video/mp4',1,repeat('f',64))" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into media_asset_variants (id,workspace_id,asset_id,request_id,filename,mime_type,file_size,sha256) values ('blocked-variant','workspace-a','asset-b','00000000-0000-0000-0000-000000000005','blocked.mp4','video/mp4',1,repeat('e',64))" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into generation_requests (id,workspace_id,created_by_user_id,provider,model_id,modality,spec,request_fingerprint,idempotency_key) values ('blocked-generation-request','workspace-b','user-a','test-provider','test-image','image','{}'::jsonb,repeat('3',64),'blocked-generation-key')" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into projects (id,workspace_id,created_by,name) values ('blocked-project','workspace-b','user-a','Blocked')" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "update projects set thumbnail_asset_id = 'asset-b' where id = 'project-a'" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "update media_assets set project_id = 'project-b' where id = 'asset-a'" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into project_generation_jobs (id,workspace_id,project_id,generation_job_id,attached_by) values ('blocked-project-job','workspace-a','project-a','generation-job-b','user-a')" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into project_editor_states (id,workspace_id,project_id,revision,schema_version,state,updated_by) values ('blocked-editor','workspace-b','project-b',1,1,'{}'::jsonb,'user-a')" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into project_render_jobs (id,workspace_id,project_id,editor_revision,editor_schema_version,editor_state,render_settings,request_fingerprint,idempotency_key,requested_by) values ('blocked-render','workspace-a','project-b',1,1,'{}'::jsonb,'{}'::jsonb,repeat('c',64),'blocked-render','user-a')" ) with pytest.raises(asyncpg.PostgresError): await admin.execute( "insert into project_generation_jobs (id,workspace_id,project_id,generation_job_id,attached_by) values ('blocked-project-job-admin','workspace-a','project-a','generation-job-b','user-a')" ) # The provider-runtime migration prevents a trusted worker recovery # process from binding one opaque worker job to two tenant jobs. await admin.execute( "update generation_jobs set external_job_id = 'worker-job-a' where id = 'generation-job-a'" ) with pytest.raises(asyncpg.PostgresError): await admin.execute( "update generation_jobs set external_job_id = 'worker-job-a' where id = 'generation-job-b'" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "update generation_jobs set output_asset_id = 'asset-b' where id = 'generation-job-a'" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "insert into social_post_targets (id,social_post_id,social_account_id,provider) values ('blocked-target','post-a','account-b','youtube')" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "update social_posts set workspace_id = 'workspace-b' where id = 'post-a'" ) # RLS USING makes an attempted write to tenant B affect zero rows. assert ( await tenant.execute("update social_jobs set status = 'failed' where id = 'job-b'") == "UPDATE 0" ) assert ( await tenant.execute("update projects set name = 'Blocked' where id = 'project-b'") == "UPDATE 0" ) assert await tenant.execute("delete from projects where id = 'project-b'") == "DELETE 0" assert ( await tenant.execute( "delete from project_generation_jobs where id = 'project-generation-b'" ) == "DELETE 0" ) assert ( await tenant.fetchval( "select count(*) from project_editor_states where project_id = 'project-b'" ) == 0 ) assert ( await tenant.execute( "update project_editor_states set revision = 2 where id = 'editor-b'" ) == "UPDATE 0" ) assert ( await tenant.execute( "update project_render_jobs set status = 'cancelled' where id = 'render-b'" ) == "UPDATE 0" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "update project_editor_states set revision = 3 where id = 'editor-a'" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "update project_render_jobs set render_settings = '{\"format\":\"webm\"}'::jsonb where id = 'render-a'" ) with pytest.raises(asyncpg.PostgresError): await tenant.execute( "update project_render_jobs set status = 'completed', completed_at = now() where id = 'render-a'" ) await admin.execute( "insert into users (id,subject,display_name) values ('viewer-a','test:viewer-a','Viewer'); " "insert into workspace_memberships (id,workspace_id,user_id,role) values ('viewer-membership-a','workspace-a','viewer-a','viewer')" ) await _tenant_context(tenant, "workspace-a", "viewer-a") assert await tenant.fetchval("select count(*) from project_editor_states") == 1 assert await tenant.fetchval("select count(*) from project_render_jobs") == 1 assert ( await tenant.execute( "update project_editor_states set revision = 2 where id = 'editor-a'" ) == "UPDATE 0" ) assert ( await tenant.execute( "update project_render_jobs set status = 'cancelled' where id = 'render-a'" ) == "UPDATE 0" ) finally: await tenant.close() await admin.close()