tests/test_postgres_rls.py

"""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()