From 93f280621bbc72c3997611f69663a9fc80dc0441 Mon Sep 17 00:00:00 2001 From: camilafagundes Date: Sat, 24 Oct 2026 00:28:43 -0300 Subject: [PATCH] fix(db): restringe privilegios e torna migrations atomicas Co-authored-by: Cursor --- migrations/0003_rls.sql | 49 ++++++++++++++++++++++++++++++++++++++-- src/infra/db/migrate.ts | 44 ++++++++++++++++++++++-------------- src/infra/db/pglite.ts | 24 +++++++++++++++++++- src/infra/db/postgres.ts | 16 ++++++++++++- src/infra/db/types.ts | 7 ++++++ 5 files changed, 119 insertions(+), 21 deletions(-) diff --git a/migrations/0003_rls.sql b/migrations/0003_rls.sql index a0b841b..eadf9ab 100644 --- a/migrations/0003_rls.sql +++ b/migrations/0003_rls.sql @@ -93,12 +93,15 @@ returns boolean language sql stable security definer -set search_path = public +set search_path = pg_catalog as $$ select exists ( - select 1 from public.spreadsheet_jobs j + select 1 + from public.spreadsheet_jobs j + join public.spreadsheet_models m on m.id = j.model_id where j.model_id = model and j.status in ('validating', 'ready', 'queued', 'processing') + and public.is_member(m.workspace_id) ) $$; @@ -208,14 +211,17 @@ create policy models_update on public.spreadsheet_models with check (public.can_manage_adapter(workspace_id)); drop policy if exists jobs_all on public.spreadsheet_jobs; +drop policy if exists jobs_select on public.spreadsheet_jobs; create policy jobs_select on public.spreadsheet_jobs for select to app_user using (user_id = public.auth_uid() and public.is_member(workspace_id) and public.module_enabled(workspace_id)); +drop policy if exists jobs_insert on public.spreadsheet_jobs; create policy jobs_insert on public.spreadsheet_jobs for insert to app_user with check (user_id = public.auth_uid() and public.is_member(workspace_id) and public.module_enabled(workspace_id)); +drop policy if exists jobs_update on public.spreadsheet_jobs; create policy jobs_update on public.spreadsheet_jobs for update to app_user using (user_id = public.auth_uid() and public.is_member(workspace_id) and public.module_enabled(workspace_id)) @@ -229,6 +235,7 @@ create policy sources_select on public.spreadsheet_job_sources where j.id = job_id and j.user_id = public.auth_uid() and public.module_enabled(j.workspace_id) )); +drop policy if exists sources_insert on public.spreadsheet_job_sources; create policy sources_insert on public.spreadsheet_job_sources for insert to app_user with check (exists ( @@ -236,6 +243,7 @@ create policy sources_insert on public.spreadsheet_job_sources where j.id = job_id and j.user_id = public.auth_uid() and public.module_enabled(j.workspace_id) )); +drop policy if exists sources_update on public.spreadsheet_job_sources; create policy sources_update on public.spreadsheet_job_sources for update to app_user using (exists ( @@ -247,6 +255,7 @@ create policy sources_update on public.spreadsheet_job_sources where j.id = job_id and j.user_id = public.auth_uid() and public.module_enabled(j.workspace_id) )); +drop policy if exists sources_delete on public.spreadsheet_job_sources; create policy sources_delete on public.spreadsheet_job_sources for delete to app_user using (exists ( @@ -263,6 +272,7 @@ create policy mappings_select on public.spreadsheet_job_mappings where s.id = job_source_id and j.user_id = public.auth_uid() )); +drop policy if exists mappings_insert on public.spreadsheet_job_mappings; create policy mappings_insert on public.spreadsheet_job_mappings for insert to app_user with check (exists ( @@ -271,6 +281,7 @@ create policy mappings_insert on public.spreadsheet_job_mappings where s.id = job_source_id and j.user_id = public.auth_uid() )); +drop policy if exists mappings_delete on public.spreadsheet_job_mappings; create policy mappings_delete on public.spreadsheet_job_mappings for delete to app_user using (exists ( @@ -282,3 +293,37 @@ create policy mappings_delete on public.spreadsheet_job_mappings grant usage on schema public to app_user; grant select, insert, update, delete on all tables in schema public to app_user; grant usage, select on all sequences in schema public to app_user; + +-- schema_migrations não tem RLS. A escrita fica só com o dono que aplica as migrations. +revoke insert, update, delete, truncate, references, trigger on table public.schema_migrations from public, app_user; + +do $priv$ +declare + versao int := current_setting('server_version_num')::int; + escrita text := 'insert, update, delete, truncate, references, trigger'; + papel text; +begin + if versao >= 170000 then + escrita := escrita || ', maintain'; + execute 'revoke maintain on table public.schema_migrations from public, app_user'; + end if; + foreach papel in array array['anon', 'authenticated', 'service_role', 'adaptador_web', 'adaptador_worker'] + loop + if exists (select 1 from pg_catalog.pg_roles where rolname = papel) then + execute format( + 'revoke %s on table public.profiles, public.workspaces, public.workspace_members, public.workspace_module_flags, public.audit_log, public.spreadsheet_adapter_settings, public.spreadsheet_models, public.spreadsheet_jobs, public.spreadsheet_job_sources, public.spreadsheet_job_mappings, public.schema_migrations from %I', + escrita, + papel + ); + if papel in ('anon', 'authenticated', 'service_role') then + execute format('revoke select on table public.schema_migrations from %I', papel); + execute format( + 'alter default privileges in schema public revoke %s on tables from %I', + escrita, + papel + ); + end if; + end if; + end loop; +end +$priv$; diff --git a/src/infra/db/migrate.ts b/src/infra/db/migrate.ts index 014cb10..58fb3a9 100644 --- a/src/infra/db/migrate.ts +++ b/src/infra/db/migrate.ts @@ -16,6 +16,8 @@ import type { Db } from "./types"; */ const MIGRATIONS_FOLDER = "migrations"; +/** Trava só durante a transação do migrate. A tabela de controle permanece id e applied_at. */ +const MIGRATION_LOCK = 872341001; export type MigrateOptions = { /** Aplica o shim de identidade. Padrão: `ADAPTER_APPLY_IDENTITY_SHIM`. */ @@ -51,27 +53,35 @@ export function assertManifestMatches(files: string[]): void { export async function migrate(db: Db, options: MigrateOptions = {}): Promise { const applyShim = options.applyIdentityShim ?? identityShimEnabled(); - - await db.exec(` - create table if not exists public.schema_migrations ( - id text primary key, - applied_at timestamptz not null default now() - ); - `); const files = await listMigrationFiles(); assertManifestMatches(files); - const appliedRows = await db.withService((query) => query.query<{ id: string }>("select id from public.schema_migrations")); - const applied = new Set(appliedRows.map((row) => row.id)); - const ran: string[] = []; + const sqlByFile = new Map(); for (const file of files) { - if (applied.has(file)) continue; - if (isShimMigration(file) && !applyShim) continue; - const sql = await readFile(path.join(process.cwd(), MIGRATIONS_FOLDER, file), "utf8"); - await db.exec(sql); - await db.withService((query) => query.query("insert into public.schema_migrations (id) values ($1)", [file])); - ran.push(file); + sqlByFile.set(file, await readFile(path.join(process.cwd(), MIGRATIONS_FOLDER, file), "utf8")); } - return ran; + + return db.withTransaction(async (tx) => { + await tx.exec(`select pg_catalog.pg_advisory_xact_lock(${MIGRATION_LOCK})`); + await tx.exec(` + create table if not exists public.schema_migrations ( + id text primary key, + applied_at timestamptz not null default now() + ); + `); + const appliedRows = await tx.query<{ id: string }>("select id from public.schema_migrations"); + const applied = new Set(appliedRows.map((row) => row.id)); + const ran: string[] = []; + for (const file of files) { + if (applied.has(file)) continue; + if (isShimMigration(file) && !applyShim) continue; + const sql = sqlByFile.get(file); + if (!sql) throw new Error(`migration ausente: ${file}`); + await tx.exec(sql); + await tx.query("insert into public.schema_migrations (id) values ($1)", [file]); + ran.push(file); + } + return ran; + }); } export { EXPECTED_MIGRATIONS, SHIM_MIGRATIONS, isShimMigration, pendingMigrations } from "./migration-manifest"; diff --git a/src/infra/db/pglite.ts b/src/infra/db/pglite.ts index 5c6f61d..71a664d 100644 --- a/src/infra/db/pglite.ts +++ b/src/infra/db/pglite.ts @@ -1,5 +1,5 @@ import { PGlite } from "@electric-sql/pglite"; -import { createMutex, type Db, type Query, type Row, assertUuid } from "./types"; +import { createMutex, type Db, type Query, type Row, type Tx, assertUuid } from "./types"; type PgClient = { exec(sql: string): Promise; @@ -44,6 +44,28 @@ export function wrapPglite(client: PgClient): Db { close() { return client.close(); }, + withTransaction(fn) { + return exclusive(async () => { + await client.exec("begin"); + const tx: Tx = { + exec(sql) { + return client.exec(sql).then(() => undefined); + }, + async query(sql: string, params: unknown[] = []) { + const result = await client.query(sql, params); + return result.rows as T[]; + }, + }; + try { + const value = await fn(tx); + await client.exec("commit"); + return value; + } catch (error) { + await client.exec("rollback"); + throw error; + } + }); + }, }; } diff --git a/src/infra/db/postgres.ts b/src/infra/db/postgres.ts index 7e169ed..e91fa12 100644 --- a/src/infra/db/postgres.ts +++ b/src/infra/db/postgres.ts @@ -1,6 +1,6 @@ import postgres from "postgres"; import type { PostgresClientOptions } from "./connection-policy"; -import { assertUuid, type Db, type Query, type Row } from "./types"; +import { assertUuid, type Db, type Query, type Row, type Tx } from "./types"; export function createPostgresDb(connectionString: string, options?: PostgresClientOptions): Db { const sql = postgres(connectionString, options ?? { max: 8, prepare: false }); @@ -33,5 +33,19 @@ export function createPostgresDb(connectionString: string, options?: PostgresCli async close() { await sql.end(); }, + async withTransaction(fn: (tx: Tx) => Promise): Promise { + const result = await sql.begin(async (tx) => { + const adapter: Tx = { + async exec(text: string) { + await tx.unsafe(text); + }, + async query(text: string, params: unknown[] = []) { + return tx.unsafe(text, params as never[]) as unknown as R[]; + }, + }; + return fn(adapter); + }); + return result as T; + }, }; } diff --git a/src/infra/db/types.ts b/src/infra/db/types.ts index 1037f18..be50c2a 100644 --- a/src/infra/db/types.ts +++ b/src/infra/db/types.ts @@ -4,10 +4,17 @@ export interface Query { query(sql: string, params?: unknown[]): Promise; } +export interface Tx { + exec(sql: string): Promise; + query(sql: string, params?: unknown[]): Promise; +} + export interface Db { exec(sql: string): Promise; withUser(userId: string, fn: (query: Query) => Promise): Promise; withService(fn: (query: Query) => Promise): Promise; + /** Uma conexão, uma transação. O callback inteiro confirma ou reverte junto. */ + withTransaction(fn: (tx: Tx) => Promise): Promise; close(): Promise; }