diff --git a/graphile/graphile-bulk-mutations/src/__tests__/pg-client.test.ts b/graphile/graphile-bulk-mutations/src/__tests__/pg-client.test.ts new file mode 100644 index 0000000000..311bc55a8a --- /dev/null +++ b/graphile/graphile-bulk-mutations/src/__tests__/pg-client.test.ts @@ -0,0 +1,35 @@ +import { queryPgClient } from '../utils/pg-client'; + +describe('queryPgClient', () => { + it('uses the native @dataplan/pg query-config contract', async () => { + const query = jest.fn(async () => ({ rows: [{ id: 1 }], rowCount: 1 })); + const client = { query }; + + await expect( + queryPgClient<{ id: number }>( + client as never, + 'UPDATE app.items SET name = $1 RETURNING id', + ['updated'] + ) + ).resolves.toEqual({ rows: [{ id: 1 }], rowCount: 1 }); + + expect(query).toHaveBeenCalledTimes(1); + expect(query).toHaveBeenCalledWith({ + text: 'UPDATE app.items SET name = $1 RETURNING id', + values: ['updated'], + }); + }); + + it('preserves query failures', async () => { + const original = new Error('database rejected mutation'); + const client = { + query: jest.fn(async () => { + throw original; + }), + }; + + await expect( + queryPgClient(client as never, 'DELETE FROM app.items', []) + ).rejects.toBe(original); + }); +}); diff --git a/graphile/graphile-bulk-mutations/src/plugins/BulkDeletePlugin.ts b/graphile/graphile-bulk-mutations/src/plugins/BulkDeletePlugin.ts index 4731702283..17874d4757 100644 --- a/graphile/graphile-bulk-mutations/src/plugins/BulkDeletePlugin.ts +++ b/graphile/graphile-bulk-mutations/src/plugins/BulkDeletePlugin.ts @@ -1,9 +1,11 @@ import '../augmentations'; -import { sideEffectWithPgClient } from '@dataplan/pg'; +import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg'; import type { GraphileConfig } from 'graphile-config'; import type { GraphQLInputType,GraphQLOutputType } from 'graphql'; +import { queryPgClient } from '../utils/pg-client'; + const version = '0.1.0'; /** @@ -105,7 +107,7 @@ export const BulkDeletePlugin: GraphileConfig.Plugin = { const $result = sideEffectWithPgClient( executor, $input, - async (pgClient: any, input: any) => { + async (pgClient: PgClient, input: any) => { if (requireWhere && (!input.where || Object.keys(input.where).length === 0)) { throw new Error( 'Bulk delete requires a non-empty where condition. Set bulkRequireWhere: false to allow unrestricted deletes.' @@ -194,12 +196,16 @@ export const BulkDeletePlugin: GraphileConfig.Plugin = { // Use RETURNING instead of RETURNING * // For delete, we capture PKs before rows are gone const text = `DELETE FROM ${compiledFrom}\nWHERE ${whereStr}\nRETURNING ${pkReturning}`; - const mutationResult = await pgClient.query(text, values); + const mutationResult = await queryPgClient>( + pgClient, + text, + values + ); const affectedCount = mutationResult.rowCount ?? 0; // For delete, rows no longer exist so we can't do a // follow-up SELECT. Return the PK values directly. - const returning = mutationResult.rows || []; + const returning = [...mutationResult.rows]; return { affectedCount, diff --git a/graphile/graphile-bulk-mutations/src/plugins/BulkInsertPlugin.ts b/graphile/graphile-bulk-mutations/src/plugins/BulkInsertPlugin.ts index 6b128024a1..9ef154e365 100644 --- a/graphile/graphile-bulk-mutations/src/plugins/BulkInsertPlugin.ts +++ b/graphile/graphile-bulk-mutations/src/plugins/BulkInsertPlugin.ts @@ -1,9 +1,10 @@ import '../augmentations'; -import { sideEffectWithPgClient } from '@dataplan/pg'; +import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg'; import type { GraphileConfig } from 'graphile-config'; import type { GraphQLInputType, GraphQLOutputType } from 'graphql'; +import { queryPgClient } from '../utils/pg-client'; import type { NestedRelationInfo } from '../utils/relations'; import { discoverNestedRelations } from '../utils/relations'; import type { ColumnSpec } from '../utils/sql-builder'; @@ -131,7 +132,7 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = { const $result = sideEffectWithPgClient( executor, $input, - async (pgClient: any, input: any) => { + async (pgClient: PgClient, input: any) => { const values = input.values; if (!values || !Array.isArray(values) || values.length === 0) { return { affectedCount: 0, returning: [] }; @@ -200,10 +201,9 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = { const allPkRows: Record[] = []; for (const batch of batches) { - const result = await pgClient.query( - batch.text, - batch.values - ); + const result = await queryPgClient< + Record + >(pgClient, batch.text, batch.values); totalAffected += result.rowCount ?? 0; if (result.rows) { allPkRows.push(...result.rows); @@ -259,7 +259,8 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = { ); for (const batch of childBatches) { - const result = await pgClient.query( + const result = await queryPgClient( + pgClient, batch.text, batch.values ); @@ -282,11 +283,12 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = { const selectParams = allPkRows.flatMap((pkRow) => pkColumns.map((col) => pkRow[col]) ); - const selectResult = await pgClient.query( + const selectResult = await queryPgClient( + pgClient, `SELECT * FROM ${compiledFrom} WHERE ${whereClause}`, selectParams ); - returning = selectResult.rows || []; + returning = [...selectResult.rows]; } return { diff --git a/graphile/graphile-bulk-mutations/src/plugins/BulkUpdatePlugin.ts b/graphile/graphile-bulk-mutations/src/plugins/BulkUpdatePlugin.ts index 8ef1170067..4c57533e31 100644 --- a/graphile/graphile-bulk-mutations/src/plugins/BulkUpdatePlugin.ts +++ b/graphile/graphile-bulk-mutations/src/plugins/BulkUpdatePlugin.ts @@ -1,9 +1,11 @@ import '../augmentations'; -import { sideEffectWithPgClient } from '@dataplan/pg'; +import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg'; import type { GraphileConfig } from 'graphile-config'; import type { GraphQLInputType, GraphQLOutputType } from 'graphql'; +import { queryPgClient } from '../utils/pg-client'; + const version = '0.1.0'; /** @@ -106,7 +108,7 @@ export const BulkUpdatePlugin: GraphileConfig.Plugin = { const $result = sideEffectWithPgClient( executor, $input, - async (pgClient: any, input: any) => { + async (pgClient: PgClient, input: any) => { if (requireWhere && (!input.where || Object.keys(input.where).length === 0)) { throw new Error( 'Bulk update requires a non-empty where condition. Set bulkRequireWhere: false to allow unrestricted updates.' @@ -212,13 +214,15 @@ export const BulkUpdatePlugin: GraphileConfig.Plugin = { // Use RETURNING instead of RETURNING * const text = `UPDATE ${compiledFrom}\nSET ${setClauses.join(', ')}\nWHERE ${whereStr}\nRETURNING ${pkReturning}`; - const mutationResult = await pgClient.query(text, values); + const mutationResult = await queryPgClient< + Record + >(pgClient, text, values); const affectedCount = mutationResult.rowCount ?? 0; // Follow-up SELECT using PKs to respect column-level grants let returning: unknown[] = []; if (mutationResult.rows && mutationResult.rows.length > 0) { - const pkRows: Record[] = mutationResult.rows; + const pkRows = mutationResult.rows; const pkConditions = pkRows.map((pkRow, rowIdx) => { return pkColumns.map((col, colIdx) => { const paramIdx = rowIdx * pkColumns.length + colIdx + 1; @@ -229,11 +233,12 @@ export const BulkUpdatePlugin: GraphileConfig.Plugin = { const selectParams = pkRows.flatMap((pkRow) => pkColumns.map((col) => pkRow[col]) ); - const selectResult = await pgClient.query( + const selectResult = await queryPgClient( + pgClient, `SELECT * FROM ${compiledFrom} WHERE ${selectWhere}`, selectParams ); - returning = selectResult.rows || []; + returning = [...selectResult.rows]; } return { diff --git a/graphile/graphile-bulk-mutations/src/plugins/BulkUpsertPlugin.ts b/graphile/graphile-bulk-mutations/src/plugins/BulkUpsertPlugin.ts index 4b261c7562..1996f48e89 100644 --- a/graphile/graphile-bulk-mutations/src/plugins/BulkUpsertPlugin.ts +++ b/graphile/graphile-bulk-mutations/src/plugins/BulkUpsertPlugin.ts @@ -1,9 +1,10 @@ import '../augmentations'; -import { sideEffectWithPgClient } from '@dataplan/pg'; +import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg'; import type { GraphileConfig } from 'graphile-config'; import type { GraphQLInputType, GraphQLOutputType } from 'graphql'; +import { queryPgClient } from '../utils/pg-client'; import type { ColumnSpec } from '../utils/sql-builder'; import { buildBulkInsertSQL } from '../utils/sql-builder'; @@ -118,7 +119,7 @@ export const BulkUpsertPlugin: GraphileConfig.Plugin = { const $result = sideEffectWithPgClient( executor, $input, - async (pgClient: any, input: any) => { + async (pgClient: PgClient, input: any) => { const values = input.values; if (!values || !Array.isArray(values) || values.length === 0) { return { affectedCount: 0, returning: [] }; @@ -177,10 +178,9 @@ export const BulkUpsertPlugin: GraphileConfig.Plugin = { const allPkRows: Record[] = []; for (const batch of batches) { - const result = await pgClient.query( - batch.text, - batch.values - ); + const result = await queryPgClient< + Record + >(pgClient, batch.text, batch.values); totalAffected += result.rowCount ?? 0; if (result.rows) { allPkRows.push(...result.rows); @@ -200,11 +200,12 @@ export const BulkUpsertPlugin: GraphileConfig.Plugin = { const selectParams = allPkRows.flatMap((pkRow) => pkColumns.map((col) => pkRow[col]) ); - const selectResult = await pgClient.query( + const selectResult = await queryPgClient( + pgClient, `SELECT * FROM ${compiledFrom} WHERE ${whereClause}`, selectParams ); - returning = selectResult.rows || []; + returning = [...selectResult.rows]; } return { diff --git a/graphile/graphile-bulk-mutations/src/utils/pg-client.ts b/graphile/graphile-bulk-mutations/src/utils/pg-client.ts new file mode 100644 index 0000000000..306d599d89 --- /dev/null +++ b/graphile/graphile-bulk-mutations/src/utils/pg-client.ts @@ -0,0 +1,10 @@ +import type { PgClient, PgClientResult } from '@dataplan/pg'; + +/** Execute SQL using @dataplan/pg's native query-config contract. */ +export function queryPgClient( + client: Pick, + text: string, + values: any[] +): Promise> { + return client.query({ text, values }); +} diff --git a/graphile/graphile-i18n/package.json b/graphile/graphile-i18n/package.json index 1b0f8f5bee..e152352d73 100644 --- a/graphile/graphile-i18n/package.json +++ b/graphile/graphile-i18n/package.json @@ -29,6 +29,7 @@ "url": "https://github.com/constructive-io/constructive/issues" }, "dependencies": { + "@constructive-io/express-context": "workspace:^", "accept-language-parser": "^1.5.0" }, "peerDependencies": { diff --git a/graphile/graphile-i18n/src/__tests__/i18n.test.ts b/graphile/graphile-i18n/src/__tests__/i18n.test.ts index 4435813e32..58811f96c0 100644 --- a/graphile/graphile-i18n/src/__tests__/i18n.test.ts +++ b/graphile/graphile-i18n/src/__tests__/i18n.test.ts @@ -8,6 +8,7 @@ * - Fallback to base table values when no translation exists */ +import { buildPgSettings } from '@constructive-io/express-context'; import type { GraphQLResponse } from 'graphile-test'; import { getConnections, seed } from 'graphile-test'; import { join } from 'path'; @@ -70,7 +71,28 @@ describe('graphile-i18n plugin', () => { db = connections.db; teardown = connections.teardown; - query = connections.query; + const baseQuery: QueryFn = connections.query; + const canonicalSettings = buildPgSettings({ + api: { + apiId: 'i18n-test-api', + databaseId: 'i18n-test-database', + dbname: 'i18n_test', + anonRole: 'postgres', + roleName: 'postgres', + schema: ['i18n_test'], + }, + token: { user_id: 'i18n-test-user' }, + requestId: 'i18n-test-request', + }); + query = (document, variables, commit, reqOptions = {}) => + baseQuery(document, variables, commit, { + ...reqOptions, + pgSettings: { + ...canonicalSettings, + ...((reqOptions.pgSettings as Record | undefined) ?? + {}), + }, + }); }); afterAll(async () => { diff --git a/graphile/graphile-i18n/src/__tests__/pg-query.test.ts b/graphile/graphile-i18n/src/__tests__/pg-query.test.ts new file mode 100644 index 0000000000..ba4c2b9ada --- /dev/null +++ b/graphile/graphile-i18n/src/__tests__/pg-query.test.ts @@ -0,0 +1,79 @@ +import { buildPgSettings } from '@constructive-io/express-context'; + +import { queryI18nWithContext } from '../pg-query'; + +const pgSettings = buildPgSettings({ + api: { + apiId: 'api-1', + databaseId: 'database-1', + dbname: 'testdb', + anonRole: 'anonymous_runtime', + roleName: 'authenticated_runtime', + schema: ['i18n_test'], + }, + token: { user_id: 'user-1' }, + requestId: 'request-1', +}); + +describe('queryI18nWithContext', () => { + it('passes the complete settings unchanged and uses the native query contract', async () => { + const original = { ...pgSettings }; + const query = jest.fn(async () => ({ + rows: [{ lang_code: 'en', title: 'Hello' }], + })); + const withPgClient = jest.fn(async (settings, callback) => + callback({ query }) + ); + + await expect( + queryI18nWithContext( + withPgClient, + pgSettings, + 1, + 'SELECT translation WHERE id = $1 AND lang = ANY($2)', + [1, ['en']] + ) + ).resolves.toEqual({ lang_code: 'en', title: 'Hello' }); + + expect(withPgClient).toHaveBeenCalledWith(pgSettings, expect.any(Function)); + expect(withPgClient.mock.calls[0][0]).not.toBeNull(); + expect(query).toHaveBeenCalledWith({ + text: 'SELECT translation WHERE id = $1 AND lang = ANY($2)', + values: [1, ['en']], + }); + expect(pgSettings).toEqual(original); + }); + + it.each([ + [ + 'missing withPgClient', + undefined, + pgSettings, + 1, + 'I18N_PG_CLIENT_CONTEXT_UNAVAILABLE', + ], + ['missing pgSettings', jest.fn(), undefined, 1, 'i18n pgSettings'], + [ + 'incomplete pgSettings', + jest.fn(), + { role: 'anonymous_runtime' }, + 1, + 'i18n pgSettings', + ], + ])( + 'fails closed for %s', + async (_label, withPgClient, settings, id, message) => { + await expect( + queryI18nWithContext(withPgClient, settings, id, 'SELECT 1', []) + ).rejects.toThrow(message); + } + ); + + it('preserves the base-row fallback when the parent id is unavailable', async () => { + const withPgClient = jest.fn(); + await expect( + queryI18nWithContext(withPgClient, pgSettings, null, 'SELECT 1', []) + ).resolves.toBeNull(); + expect(withPgClient).not.toHaveBeenCalled(); + }); +}); diff --git a/graphile/graphile-i18n/src/pg-query.ts b/graphile/graphile-i18n/src/pg-query.ts new file mode 100644 index 0000000000..c29df27358 --- /dev/null +++ b/graphile/graphile-i18n/src/pg-query.ts @@ -0,0 +1,36 @@ +import { + assertCompletePgSettings, + type PgSettings, +} from '@constructive-io/express-context'; +import type { PgClient } from '@dataplan/pg'; + +export type GraphileWithPgClient = ( + pgSettings: PgSettings, + callback: (client: PgClient) => Promise +) => Promise; + +export async function queryI18nWithContext( + withPgClient: unknown, + pgSettings: unknown, + id: unknown, + text: string, + values: any[] +): Promise | null> { + if (typeof withPgClient !== 'function') { + throw new Error('I18N_PG_CLIENT_CONTEXT_UNAVAILABLE'); + } + assertCompletePgSettings(pgSettings, 'i18n pgSettings'); + if (id === null || id === undefined) { + // Preserve the plugin's existing base-row fallback when the parent has no + // usable key; there is no request-lane SQL to authorize in this case. + return null; + } + + return (withPgClient as GraphileWithPgClient)(pgSettings, async (client) => { + const { rows } = await client.query>({ + text, + values, + }); + return rows[0] ?? null; + }); +} diff --git a/graphile/graphile-i18n/src/plugin.ts b/graphile/graphile-i18n/src/plugin.ts index 0fb1af80d6..408bc76fe5 100644 --- a/graphile/graphile-i18n/src/plugin.ts +++ b/graphile/graphile-i18n/src/plugin.ts @@ -25,6 +25,7 @@ import { TYPES } from '@dataplan/pg'; import { context as grafastContext, lambda, object } from 'grafast'; import type { GraphileConfig } from 'graphile-config'; +import { queryI18nWithContext } from './pg-query'; import type { I18nPluginOptions, I18nTableInfo, TranslatableField } from './types'; // ─── Namespace Augmentations ───────────────────────────────────────────────── @@ -266,32 +267,29 @@ export function createI18nPlugin(options: I18nPluginOptions = {}): GraphileConfi $baseCols[column] = $parent.get(column); } const $withPgClient = (grafastContext() as any).get('withPgClient'); + const $pgSettings = (grafastContext() as any).get('pgSettings'); const $langCodes = (grafastContext() as any).get('langCodes'); // Combine all inputs into a single step const $input = object({ id: $id, withPgClient: $withPgClient, + pgSettings: $pgSettings, langCodes: $langCodes, ...$baseCols, }); return lambda($input, async (input: any) => { - const { id, withPgClient, langCodes: ctxLangCodes, ...baseCols } = input; + const { id, withPgClient, pgSettings, langCodes: ctxLangCodes, ...baseCols } = input; const langs: string[] = ctxLangCodes ?? defaultLanguages; - if (!withPgClient || !id) { - const result: Record = { [langCodeGqlField]: null }; - for (const { gqlName, column } of baseColNames) { - result[gqlName] = baseCols[column] ?? null; - } - return result; - } - - const row = await withPgClient(null, async (client: any) => { - const { rows } = await client.query(sqlQuery, [id, langs]); - return rows[0] ?? null; - }); + const row = await queryI18nWithContext( + withPgClient, + pgSettings, + id, + sqlQuery, + [id, langs] + ); if (!row) { const result: Record = { [langCodeGqlField]: null }; diff --git a/graphile/graphile-llm/src/__tests__/graphile-llm.test.ts b/graphile/graphile-llm/src/__tests__/graphile-llm.test.ts index 30f2b58439..8248e28aec 100644 --- a/graphile/graphile-llm/src/__tests__/graphile-llm.test.ts +++ b/graphile/graphile-llm/src/__tests__/graphile-llm.test.ts @@ -1,4 +1,5 @@ import OllamaClient from '@agentic-kit/ollama'; +import { buildPgSettings } from '@constructive-io/express-context'; import type { GraphileConfig } from 'graphile-config'; import { ConnectionFilterPreset } from 'graphile-connection-filter'; import { createPgvectorAdapter } from 'graphile-search/adapters/pgvector'; @@ -45,6 +46,8 @@ async function ensureNomicModel(): Promise { type QueryFn = ( query: string, variables?: Record, + commit?: boolean, + reqOptions?: Record ) => Promise>; // ============================================================================= @@ -173,7 +176,29 @@ describe('graphile-llm schema enrichment', () => { db = connections.db; teardown = connections.teardown; - query = connections.query; + const baseQuery: QueryFn = connections.query; + const canonicalSettings = buildPgSettings({ + api: { + apiId: 'llm-rag-test-api', + databaseId: 'llm-rag-test-database', + dbname: 'llm_test', + anonRole: 'postgres', + roleName: 'postgres', + schema: ['llm_test'], + }, + token: { user_id: 'llm-rag-test-user' }, + requestId: 'llm-rag-test-request', + }); + query = ( + document: string, + variables?: Record, + commit?: boolean, + reqOptions: Record = {} + ) => + baseQuery(document, variables, commit, { + ...reqOptions, + pgSettings: canonicalSettings, + }); }); afterAll(async () => { @@ -706,7 +731,29 @@ describe('RAG plugin schema enrichment', () => { db = connections.db; teardown = connections.teardown; - query = connections.query; + const baseQuery: QueryFn = connections.query; + const canonicalSettings = buildPgSettings({ + api: { + apiId: 'llm-rag-test-api', + databaseId: 'llm-rag-test-database', + dbname: 'llm_test', + anonRole: 'postgres', + roleName: 'postgres', + schema: ['llm_test'], + }, + token: { user_id: 'llm-rag-test-user' }, + requestId: 'llm-rag-test-request', + }); + query = ( + document: string, + variables?: Record, + commit?: boolean, + reqOptions: Record = {} + ) => + baseQuery(document, variables, commit, { + ...reqOptions, + pgSettings: canonicalSettings, + }); }); afterAll(async () => { diff --git a/graphile/graphile-llm/src/__tests__/request-context.test.ts b/graphile/graphile-llm/src/__tests__/request-context.test.ts new file mode 100644 index 0000000000..34c483b439 --- /dev/null +++ b/graphile/graphile-llm/src/__tests__/request-context.test.ts @@ -0,0 +1,169 @@ +import { buildPgSettings } from '@constructive-io/express-context'; + +import { + getLlmBillingConfig, + invalidateLlmBillingConfig, +} from '../config-cache'; +import { buildMeteringContext } from '../plugins/metering-plugin'; +import { withGraphileRequestPgClient } from '../request-context'; + +const api = { + apiId: 'api-1', + databaseId: 'database-1', + dbname: 'tenant_db', + anonRole: 'anonymous_runtime', + roleName: 'authenticated_runtime', + schema: ['app_public'], +}; + +const pgSettings = buildPgSettings({ + api, + token: { user_id: 'user-1' }, + requestId: 'request-1', +}); + +describe('graphile-llm request context', () => { + afterEach(() => invalidateLlmBillingConfig()); + + it('passes complete settings unchanged and uses a native PgClient callback', async () => { + const original = { ...pgSettings }; + const query = jest.fn(async () => ({ rows: [{ ok: true }], rowCount: 1 })); + const withPgClient = jest.fn(async (settings, callback) => + callback({ query }) + ); + + await expect( + withGraphileRequestPgClient( + withPgClient, + pgSettings, + async (client) => + client.query({ text: 'SELECT $1::text', values: ['ok'] }), + 'RAG' + ) + ).resolves.toMatchObject({ rows: [{ ok: true }] }); + + expect(withPgClient).toHaveBeenCalledWith(pgSettings, expect.any(Function)); + expect(query).toHaveBeenCalledWith({ + text: 'SELECT $1::text', + values: ['ok'], + }); + expect(pgSettings).toEqual(original); + }); + + it.each([ + [ + 'missing withPgClient', + undefined, + pgSettings, + 'RAG_PG_CLIENT_CONTEXT_UNAVAILABLE', + ], + ['missing pgSettings', jest.fn(), undefined, 'RAG pgSettings'], + [ + 'incomplete pgSettings', + jest.fn(), + { role: 'anonymous_runtime' }, + 'RAG pgSettings', + ], + ])('fails closed for %s', async (_label, withPgClient, settings, message) => { + await expect( + withGraphileRequestPgClient( + withPgClient, + settings, + async (): Promise => undefined, + 'RAG' + ) + ).rejects.toThrow(message); + }); + + it('uses native query configs for metering metadata resolution', async () => { + const query = jest.fn(async ({ text }: { text: string }) => { + if (text.includes('to_regclass')) { + return { rows: [{ relation: 'provisioned' }], rowCount: 1 }; + } + if (text.includes('billing_module')) { + return { + rows: [ + { + public_schema: 'billing_public', + private_schema: 'billing_private', + record_usage_function: 'record_usage', + }, + ], + rowCount: 1, + }; + } + return { + rows: [{ schema: 'log_private', table_name: 'usage_log_inference' }], + rowCount: 1, + }; + }); + + await expect( + getLlmBillingConfig({ query } as never, 'database-native-contract') + ).resolves.toMatchObject({ + billing: { recordUsageFunction: 'record_usage' }, + inferenceLog: { tableName: 'usage_log_inference' }, + }); + + expect(query).toHaveBeenCalledTimes(4); + for (const [queryConfig] of query.mock.calls) { + expect(queryConfig).toEqual({ + text: expect.any(String), + values: expect.any(Array), + }); + } + }); + + it('preserves metering metadata query failures', async () => { + const original = new Error('metadata query failed'); + const query = jest.fn(async () => { + throw original; + }); + + await expect( + getLlmBillingConfig({ query } as never, 'database-error-contract') + ).rejects.toBe(original); + }); + + it('treats absent optional module relations as unprovisioned', async () => { + const query = jest.fn(async () => ({ + rows: [{ relation: null as string | null }], + rowCount: 1, + })); + + await expect( + getLlmBillingConfig({ query } as never, 'database-unprovisioned-contract') + ).resolves.toEqual({ billing: null, inferenceLog: null }); + expect(query).toHaveBeenCalledTimes(2); + }); + + it('fails closed for invalid metering context but stays optional without identity', async () => { + await expect( + buildMeteringContext( + { pgSettings }, + (settings) => settings['jwt.claims.user_id'] || null + ) + ).rejects.toThrow('LLM_METERING_PG_CLIENT_CONTEXT_UNAVAILABLE'); + + await expect( + buildMeteringContext( + { pgSettings: { role: 'anonymous_runtime' }, withPgClient: jest.fn() }, + () => null + ) + ).rejects.toThrow('LLM_METERING pgSettings'); + + const anonymousSettings = buildPgSettings({ + api, + token: null, + requestId: 'request-anonymous', + }); + const withPgClient = jest.fn(); + await expect( + buildMeteringContext( + { pgSettings: anonymousSettings, withPgClient }, + (settings) => settings['jwt.claims.user_id'] || null + ) + ).resolves.toBeNull(); + expect(withPgClient).not.toHaveBeenCalled(); + }); +}); diff --git a/graphile/graphile-llm/src/config-cache.ts b/graphile/graphile-llm/src/config-cache.ts index c3a5ae82fb..f3b0784ae2 100644 --- a/graphile/graphile-llm/src/config-cache.ts +++ b/graphile/graphile-llm/src/config-cache.ts @@ -17,6 +17,7 @@ * billing piece. */ +import type { PgClient as DataplanPgClient } from '@dataplan/pg'; import { ModuleConfigCache } from 'graphile-cache'; // ─── Types ────────────────────────────────────────────────────────────────── @@ -25,9 +26,7 @@ import { ModuleConfigCache } from 'graphile-cache'; * Generic pg client interface matching what Graphile's withPgClient provides. * Avoids a hard dependency on the `pg` package. */ -export interface PgClient { - query(sql: string, values?: unknown[]): Promise<{ rows: Record[] }>; -} +export type PgClient = DataplanPgClient; /** * Billing function metadata resolved from the billing_module metaschema table. @@ -105,61 +104,64 @@ const billingCache = new ModuleConfigCache({ // ─── Resolution Functions ─────────────────────────────────────────────────── -/** - * SQL to check if a schema exists. Used as a guard before querying - * metaschema tables that may not be provisioned. - */ -const SCHEMA_EXISTS_SQL = ` - SELECT 1 FROM information_schema.schemata WHERE schema_name = $1 LIMIT 1 +/** Check the exact optional module relation before querying it. */ +const RELATION_EXISTS_SQL = ` + SELECT pg_catalog.to_regclass($1) AS relation `; async function resolveInferenceLogConfig( pgClient: PgClient, databaseId: string ): Promise { - try { - const schemaCheck = await pgClient.query(SCHEMA_EXISTS_SQL, ['metaschema_modules_public']); - if (schemaCheck.rows.length === 0) return null; - - const result = await pgClient.query(INFERENCE_LOG_MODULE_SQL, [databaseId]); - const row = result.rows[0]; - if (!row?.schema || !row?.table_name) return null; - - return { - schema: row.schema as string, - tableName: row.table_name as string - }; - } catch { - return null; + const relationCheck = await pgClient.query<{ relation: string | null }>({ + text: RELATION_EXISTS_SQL, + values: ['metaschema_modules_public.inference_log_module'], + }); + if (!relationCheck.rows[0]?.relation) return null; + + const result = await pgClient.query>({ + text: INFERENCE_LOG_MODULE_SQL, + values: [databaseId], + }); + const row = result.rows[0]; + if (!row) return null; + if (!row.schema || !row.table_name) { + throw new Error('LLM_INFERENCE_LOG_CONFIG_INCOMPLETE'); } + + return { + schema: row.schema as string, + tableName: row.table_name as string, + }; } async function resolveBillingConfig( pgClient: PgClient, databaseId: string ): Promise { - try { - // Guard: check if the metaschema_modules_public schema exists. - // If the database doesn't have the billing module provisioned, - // this schema (or the billing_module table) won't exist. - const schemaCheck = await pgClient.query(SCHEMA_EXISTS_SQL, ['metaschema_modules_public']); - if (schemaCheck.rows.length === 0) return null; - - const result = await pgClient.query(BILLING_MODULE_SQL, [databaseId]); - const row = result.rows[0]; - if (!row?.record_usage_function) return null; - - return { - publicSchema: row.public_schema as string, - privateSchema: row.private_schema as string, - recordUsageFunction: row.record_usage_function as string, - // The check_billing_quota function name follows the inflection pattern - checkBillingQuotaFunction: 'check_billing_quota' - }; - } catch { - // Schema/table doesn't exist or query failed — billing not available - return null; + const relationCheck = await pgClient.query<{ relation: string | null }>({ + text: RELATION_EXISTS_SQL, + values: ['metaschema_modules_public.billing_module'], + }); + if (!relationCheck.rows[0]?.relation) return null; + + const result = await pgClient.query>({ + text: BILLING_MODULE_SQL, + values: [databaseId], + }); + const row = result.rows[0]; + if (!row) return null; + if (!row.public_schema || !row.private_schema || !row.record_usage_function) { + throw new Error('LLM_BILLING_CONFIG_INCOMPLETE'); } + + return { + publicSchema: row.public_schema as string, + privateSchema: row.private_schema as string, + recordUsageFunction: row.record_usage_function as string, + // The check_billing_quota function name follows the inflection pattern + checkBillingQuotaFunction: 'check_billing_quota', + }; } // ─── Public API ───────────────────────────────────────────────────────────── diff --git a/graphile/graphile-llm/src/plugins/metering-plugin.ts b/graphile/graphile-llm/src/plugins/metering-plugin.ts index 754f4aabfd..78c8c89f17 100644 --- a/graphile/graphile-llm/src/plugins/metering-plugin.ts +++ b/graphile/graphile-llm/src/plugins/metering-plugin.ts @@ -39,6 +39,10 @@ import type { PgClient } from '../config-cache'; import { getLlmBillingConfig } from '../config-cache'; import type { MeteringContext, MeteringOptions, WithPgClient } from '../metering'; import { meteredEmbed } from '../metering'; +import { + assertGraphileRequestContext, + withGraphileRequestPgClient, +} from '../request-context'; import type { EmbedderFunction, MeteringConfig } from '../types'; // ─── TypeScript Augmentation ──────────────────────────────────────────────── @@ -61,31 +65,35 @@ function defaultResolveEntityId(pgSettings: Record): string | nu return pgSettings['jwt.claims.user_id'] ?? null; } -async function buildMeteringContext( +export async function buildMeteringContext( graphqlContext: any, resolveEntityId: (pgSettings: Record) => string | null ): Promise { - const pgSettings: Record = graphqlContext?.pgSettings ?? {}; + const pgSettings = graphqlContext?.pgSettings; + // Metering is a request plugin; malformed or missing request context is not + // equivalent to an unprovisioned optional billing module. + const withPgClient: WithPgClient | undefined = graphqlContext?.withPgClient; + // Validate before reading identity so an absent context cannot silently + // downgrade a request to the unmetered path. + assertGraphileRequestContext(withPgClient, pgSettings, 'LLM_METERING'); const entityId = resolveEntityId(pgSettings); const databaseId = pgSettings['jwt.claims.database_id'] ?? null; const requestId = pgSettings['request.id'] ?? null; const actorId = pgSettings['jwt.claims.user_id'] ?? null; if (!entityId || !databaseId) return null; - const withPgClient: WithPgClient | undefined = graphqlContext?.withPgClient; - if (!withPgClient) return null; - let billingConfig = null; let inferenceLogConfig = null; - try { - await withPgClient(pgSettings, async (pgClient: PgClient) => { + await withGraphileRequestPgClient( + withPgClient, + pgSettings, + async (pgClient: PgClient) => { const entry = await getLlmBillingConfig(pgClient, databaseId); billingConfig = entry.billing; inferenceLogConfig = entry.inferenceLog; - }); - } catch { - return null; - } + }, + 'LLM_METERING' + ); if (!billingConfig) return null; diff --git a/graphile/graphile-llm/src/plugins/rag-plugin.ts b/graphile/graphile-llm/src/plugins/rag-plugin.ts index 3c1a3e15cf..6f9b9e8b7d 100644 --- a/graphile/graphile-llm/src/plugins/rag-plugin.ts +++ b/graphile/graphile-llm/src/plugins/rag-plugin.ts @@ -24,7 +24,13 @@ import { context as grafastContext, lambda, object } from 'grafast'; import type { GraphileConfig } from 'graphile-config'; import { extendSchema, gql } from 'graphile-utils'; -import type { ChatFunction, ChunkTableInfo, EmbedderFunction, RagDefaults } from '../types'; +import { withGraphileRequestPgClient } from '../request-context'; +import type { + ChatFunction, + ChunkTableInfo, + EmbedderFunction, + RagDefaults, +} from '../types'; // ─── TypeScript Augmentation ──────────────────────────────────────────────── @@ -307,20 +313,34 @@ export function createLlmRagPlugin( }> = []; if (chunkTables.length > 0) { - await withPgClient(pgSettings, async (pgClient: any) => { - for (const table of chunkTables) { - const query = buildChunkSearchSql(table, vectorString, limit, maxDistance); - const result = await pgClient.query(query); - for (const row of result.rows) { - allChunks.push({ - content: row.content, - parent_id: row.parent_id, - distance: parseFloat(row.distance), - table_name: table.parentCodecName - }); + await withGraphileRequestPgClient( + withPgClient, + pgSettings, + async (pgClient) => { + for (const table of chunkTables) { + const query = buildChunkSearchSql( + table, + vectorString, + limit, + maxDistance + ); + const result = await pgClient.query<{ + content: string; + parent_id: string; + distance: string; + }>(query); + for (const row of result.rows) { + allChunks.push({ + content: row.content, + parent_id: row.parent_id, + distance: parseFloat(row.distance), + table_name: table.parentCodecName, + }); + } } - } - }); + }, + 'RAG' + ); } // Sort by distance (ascending) and take top N diff --git a/graphile/graphile-llm/src/request-context.ts b/graphile/graphile-llm/src/request-context.ts new file mode 100644 index 0000000000..0e933250f5 --- /dev/null +++ b/graphile/graphile-llm/src/request-context.ts @@ -0,0 +1,27 @@ +import { + assertCompletePgSettings, + type PgSettings, +} from '@constructive-io/express-context'; +import type { PgClient, WithPgClient } from '@dataplan/pg'; + +export function assertGraphileRequestContext( + withPgClient: unknown, + pgSettings: unknown, + label: string +): asserts pgSettings is PgSettings { + if (typeof withPgClient !== 'function') { + throw new Error(`${label}_PG_CLIENT_CONTEXT_UNAVAILABLE`); + } + assertCompletePgSettings(pgSettings, `${label} pgSettings`); +} + +/** Run request-lane SQL with the complete Graphile request settings. */ +export async function withGraphileRequestPgClient( + withPgClient: unknown, + pgSettings: unknown, + callback: (client: PgClient) => T | Promise, + label: string +): Promise { + assertGraphileRequestContext(withPgClient, pgSettings, label); + return (withPgClient as WithPgClient)(pgSettings as PgSettings, callback); +} diff --git a/graphile/graphile-settings/__tests__/PublicKeySignature.test.ts b/graphile/graphile-settings/__tests__/PublicKeySignature.test.ts index cc833a3333..e7f55b3acb 100644 --- a/graphile/graphile-settings/__tests__/PublicKeySignature.test.ts +++ b/graphile/graphile-settings/__tests__/PublicKeySignature.test.ts @@ -1,8 +1,15 @@ +import { buildPgSettings } from '@constructive-io/express-context'; + import type { PublicKeyChallengeConfig } from '../src/plugins/PublicKeySignature'; -import { PublicKeySignature } from '../src/plugins/PublicKeySignature'; +import { + PublicKeySignature, + queryPublicKeyFunction, + withAnonymousPublicKeyClient, +} from '../src/plugins/PublicKeySignature'; const defaultConfig: PublicKeyChallengeConfig = { schema: 'app_private', + anonymousRole: 'anonymous_runtime', crypto_network: 'btc', sign_up_with_key: 'sign_up_with_key', sign_in_request_challenge: 'sign_in_request_challenge', @@ -29,6 +36,7 @@ describe('PublicKeySignature plugin factory', () => { it('accepts custom config values', () => { const customConfig: PublicKeyChallengeConfig = { schema: 'custom_schema', + anonymousRole: 'custom_anonymous', crypto_network: 'eth', sign_up_with_key: 'custom_signup', sign_in_request_challenge: 'custom_challenge', @@ -59,6 +67,15 @@ describe('PublicKeySignature config validation', () => { expect(() => PublicKeySignature({ ...defaultConfig, schema: 'DROP TABLE' })).toThrow(/invalid schema/); }); + it('throws on invalid anonymous role', () => { + expect(() => + PublicKeySignature({ + ...defaultConfig, + anonymousRole: 'anonymous; RESET ALL', + }) + ).toThrow(/invalid anonymousRole/); + }); + it('throws on invalid function name', () => { expect(() => PublicKeySignature({ ...defaultConfig, sign_up_with_key: 'evil"; DROP' })).toThrow( /invalid sign_up_with_key/, @@ -87,3 +104,120 @@ describe('PublicKeySignature config validation', () => { expect(() => PublicKeySignature(defaultConfig)).not.toThrow(); }); }); + +describe('PublicKeySignature request context', () => { + const pgSettings = buildPgSettings({ + api: { + apiId: 'api-1', + databaseId: 'database-1', + dbname: 'tenant_db', + anonRole: 'anonymous_runtime', + roleName: 'authenticated_runtime', + schema: ['app_public'], + }, + token: { + id: 'token-1', + user_id: 'user-1', + entity_id: 'entity-1', + }, + requestId: 'request-1', + dependencySchemas: ['app_shared'], + }); + + it('copies every setting and replaces only the role', async () => { + const original = { ...pgSettings }; + const query = jest.fn(); + const callback = jest.fn(async () => 'ok'); + const withPgClient = jest.fn(async (settings, fn) => fn({ query })); + + await expect( + withAnonymousPublicKeyClient( + withPgClient, + pgSettings, + 'anonymous_runtime', + callback + ) + ).resolves.toBe('ok'); + + const anonymousSettings = withPgClient.mock.calls[0][0]; + expect(anonymousSettings).toEqual({ + ...pgSettings, + role: 'anonymous_runtime', + }); + expect(anonymousSettings).not.toBe(pgSettings); + expect(Object.keys(anonymousSettings)).toEqual(Object.keys(pgSettings)); + expect(anonymousSettings['jwt.claims.api_id']).toBe('api-1'); + expect(anonymousSettings['jwt.claims.database_id']).toBe('database-1'); + expect(anonymousSettings['request.id']).toBe('request-1'); + expect(anonymousSettings.row_security).toBe('on'); + expect(anonymousSettings.search_path).toBe( + 'pg_catalog, "app_shared", "app_public"' + ); + expect(anonymousSettings['jwt.claims.email']).toBe(''); + expect(pgSettings).toEqual(original); + }); + + it.each([ + [ + 'missing withPgClient', + undefined, + pgSettings, + 'PUBLIC_KEY_PG_CLIENT_CONTEXT_UNAVAILABLE', + ], + ['missing settings', jest.fn(), undefined, 'PublicKeySignature pgSettings'], + ['null settings', jest.fn(), null, 'PublicKeySignature pgSettings'], + ['array settings', jest.fn(), [], 'PublicKeySignature pgSettings'], + [ + 'incomplete settings', + jest.fn(), + { role: 'authenticated_runtime' }, + 'PublicKeySignature pgSettings', + ], + ])('fails closed for %s', async (_label, withPgClient, settings, message) => { + await expect( + withAnonymousPublicKeyClient( + withPgClient, + settings, + 'anonymous_runtime', + async (): Promise => undefined + ) + ).rejects.toThrow(message); + }); + + it('uses the native query config and validates database identifiers', async () => { + const query = jest.fn(async () => ({ + rows: [{ sign_in_request_challenge: 'challenge' }], + rowCount: 1, + })); + + await expect( + queryPublicKeyFunction( + { query } as never, + 'app_private', + 'sign_in_request_challenge', + ['public-key'] + ) + ).resolves.toMatchObject({ rowCount: 1 }); + + expect(query).toHaveBeenCalledWith({ + text: 'SELECT * FROM app_private.sign_in_request_challenge($1)', + values: ['public-key'], + }); + expect(() => + queryPublicKeyFunction( + { query } as never, + 'unsafe.schema', + 'sign_in_request_challenge', + [] + ) + ).toThrow(/invalid schema/); + expect(() => + queryPublicKeyFunction( + { query } as never, + 'app_private', + 'unsafe_function()', + [] + ) + ).toThrow(/invalid function/); + }); +}); diff --git a/graphile/graphile-settings/__tests__/request-context.integration.test.ts b/graphile/graphile-settings/__tests__/request-context.integration.test.ts new file mode 100644 index 0000000000..54e5fefa37 --- /dev/null +++ b/graphile/graphile-settings/__tests__/request-context.integration.test.ts @@ -0,0 +1,344 @@ +import { join } from 'node:path'; + +import { buildPgSettings } from '@constructive-io/express-context'; +import { createI18nPlugin } from 'graphile-i18n'; +import type { GraphQLResponse } from 'graphile-test'; +import { getConnections, seed } from 'graphile-test'; + +import { PublicKeySignature } from '../src/plugins/PublicKeySignature'; + +const api = { + apiId: 'api-1', + databaseId: 'database-1', + dbname: 'request_context_db', + anonRole: 'anonymous', + roleName: 'authenticated', + schema: ['request_context_test'], +}; + +const settings = (userId: string | null, requestId: string) => + buildPgSettings({ + api, + token: userId ? { id: 'token-1', user_id: userId } : null, + requestId, + }); + +describe('complete Graphile request context integration', () => { + let db: any; + let teardown: () => Promise; + let query: ( + document: string, + variables?: Record, + commit?: boolean, + reqOptions?: Record + ) => Promise>; + + beforeAll(async () => { + const connections = await getConnections( + { + schemas: ['request_context_test'], + authRole: 'authenticated', + preset: { + plugins: [ + createI18nPlugin({ defaultLanguages: ['en'] }), + PublicKeySignature({ + schema: 'request_context_test', + anonymousRole: 'anonymous', + crypto_network: 'test', + sign_up_with_key: 'sign_up_with_key', + sign_in_request_challenge: 'sign_in_request_challenge', + sign_in_record_failure: 'sign_in_record_failure', + sign_in_with_challenge: 'sign_in_with_challenge', + }), + ], + }, + }, + [ + seed.fn(async ({ admin, config, connect }) => { + await admin.streamSql( + `DO $roles$ + BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'anonymous') THEN + EXECUTE 'CREATE ROLE anonymous'; + END IF; + IF NOT EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = 'authenticated') THEN + EXECUTE 'CREATE ROLE authenticated'; + END IF; + END + $roles$;`, + config.database + ); + const appUser = connect.connections?.app?.user; + if (!appUser) + throw new Error('request-context test requires an app user'); + await admin.grantRole('anonymous', appUser, config.database); + await admin.grantRole('authenticated', appUser, config.database); + }), + seed.sqlfile([join(__dirname, 'request-context.setup.sql')]), + ] + ); + db = connections.db; + teardown = connections.teardown; + query = connections.query; + }, 30_000); + + afterAll(async () => { + if (teardown) await teardown(); + }); + + beforeEach(async () => { + if (db) await db.beforeEach(); + }); + + afterEach(async () => { + if (db) await db.afterEach(); + }); + + it('preserves the PublicKeySignature mutation schema contract', async () => { + const result = await query<{ + mutationType: { + fields: Array<{ + name: string; + args: Array<{ + name: string; + type: { + kind: string; + name: string | null; + ofType: { kind: string; name: string | null } | null; + }; + }>; + type: { kind: string; name: string | null }; + }>; + } | null; + }>( + ` + query PublicKeySchemaContract { + mutationType: __type(name: "Mutation") { + fields { + name + args { + name + type { + kind + name + ofType { kind name } + } + } + type { kind name } + } + } + } + `, + undefined, + false, + { pgSettings: settings(null, 'schema-contract') } + ); + + expect(result.errors).toBeUndefined(); + const fields = result.data?.mutationType?.fields + .filter((field) => + [ + 'createUserAccountWithPublicKey', + 'getMessageForSigning', + 'verifyMessageForSigning', + ].includes(field.name) + ) + .sort((a, b) => a.name.localeCompare(b.name)); + + expect(fields).toEqual([ + { + name: 'createUserAccountWithPublicKey', + args: [ + { + name: 'input', + type: { + kind: 'INPUT_OBJECT', + name: 'CreateUserAccountWithPublicKeyInput', + ofType: null, + }, + }, + ], + type: { + kind: 'OBJECT', + name: 'createUserAccountWithPublicKeyPayload', + }, + }, + { + name: 'getMessageForSigning', + args: [ + { + name: 'input', + type: { + kind: 'INPUT_OBJECT', + name: 'GetMessageForSigningInput', + ofType: null, + }, + }, + ], + type: { + kind: 'OBJECT', + name: 'getMessageForSigningPayload', + }, + }, + { + name: 'verifyMessageForSigning', + args: [ + { + name: 'input', + type: { + kind: 'INPUT_OBJECT', + name: 'VerifyMessageForSigningInput', + ofType: null, + }, + }, + ], + type: { + kind: 'OBJECT', + name: 'verifyMessageForSigningPayload', + }, + }, + ]); + }); + + it('keeps normal, F13 and anonymous PublicKey lanes isolated', async () => { + const authenticated = settings('user-1', 'authenticated-request'); + const authenticatedProbe = await query<{ + contextProbe: Record; + }>( + ` + query AuthenticatedContext { + contextProbe + } + `, + undefined, + false, + { pgSettings: authenticated } + ); + + expect(authenticatedProbe.errors).toBeUndefined(); + expect(authenticatedProbe.data?.contextProbe).toMatchObject({ + currentUser: 'authenticated', + userId: 'user-1', + apiId: 'api-1', + databaseId: 'database-1', + requestId: 'authenticated-request', + readOnly: 'off', + rowSecurity: 'on', + searchPath: 'pg_catalog, "request_context_test"', + }); + + const anonymous = settings(null, 'anonymous-request'); + const anonymousProbe = await query<{ + contextProbe: Record; + }>( + ` + query AnonymousContext { + contextProbe + } + `, + undefined, + false, + { pgSettings: anonymous } + ); + + expect(anonymousProbe.errors).toBeUndefined(); + expect(anonymousProbe.data?.contextProbe).toMatchObject({ + currentUser: 'anonymous', + userId: '', + apiId: 'api-1', + databaseId: 'database-1', + requestId: 'anonymous-request', + }); + + const f13Settings = settings('user-1', 'f13-request'); + const i18nResult = await query<{ + postByRowId: { localeStrings: { title: string } } | null; + }>( + ` + query F13Context { + postByRowId(rowId: 1) { + localeStrings { + title + } + } + } + `, + undefined, + false, + { pgSettings: f13Settings } + ); + + expect(i18nResult.errors).toBeUndefined(); + expect(i18nResult.data?.postByRowId?.localeStrings.title).toBe( + 'Context-approved translation' + ); + + const publicKeyResult = await query<{ + getMessageForSigning: { message: string } | null; + }>( + ` + mutation PublicKeyAnonymousLane { + getMessageForSigning(input: { publicKey: "public-key-1" }) { + message + } + } + `, + undefined, + false, + { pgSettings: authenticated } + ); + + expect(publicKeyResult.errors).toBeUndefined(); + const publicKeyContext = JSON.parse( + publicKeyResult.data?.getMessageForSigning?.message ?? '{}' + ); + expect(publicKeyContext).toMatchObject({ + currentUser: 'anonymous', + userId: 'user-1', + apiId: 'api-1', + databaseId: 'database-1', + requestId: 'authenticated-request', + readOnly: 'off', + rowSecurity: 'on', + searchPath: 'pg_catalog, "request_context_test"', + }); + + const rollbackResult = await query( + ` + mutation PublicKeyRollback { + createUserAccountWithPublicKey(input: { publicKey: "force-rollback" }) { + message + } + } + `, + undefined, + false, + { pgSettings: authenticated } + ); + expect(rollbackResult.errors?.[0]?.message).toContain( + 'forced public-key rollback' + ); + + const audit = await db.client.query( + 'SELECT count(*)::int AS count FROM request_context_test.public_key_audit' + ); + expect(audit.rows[0].count).toBe(0); + + const afterRollback = await query<{ contextProbe: Record }>( + ` + query AfterRollback { + contextProbe + } + `, + undefined, + false, + { pgSettings: anonymous } + ); + expect(afterRollback.errors).toBeUndefined(); + expect(afterRollback.data?.contextProbe).toMatchObject({ + currentUser: 'anonymous', + userId: '', + requestId: 'anonymous-request', + }); + }); +}); diff --git a/graphile/graphile-settings/__tests__/request-context.setup.sql b/graphile/graphile-settings/__tests__/request-context.setup.sql new file mode 100644 index 0000000000..8cfb83376f --- /dev/null +++ b/graphile/graphile-settings/__tests__/request-context.setup.sql @@ -0,0 +1,126 @@ +CREATE SCHEMA request_context_test; +GRANT USAGE ON SCHEMA request_context_test TO anonymous, authenticated; + +CREATE FUNCTION request_context_test.context_probe() +RETURNS jsonb +LANGUAGE sql +STABLE +AS $$ + SELECT jsonb_build_object( + 'currentUser', current_user, + 'userId', current_setting('jwt.claims.user_id', true), + 'apiId', current_setting('jwt.claims.api_id', true), + 'databaseId', current_setting('jwt.claims.database_id', true), + 'requestId', current_setting('request.id', true), + 'readOnly', current_setting('transaction_read_only'), + 'rowSecurity', current_setting('row_security'), + 'searchPath', current_setting('search_path') + ) +$$; +GRANT EXECUTE ON FUNCTION request_context_test.context_probe() TO anonymous, authenticated; + +CREATE TABLE request_context_test.posts ( + id integer PRIMARY KEY, + title text NOT NULL +); +COMMENT ON TABLE request_context_test.posts IS E'@i18n posts_translations'; + +CREATE TABLE request_context_test.posts_translations ( + id integer PRIMARY KEY, + post_id integer NOT NULL REFERENCES request_context_test.posts(id), + lang_code text NOT NULL, + title text NOT NULL, + UNIQUE (post_id, lang_code) +); + +INSERT INTO request_context_test.posts (id, title) +VALUES (1, 'Base title'); +INSERT INTO request_context_test.posts_translations (id, post_id, lang_code, title) +VALUES (1, 1, 'en', 'Context-approved translation'); + +ALTER TABLE request_context_test.posts ENABLE ROW LEVEL SECURITY; +ALTER TABLE request_context_test.posts_translations ENABLE ROW LEVEL SECURITY; + +CREATE POLICY complete_request_context_posts +ON request_context_test.posts +FOR SELECT +TO authenticated +USING ( + current_user = 'authenticated' + AND current_setting('jwt.claims.user_id', true) = 'user-1' + AND current_setting('jwt.claims.api_id', true) = 'api-1' + AND current_setting('jwt.claims.database_id', true) = 'database-1' + AND current_setting('request.id', true) = 'f13-request' + AND current_setting('transaction_read_only') = 'off' + AND current_setting('row_security') = 'on' + AND current_setting('search_path') = 'pg_catalog, "request_context_test"' +); + +CREATE POLICY complete_request_context_translations +ON request_context_test.posts_translations +FOR SELECT +TO authenticated +USING ( + current_user = 'authenticated' + AND current_setting('jwt.claims.user_id', true) = 'user-1' + AND current_setting('jwt.claims.api_id', true) = 'api-1' + AND current_setting('jwt.claims.database_id', true) = 'database-1' + AND current_setting('request.id', true) = 'f13-request' + AND current_setting('transaction_read_only') = 'off' + AND current_setting('row_security') = 'on' + AND current_setting('search_path') = 'pg_catalog, "request_context_test"' +); + +GRANT SELECT ON request_context_test.posts TO anonymous, authenticated; +GRANT SELECT ON request_context_test.posts_translations TO anonymous, authenticated; + +CREATE TABLE request_context_test.public_key_audit ( + public_key text NOT NULL +); +GRANT SELECT, INSERT ON request_context_test.public_key_audit TO anonymous, authenticated; + +CREATE FUNCTION request_context_test.sign_up_with_key(public_key text) +RETURNS TABLE(sign_up_with_key text) +LANGUAGE plpgsql +VOLATILE +AS $$ +BEGIN + INSERT INTO request_context_test.public_key_audit VALUES (public_key); + IF public_key = 'force-rollback' THEN + RAISE EXCEPTION 'forced public-key rollback'; + END IF; + RETURN QUERY SELECT public_key; +END +$$; + +CREATE FUNCTION request_context_test.sign_in_request_challenge(public_key text) +RETURNS TABLE(sign_in_request_challenge text) +LANGUAGE sql +STABLE +AS $$ + SELECT jsonb_build_object( + 'currentUser', current_user, + 'userId', current_setting('jwt.claims.user_id', true), + 'apiId', current_setting('jwt.claims.api_id', true), + 'databaseId', current_setting('jwt.claims.database_id', true), + 'requestId', current_setting('request.id', true), + 'readOnly', current_setting('transaction_read_only'), + 'rowSecurity', current_setting('row_security'), + 'searchPath', current_setting('search_path'), + 'publicKey', public_key + )::text +$$; + +CREATE FUNCTION request_context_test.sign_in_record_failure(public_key text) +RETURNS void +LANGUAGE sql +VOLATILE +AS $$ SELECT NULL::void $$; + +CREATE FUNCTION request_context_test.sign_in_with_challenge(public_key text, message text) +RETURNS TABLE(access_token text, access_token_expires_at timestamptz) +LANGUAGE sql +VOLATILE +AS $$ SELECT public_key || message, now() + interval '1 hour' $$; + +GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA request_context_test TO anonymous, authenticated; diff --git a/graphile/graphile-settings/package.json b/graphile/graphile-settings/package.json index 409a7e679b..77eb019e1d 100644 --- a/graphile/graphile-settings/package.json +++ b/graphile/graphile-settings/package.json @@ -31,6 +31,7 @@ "dependencies": { "@aws-sdk/client-s3": "^3.1052.0", "@constructive-io/bucket-provisioner": "workspace:^", + "@constructive-io/express-context": "workspace:^", "@constructive-io/graphql-env": "workspace:^", "@constructive-io/graphql-types": "workspace:^", "@constructive-io/s3-streamer": "workspace:^", @@ -69,7 +70,6 @@ "lru-cache": "^11.2.7", "mime-bytes": "workspace:^", "pg": "^8.21.0", - "pg-query-context": "workspace:^", "pg-sql2": "5.0.1", "postgraphile": "5.1.4", "request-ip": "^3.3.0", diff --git a/graphile/graphile-settings/src/plugins/PublicKeySignature.ts b/graphile/graphile-settings/src/plugins/PublicKeySignature.ts index 2d01aa9712..26a4df6db3 100644 --- a/graphile/graphile-settings/src/plugins/PublicKeySignature.ts +++ b/graphile/graphile-settings/src/plugins/PublicKeySignature.ts @@ -1,13 +1,20 @@ // import Networks from '@pyramation/crypto-networks'; // import { verifyMessage } from '@pyramation/crypto-keys'; +import { + assertCompletePgSettings, + type PgSettings, + withPgSettingsRole, +} from '@constructive-io/express-context'; +import type { PgClient, PgClientResult, WithPgClient } from '@dataplan/pg'; import { QuoteUtils } from '@pgsql/quotes'; import { context as grafastContext, lambda, object } from 'grafast'; import type { GraphileConfig } from 'graphile-config'; import { extendSchema, gql } from 'graphile-utils'; -import pgQueryWithContext from 'pg-query-context'; export interface PublicKeyChallengeConfig { schema: string; + /** Exact anonymous role configured for this Graphile API surface. */ + anonymousRole: string; crypto_network: string; // crypto_network: keyof typeof Networks; sign_up_with_key: string; @@ -20,13 +27,15 @@ const SAFE_IDENTIFIER = /^[a-z_][a-z0-9_]*$/; const SAFE_CRYPTO_NETWORK = /^[a-z0-9_-]{1,64}$/i; function validateIdentifier(name: string, label: string): void { - if (!SAFE_IDENTIFIER.test(name)) { - throw new Error(`PublicKeySignature: invalid ${label} "${name}" — must match /^[a-z_][a-z0-9_]*$/`); + if (typeof name !== 'string' || !SAFE_IDENTIFIER.test(name)) { + throw new Error( + `PublicKeySignature: invalid ${label} "${name}" — must match /^[a-z_][a-z0-9_]*$/` + ); } } function validateCryptoNetwork(name: string): void { - if (!SAFE_CRYPTO_NETWORK.test(name)) { + if (typeof name !== 'string' || !SAFE_CRYPTO_NETWORK.test(name)) { throw new Error( 'PublicKeySignature: invalid crypto_network — must match /^[a-z0-9_-]{1,64}$/i', ); @@ -38,9 +47,50 @@ const MAX_MESSAGE_LENGTH = 4096; const MAX_SIGNATURE_LENGTH = 1024; const ENABLE_SIGNATURE_VERIFICATION = process.env.ENABLE_SIGNATURE_VERIFICATION === 'true'; -export const PublicKeySignature = (pubkey_challenge: PublicKeyChallengeConfig): GraphileConfig.Plugin => { +/** + * Run a public-key authentication operation with the request's complete GUC + * context while retaining the deliberately anonymous database role. + */ +export async function withAnonymousPublicKeyClient( + withPgClient: unknown, + pgSettings: unknown, + anonymousRole: string, + callback: (pgClient: PgClient) => T | Promise +): Promise { + if (typeof withPgClient !== 'function') { + throw new Error('PUBLIC_KEY_PG_CLIENT_CONTEXT_UNAVAILABLE'); + } + assertCompletePgSettings(pgSettings, 'PublicKeySignature pgSettings'); + validateIdentifier(anonymousRole, 'anonymousRole'); + + const anonymousSettings: PgSettings = withPgSettingsRole( + pgSettings, + anonymousRole + ); + return (withPgClient as WithPgClient)(anonymousSettings, callback); +} + +/** Use @dataplan/pg's native query-config contract for every public-key call. */ +export function queryPublicKeyFunction( + pgClient: Pick, + schema: string, + functionName: string, + values: any[] +): Promise> { + validateIdentifier(schema, 'schema'); + validateIdentifier(functionName, 'function'); + return pgClient.query({ + text: `SELECT * FROM ${QuoteUtils.quoteQualifiedIdentifier(schema, functionName)}(${values.map((_, index) => `$${index + 1}`).join(', ')})`, + values, + }); +} + +export const PublicKeySignature = ( + pubkey_challenge: PublicKeyChallengeConfig +): GraphileConfig.Plugin => { const { schema, + anonymousRole, crypto_network, sign_up_with_key, sign_in_request_challenge, @@ -49,6 +99,7 @@ export const PublicKeySignature = (pubkey_challenge: PublicKeyChallengeConfig): } = pubkey_challenge; validateIdentifier(schema, 'schema'); + validateIdentifier(anonymousRole, 'anonymousRole'); validateIdentifier(sign_up_with_key, 'sign_up_with_key'); validateIdentifier(sign_in_request_challenge, 'sign_in_request_challenge'); validateIdentifier(sign_in_record_failure, 'sign_in_record_failure'); @@ -103,69 +154,94 @@ export const PublicKeySignature = (pubkey_challenge: PublicKeyChallengeConfig): createUserAccountWithPublicKey(_$mutation: any, fieldArgs: any) { const $input = fieldArgs.getRaw('input'); const $withPgClient = (grafastContext() as any).get('withPgClient'); - const $combined = object({ input: $input, withPgClient: $withPgClient }); - - return lambda($combined, async ({ input, withPgClient }: any) => { - if (!input.publicKey || typeof input.publicKey !== 'string' || input.publicKey.length > MAX_PUBLIC_KEY_LENGTH) { - throw new Error('INVALID_PUBLIC_KEY'); - } + const $pgSettings = (grafastContext() as any).get('pgSettings'); + const $combined = object({ + input: $input, + withPgClient: $withPgClient, + pgSettings: $pgSettings, + }); - return withPgClient(null, async (pgClient: any) => { - await pgClient.query('BEGIN'); - try { - await pgQueryWithContext({ - client: pgClient, - context: { role: 'anonymous' }, - query: `SELECT * FROM ${QuoteUtils.quoteQualifiedIdentifier(schema, sign_up_with_key)}($1)`, - variables: [input.publicKey], - skipTransaction: true - }); - - const { - rows: [{ [sign_in_request_challenge]: message }] - } = await pgQueryWithContext({ - client: pgClient, - context: { role: 'anonymous' }, - query: `SELECT * FROM ${QuoteUtils.quoteQualifiedIdentifier(schema, sign_in_request_challenge)}($1)`, - variables: [input.publicKey], - skipTransaction: true - }); - - await pgClient.query('COMMIT'); - return { message }; - } catch (err) { - await pgClient.query('ROLLBACK'); - throw err; + return lambda( + $combined, + async ({ input, withPgClient, pgSettings }: any) => { + if ( + !input.publicKey || + typeof input.publicKey !== 'string' || + input.publicKey.length > MAX_PUBLIC_KEY_LENGTH + ) { + throw new Error('INVALID_PUBLIC_KEY'); } - }); - }); + + return withAnonymousPublicKeyClient( + withPgClient, + pgSettings, + anonymousRole, + async (pgClient) => { + await queryPublicKeyFunction( + pgClient, + schema, + sign_up_with_key, + [input.publicKey] + ); + + const { + rows: [{ [sign_in_request_challenge]: message }], + } = await queryPublicKeyFunction>( + pgClient, + schema, + sign_in_request_challenge, + [input.publicKey] + ); + + return { message }; + } + ); + } + ); }, getMessageForSigning(_$mutation: any, fieldArgs: any) { const $input = fieldArgs.getRaw('input'); const $withPgClient = (grafastContext() as any).get('withPgClient'); - const $combined = object({ input: $input, withPgClient: $withPgClient }); - - return lambda($combined, async ({ input, withPgClient }: any) => { - if (!input.publicKey || typeof input.publicKey !== 'string' || input.publicKey.length > MAX_PUBLIC_KEY_LENGTH) { - throw new Error('INVALID_PUBLIC_KEY'); - } - - return withPgClient(null, async (pgClient: any) => { - const { - rows: [{ [sign_in_request_challenge]: message }] - } = await pgQueryWithContext({ - client: pgClient, - context: { role: 'anonymous' }, - query: `SELECT * FROM ${QuoteUtils.quoteQualifiedIdentifier(schema, sign_in_request_challenge)}($1)`, - variables: [input.publicKey] - }); + const $pgSettings = (grafastContext() as any).get('pgSettings'); + const $combined = object({ + input: $input, + withPgClient: $withPgClient, + pgSettings: $pgSettings, + }); - if (!message) throw new Error('NO_ACCOUNT_EXISTS'); + return lambda( + $combined, + async ({ input, withPgClient, pgSettings }: any) => { + if ( + !input.publicKey || + typeof input.publicKey !== 'string' || + input.publicKey.length > MAX_PUBLIC_KEY_LENGTH + ) { + throw new Error('INVALID_PUBLIC_KEY'); + } - return { message }; - }); - }); + return withAnonymousPublicKeyClient( + withPgClient, + pgSettings, + anonymousRole, + async (pgClient) => { + const { + rows: [{ [sign_in_request_challenge]: message }], + } = await queryPublicKeyFunction>( + pgClient, + schema, + sign_in_request_challenge, + [input.publicKey] + ); + + if (!message) throw new Error('NO_ACCOUNT_EXISTS'); + + return { message }; + } + ); + } + ); }, // NOTE: Verification remains behind a feature flag until crypto @@ -173,57 +249,73 @@ export const PublicKeySignature = (pubkey_challenge: PublicKeyChallengeConfig): verifyMessageForSigning(_$mutation: any, fieldArgs: any) { const $input = fieldArgs.getRaw('input'); const $withPgClient = (grafastContext() as any).get('withPgClient'); - const $combined = object({ input: $input, withPgClient: $withPgClient }); + const $pgSettings = (grafastContext() as any).get('pgSettings'); + const $combined = object({ + input: $input, + withPgClient: $withPgClient, + pgSettings: $pgSettings, + }); - return lambda($combined, async ({ input, withPgClient }: any) => { - const { publicKey, message, signature: _signature } = input; + return lambda( + $combined, + async ({ input, withPgClient, pgSettings }: any) => { + const { publicKey, message, signature: _signature } = input; + + if ( + !publicKey || + typeof publicKey !== 'string' || + publicKey.length > MAX_PUBLIC_KEY_LENGTH + ) { + throw new Error('INVALID_PUBLIC_KEY'); + } + if ( + !message || + typeof message !== 'string' || + message.length > MAX_MESSAGE_LENGTH + ) { + throw new Error('INVALID_MESSAGE'); + } + if ( + !_signature || + typeof _signature !== 'string' || + _signature.length > MAX_SIGNATURE_LENGTH + ) { + throw new Error('INVALID_SIGNATURE'); + } - if (!publicKey || typeof publicKey !== 'string' || publicKey.length > MAX_PUBLIC_KEY_LENGTH) { - throw new Error('INVALID_PUBLIC_KEY'); - } - if (!message || typeof message !== 'string' || message.length > MAX_MESSAGE_LENGTH) { - throw new Error('INVALID_MESSAGE'); - } - if (!_signature || typeof _signature !== 'string' || _signature.length > MAX_SIGNATURE_LENGTH) { - throw new Error('INVALID_SIGNATURE'); - } + if (!ENABLE_SIGNATURE_VERIFICATION) { + // Fail closed without mutating lockout counters while verification + // is disabled. + throw new Error('FEATURE_DISABLED'); + } - if (!ENABLE_SIGNATURE_VERIFICATION) { - // Fail closed without mutating lockout counters while verification - // is disabled. - throw new Error('FEATURE_DISABLED'); + return withAnonymousPublicKeyClient( + withPgClient, + pgSettings, + anonymousRole, + async (pgClient) => { + const { + rows: [token], + } = await queryPublicKeyFunction>( + pgClient, + schema, + sign_in_with_challenge, + [publicKey, message] + ); + + if (!token?.access_token) throw new Error('BAD_SIGNIN'); + + return { + access_token: token.access_token, + access_token_expires_at: token.access_token_expires_at, + }; + } + ); } - - return withPgClient(null, async (pgClient: any) => { - // Only the success path needs a transaction (multi-step) - await pgClient.query('BEGIN'); - try { - const { - rows: [token] - } = await pgQueryWithContext({ - client: pgClient, - context: { role: 'anonymous' }, - query: `SELECT * FROM ${QuoteUtils.quoteQualifiedIdentifier(schema, sign_in_with_challenge)}($1, $2)`, - variables: [publicKey, message], - skipTransaction: true - }); - - if (!token?.access_token) throw new Error('BAD_SIGNIN'); - - await pgClient.query('COMMIT'); - return { - access_token: token.access_token, - access_token_expires_at: token.access_token_expires_at - }; - } catch (err) { - await pgClient.query('ROLLBACK'); - throw err; - } - }); - }); - } - } - } + ); + }, + }, + }, })); }; diff --git a/graphile/graphile-test/src/context.ts b/graphile/graphile-test/src/context.ts index 677eeae4f6..964549451c 100644 --- a/graphile/graphile-test/src/context.ts +++ b/graphile/graphile-test/src/context.ts @@ -264,9 +264,10 @@ export const runGraphQLInContext = async ({ // Provide a custom withPgClient function that uses the test client // This ensures GraphQL operations run within the test transaction // instead of getting a new connection from the pool + const isInTransaction = !input.useRoot; const withPgClientKey = pgService.withPgClientKey ?? 'withPgClient'; contextValue[withPgClientKey] = async ( - _pgSettings: Record | null, + requestedPgSettings: Record | null, callback: (client: Client) => T | Promise ): Promise => { // Augment the client with withTransaction if it doesn't already have it. @@ -291,14 +292,40 @@ export const runGraphQLInContext = async ({ } }; } - return callback(pgClient); + const callbackSettings = requestedPgSettings ?? pgSettings; + if (!isInTransaction) { + await client.query('BEGIN'); + try { + await setContextOnClient( + client, + callbackSettings, + callbackSettings.role ?? pgSettings.role + ); + const result = await callback(client); + await client.query('COMMIT'); + return result; + } catch (error) { + await client.query('ROLLBACK').catch(() => {}); + throw error; + } + } + + await setContextOnClient( + client, + callbackSettings, + callbackSettings.role ?? pgSettings.role + ); + const result = await callback(client); + // Errors are rolled back by the existing execution savepoint below. On a + // successful derivative lane, explicitly restore the primary request + // context before returning control to the rest of the GraphQL operation. + await setContextOnClient(client, pgSettings, pgSettings.role); + return result; }; // Check if we're in a transaction by looking at the test client's transaction state // When useRoot is true, we might not be in a transaction // pgsql-test's `db` client is in a transaction, but `pg` (root) client may not be - const isInTransaction = !input.useRoot; - // Wrap the entire query execution in a savepoint if we're in a transaction // This matches v4 PostGraphile behavior where each mutation is wrapped in a savepoint // allowing the transaction to continue after a database error diff --git a/graphql/env/__tests__/__snapshots__/merge.test.ts.snap b/graphql/env/__tests__/__snapshots__/merge.test.ts.snap index 6383de2044..27e9075742 100644 --- a/graphql/env/__tests__/__snapshots__/merge.test.ts.snap +++ b/graphql/env/__tests__/__snapshots__/merge.test.ts.snap @@ -70,6 +70,7 @@ exports[`getEnvOptions merges pgpm defaults, graphql defaults, config, env, and }, "graphile": { "extends": [], + "introspectionDependencySchemas": [], "preset": {}, "schema": [ "override_schema", diff --git a/graphql/server/src/middleware/__tests__/graphile-request-context.test.ts b/graphql/server/src/middleware/__tests__/graphile-request-context.test.ts new file mode 100644 index 0000000000..54c07a9efe --- /dev/null +++ b/graphql/server/src/middleware/__tests__/graphile-request-context.test.ts @@ -0,0 +1,122 @@ +import { buildPgSettings } from '@constructive-io/express-context'; +import type { Request } from 'express'; + +import { getGraphileRequestPgSettings } from '../graphile-request-context'; + +const baseApi = { + apiId: 'api-1', + databaseId: 'database-1', + dbname: 'tenant_db', + anonRole: 'anonymous_runtime', + roleName: 'authenticated_runtime', + schema: ['tenant_api'], + isPublic: true, +}; + +function makeRequest( + overrides: Record = {}, + headers: Record = {} +): Request { + const normalizedHeaders = Object.fromEntries( + Object.entries(headers).map(([key, value]) => [key.toLowerCase(), value]) + ); + return { + get: (name: string) => normalizedHeaders[name.toLowerCase()], + ...overrides, + } as unknown as Request; +} + +describe('Graphile canonical request context', () => { + it.each([ + ['anonymous', null], + ['authenticated', { id: 'token-1', user_id: 'user-1' }], + ])('reuses the exact canonical object for %s requests', (_label, token) => { + const pgSettings = buildPgSettings({ + api: baseApi, + token, + requestId: 'request-1', + }); + const req = makeRequest({ + api: baseApi, + token, + constructive: { pgSettings }, + }); + + expect(getGraphileRequestPgSettings(req)).toBe(pgSettings); + }); + + it('does not derive identity from unauthenticated private headers', () => { + const privateApi = { ...baseApi, isPublic: false }; + const pgSettings = buildPgSettings({ + api: privateApi, + token: null, + requestId: 'request-private', + clientIp: '192.0.2.8', + }); + const req = makeRequest( + { + api: privateApi, + token: null, + constructive: { pgSettings }, + }, + { + 'X-Actor-Id': 'actor-1', + 'X-Entity-Id': 'entity-1', + 'X-Organization-Id': 'organization-1', + } + ); + + expect(getGraphileRequestPgSettings(req)).toBe(pgSettings); + expect(pgSettings.role).toBe('anonymous_runtime'); + expect(pgSettings['jwt.claims.user_id']).toBe(''); + expect(pgSettings['jwt.claims.entity_id']).toBe(''); + expect(pgSettings['jwt.claims.organization_id']).toBe(''); + }); + + it('does not trust private identity headers on a public surface', () => { + const pgSettings = buildPgSettings({ + api: baseApi, + token: null, + requestId: 'request-public', + }); + const req = makeRequest( + { api: baseApi, token: null, constructive: { pgSettings } }, + { 'X-Actor-Id': 'attacker-controlled' } + ); + + expect(getGraphileRequestPgSettings(req)).toBe(pgSettings); + expect(pgSettings['jwt.claims.user_id']).toBe(''); + }); + + it('does not replace authenticated identity on a private surface', () => { + const privateApi = { ...baseApi, isPublic: false }; + const token = { id: 'token-1', user_id: 'token-user' }; + const pgSettings = buildPgSettings({ + api: privateApi, + token, + requestId: 'request-authenticated-private', + }); + const req = makeRequest( + { api: privateApi, token, constructive: { pgSettings } }, + { 'X-Actor-Id': 'header-user' } + ); + + expect(getGraphileRequestPgSettings(req)).toBe(pgSettings); + expect(pgSettings['jwt.claims.user_id']).toBe('token-user'); + }); + + it.each([ + ['missing request', undefined], + ['missing constructive context', makeRequest()], + [ + 'incomplete settings', + makeRequest({ + constructive: { pgSettings: { role: 'anonymous_runtime' } }, + }), + ], + ])('fails closed for %s', (_label, req) => { + expect(() => getGraphileRequestPgSettings(req)).toThrow( + /req\.constructive\.pgSettings/ + ); + }); +}); diff --git a/graphql/server/src/middleware/graphile-request-context.ts b/graphql/server/src/middleware/graphile-request-context.ts new file mode 100644 index 0000000000..cd73592112 --- /dev/null +++ b/graphql/server/src/middleware/graphile-request-context.ts @@ -0,0 +1,19 @@ +import { + assertCompletePgSettings, + type PgSettings, +} from '@constructive-io/express-context'; +import type { Request } from 'express'; + +/** + * Read the canonical request context assembled by express-context. + * + * Identity-bearing private headers remain inert until an authenticated + * internal-ingress boundary owns their translation into trusted claims. + */ +export function getGraphileRequestPgSettings( + req: Request | undefined +): PgSettings { + const canonical = req?.constructive?.pgSettings; + assertCompletePgSettings(canonical, 'req.constructive.pgSettings'); + return canonical; +} diff --git a/graphql/server/src/middleware/graphile.ts b/graphql/server/src/middleware/graphile.ts index e6de98f7ad..2a4618b7b3 100644 --- a/graphql/server/src/middleware/graphile.ts +++ b/graphql/server/src/middleware/graphile.ts @@ -21,6 +21,7 @@ import { HandlerCreationError } from '../errors/api-errors'; import { respondWithGraphQLError } from '../errors/graphql-response'; import { AuthCookiePlugin } from '../plugins/auth-cookie-plugin'; import type { DatabaseSettings } from '../types'; +import { getGraphileRequestPgSettings } from './graphile-request-context'; import { observeGraphileBuild } from './observability/graphile-build-stats'; const maskErrorLog = new Logger('graphile:maskError'); @@ -163,7 +164,6 @@ const reqLabel = (req: Request): string => (req.requestId ? `[${req.requestId}]` const buildPreset = ( pool: import('pg').Pool, schemas: string[], - anonRole: string, roleName: string, databaseSettings?: DatabaseSettings, apiId?: string, @@ -209,110 +209,11 @@ const buildPreset = ( grafast: { explain: process.env.NODE_ENV === 'development', context: (requestContext: Partial) => { - // In grafserv/express/v4, the request is available at requestContext.expressv4.req - const req = (requestContext as { expressv4?: { req?: Request } })?.expressv4?.req; - const context: Record = {}; - - if (req) { - if (req.databaseId) { - context['jwt.claims.database_id'] = req.databaseId; - } - // API provenance — which API surface this request arrived through. - // Derived server-side by resolving the hostname through the scoped - // routing plane (resolve_route -> api_id); never taken from - // client-supplied headers, body, or token payload. - if (req.api?.apiId) { - context['jwt.claims.api_id'] = req.api.apiId; - } - if (req.clientIp) { - context['jwt.claims.ip_address'] = req.clientIp; - } - if (req.get('origin')) { - context['jwt.claims.origin'] = req.get('origin') as string; - } - if (req.get('User-Agent')) { - context['jwt.claims.user_agent'] = req.get('User-Agent') as string; - } - if (req.deviceToken) { - context['jwt.claims.device_token'] = req.deviceToken; - } - - if (req.token?.user_id) { - const pgSettings: Record = { - role: roleName, - 'jwt.claims.token_id': req.token.id, - 'jwt.claims.user_id': req.token.user_id, - ...context - }; - - if (req.token.session_id) { - pgSettings['jwt.claims.session_id'] = req.token.session_id; - } - - // Propagate credential metadata as JWT claims so PG functions - // can read them via current_setting('jwt.claims.access_level') etc. - if (req.token.access_level) { - pgSettings['jwt.claims.access_level'] = req.token.access_level; - } - if (req.token.kind) { - pgSettings['jwt.claims.kind'] = req.token.kind; - } - - // Principal identity — always set; equals user_id for human sessions - pgSettings['jwt.claims.principal_id'] = req.token.principal_id || req.token.user_id; - - // Enforce read-only transactions for read_only credentials - if (req.token.access_level === 'read_only') { - pgSettings['default_transaction_read_only'] = 'on'; - } - - if (req.requestId) { - pgSettings['request.id'] = req.requestId; - } - - return { pgSettings }; - } - - // Private (in-cluster) surface: there is no token — identity - // arrives on the trusted internal X-* headers stamped by the - // dispatching worker/sync gateway (the same vocabulary as - // X-Database-Id above). Map it into per-request claims so writes - // made through this surface carry actor attribution. Never applied - // on the public surface, where client-supplied identity headers - // must not assert identity. - const headerActorId = req.get('X-Actor-Id'); - if (req.api?.isPublic === false && headerActorId) { - const pgSettings: Record = { - role: roleName, - 'jwt.claims.user_id': headerActorId, - 'jwt.claims.principal_id': headerActorId, - ...context - }; - const headerEntityId = req.get('X-Entity-Id'); - if (headerEntityId) { - pgSettings['jwt.claims.entity_id'] = headerEntityId; - } - const headerOrganizationId = req.get('X-Organization-Id'); - if (headerOrganizationId) { - pgSettings['jwt.claims.organization_id'] = headerOrganizationId; - } - if (req.requestId) { - pgSettings['request.id'] = req.requestId; - } - return { pgSettings }; - } - } - - const anonSettings: Record = { - role: anonRole, - ...context - }; - if (req?.requestId) { - anonSettings['request.id'] = req.requestId; - } - + // In grafserv/express/v4, the request is available at requestContext.expressv4.req + const req = (requestContext as { expressv4?: { req?: Request } }) + ?.expressv4?.req; return { - pgSettings: anonSettings + pgSettings: getGraphileRequestPgSettings(req), }; } } @@ -402,8 +303,17 @@ export const graphile = (opts: ConstructiveOptions): RequestHandler => { const pool = getPgPool(pgConfig); // Create promise and store in in-flight map BEFORE try block - const compute = api.apiId ? await req.constructive?.useModule('compute') : undefined; - const preset = buildPreset(pool, schema || [], anonRole, roleName, api.databaseSettings, api.apiId, compute); + const compute = api.apiId + ? await req.constructive?.useModule('compute') + : undefined; + const preset = buildPreset( + pool, + schema || [], + roleName, + api.databaseSettings, + api.apiId, + compute + ); const creationPromise = observeGraphileBuild( { cacheKey: key, diff --git a/graphql/server/src/middleware/types.ts b/graphql/server/src/middleware/types.ts index 5b0868f764..146074f3fb 100644 --- a/graphql/server/src/middleware/types.ts +++ b/graphql/server/src/middleware/types.ts @@ -1,14 +1,7 @@ -import type { ApiStructure } from '../types'; - -export type ConstructiveAPIToken = { - id?: string; - user_id?: string; - principal_id?: string; - session_id?: string; - access_level?: string; - kind?: string; - [key: string]: unknown; -}; +import type { + ApiStructure, + ConstructiveAPIToken, +} from '@constructive-io/express-context'; declare global { namespace Express { diff --git a/graphql/server/src/server.ts b/graphql/server/src/server.ts index ea7f1e94c8..6f2a826ea9 100644 --- a/graphql/server/src/server.ts +++ b/graphql/server/src/server.ts @@ -165,6 +165,7 @@ class Server { app.use(authenticate); app.use(createContextMiddleware({ pg: effectiveOpts.pg, + dependencySchemas: effectiveOpts.graphile?.introspectionDependencySchemas, loaders: createDefaultRegistry(), routingSchema: getRoutingSchema(effectiveOpts) })); diff --git a/graphql/types/src/graphile.ts b/graphql/types/src/graphile.ts index 72fff4c739..bf46b3c220 100644 --- a/graphql/types/src/graphile.ts +++ b/graphql/types/src/graphile.ts @@ -6,6 +6,8 @@ import type { GraphileConfig } from 'graphile-config'; export interface GraphileOptions { /** Database schema(s) to expose through GraphQL */ schema?: string | string[]; + /** Ordered extension/shared schemas required by Graphile and request SQL. */ + introspectionDependencySchemas?: string[]; /** Additional presets to extend */ extends?: GraphileConfig.Preset[]; /** Preset overrides */ @@ -51,6 +53,7 @@ export interface ApiOptions { */ export const graphileDefaults: GraphileOptions = { schema: [], + introspectionDependencySchemas: [], extends: [], preset: {} }; diff --git a/packages/express-context/__tests__/context-pg-settings.test.ts b/packages/express-context/__tests__/context-pg-settings.test.ts new file mode 100644 index 0000000000..a24a21cc0b --- /dev/null +++ b/packages/express-context/__tests__/context-pg-settings.test.ts @@ -0,0 +1,46 @@ +import type { Request } from 'express'; + +import { buildContext } from '../src/context'; + +jest.mock('pg-cache', () => ({ + getPgPool: jest.fn(() => ({ query: jest.fn(), connect: jest.fn() })), +})); + +describe('buildContext pgSettings forwarding', () => { + it('forwards server-owned HTTP metadata into the canonical builder', () => { + const headers: Record = { + origin: 'https://app.example.test', + 'user-agent': 'context-test/1.0', + }; + const req = { + api: { + apiId: 'api-1', + databaseId: 'database-1', + dbname: 'tenant_db', + anonRole: 'anonymous_runtime', + roleName: 'authenticated_runtime', + schema: ['tenant_api'], + }, + token: { user_id: 'user-1' }, + requestId: 'request-1', + clientIp: '192.0.2.4', + deviceToken: 'device-1', + get: (name: string) => headers[name.toLowerCase()], + } as unknown as Request; + + const context = buildContext(req, { dependencySchemas: ['shared_api'] }); + + expect(context?.pgSettings).toMatchObject({ + role: 'authenticated_runtime', + 'request.id': 'request-1', + 'jwt.claims.user_id': 'user-1', + 'jwt.claims.api_id': 'api-1', + 'jwt.claims.database_id': 'database-1', + 'jwt.claims.ip_address': '192.0.2.4', + 'jwt.claims.origin': 'https://app.example.test', + 'jwt.claims.user_agent': 'context-test/1.0', + 'jwt.claims.device_token': 'device-1', + search_path: 'pg_catalog, "shared_api", "tenant_api"', + }); + }); +}); diff --git a/packages/express-context/__tests__/pg-settings.test.ts b/packages/express-context/__tests__/pg-settings.test.ts index 6b87b83b15..62242086f4 100644 --- a/packages/express-context/__tests__/pg-settings.test.ts +++ b/packages/express-context/__tests__/pg-settings.test.ts @@ -1,51 +1,232 @@ -import { buildPgSettings } from '../src/pg-settings'; +import { + assertCompletePgSettings, + buildPgSettings, + REQUIRED_PG_SETTING_KEYS, + SECURITY_GUC_KEYS, + withPgSettingsRole, + withTrustedPgClaims, +} from '../src/pg-settings'; import type { ApiStructure, ConstructiveAPIToken } from '../src/types'; const api: ApiStructure = { - apiId: '6c9997a4-591b-4cb3-9313-4ef45d6f134e', + apiId: 'api-1', dbname: 'testdb', - anonRole: 'anonymous', - roleName: 'authenticated', - schema: ['public'], + anonRole: 'anonymous_runtime', + roleName: 'authenticated_runtime', + schema: ['public', 'app'], domains: [], - databaseId: '80a2eaaf-f77e-4bfe-8506-df929ef1b8d9', - isPublic: true + databaseId: 'database-1', + isPublic: true, }; -describe('buildPgSettings — jwt.claims.api_id provenance', () => { - it('sets jwt.claims.api_id from the resolved api for anonymous requests', () => { - const settings = buildPgSettings({ api, token: null, requestId: 'r1' }); +const token: ConstructiveAPIToken = { + id: 'token-1', + user_id: 'user-1', + principal_id: 'principal-1', + session_id: 'session-1', + access_level: 'read_only', + kind: 'api_token', + email: 'primary@example.test', + user_email: 'user@example.test', + entity_id: 'entity-1', + organization_id: 'organization-1', + tenant_id: 'tenant-1', + role_type: 'member', +}; + +describe('buildPgSettings', () => { + it('builds a value-complete anonymous request context', () => { + const settings = buildPgSettings({ + api, + token: null, + requestId: 'request-1', + }); - expect(settings['jwt.claims.api_id']).toBe(api.apiId); - expect(settings['role']).toBe('anonymous'); + expect(Object.keys(settings)).toEqual( + expect.arrayContaining(REQUIRED_PG_SETTING_KEYS) + ); + expect(settings.role).toBe('anonymous_runtime'); + expect(settings['request.id']).toBe('request-1'); + expect(settings.transaction_read_only).toBe('off'); + expect(settings.row_security).toBe('on'); + expect(settings.search_path).toBe('pg_catalog, "public", "app"'); + for (const key of SECURITY_GUC_KEYS) { + expect(typeof settings[key]).toBe('string'); + } + expect(settings['jwt.claims.user_id']).toBe(''); + expect(settings['jwt.claims.api_id']).toBe('api-1'); + expect(settings['jwt.claims.database_id']).toBe('database-1'); + expect(() => assertCompletePgSettings(settings)).not.toThrow(); }); - it('sets jwt.claims.api_id from the resolved api for authenticated requests', () => { - const token = { user_id: 'u1' } as ConstructiveAPIToken; - const settings = buildPgSettings({ api, token, requestId: 'r1' }); + it('maps every supported authenticated claim and trusted request fact', () => { + const settings = buildPgSettings({ + api, + token, + requestId: 'request-2', + clientIp: '192.0.2.10', + origin: 'https://app.example.test', + userAgent: 'test-agent/1.0', + deviceToken: 'device-1', + }); - expect(settings['jwt.claims.api_id']).toBe(api.apiId); - expect(settings['role']).toBe('authenticated'); - expect(settings['jwt.claims.user_id']).toBe('u1'); + expect(settings).toMatchObject({ + role: 'authenticated_runtime', + 'request.id': 'request-2', + transaction_read_only: 'on', + row_security: 'on', + 'jwt.claims.token_id': 'token-1', + 'jwt.claims.user_id': 'user-1', + 'jwt.claims.principal_id': 'principal-1', + 'jwt.claims.session_id': 'session-1', + 'jwt.claims.access_level': 'read_only', + 'jwt.claims.kind': 'api_token', + 'jwt.claims.email': 'primary@example.test', + 'jwt.claims.user_email': 'user@example.test', + 'jwt.claims.entity_id': 'entity-1', + 'jwt.claims.organization_id': 'organization-1', + 'jwt.claims.tenant_id': 'tenant-1', + 'jwt.claims.role_type': 'member', + 'jwt.claims.api_id': 'api-1', + 'jwt.claims.database_id': 'database-1', + 'jwt.claims.ip_address': '192.0.2.10', + 'jwt.claims.origin': 'https://app.example.test', + 'jwt.claims.user_agent': 'test-agent/1.0', + 'jwt.claims.device_token': 'device-1', + }); }); - it('omits jwt.claims.api_id when the api has no apiId (non-API surface)', () => { + it('represents every unavailable claim with an empty string', () => { const settings = buildPgSettings({ - api: { ...api, apiId: undefined }, + api: { ...api, apiId: undefined, databaseId: undefined }, token: null, - requestId: 'r1' + requestId: '', }); - expect(settings['jwt.claims.api_id']).toBeUndefined(); + for (const key of SECURITY_GUC_KEYS) { + expect(settings[key]).toBe(''); + } + }); + + it('does not retain authenticated claims or read-only state in a later anonymous request', () => { + const authenticated = buildPgSettings({ + api, + token, + requestId: 'authenticated', + }); + const anonymous = buildPgSettings({ + api, + token: null, + requestId: 'anonymous', + }); + + expect(authenticated['jwt.claims.user_id']).toBe('user-1'); + expect(authenticated.transaction_read_only).toBe('on'); + expect(anonymous['jwt.claims.user_id']).toBe(''); + expect(anonymous['jwt.claims.token_id']).toBe(''); + expect(anonymous.transaction_read_only).toBe('off'); + }); + + it('returns an independent object for every request', () => { + const first = buildPgSettings({ api, token: null, requestId: 'first' }); + const second = buildPgSettings({ api, token: null, requestId: 'second' }); + + expect(first).not.toBe(second); + first['jwt.claims.user_id'] = 'mutated'; + expect(second['jwt.claims.user_id']).toBe(''); + }); + + it('builds a deterministic, deduplicated and quoted search path', () => { + const settings = buildPgSettings({ + api, + token: null, + requestId: 'request-3', + dependencySchemas: ['shared', 'strange"name', 'public', 'shared'], + }); + + expect(settings.search_path).toBe( + 'pg_catalog, "shared", "strange""name", "public", "app"' + ); + }); + + it('derives principal_id from user_id when no explicit principal exists', () => { + const settings = buildPgSettings({ + api, + token: { user_id: 'user-fallback' }, + requestId: 'request-4', + }); + + expect(settings['jwt.claims.principal_id']).toBe('user-fallback'); + }); + + it('accepts allowlisted trusted claims without mutating either input', () => { + const trustedClaims = { 'jwt.claims.entity_id': 'trusted-entity' } as const; + const settings = buildPgSettings({ + api, + token: null, + requestId: 'request-5', + trustedClaims, + }); + const derived = withTrustedPgClaims(settings, { + 'jwt.claims.user_id': 'trusted-user', + }); + + expect(settings['jwt.claims.entity_id']).toBe('trusted-entity'); + expect(settings['jwt.claims.user_id']).toBe(''); + expect(derived['jwt.claims.user_id']).toBe('trusted-user'); + expect(derived).not.toBe(settings); + expect(trustedClaims).toEqual({ 'jwt.claims.entity_id': 'trusted-entity' }); + }); + + it.each([ + ['arbitrary setting', { role: 'postgres' }], + ['non-string value', { 'jwt.claims.user_id': null }], + ['array', []], + ['null', null], + ])('rejects invalid trusted claims: %s', (_label, trustedClaims) => { + expect(() => + buildPgSettings({ + api, + token: null, + requestId: 'request-6', + trustedClaims: trustedClaims as never, + }) + ).toThrow(TypeError); + }); + + it('rejects symbol and accessor properties in trusted claims', () => { + const symbolClaims = { 'jwt.claims.user_id': 'user' } as Record< + PropertyKey, + unknown + >; + symbolClaims[Symbol('claim')] = 'hidden'; + expect(() => + withTrustedPgClaims( + buildPgSettings({ api, token: null, requestId: 'request-7' }), + symbolClaims + ) + ).toThrow('must not contain symbol properties'); + + const accessorClaims = {}; + Object.defineProperty(accessorClaims, 'jwt.claims.user_id', { + enumerable: true, + get: () => 'user', + }); + expect(() => + withTrustedPgClaims( + buildPgSettings({ api, token: null, requestId: 'request-8' }), + accessorClaims + ) + ).toThrow('must be a string data property'); }); - it('is derived only from the resolved api, never from the token', () => { - const token = { - user_id: 'u1', - api_id: 'attacker-controlled' - } as unknown as ConstructiveAPIToken; - const settings = buildPgSettings({ api, token, requestId: 'r1' }); + it('copies a complete context when switching role and rejects invalid roles', () => { + const settings = buildPgSettings({ api, token, requestId: 'request-9' }); + const anonymous = withPgSettingsRole(settings, 'anonymous_runtime'); - expect(settings['jwt.claims.api_id']).toBe(api.apiId); + expect(anonymous).toEqual({ ...settings, role: 'anonymous_runtime' }); + expect(anonymous).not.toBe(settings); + expect(settings.role).toBe('authenticated_runtime'); + expect(() => withPgSettingsRole(settings, '')).toThrow('non-empty string'); }); }); diff --git a/packages/express-context/src/context.ts b/packages/express-context/src/context.ts index 82d87de9d3..87ece44dca 100644 --- a/packages/express-context/src/context.ts +++ b/packages/express-context/src/context.ts @@ -34,6 +34,8 @@ export interface ContextMiddlewareOptions { loaders?: LoaderRegistry; /** Routing-plane schema loaders query (defaults to routing_public) */ routingSchema?: string; + /** Ordered, audited extension/shared schemas required by request SQL. */ + dependencySchemas?: readonly string[]; } /** @@ -77,7 +79,11 @@ export function buildContext( api, token, requestId, - clientIp: req.clientIp + clientIp: req.clientIp, + origin: req.get('origin'), + userAgent: req.get('User-Agent'), + deviceToken: req.deviceToken, + dependencySchemas: opts.dependencySchemas, }); const tenantPool: Pool = getPgPool({ diff --git a/packages/express-context/src/index.ts b/packages/express-context/src/index.ts index 013e195f5d..d02109b158 100644 --- a/packages/express-context/src/index.ts +++ b/packages/express-context/src/index.ts @@ -64,9 +64,23 @@ export type { export type { BillingClient, InferenceLogEntry } from './billing-client'; export { createBillingClient } from './billing-client'; -// pgSettings builder -export type { PgSettingsInput } from './pg-settings'; -export { buildPgSettings } from './pg-settings'; +// pgSettings builder and validation contract +export type { + PgSettings, + PgSettingsInput, + RequiredPgSettingKey, + SecurityGucKey, + TrustedPgClaims, +} from './pg-settings'; +export { + assertCompletePgSettings, + assertPgSettings, + buildPgSettings, + REQUIRED_PG_SETTING_KEYS, + SECURITY_GUC_KEYS, + withPgSettingsRole, + withTrustedPgClaims, +} from './pg-settings'; // withPgClient helper export { withPgClient } from './pg-client'; diff --git a/packages/express-context/src/pg-settings.ts b/packages/express-context/src/pg-settings.ts index cb86336456..4044ca92ca 100644 --- a/packages/express-context/src/pg-settings.ts +++ b/packages/express-context/src/pg-settings.ts @@ -1,76 +1,217 @@ /** - * pg-settings — Build pgSettings from resolved API + auth token + * Canonical PostgreSQL request settings. * - * pgSettings are key-value pairs passed to PostgreSQL via SET LOCAL - * within each transaction. They carry the JWT claims, role, database_id, - * and request_id so that RLS policies and SQL functions can reference - * the current user context via `current_setting('jwt.claims.user_id')`. - * - * This module extracts the pgSettings construction so it's reusable - * across the PostGraphile server, LLM sidecar, or any Express service. + * Every request receives a value-complete security context. Missing claims are + * represented by empty strings so a reused execution path cannot accidentally + * retain facts from an earlier request. */ import type { ApiStructure, ConstructiveAPIToken } from './types'; +export const SECURITY_GUC_KEYS = [ + 'jwt.claims.access_level', + 'jwt.claims.api_id', + 'jwt.claims.database_id', + 'jwt.claims.device_token', + 'jwt.claims.email', + 'jwt.claims.entity_id', + 'jwt.claims.ip_address', + 'jwt.claims.kind', + 'jwt.claims.organization_id', + 'jwt.claims.origin', + 'jwt.claims.principal_id', + 'jwt.claims.role_type', + 'jwt.claims.session_id', + 'jwt.claims.tenant_id', + 'jwt.claims.token_id', + 'jwt.claims.user_agent', + 'jwt.claims.user_email', + 'jwt.claims.user_id', +] as const; + +export const REQUIRED_PG_SETTING_KEYS = [ + ...SECURITY_GUC_KEYS, + 'role', + 'request.id', + 'transaction_read_only', + 'search_path', + 'row_security', +] as const; + +export type SecurityGucKey = (typeof SECURITY_GUC_KEYS)[number]; +export type RequiredPgSettingKey = (typeof REQUIRED_PG_SETTING_KEYS)[number]; +export type PgSettings = Record; +export type TrustedPgClaims = Partial>; + +const SECURITY_GUC_KEY_SET: ReadonlySet = new Set(SECURITY_GUC_KEYS); + +function assertStringDataProperties( + value: unknown, + label: string +): asserts value is Record { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new TypeError(`${label} must be an object of string data properties`); + } + + for (const key of Reflect.ownKeys(value)) { + if (typeof key !== 'string') { + throw new TypeError(`${label} must not contain symbol properties`); + } + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if ( + !descriptor || + !('value' in descriptor) || + typeof descriptor.value !== 'string' + ) { + throw new TypeError(`${label}.${key} must be a string data property`); + } + } +} + +/** Validate the value shape accepted by Graphile's `withPgClient`. */ +export function assertPgSettings( + value: unknown, + label = 'pgSettings' +): asserts value is PgSettings { + assertStringDataProperties(value, label); +} + +/** Validate that a request carries the complete canonical settings contract. */ +export function assertCompletePgSettings( + value: unknown, + label = 'pgSettings' +): asserts value is PgSettings { + assertStringDataProperties(value, label); + for (const key of REQUIRED_PG_SETTING_KEYS) { + if (!Object.prototype.hasOwnProperty.call(value, key)) { + throw new TypeError(`${label} is missing required setting '${key}'`); + } + } +} + +function copyTrustedClaims(claims: unknown, label: string): TrustedPgClaims { + assertStringDataProperties(claims, label); + const copy: TrustedPgClaims = {}; + for (const key of Object.keys(claims)) { + if (!SECURITY_GUC_KEY_SET.has(key)) { + throw new TypeError( + `${label} contains unsupported security GUC '${key}'` + ); + } + copy[key as SecurityGucKey] = claims[key]; + } + return copy; +} + +/** Add server-owned claims without admitting arbitrary PostgreSQL settings. */ +export function withTrustedPgClaims( + pgSettings: unknown, + trustedClaims: unknown +): PgSettings { + assertCompletePgSettings(pgSettings); + return { + ...pgSettings, + ...copyTrustedClaims(trustedClaims, 'trustedClaims'), + }; +} + +/** Copy a complete request context while changing only its execution role. */ +export function withPgSettingsRole( + pgSettings: unknown, + role: string +): PgSettings { + assertCompletePgSettings(pgSettings); + if (typeof role !== 'string' || role.length === 0) { + throw new TypeError('role must be a non-empty string'); + } + return { ...pgSettings, role }; +} + export interface PgSettingsInput { - /** Resolved API config (provides role names, database_id) */ + /** Resolved API config (provides role names, database_id, physical schemas). */ api: ApiStructure; - /** Authenticated token (null for anonymous) */ + /** Authenticated token (null for anonymous). */ token: ConstructiveAPIToken | null; - /** Per-request correlation ID */ + /** Per-request correlation ID. */ requestId: string; - /** Client IP address (from request-ip middleware) */ + /** Client IP address resolved by server middleware. */ clientIp?: string; + /** Origin header captured by the server. */ + origin?: string; + /** User-Agent header captured by the server. */ + userAgent?: string; + /** Trusted device cookie resolved by authentication middleware. */ + deviceToken?: string; + /** Server-derived claims for an existing trusted private surface. */ + trustedClaims?: TrustedPgClaims; + /** Ordered, audited extension/shared schemas required by request SQL. */ + dependencySchemas?: readonly string[]; } -/** - * Build pgSettings from the resolved API + auth token. - * - * These settings are applied via SET LOCAL in each transaction, - * making them available to RLS policies and SQL functions. - */ -export function buildPgSettings(input: PgSettingsInput): Record { - const { api, token, requestId, clientIp } = input; - const settings: Record = {}; - - // Role: from token (authenticated) or api (anonymous fallback) - if (token?.user_id) { - settings['role'] = api.roleName || 'authenticated'; - settings['jwt.claims.user_id'] = token.user_id; - } else { - settings['role'] = api.anonRole || 'anonymous'; - } +const quoteIdentifier = (identifier: string): string => + `"${identifier.replace(/"/g, '""')}"`; - // Session claims - if (token?.session_id) { - settings['jwt.claims.session_id'] = token.session_id; - } +function setClaim( + settings: PgSettings, + key: SecurityGucKey, + value: unknown +): void { + if (typeof value === 'string') settings[key] = value; +} - // Principal identity (service accounts / bots) - if (token?.principal_id) { - settings['jwt.claims.principal_id'] = token.principal_id; - } +/** Build a fresh, complete PostgreSQL security context for one request. */ +export function buildPgSettings(input: PgSettingsInput): PgSettings { + const { api, token, requestId, clientIp, origin, userAgent, deviceToken } = + input; + const settings: PgSettings = Object.fromEntries( + SECURITY_GUC_KEYS.map((key) => [key, '']) + ); - // Database context - if (api.databaseId) { - settings['jwt.claims.database_id'] = api.databaseId; - } + settings.role = token?.user_id + ? api.roleName || 'authenticated' + : api.anonRole || 'anonymous'; - // API provenance — which API surface this request arrived through. - // Derived server-side by resolving the hostname through the scoped routing - // plane (resolve_route -> api_id); never taken from client-supplied headers, - // body, or token payload. - if (api.apiId) { - settings['jwt.claims.api_id'] = api.apiId; + setClaim(settings, 'jwt.claims.token_id', token?.id); + setClaim(settings, 'jwt.claims.user_id', token?.user_id); + setClaim(settings, 'jwt.claims.session_id', token?.session_id); + setClaim(settings, 'jwt.claims.access_level', token?.access_level); + setClaim(settings, 'jwt.claims.kind', token?.kind); + setClaim(settings, 'jwt.claims.email', token?.email); + setClaim(settings, 'jwt.claims.user_email', token?.user_email); + setClaim(settings, 'jwt.claims.entity_id', token?.entity_id); + setClaim(settings, 'jwt.claims.organization_id', token?.organization_id); + setClaim(settings, 'jwt.claims.tenant_id', token?.tenant_id); + setClaim(settings, 'jwt.claims.role_type', token?.role_type); + setClaim( + settings, + 'jwt.claims.principal_id', + token?.principal_id || token?.user_id + ); + setClaim(settings, 'jwt.claims.database_id', api.databaseId); + setClaim(settings, 'jwt.claims.api_id', api.apiId); + setClaim(settings, 'jwt.claims.ip_address', clientIp); + setClaim(settings, 'jwt.claims.origin', origin); + setClaim(settings, 'jwt.claims.user_agent', userAgent); + setClaim(settings, 'jwt.claims.device_token', deviceToken); + + if (input.trustedClaims !== undefined) { + Object.assign( + settings, + copyTrustedClaims(input.trustedClaims, 'trustedClaims') + ); } - // Distributed tracing settings['request.id'] = requestId; + settings.transaction_read_only = + token?.access_level === 'read_only' ? 'on' : 'off'; + settings.row_security = 'on'; - // Client metadata (for audit functions) - if (clientIp) { - settings['jwt.claims.ip_address'] = clientIp; - } + const physicalSchemas = [...(input.dependencySchemas ?? []), ...api.schema]; + settings.search_path = [ + 'pg_catalog', + ...[...new Set(physicalSchemas)].map(quoteIdentifier), + ].join(', '); + assertCompletePgSettings(settings, 'built pgSettings'); return settings; } diff --git a/packages/express-context/src/types.ts b/packages/express-context/src/types.ts index 4316018209..fd6258357c 100644 --- a/packages/express-context/src/types.ts +++ b/packages/express-context/src/types.ts @@ -163,6 +163,12 @@ export type ConstructiveAPIToken = { session_id?: string; access_level?: string; kind?: string; + email?: string; + user_email?: string; + entity_id?: string; + organization_id?: string; + tenant_id?: string; + role_type?: string; [key: string]: unknown; }; @@ -340,6 +346,7 @@ declare global { clientIp?: string; requestId?: string; token?: ConstructiveAPIToken; + deviceToken?: string; constructive?: ConstructiveContext; } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9ba0972089..7e4a836327 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -620,6 +620,9 @@ importers: graphile/graphile-i18n: dependencies: + '@constructive-io/express-context': + specifier: workspace:^ + version: link:../../packages/express-context/dist '@dataplan/pg': specifier: 1.1.1 version: 1.1.1(@dataplan/json@1.0.1(grafast@1.1.2(graphql@16.13.0)))(grafast@1.1.2(graphql@16.13.0))(graphile-config@1.1.0)(graphql@16.13.0)(pg-sql2@5.0.1)(pg@8.21.0) @@ -1210,6 +1213,9 @@ importers: '@constructive-io/bucket-provisioner': specifier: workspace:^ version: link:../../packages/bucket-provisioner/dist + '@constructive-io/express-context': + specifier: workspace:^ + version: link:../../packages/express-context/dist '@constructive-io/graphql-env': specifier: workspace:^ version: link:../../graphql/env/dist @@ -1324,9 +1330,6 @@ importers: pg: specifier: ^8.21.0 version: 8.21.0 - pg-query-context: - specifier: workspace:^ - version: link:../../postgres/pg-query-context/dist pg-sql2: specifier: 5.0.1 version: 5.0.1 @@ -3442,6 +3445,9 @@ importers: makage: specifier: ^0.3.0 version: 0.3.0 + pgsql-test: + specifier: workspace:^ + version: link:../pgsql-test/dist publishDirectory: dist postgres/pg-seed: diff --git a/postgres/pg-query-context/package.json b/postgres/pg-query-context/package.json index 62c4b6f241..ce07870dc9 100644 --- a/postgres/pg-query-context/package.json +++ b/postgres/pg-query-context/package.json @@ -33,7 +33,8 @@ }, "devDependencies": { "@types/pg": "^8.20.4", - "makage": "^0.3.0" + "makage": "^0.3.0", + "pgsql-test": "workspace:^" }, "keywords": [ "postgresql", diff --git a/postgres/pg-query-context/src/__tests__/index.test.ts b/postgres/pg-query-context/src/__tests__/index.test.ts new file mode 100644 index 0000000000..2cb18a32cb --- /dev/null +++ b/postgres/pg-query-context/src/__tests__/index.test.ts @@ -0,0 +1,174 @@ +import type { Pool, PoolClient } from 'pg'; + +import pgQueryContext, { + UNSAFE_POOLED_CONTEXT_ERROR_CODE, + UnsafePooledContextError, + withPgClient, +} from '../index'; + +const SETTINGS_SQL = + 'SELECT pg_catalog.set_config(setting->>0, setting->>1, true) ' + + 'FROM pg_catalog.json_array_elements($1::json) AS setting'; + +const makePool = () => { + const client = { + query: jest.fn(async () => ({ rows: [] as unknown[] })), + release: jest.fn(), + } as unknown as PoolClient; + const pool = { + connect: jest.fn(async () => client), + totalCount: 1, + } as unknown as Pool; + return { client, pool }; +}; + +describe('pg query context', () => { + it('applies the complete ordered context in one parameterized round trip', async () => { + const { client, pool } = makePool(); + const context = { + 'jwt.claims.user_id': '', + role: 'tenant_runtime', + transaction_read_only: 'off', + search_path: 'pg_catalog, "tenant_api"', + row_security: 'on', + }; + const callback = jest.fn(async () => 'ok'); + + await expect(withPgClient(pool, context, callback)).resolves.toBe('ok'); + + expect(client.query).toHaveBeenNthCalledWith(1, 'BEGIN'); + expect(client.query).toHaveBeenNthCalledWith(2, SETTINGS_SQL, [ + JSON.stringify(Object.entries(context)), + ]); + expect(client.query).toHaveBeenNthCalledWith(3, 'COMMIT'); + expect(callback).toHaveBeenCalledWith(client); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it('does not issue a context query for an empty context', async () => { + const { client, pool } = makePool(); + + await withPgClient(pool, {}, async (): Promise => undefined); + + expect(client.query).toHaveBeenCalledTimes(2); + expect(client.query).toHaveBeenNthCalledWith(1, 'BEGIN'); + expect(client.query).toHaveBeenNthCalledWith(2, 'COMMIT'); + }); + + it('fails closed instead of coercing non-string security settings', async () => { + const { client, pool } = makePool(); + + await expect( + withPgClient( + pool, + { 'jwt.claims.user_id': null } as unknown as Record, + async (): Promise => undefined + ) + ).rejects.toThrow( + "PostgreSQL context setting 'jwt.claims.user_id' must be a string" + ); + + expect(client.query).toHaveBeenNthCalledWith(1, 'BEGIN'); + expect(client.query).toHaveBeenNthCalledWith(2, 'ROLLBACK'); + expect(client.query).toHaveBeenCalledTimes(2); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it('rolls back and releases when the batched context is rejected', async () => { + const { client, pool } = makePool(); + (client.query as jest.Mock) + .mockResolvedValueOnce({ rows: [] }) + .mockRejectedValueOnce(new Error('invalid role')) + .mockResolvedValueOnce({ rows: [] }); + + await expect( + withPgClient( + pool, + { role: 'missing_role' }, + async (): Promise => undefined + ) + ).rejects.toThrow('invalid role'); + + expect(client.query).toHaveBeenNthCalledWith(3, 'ROLLBACK'); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it('preserves callback failures while rolling back and releasing', async () => { + const { client, pool } = makePool(); + const original = new Error('callback failed'); + + await expect( + withPgClient(pool, { role: 'tenant_runtime' }, async () => { + throw original; + }) + ).rejects.toBe(original); + + expect(client.query).toHaveBeenNthCalledWith(3, 'ROLLBACK'); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it('uses the same single context batch for the one-query API', async () => { + const { client, pool } = makePool(); + (client.query as jest.Mock).mockImplementation(async (query: unknown) => ({ + rows: [ + query === 'SELECT tenant_id FROM documents' + ? { tenant_id: 'a' } + : undefined, + ].filter(Boolean), + })); + + await pgQueryContext({ + client: pool, + context: { role: 'tenant_runtime', 'jwt.claims.tenant_id': 'a' }, + query: 'SELECT tenant_id FROM documents', + }); + + expect(client.query).toHaveBeenNthCalledWith(2, SETTINGS_SQL, [ + JSON.stringify([ + ['role', 'tenant_runtime'], + ['jwt.claims.tenant_id', 'a'], + ]), + ]); + expect(client.query).toHaveBeenCalledTimes(4); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it('rejects transaction-local context through a pool without a transaction', async () => { + const { pool } = makePool(); + + await expect( + withPgClient( + pool, + { role: 'tenant_runtime' }, + async (): Promise => undefined, + { skipTransaction: true } + ) + ).rejects.toMatchObject({ + name: UnsafePooledContextError.name, + code: UNSAFE_POOLED_CONTEXT_ERROR_CODE, + }); + + expect(pool.connect).not.toHaveBeenCalled(); + + await expect( + pgQueryContext({ + client: pool, + context: { 'jwt.claims.tenant_id': 'tenant-a' }, + query: 'SELECT 1', + skipTransaction: true, + }) + ).rejects.toBeInstanceOf(UnsafePooledContextError); + expect(pool.connect).not.toHaveBeenCalled(); + }); + + it('allows transaction-free pooled execution only when no context is requested', async () => { + const { client, pool } = makePool(); + + await expect( + withPgClient(pool, {}, async () => 'ok', { skipTransaction: true }) + ).resolves.toBe('ok'); + + expect(client.query).not.toHaveBeenCalled(); + expect(client.release).toHaveBeenCalledTimes(1); + }); +}); diff --git a/postgres/pg-query-context/src/__tests__/postgres.integration.test.ts b/postgres/pg-query-context/src/__tests__/postgres.integration.test.ts new file mode 100644 index 0000000000..1b2e935aef --- /dev/null +++ b/postgres/pg-query-context/src/__tests__/postgres.integration.test.ts @@ -0,0 +1,201 @@ +import { Pool, type PoolClient } from 'pg'; +import { getConnections } from 'pgsql-test'; + +import pgQueryContext, { withPgClient } from '../index'; + +interface SessionState { + role: string; + transaction_read_only: string; + search_path: string; + row_security: string; + user_id: string; +} + +async function readSessionState(client: PoolClient): Promise { + const result = await client.query(` + SELECT + current_setting('role') AS role, + current_setting('transaction_read_only') AS transaction_read_only, + current_setting('search_path') AS search_path, + current_setting('row_security') AS row_security, + current_setting('jwt.claims.user_id', true) AS user_id + `); + return result.rows[0]; +} + +describe('pg-query-context transaction-local integration', () => { + let db: Awaited>['db']; + let teardown: Awaited>['teardown']; + let singleClientPool: Pool; + + beforeAll(async () => { + ({ db, teardown } = await getConnections({}, [])); + singleClientPool = new Pool({ ...db.config, max: 1 }); + }); + + afterAll(async () => { + if (singleClientPool) await singleClientPool.end(); + if (teardown) await teardown(); + }); + + beforeEach(async () => { + if (db) await db.beforeEach(); + }); + + afterEach(async () => { + if (db) await db.afterEach(); + }); + + it('applies a complete context and restores state after rollback', async () => { + const { rows: identityRows } = await db.client.query<{ + current_user: string; + }>('SELECT current_user'); + const currentUser = identityRows[0].current_user; + const context = { + role: currentUser, + 'jwt.claims.user_id': 'integration-user', + 'jwt.claims.api_id': 'integration-api', + 'jwt.claims.database_id': 'integration-database', + 'request.id': 'integration-request', + transaction_read_only: 'on', + row_security: 'on', + search_path: 'pg_catalog, public', + }; + + const result = await pgQueryContext({ + client: db.client, + context, + skipTransaction: true, + query: ` + SELECT + current_user, + current_setting('jwt.claims.user_id', true) AS user_id, + current_setting('jwt.claims.api_id', true) AS api_id, + current_setting('jwt.claims.database_id', true) AS database_id, + current_setting('request.id', true) AS request_id, + current_setting('transaction_read_only') AS read_only, + current_setting('row_security') AS row_security, + current_setting('search_path') AS search_path + `, + }); + + expect(result.rows[0]).toEqual({ + current_user: currentUser, + user_id: 'integration-user', + api_id: 'integration-api', + database_id: 'integration-database', + request_id: 'integration-request', + read_only: 'on', + row_security: 'on', + search_path: 'pg_catalog, public', + }); + + await db.rollback(); + const restored = await db.client.query<{ + user_id: string; + request_id: string; + }>(` + SELECT + current_setting('jwt.claims.user_id', true) AS user_id, + current_setting('request.id', true) AS request_id + `); + // PostgreSQL retains an empty placeholder for a custom GUC after its first + // transaction-local use; importantly, the request values themselves do not + // survive the rollback. + expect(restored.rows[0]).toEqual({ user_id: '', request_id: '' }); + await db.savepoint(); + }); + + it('restores a reused backend baseline after commit and rollback', async () => { + const baselineClient = await singleClientPool.connect(); + let runtimeRole: string; + try { + const identity = await baselineClient.query<{ current_user: string }>( + 'SELECT current_user' + ); + runtimeRole = identity.rows[0].current_user; + + await baselineClient.query('RESET ROLE'); + await baselineClient.query('SET transaction_read_only TO off'); + await baselineClient.query('SET search_path TO public'); + await baselineClient.query('SET row_security TO off'); + await baselineClient.query( + "SELECT pg_catalog.set_config('jwt.claims.user_id', 'baseline-user', false)" + ); + } finally { + baselineClient.release(); + } + + const committedInside = await withPgClient( + singleClientPool, + { + role: runtimeRole, + transaction_read_only: 'on', + search_path: 'pg_catalog', + row_security: 'on', + 'jwt.claims.user_id': '', + }, + readSessionState + ); + + expect(committedInside).toEqual({ + role: runtimeRole, + transaction_read_only: 'on', + search_path: 'pg_catalog', + row_security: 'on', + user_id: '', + }); + + const afterCommitClient = await singleClientPool.connect(); + try { + await expect(readSessionState(afterCommitClient)).resolves.toEqual({ + role: 'none', + transaction_read_only: 'off', + search_path: 'public', + row_security: 'off', + user_id: 'baseline-user', + }); + } finally { + afterCommitClient.release(); + } + + let rolledBackInside: SessionState | undefined; + await expect( + withPgClient( + singleClientPool, + { + role: runtimeRole, + transaction_read_only: 'on', + search_path: 'pg_catalog', + row_security: 'on', + 'jwt.claims.user_id': 'rollback-canary', + }, + async (client) => { + rolledBackInside = await readSessionState(client); + throw new Error('force rollback'); + } + ) + ).rejects.toThrow('force rollback'); + + expect(rolledBackInside).toEqual({ + role: runtimeRole, + transaction_read_only: 'on', + search_path: 'pg_catalog', + row_security: 'on', + user_id: 'rollback-canary', + }); + + const afterRollbackClient = await singleClientPool.connect(); + try { + await expect(readSessionState(afterRollbackClient)).resolves.toEqual({ + role: 'none', + transaction_read_only: 'off', + search_path: 'public', + row_security: 'off', + user_id: 'baseline-user', + }); + } finally { + afterRollbackClient.release(); + } + }); +}); diff --git a/postgres/pg-query-context/src/index.ts b/postgres/pg-query-context/src/index.ts index 188d8670f2..eebcdb608c 100644 --- a/postgres/pg-query-context/src/index.ts +++ b/postgres/pg-query-context/src/index.ts @@ -2,18 +2,58 @@ import { ClientBase, Pool, PoolClient, QueryResult } from 'pg'; // --- Internal helpers --- -function setContext(ctx: Record): { query: string; values: string[] }[] { - return Object.keys(ctx || {}).reduce<{ query: string; values: string[] }[]>((m, el) => { - m.push({ query: 'SELECT set_config($1, $2, true)', values: [el, ctx[el]] }); - return m; - }, []); +export const UNSAFE_POOLED_CONTEXT_ERROR_CODE = + 'PG_QUERY_CONTEXT_UNSAFE_POOLED_CONTEXT'; + +export class UnsafePooledContextError extends Error { + readonly code = UNSAFE_POOLED_CONTEXT_ERROR_CODE; + + constructor() { + super( + 'Transaction-local PostgreSQL context cannot be applied through a pool ' + + 'when skipTransaction is enabled' + ); + this.name = 'UnsafePooledContextError'; + } +} + +function assertContextHasTransaction( + usesPool: boolean, + skipTransaction: boolean, + context: Record +): void { + if (usesPool && skipTransaction && Object.keys(context).length > 0) { + throw new UnsafePooledContextError(); + } } -async function execContext(client: ClientBase, ctx: Record): Promise { - const local = setContext(ctx); - for (const { query, values } of local) { - await client.query(query, values); +function isPgPool(client: Pool | ClientBase): client is Pool { + return ( + typeof (client as Pool).connect === 'function' && + typeof (client as Pool).totalCount === 'number' + ); +} + +async function execContext( + client: ClientBase, + ctx: Record +): Promise { + const entries = Object.entries(ctx || {}); + if (entries.length === 0) return; + + for (const [key, value] of entries) { + if (typeof value !== 'string') { + throw new TypeError( + `PostgreSQL context setting '${key}' must be a string` + ); + } } + + await client.query( + 'SELECT pg_catalog.set_config(setting->>0, setting->>1, true) ' + + 'FROM pg_catalog.json_array_elements($1::json) AS setting', + [JSON.stringify(entries)] + ); } // --- Single-query API (original) --- @@ -27,10 +67,12 @@ export interface ExecOptions { } async function pgQueryContext({ client, context = {}, query = '', variables = [], skipTransaction = false }: ExecOptions): Promise { - const isPool = 'connect' in client; + const isPool = isPgPool(client); const shouldRelease = isPool; let pgClient: ClientBase | PoolClient | null = null; + assertContextHasTransaction(isPool, skipTransaction, context); + try { pgClient = isPool ? await (client as Pool).connect() : client as ClientBase; @@ -80,6 +122,7 @@ export async function withPgClient( fn: (client: PoolClient) => Promise, opts: WithPgClientOptions = {}, ): Promise { + assertContextHasTransaction(true, opts.skipTransaction === true, context); const client = await pool.connect(); try { if (!opts.skipTransaction) {