diff --git a/apps/api/src/app/controllers/sf-bulk-api.controller.ts b/apps/api/src/app/controllers/sf-bulk-api.controller.ts index c11aa74f0..9b3b7a7ac 100644 --- a/apps/api/src/app/controllers/sf-bulk-api.controller.ts +++ b/apps/api/src/app/controllers/sf-bulk-api.controller.ts @@ -142,13 +142,15 @@ const closeOrAbortJob = createRoute(routeDefinition.closeOrAbortJob.validators, const addBatchToJob = createRoute( routeDefinition.addBatchToJob.validators, - async ({ body, params, query, jetstreamConn }, _, res, next) => { + async ({ body, params, query, jetstreamConn }, req, res, next) => { try { const jobId = params.jobId; const csv = body; const closeJob = query.closeJob; + // gzipBodyPassthrough left the body compressed - Salesforce decodes it, this server never does + const contentEncoding = req.get(HTTP.HEADERS.CONTENT_ENCODING)?.toLowerCase() === 'gzip' ? 'gzip' : undefined; - const results = await jetstreamConn.bulk.addBatchToJob(csv, jobId, closeJob); + const results = await jetstreamConn.bulk.addBatchToJob(csv, jobId, closeJob, HTTP.CONTENT_TYPE.CSV, contentEncoding); sendJson(res, results); } catch (ex) { diff --git a/apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts b/apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts new file mode 100644 index 000000000..ba31581e3 --- /dev/null +++ b/apps/api/src/app/utils/__tests__/gzip-body-passthrough.middleware.spec.ts @@ -0,0 +1,91 @@ +import { HTTP } from '@jetstream/shared/constants'; +import { raw } from 'body-parser'; +import express from 'express'; +import type { AddressInfo } from 'node:net'; +import { gzipSync } from 'node:zlib'; +import { gzipBodyPassthrough } from '../gzip-body-passthrough.middleware'; + +const CSV = 'Name\nAcme\nGlobex\n'; + +/** Mirrors main.ts: passthrough first, then the global parser that rejects every Content-Encoding */ +function createServer(limitBytes = 1024) { + const app = express(); + app.post('/api/bulk/:jobId', gzipBodyPassthrough({ limitBytes })); + app.use(raw({ limit: '30mb', type: ['text/csv'], inflate: false })); + app.post('/api/bulk/:jobId', (req, res) => { + res.json({ + isBuffer: Buffer.isBuffer(req.body), + bodyBase64: Buffer.isBuffer(req.body) ? req.body.toString('base64') : null, + }); + }); + app.use((err: { status?: number; type?: string }, _req: express.Request, res: express.Response, _next: express.NextFunction) => { + res.status(err.status ?? 500).json({ type: err.type }); + }); + return new Promise<{ url: string; close: () => void }>((resolve) => { + const server = app.listen(0, () => { + const { port } = server.address() as AddressInfo; + resolve({ url: `http://127.0.0.1:${port}`, close: () => server.close() }); + }); + }); +} + +function post(url: string, body: BodyInit, headers: Record) { + return fetch(`${url}/api/bulk/750000000000001`, { method: 'POST', body, headers }); +} + +describe('gzipBodyPassthrough', () => { + let server: Awaited>; + + afterEach(() => { + server?.close(); + }); + + test('passes gzipped csv through as the original compressed bytes', async () => { + server = await createServer(); + const compressed = gzipSync(CSV); + const response = await post(server.url, compressed, { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip', + }); + expect(response.status).toBe(200); + const result = await response.json(); + expect(result.isBuffer).toBe(true); + expect(Buffer.from(result.bodyBase64, 'base64').equals(compressed)).toBe(true); + }); + + test('leaves uncompressed csv to the regular parser', async () => { + server = await createServer(); + const response = await post(server.url, CSV, { [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV }); + expect(response.status).toBe(200); + const result = await response.json(); + expect(Buffer.from(result.bodyBase64, 'base64').toString('utf8')).toBe(CSV); + }); + + test('rejects a body that is not gzip', async () => { + server = await createServer(); + const response = await post(server.url, CSV, { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip', + }); + expect(response.status).toBe(400); + expect((await response.json()).type).toBe('encoding.invalid'); + }); + + test('rejects a compressed body over the limit', async () => { + server = await createServer(16); + const response = await post(server.url, gzipSync('x'.repeat(10_000) + CSV), { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'gzip', + }); + expect(response.status).toBe(413); + }); + + test('leaves other encodings to the global parser, which rejects them', async () => { + server = await createServer(); + const response = await post(server.url, CSV, { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + [HTTP.HEADERS.CONTENT_ENCODING]: 'br', + }); + expect(response.status).toBe(415); + }); +}); diff --git a/apps/api/src/app/utils/gzip-body-passthrough.middleware.ts b/apps/api/src/app/utils/gzip-body-passthrough.middleware.ts new file mode 100644 index 000000000..f1a25e5e6 --- /dev/null +++ b/apps/api/src/app/utils/gzip-body-passthrough.middleware.ts @@ -0,0 +1,48 @@ +import { HTTP } from '@jetstream/shared/constants'; +import express from 'express'; + +function createBodyError(status: number, type: string, message: string) { + return Object.assign(new Error(message), { status, type, expose: true }); +} + +/** + * Reads a gzip-encoded CSV body into a Buffer WITHOUT decompressing it, so it can be forwarded to Salesforce + * as-is (Bulk API decodes gzip request bodies itself). + * + * The global body parsers reject every Content-Encoding (inflate: false) to keep untrusted input away from the + * zlib/brotli decoders, so this must be mounted before them - they skip any request whose body was already read. + * An over-limit body is drained before responding (as body-parser does) so the client receives the 413. + */ +export function gzipBodyPassthrough({ limitBytes }: { limitBytes: number }) { + return (req: express.Request, _res: express.Response, next: express.NextFunction) => { + if (req.get(HTTP.HEADERS.CONTENT_ENCODING)?.toLowerCase() !== 'gzip' || !req.is('text/csv')) { + next(); + return; + } + + const chunks: Buffer[] = []; + let receivedBytes = 0; + + req.on('data', (chunk: Buffer) => { + receivedBytes += chunk.length; + if (receivedBytes <= limitBytes) { + chunks.push(chunk); + } + }); + req.on('error', next); + req.on('end', () => { + if (receivedBytes > limitBytes) { + next(createBodyError(413, 'entity.too.large', 'request entity too large')); + return; + } + const body = Buffer.concat(chunks); + // gzip magic bytes - reject anything else here rather than forwarding garbage to Salesforce + if (body[0] !== 0x1f || body[1] !== 0x8b) { + next(createBodyError(400, 'encoding.invalid', 'request body is not valid gzip')); + return; + } + req.body = body; + next(); + }); + }; +} diff --git a/apps/api/src/main.ts b/apps/api/src/main.ts index 573f93f99..16a041753 100644 --- a/apps/api/src/main.ts +++ b/apps/api/src/main.ts @@ -64,6 +64,7 @@ import { setCacheControlForApiRoutes, setPermissionPolicy, } from './app/routes/route.middleware'; +import { gzipBodyPassthrough } from './app/utils/gzip-body-passthrough.middleware'; import { healthCheck, uncaughtErrorHandler } from './app/utils/response.handlers'; import { buildAppCspDirectives, buildCspDirectives, buildHstsConfig } from './app/utils/security-headers'; import { handleWorkerExit, primaryClusterInitSideEffects, shutdownPrimaryGracefully } from './app/utils/server-process.utils'; @@ -319,8 +320,11 @@ if (ENV.NODE_ENV === 'production' && !ENV.CI && cluster.isPrimary) { // handles the report payload instead of the 20mb global JSON parser, and responds 204. app.use('/api/csp-report', cspReportRoutes); - // inflate: false - no client sends compressed request bodies; rejecting Content-Encoding (415) keeps - // untrusted input away from zlib/brotli decoders (e.g. CVE-2026-111497) + // Gzipped bulk batches are read as raw bytes and forwarded to Salesforce, which decodes them + app.post('/api/bulk/:jobId', gzipBodyPassthrough({ limitBytes: 30 * 1024 * 1024 })); + + // inflate: false - rejecting Content-Encoding (415) keeps untrusted input away from zlib/brotli decoders + // (e.g. CVE-2026-111497); the one compressed body we accept is passed through above, never decoded app.use(raw({ limit: '30mb', type: ['text/csv'], inflate: false })); app.use(raw({ limit: '30mb', type: ['application/zip'], inflate: false })); app.use(json({ limit: '20mb', type: ['json'], inflate: false })); diff --git a/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts b/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts index 657244326..2b9d045a5 100644 --- a/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts +++ b/apps/jetstream-e2e/src/tests/api/bulk.api.spec.ts @@ -1,5 +1,6 @@ import { CreateJobRequest } from '@jetstream/api-types'; import { BulkJob, BulkJobBatchInfo, BulkJobWithBatches } from '@jetstream/types'; +import { gzipSync } from 'node:zlib'; import { expect, test } from '../../fixtures/fixtures'; test.describe.configure({ mode: 'parallel' }); @@ -54,6 +55,43 @@ test.describe('API - Bulk', () => { expect(getJobResponse2.state).toEqual('Closed'); }); + test('Bulk Job - Add gzipped Batch', async ({ apiRequestUtils }) => { + const createJobResponse = await apiRequestUtils.makeRequest('POST', `/api/bulk`, { + externalId: null, + serialMode: false, + sObject: 'LEAD', + type: 'INSERT', + } as CreateJobRequest); + + const timestamp = new Date().getTime(); + // The job uses Salesforce's default comma delimiter + const csv = [ + `LastName,FirstName,Company,Email,Status`, + `Gzip-${timestamp},Kathy,TNR Corp. ${timestamp},ksynder@tnr.${timestamp}.net,Working - Contacted`, + ].join('\n'); + + const addBatchToJobResponse = await apiRequestUtils.makeRequest( + 'POST', + `/api/bulk/${createJobResponse.id}?${new URLSearchParams({ closeJob: 'true' })}`, + gzipSync(csv), + { 'Content-Type': 'text/csv; charset=UTF-8', 'Content-Encoding': 'gzip' }, + ); + + expect(addBatchToJobResponse.id).toBeTruthy(); + + // Salesforce decodes the batch itself - a successful insert proves the bytes arrived intact + await expect + .poll( + async () => { + const { batches } = await apiRequestUtils.makeRequest('GET', `/api/bulk/${createJobResponse.id}`); + const [{ state, stateMessage, numberRecordsProcessed, numberRecordsFailed }] = batches; + return { state, stateMessage, numberRecordsProcessed, numberRecordsFailed }; + }, + { timeout: 90_000 }, + ) + .toMatchObject({ state: 'Completed', numberRecordsProcessed: 1, numberRecordsFailed: 0 }); + }); + test('Bulk Job - Errors', async ({ apiRequestUtils }) => { const createJobResponse = await apiRequestUtils.makeRequestRaw('POST', `/api/bulk`, { externalId: null, diff --git a/libs/salesforce-api/src/lib/api-bulk.ts b/libs/salesforce-api/src/lib/api-bulk.ts index 4fde58c6d..02cc9ae92 100644 --- a/libs/salesforce-api/src/lib/api-bulk.ts +++ b/libs/salesforce-api/src/lib/api-bulk.ts @@ -68,13 +68,19 @@ export class ApiBulk extends SalesforceApi { jobId: string, closeJob = false, contentType: typeof HTTP.CONTENT_TYPE.CSV | typeof HTTP.CONTENT_TYPE.ZIP_CSV = HTTP.CONTENT_TYPE.CSV, + /** Set when `body` is already gzip-compressed - Salesforce decodes it */ + contentEncoding?: 'gzip', ) { const result = await this.apiRequest<{ batchInfo: BulkJobBatchInfoUntyped }>({ sessionInfo: this.sessionInfo, url: this.getBulkApiUrl(`/job/${jobId}/batch`), method: 'POST', body, - headers: { [HTTP.HEADERS.CONTENT_TYPE]: contentType, Accept: HTTP.CONTENT_TYPE.XML }, + headers: { + [HTTP.HEADERS.CONTENT_TYPE]: contentType, + Accept: HTTP.CONTENT_TYPE.XML, + ...(contentEncoding && { [HTTP.HEADERS.CONTENT_ENCODING]: contentEncoding }), + }, outputType: 'xml', rawBody: true, }).then(({ batchInfo }) => bulkApiEnsureTyped(batchInfo)); diff --git a/libs/shared/constants/src/lib/shared-constants.ts b/libs/shared/constants/src/lib/shared-constants.ts index 1050f9d96..19982e8dc 100644 --- a/libs/shared/constants/src/lib/shared-constants.ts +++ b/libs/shared/constants/src/lib/shared-constants.ts @@ -102,6 +102,7 @@ export const HTTP = { X_CLIENT_RUNTIME: 'X-Client-Runtime', X_CLIENT_INSTALL: 'X-Client-Install', CONTENT_TYPE: 'Content-Type', + CONTENT_ENCODING: 'Content-Encoding', X_MOCK_KEY: 'X-MOCK-KEY', X_FORWARDED_FOR: 'X-FORWARDED-FOR', CONTENT_DISPOSITION: 'Content-Disposition', diff --git a/libs/shared/data/src/lib/client-data.ts b/libs/shared/data/src/lib/client-data.ts index 2ce491ae5..aa345af7d 100644 --- a/libs/shared/data/src/lib/client-data.ts +++ b/libs/shared/data/src/lib/client-data.ts @@ -23,6 +23,7 @@ import type { import { logger } from '@jetstream/shared/client-logger'; import { HTTP, MIME_TYPES } from '@jetstream/shared/constants'; import { + gzipEncodeText, hasIncompleteSubqueries, isIncompleteSubqueryResult, isSubqueryResult, @@ -99,7 +100,7 @@ import { parseISO } from 'date-fns/parseISO'; import isFunction from 'lodash/isFunction'; import isNil from 'lodash/isNil'; import orderBy from 'lodash/orderBy'; -import { handleExternalRequest, handleRequest, transformListMetadataResponse } from './client-data-data-helper'; +import { AxiosAdapterConfig, handleExternalRequest, handleRequest, transformListMetadataResponse } from './client-data-data-helper'; import { getStandardValueSetTypes } from './standardValueSet'; export interface PaginationCursorParams { @@ -1327,19 +1328,42 @@ export async function bulkApiAbortJob(org: SalesforceOrgUi, jobId: string): Prom return handleRequest({ method: 'DELETE', url: `/api/bulk/${jobId}/abort` }, { org }).then(unwrapResponseIgnoreCache); } +const BULK_BATCH_GZIP_MIN_LENGTH = 64 * 1024; + +/** + * Gzip large batches when they travel over the network to the Jetstream server, which forwards the compressed + * bytes to Salesforce untouched. Desktop, extension and canvas set a custom adapter and call Salesforce + * in-process, so their batches are left as plain text. + */ +async function getBulkBatchRequestBody(csv: string): Promise<{ data: string | Uint8Array; contentEncoding?: 'gzip' }> { + if (AxiosAdapterConfig.adapter || csv.length < BULK_BATCH_GZIP_MIN_LENGTH || typeof CompressionStream === 'undefined') { + return { data: csv }; + } + try { + return { data: await gzipEncodeText(csv), contentEncoding: 'gzip' }; + } catch (ex) { + logger.warn('[BULK] Failed to gzip batch, sending uncompressed', ex); + return { data: csv }; + } +} + export async function bulkApiAddBatchToJob( org: SalesforceOrgUi, jobId: string, csv: string, closeJob?: boolean, ): Promise { + const { data, contentEncoding } = await getBulkBatchRequestBody(csv); return handleRequest( { method: 'POST', url: `/api/bulk/${jobId}`, - data: csv, + data, params: { closeJob }, - headers: { [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV }, + headers: { + [HTTP.HEADERS.CONTENT_TYPE]: HTTP.CONTENT_TYPE.CSV, + ...(contentEncoding && { [HTTP.HEADERS.CONTENT_ENCODING]: contentEncoding }), + }, }, { org }, ) diff --git a/libs/shared/utils/src/lib/compression.ts b/libs/shared/utils/src/lib/compression.ts index b4eb16a7b..1c97ce0a4 100644 --- a/libs/shared/utils/src/lib/compression.ts +++ b/libs/shared/utils/src/lib/compression.ts @@ -2,8 +2,11 @@ const TEXT_ENCODER = new TextEncoder(); const TEXT_DECODER = new TextDecoder(); export async function gzipEncode(value: unknown): Promise> { - const json = JSON.stringify(value); - const stream = new Blob([TEXT_ENCODER.encode(json) as BlobPart]).stream().pipeThrough(new CompressionStream('gzip')); + return gzipEncodeText(JSON.stringify(value)); +} + +export async function gzipEncodeText(value: string): Promise> { + const stream = new Blob([TEXT_ENCODER.encode(value) as BlobPart]).stream().pipeThrough(new CompressionStream('gzip')); const buffer = await new Response(stream).arrayBuffer(); return new Uint8Array(buffer); }