From 6bef3b676f4fe74fff64b9bd64d47f78de1c74d7 Mon Sep 17 00:00:00 2001 From: Dan Lynch Date: Sun, 2 Aug 2026 19:38:00 +0000 Subject: [PATCH] fix(presigned-url): apply request pgSettings inside a transaction so the database claim survives Presigned upload/download resolvers ran multi-statement request work under withPgClient(pgSettings, cb), which applies jwt claims as transaction-LOCAL set_config without opening an explicit transaction. Each statement ran in its own autocommit transaction, so jwt.claims.database_id was gone by the next statement and jwt_private.current_database_id() raised DATABASE_CLAIM_REQUIRED on uploadAppFile. Add withRequestPgClient: acquire the client in the system lane (null pgSettings), open one transaction, apply the request settings inside it, then run the callback. Route resolveDatabaseId, bucket/file reads, single/bulk uploads, and delete-middleware reads through it. The system-lane physical_name record-write now carries the tenant database_id claim too, since the buckets catalog-sync trigger calls current_database_id(). --- .../__tests__/request-pg-client.test.ts | 175 ++++++++++++++++++ .../src/download-url-field.ts | 7 +- .../src/plugin.ts | 94 +++++----- .../src/request-pg-client.ts | 56 ++++++ 4 files changed, 287 insertions(+), 45 deletions(-) create mode 100644 graphile/graphile-presigned-url-plugin/__tests__/request-pg-client.test.ts create mode 100644 graphile/graphile-presigned-url-plugin/src/request-pg-client.ts diff --git a/graphile/graphile-presigned-url-plugin/__tests__/request-pg-client.test.ts b/graphile/graphile-presigned-url-plugin/__tests__/request-pg-client.test.ts new file mode 100644 index 000000000..7da9392af --- /dev/null +++ b/graphile/graphile-presigned-url-plugin/__tests__/request-pg-client.test.ts @@ -0,0 +1,175 @@ +/** + * Regression tests for withRequestPgClient. + * + * The bug: the grafast context's `withPgClient(pgSettings, cb)` applies the + * request's jwt claims as transaction-LOCAL `set_config(key, value, is_local => + * true)` but does NOT open an explicit transaction around `cb`. When `cb` runs + * more than one statement, each executes in its own implicit (autocommit) + * transaction, so a LOCAL setting applied for the set_config statement is gone + * by the next statement. `jwt_private.current_database_id()` then raised + * DATABASE_CLAIM_REQUIRED on presigned uploads even though the request carried + * `jwt.claims.database_id`. + * + * The fix: withRequestPgClient acquires the client WITHOUT settings, opens ONE + * explicit transaction, and applies the settings inside it — so every statement + * in `cb` observes the same claims. + * + * These tests use a small in-memory model of the node-postgres adaptor's + * transaction-LOCAL semantics so the regression needs no live database. + */ + +import { type WithPgClient,withRequestPgClient } from '../src/request-pg-client'; + +const SET_CONFIG_SQL_RE = /set_config\(el->>0, el->>1, true\)/; + +/** + * Faithful-enough model of the adaptor client around `set_config(..., is_local + * => true)`: a LOCAL setting lives for the enclosing transaction, or — with no + * open transaction — only for the single statement that set it (autocommit). + */ +class FakePg { + txDepth = 0; + calls: string[] = []; + private txLocal = new Map(); + private stmtLocal = new Map(); + + private effective(key: string): string | null { + return this.txLocal.get(key) ?? this.stmtLocal.get(key) ?? null; + } + + async query(opts: { text: string; values?: unknown[] }): Promise<{ rows: Array> }> { + this.calls.push(opts.text.replace(/\s+/g, ' ').trim()); + + if (SET_CONFIG_SQL_RE.test(opts.text)) { + const entries = JSON.parse(String(opts.values?.[0] ?? '[]')) as Array<[string, string]>; + for (const [key, value] of entries) { + if (this.txDepth > 0) this.txLocal.set(key, value); + else this.stmtLocal.set(key, value); + } + if (this.txDepth === 0) this.stmtLocal.clear(); + return { rows: [{}] }; + } + + const match = /current_setting\('([^']+)'/.exec(opts.text); + const rows: Array> = match + ? [{ v: this.effective(match[1]) }] + : [{}]; + if (this.txDepth === 0) this.stmtLocal.clear(); + return { rows }; + } + + async withTransaction(cb: (tx: FakePg) => Promise): Promise { + this.txDepth++; + this.calls.push('BEGIN'); + try { + const result = await cb(this); + this.calls.push('COMMIT'); + return result; + } catch (err) { + this.calls.push('ROLLBACK'); + throw err; + } finally { + this.txDepth--; + if (this.txDepth === 0) this.txLocal.clear(); + } + } +} + +function makeWithPgClient(pg: FakePg): { withPgClient: WithPgClient; settingsSeen: Array | null> } { + const settingsSeen: Array | null> = []; + const withPgClient = (async (pgSettings, cb) => { + settingsSeen.push(pgSettings); + // The adaptor applies request pgSettings via a LOCAL set_config in the same + // lane WITHOUT opening an explicit transaction — this is the buggy path. + if (pgSettings) { + await pg.query({ + text: 'SELECT set_config(el->>0, el->>1, true) FROM json_array_elements($1::json) el', + values: [JSON.stringify(Object.entries(pgSettings))], + }); + } + return cb(pg as never); + }) as WithPgClient; + return { withPgClient, settingsSeen }; +} + +const readClaim = "SELECT current_setting('jwt.claims.database_id', true) AS v"; + +describe('withRequestPgClient', () => { + it('reproduces the bug: a LOCAL claim is lost across autocommit statements', async () => { + const pg = new FakePg(); + const { withPgClient } = makeWithPgClient(pg); + + // Directly using withPgClient(pgSettings, cb) — no surrounding transaction. + const claim = await withPgClient({ 'jwt.claims.database_id': 'db-1' }, async (client) => { + const result = await client.query({ text: readClaim }); + return result.rows[0]?.v ?? null; + }); + + expect(claim).toBeNull(); + }); + + it('keeps the claim visible across every statement inside the transaction', async () => { + const pg = new FakePg(); + const { withPgClient } = makeWithPgClient(pg); + + const claim = await withRequestPgClient( + withPgClient, + { role: 'authenticated', 'jwt.claims.database_id': 'db-1' }, + async (tx) => { + await tx.query({ text: 'SELECT 1' }); + const result = await tx.query({ text: readClaim }); + return result.rows[0]?.v ?? null; + }, + ); + + expect(claim).toBe('db-1'); + }); + + it('acquires the client in the system lane (null pgSettings) and wraps work in one transaction', async () => { + const pg = new FakePg(); + const { withPgClient, settingsSeen } = makeWithPgClient(pg); + + await withRequestPgClient( + withPgClient, + { 'jwt.claims.database_id': 'db-1' }, + async (tx) => tx.query({ text: readClaim }), + ); + + // The helper must NOT pass request pgSettings to withPgClient — it applies + // them itself inside the transaction. + expect(settingsSeen).toEqual([null]); + + // BEGIN → set_config → work → COMMIT, in that order. + expect(pg.calls[0]).toBe('BEGIN'); + expect(pg.calls[1]).toMatch(SET_CONFIG_SQL_RE); + expect(pg.calls[2]).toBe(readClaim); + expect(pg.calls[pg.calls.length - 1]).toBe('COMMIT'); + }); + + it('opens a transaction but issues no set_config when there are no settings', async () => { + for (const settings of [null, {}] as Array | null>) { + const pg = new FakePg(); + const { withPgClient } = makeWithPgClient(pg); + + await withRequestPgClient(withPgClient, settings, async (tx) => tx.query({ text: 'SELECT 1' })); + + expect(pg.calls).toContain('BEGIN'); + expect(pg.calls).toContain('COMMIT'); + expect(pg.calls.filter((c) => SET_CONFIG_SQL_RE.test(c))).toHaveLength(0); + } + }); + + it('propagates callback errors (never swallows) and rolls back', async () => { + const pg = new FakePg(); + const { withPgClient } = makeWithPgClient(pg); + + await expect( + withRequestPgClient(withPgClient, { 'jwt.claims.database_id': 'db-1' }, async () => { + throw new Error('boom'); + }), + ).rejects.toThrow('boom'); + + expect(pg.calls).toContain('ROLLBACK'); + expect(pg.calls).not.toContain('COMMIT'); + }); +}); diff --git a/graphile/graphile-presigned-url-plugin/src/download-url-field.ts b/graphile/graphile-presigned-url-plugin/src/download-url-field.ts index 5d0d334f6..b0d2bc9a7 100644 --- a/graphile/graphile-presigned-url-plugin/src/download-url-field.ts +++ b/graphile/graphile-presigned-url-plugin/src/download-url-field.ts @@ -25,6 +25,7 @@ import { Logger } from '@pgpmjs/logger'; import { context as grafastContext, lambda, object } from 'grafast'; import type { GraphileConfig } from 'graphile-config'; +import { withRequestPgClient } from './request-pg-client'; import { generatePresignedGetUrl } from './s3-signer'; import { loadAllStorageModules, resolveStorageConfigFromCodec, storedPhysicalName } from './storage-module-cache'; import type { PresignedUrlPluginOptions, S3Config, StorageModuleConfig } from './types'; @@ -149,11 +150,11 @@ export function createDownloadUrlPlugin( let downloadUrlExpirySeconds = 3600; try { if (withPgClient && pgSettings) { - const databaseId = await withPgClient(pgSettings, async (pgClient: any) => { + const databaseId = await withRequestPgClient(withPgClient, pgSettings, async (pgClient) => { const dbResult = await pgClient.query({ text: `SELECT jwt_private.current_database_id() AS id`, }); - return dbResult.rows[0]?.id ?? null; + return (dbResult.rows[0]?.id as string | undefined) ?? null; }); // Module registration is server config, not user data: // resolve it without the request role's pgSettings. @@ -164,7 +165,7 @@ export function createDownloadUrlPlugin( ) : null; const resolved = config && bucketId - ? await withPgClient(pgSettings, async (pgClient: any) => { + ? await withRequestPgClient(withPgClient, pgSettings, async (pgClient) => { // Look up the stored physical coordinate for scoped S3 resolution const bucketResult = await pgClient.query({ text: `SELECT key, physical_name FROM ${config.bucketsQualifiedName} WHERE id = $1 LIMIT 1`, diff --git a/graphile/graphile-presigned-url-plugin/src/plugin.ts b/graphile/graphile-presigned-url-plugin/src/plugin.ts index cea4da585..6d27b29b1 100644 --- a/graphile/graphile-presigned-url-plugin/src/plugin.ts +++ b/graphile/graphile-presigned-url-plugin/src/plugin.ts @@ -23,6 +23,7 @@ import { Logger } from '@pgpmjs/logger'; import { access, context as grafastContext, lambda, object } from 'grafast'; import type { GraphileConfig } from 'graphile-config'; +import { type WithPgClient,withRequestPgClient } from './request-pg-client'; import { deleteS3Object,generatePresignedPutUrl } from './s3-signer'; import { getBucketConfig, isS3BucketProvisioned, loadAllStorageModules, markS3BucketProvisioned,resolveStorageConfigFromCodec, storedPhysicalName } from './storage-module-cache'; import type { BucketConfig,PresignedUrlPluginOptions, S3Config, StorageModuleConfig } from './types'; @@ -143,14 +144,19 @@ function resolveS3ForDatabase( * value is the durable coordinate: route resolution and every later read use * it verbatim; nothing is recomputed. * - * The record write runs in the system lane (`withPgClient(null, ...)`) — it is - * server bookkeeping, not request data, and anonymous request roles cannot - * UPDATE bucket rows under RLS. `bucket` (the cached config) is mutated in place - * so subsequent reads observe the recorded name without a DB round-trip. + * The record write runs in the system lane (privileged role, so it bypasses the + * RLS that stops request roles from UPDATE-ing bucket rows) — it is server + * bookkeeping, not request data. It still carries the tenant `database_id` + * claim, because the buckets table's catalog-sync trigger calls + * `jwt_private.current_database_id()` and would otherwise raise + * DATABASE_CLAIM_REQUIRED; `withRequestPgClient` applies that claim inside the + * write's transaction without switching off the privileged role. + * `bucket` (the cached config) is mutated in place so subsequent reads observe + * the recorded name without a DB round-trip. */ async function provisionAndRecordPhysicalBucket( options: PresignedUrlPluginOptions, - withPgClient: (pgSettings: null, cb: (client: any) => Promise) => Promise, + withPgClient: WithPgClient, storageConfig: StorageModuleConfig, databaseId: string, bucket: BucketConfig, @@ -167,7 +173,9 @@ async function provisionAndRecordPhysicalBucket( // Record the physical coordinate on the source row. The `physical_name IS NULL` // guard keeps this idempotent and race-safe across concurrent first uploads. - await withPgClient(null, (client: any) => + // The catalog-sync trigger on this UPDATE needs `jwt.claims.database_id`, so the + // write runs under the resolved database claim (privileged role preserved). + await withRequestPgClient(withPgClient, { 'jwt.claims.database_id': databaseId }, (client) => client.query({ text: `UPDATE ${storageConfig.bucketsQualifiedName} SET physical_name = $1 @@ -322,7 +330,10 @@ export function createPresignedUrlPlugin( }); return lambda($combined, async (vals: any) => { - const databaseId = await vals.withPgClient(vals.pgSettings, (pgClient: any) => + // Request-lane reads/writes run under the request role's pgSettings + // inside an explicit transaction so the jwt claims stay applied + // across every statement (see withRequestPgClient). + const databaseId = await withRequestPgClient(vals.withPgClient, vals.pgSettings, (pgClient) => resolveDatabaseId(pgClient), ); if (!databaseId) throw new Error('DATABASE_NOT_FOUND'); @@ -336,7 +347,7 @@ export function createPresignedUrlPlugin( if (!storageConfig) throw new Error('STORAGE_MODULE_NOT_FOUND'); // Bucket config read under the request role (RLS-gated visibility). - const bucket = await vals.withPgClient(vals.pgSettings, (pgClient: any) => + const bucket = await withRequestPgClient(vals.withPgClient, vals.pgSettings, (pgClient) => getBucketConfig(pgClient, storageConfig, databaseId, vals.bucketKey, vals.ownerId || undefined), ); if (!bucket) throw new Error('BUCKET_NOT_FOUND'); @@ -349,16 +360,14 @@ export function createPresignedUrlPlugin( const s3ForDb = resolveS3ForDatabase(options, storageConfig, physicalName); // File row INSERT under the request role (RLS enforced). - return vals.withPgClient(vals.pgSettings, (pgClient: any) => - pgClient.withTransaction((txClient: any) => - processSingleFile(options, txClient, storageConfig, databaseId, bucket, s3ForDb, { - contentHash: vals.contentHash, - contentType: vals.contentType, - size: vals.size, - filename: vals.filename, - key: vals.customKey, - }), - ), + return withRequestPgClient(vals.withPgClient, vals.pgSettings, (txClient) => + processSingleFile(options, txClient, storageConfig, databaseId, bucket, s3ForDb, { + contentHash: vals.contentHash, + contentType: vals.contentType, + size: vals.size, + filename: vals.filename, + key: vals.customKey, + }), ); }); }, @@ -435,7 +444,10 @@ export function createPresignedUrlPlugin( }); return lambda($combined, async (vals: any) => { - const databaseId = await vals.withPgClient(vals.pgSettings, (pgClient: any) => + // Request-lane reads/writes run under the request role's pgSettings + // inside an explicit transaction so the jwt claims stay applied + // across every statement (see withRequestPgClient). + const databaseId = await withRequestPgClient(vals.withPgClient, vals.pgSettings, (pgClient) => resolveDatabaseId(pgClient), ); if (!databaseId) throw new Error('DATABASE_NOT_FOUND'); @@ -449,7 +461,7 @@ export function createPresignedUrlPlugin( if (!storageConfig) throw new Error('STORAGE_MODULE_NOT_FOUND'); // Bucket config read under the request role (RLS-gated visibility). - const bucket = await vals.withPgClient(vals.pgSettings, (pgClient: any) => + const bucket = await withRequestPgClient(vals.withPgClient, vals.pgSettings, (pgClient) => getBucketConfig(pgClient, storageConfig, databaseId, vals.bucketKey, vals.ownerId || undefined), ); if (!bucket) throw new Error('BUCKET_NOT_FOUND'); @@ -476,23 +488,21 @@ export function createPresignedUrlPlugin( const s3ForDb = resolveS3ForDatabase(options, storageConfig, physicalName); // File row INSERTs under the request role (RLS enforced). - return vals.withPgClient(vals.pgSettings, (pgClient: any) => - pgClient.withTransaction(async (txClient: any) => { - const results = []; - for (const file of filesArray) { - results.push( - await processSingleFile(options, txClient, storageConfig, databaseId, bucket, s3ForDb, { - contentHash: file.contentHash, - contentType: file.contentType, - size: file.size, - filename: file.filename, - key: file.key, - }), - ); - } - return { files: results }; - }), - ); + return withRequestPgClient(vals.withPgClient, vals.pgSettings, async (txClient) => { + const results = []; + for (const file of filesArray) { + results.push( + await processSingleFile(options, txClient, storageConfig, databaseId, bucket, s3ForDb, { + contentHash: file.contentHash, + contentType: file.contentType, + size: file.size, + filename: file.filename, + key: file.key, + }), + ); + } + return { files: results }; + }); }); }, }, @@ -557,7 +567,7 @@ export function createPresignedUrlPlugin( if (withPgClient) { try { - const databaseId = await withPgClient(pgSettings, (pgClient: any) => resolveDatabaseId(pgClient)); + const databaseId = await withRequestPgClient(withPgClient, pgSettings, (pgClient) => resolveDatabaseId(pgClient)); // Module registration is server config, not user data: // resolve it without the request role's pgSettings. const allConfigs = databaseId @@ -566,7 +576,7 @@ export function createPresignedUrlPlugin( const storageConfig = resolveStorageConfigFromCodec(capturedCodec, allConfigs); if (storageConfig) { - await withPgClient(pgSettings, async (pgClient: any) => { + await withRequestPgClient(withPgClient, pgSettings, async (pgClient) => { // Read the file row (RLS enforced) const result = await pgClient.query({ text: `SELECT key, bucket_id FROM ${storageConfig.filesQualifiedName} WHERE id = $1 LIMIT 1`, @@ -593,7 +603,7 @@ export function createPresignedUrlPlugin( if (withPgClient) { try { - const databaseId = await withPgClient(pgSettings, (pgClient: any) => resolveDatabaseId(pgClient)); + const databaseId = await withRequestPgClient(withPgClient, pgSettings, (pgClient) => resolveDatabaseId(pgClient)); // Module registration is server config, not user data: // resolve it without the request role's pgSettings. const allConfigs = databaseId @@ -601,13 +611,13 @@ export function createPresignedUrlPlugin( : []; const storageConfig = resolveStorageConfigFromCodec(capturedCodec, allConfigs); - if (storageConfig) await withPgClient(pgSettings, async (pgClient: any) => { + if (storageConfig) await withRequestPgClient(withPgClient, pgSettings, async (pgClient) => { // Check refcount: any other file with the same key in this bucket? const refResult = await pgClient.query({ text: `SELECT COUNT(*)::int AS ref_count FROM ${storageConfig.filesQualifiedName} WHERE key = $1 AND bucket_id = $2`, values: [fileRow!.key, fileRow!.bucket_id], }); - const refCount = refResult.rows[0]?.ref_count ?? 0; + const refCount = (refResult.rows[0]?.ref_count as number | undefined) ?? 0; if (refCount > 0) { log.info(`File deleted from DB; S3 key ${fileRow!.key} still referenced by ${refCount} file(s)`); diff --git a/graphile/graphile-presigned-url-plugin/src/request-pg-client.ts b/graphile/graphile-presigned-url-plugin/src/request-pg-client.ts new file mode 100644 index 000000000..62f8836a6 --- /dev/null +++ b/graphile/graphile-presigned-url-plugin/src/request-pg-client.ts @@ -0,0 +1,56 @@ +/** + * Run a callback under the request's pgSettings inside ONE explicit transaction. + * + * The grafast context's `withPgClient(pgSettings, cb)` applies pgSettings as + * transaction-LOCAL `set_config(key, value, is_local => true)`. When `cb` issues + * more than one statement WITHOUT an explicit surrounding transaction, each + * statement runs in its own implicit (autocommit) transaction, so a LOCAL + * setting applied for one statement is already gone by the next. The request + * role's jwt claims (notably `jwt.claims.database_id`) then vanish between + * statements, and `jwt_private.current_database_id()` raises + * DATABASE_CLAIM_REQUIRED even though the request carried the claim. + * + * This helper acquires the client without settings (`withPgClient(null, ...)`), + * opens a single explicit transaction, and applies the request settings inside + * it — so every statement in `cb` observes the same role and jwt claims. It + * mirrors the `pg-query-context` pattern used elsewhere in the codebase for + * manual, multi-statement RLS work. + */ + +export interface RequestPgClient { + query(opts: { text: string; values?: unknown[] }): Promise<{ rows: Array> }>; + withTransaction(cb: (tx: RequestPgClient) => Promise): Promise; +} + +export type WithPgClient = ( + pgSettings: Record | null, + cb: (client: RequestPgClient) => Promise, +) => Promise; + +async function applyRequestSettings( + tx: RequestPgClient, + pgSettings: Record | null, +): Promise { + if (!pgSettings) return; + const entries = Object.entries(pgSettings) + .filter(([, value]) => value != null) + .map(([key, value]) => [key, String(value)]); + if (entries.length === 0) return; + await tx.query({ + text: 'SELECT set_config(el->>0, el->>1, true) FROM json_array_elements($1::json) el', + values: [JSON.stringify(entries)], + }); +} + +export function withRequestPgClient( + withPgClient: WithPgClient, + pgSettings: Record | null, + cb: (tx: RequestPgClient) => Promise, +): Promise { + return withPgClient(null, (client) => + client.withTransaction(async (tx) => { + await applyRequestSettings(tx, pgSettings); + return cb(tx); + }), + ); +}