From 9a88bbf08d6c86db5bafb2cf59b77bfcb9151cad Mon Sep 17 00:00:00 2001 From: Aditya Reddy <82123567+adityareddy-dev@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:54:28 -0500 Subject: [PATCH 1/3] api: fix parsed ClickHouse proxy bodies --- .changeset/proxy-parsed-bodies.md | 7 + .../api/__tests__/clickhouseProxy.int.test.ts | 206 ++++++++++++++++-- .../api/src/routers/api/clickhouseProxy.ts | 18 +- .../api/src/utils/__tests__/proxyBody.test.ts | 132 +++++++++++ packages/api/src/utils/proxyBody.ts | 73 +++++++ 5 files changed, 399 insertions(+), 37 deletions(-) create mode 100644 .changeset/proxy-parsed-bodies.md create mode 100644 packages/api/src/utils/__tests__/proxyBody.test.ts create mode 100644 packages/api/src/utils/proxyBody.ts diff --git a/.changeset/proxy-parsed-bodies.md b/.changeset/proxy-parsed-bodies.md new file mode 100644 index 0000000000..f1fc139e10 --- /dev/null +++ b/.changeset/proxy-parsed-bodies.md @@ -0,0 +1,7 @@ +--- +'@hyperdx/api': patch +--- + +Forward parsed JSON and form request bodies through the ClickHouse proxy, +including charset parameters and repeated form fields. Keep request framing in +sync with serialized bytes and preserve unparsed multipart streams. diff --git a/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts b/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts index 23d5c2e5ca..efb774e218 100644 --- a/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts +++ b/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts @@ -1,5 +1,6 @@ import { buildLogComment } from '@hyperdx/common-utils/dist/clickhouse'; import http from 'http'; +import { gzipSync } from 'zlib'; import { getLoggedInAgent, getServer } from '@/fixtures'; import Connection from '@/models/connection'; @@ -16,14 +17,11 @@ type CapturedRequest = { }; /** - * Pins what /clickhouse-proxy forwards upstream: text/plain re-injection, the - * raw-stream passthrough of multipart/form-data bodies, which + * Pins what /clickhouse-proxy forwards upstream: parsed body re-injection with + * valid framing, the raw-stream passthrough of multipart/form-data bodies, which * `@clickhouse/client-web` emits when query params exceed its URL budget * (regression context: ClickHouse support-escalation #8482), and the * decode/re-encode the query string goes through in `pathRewrite`. - * - * Known-latent cases (charset-suffixed JSON, urlencoded) are not covered; - * tracked in https://github.com/hyperdxio/hyperdx/issues/2942. */ describe('clickhouse-proxy forwarding', () => { const server = getServer(); @@ -66,6 +64,7 @@ describe('clickhouse-proxy forwarding', () => { }); afterAll(async () => { + upstream.closeAllConnections(); await new Promise((resolve, reject) => upstream.close(err => (err ? reject(err) : resolve())), ); @@ -101,6 +100,156 @@ describe('clickhouse-proxy forwarding', () => { expect(captured[0].headers['x-clickhouse-user']).toBe('default'); }); + it.each(['application/json', 'application/json; charset=utf-8'])( + 're-serializes %s and updates Content-Length to its UTF-8 byte length', + async contentType => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const body = { + query: "SELECT '東京 🍰' FORMAT JSON", + param_name: 'café', + }; + const original = JSON.stringify(body, null, 2); + const serialized = JSON.stringify(body); + expect(Buffer.byteLength(original)).toBeGreaterThan( + Buffer.byteLength(serialized), + ); + expect(Buffer.byteLength(serialized)).toBeGreaterThan(serialized.length); + + await agent + .post('/clickhouse-proxy/?query_id=test-json') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', contentType) + .send(original) + .timeout({ response: 5000, deadline: 10000 }) + .expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body).toEqual(Buffer.from(serialized)); + expect(captured[0].headers['content-type']).toBe(contentType); + expect(captured[0].headers['content-length']).toBe( + String(Buffer.byteLength(serialized)), + ); + expect(captured[0].headers['transfer-encoding']).toBeUndefined(); + }, + ); + + it('forwards parsed UTF-16LE JSON as UTF-8 with matching headers', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const body = { query: "SELECT '東京 🍰' FORMAT JSON" }; + const original = Buffer.from(JSON.stringify(body, null, 2), 'utf16le'); + const serialized = Buffer.from(JSON.stringify(body)); + + const proxyRequest = agent + .post('/clickhouse-proxy/?query_id=test-utf16-json') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json; charset=utf-16le') + .set('Content-Length', String(original.length)) + .timeout({ response: 5000, deadline: 10000 }); + proxyRequest.write(original); + await proxyRequest.expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body).toEqual(serialized); + expect(captured[0].headers['content-type']).toMatch( + /^application\/json;\s*charset=utf-8$/i, + ); + expect(captured[0].headers['content-length']).toBe( + String(serialized.length), + ); + expect(captured[0].headers['content-encoding']).toBeUndefined(); + }); + + it('forwards parsed gzip JSON without the original content encoding', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const body = { query: "SELECT 'café' FORMAT JSON" }; + const original = gzipSync(Buffer.from(JSON.stringify(body, null, 2))); + const serialized = Buffer.from(JSON.stringify(body)); + + const proxyRequest = agent + .post('/clickhouse-proxy/?query_id=test-gzip-json') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json; charset=utf-8') + .set('Content-Encoding', 'gzip') + .set('Content-Length', String(original.length)) + .timeout({ response: 5000, deadline: 10000 }); + proxyRequest.write(original); + await proxyRequest.expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body).toEqual(serialized); + expect(captured[0].headers['content-type']).toBe( + 'application/json; charset=utf-8', + ); + expect(captured[0].headers['content-length']).toBe( + String(serialized.length), + ); + expect(captured[0].headers['content-encoding']).toBeUndefined(); + }); + + it.each([ + 'application/x-www-form-urlencoded', + 'application/x-www-form-urlencoded; charset=utf-8', + ])( + 're-serializes %s without losing repeated form keys', + async contentType => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const original = + 'query=SELECT%20%27caf%C3%A9%27¶m=first¶m=second%20value&empty=&literal=%2B%26%3D'; + const entries = [ + ['query', "SELECT 'café'"], + ['param', 'first'], + ['param', 'second value'], + ['empty', ''], + ['literal', '+&='], + ]; + const serialized = new URLSearchParams(entries).toString(); + + await agent + .post('/clickhouse-proxy/?query_id=test-urlencoded') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', contentType) + .send(original) + .timeout({ response: 5000, deadline: 10000 }) + .expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body.toString()).toBe(serialized); + expect([...new URLSearchParams(captured[0].body.toString())]).toEqual( + entries, + ); + expect(captured[0].headers['content-type']).toBe(contentType); + expect(captured[0].headers['content-length']).toBe( + String(Buffer.byteLength(serialized)), + ); + }, + ); + + it('keeps chunked framing when re-injecting parsed JSON', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const body = { query: "SELECT 'café' FORMAT JSON" }; + const original = JSON.stringify(body, null, 2); + const proxyRequest = agent + .post('/clickhouse-proxy/?query_id=test-chunked-json') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json; charset=utf-8') + .set('Transfer-Encoding', 'chunked') + .timeout({ response: 5000, deadline: 10000 }); + + proxyRequest.write(original.slice(0, 10)); + proxyRequest.write(original.slice(10)); + await proxyRequest.expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body.toString()).toBe(JSON.stringify(body)); + expect(captured[0].headers['transfer-encoding']).toBe('chunked'); + expect(captured[0].headers['content-length']).toBeUndefined(); + }); + it('forwards a large multipart/form-data body byte-for-byte (unparsed stream passthrough)', async () => { const { agent, team } = await getLoggedInAgent(server); const connectionId = await createConnection(team._id.toString()); @@ -120,25 +269,40 @@ describe('clickhouse-proxy forwarding', () => { `some.rather.long.map.attribute.key.number.${i}\r\n`, ); } - parts.push(`--${boundary}--\r\n`); - const body = parts.join(''); + const body = Buffer.concat([ + Buffer.from(parts.join('')), + Buffer.from( + `--${boundary}\r\n` + + `Content-Disposition: form-data; name="file"; filename="bytes.bin"\r\n` + + `Content-Type: application/octet-stream\r\n\r\n`, + ), + Buffer.from([0x00, 0xff, 0x80, 0x0d, 0x0a]), + Buffer.from(`\r\n--${boundary}--\r\n`), + ]); expect(Buffer.byteLength(body)).toBeGreaterThan(4096); - await agent - .post('/clickhouse-proxy/?query_id=test-multipart') - .set('x-hyperdx-connection-id', connectionId) - .set('Content-Type', `multipart/form-data; boundary=${boundary}`) - .send(body) - .expect(200); + const consoleError = jest.spyOn(console, 'error'); + try { + await agent + .post('/clickhouse-proxy/?query_id=test-multipart') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', `multipart/form-data; boundary=${boundary}`) + .send(body) + .timeout({ response: 5000, deadline: 10000 }) + .expect(200); - expect(captured).toHaveLength(1); - expect(captured[0].headers['content-type']).toBe( - `multipart/form-data; boundary=${boundary}`, - ); - expect(captured[0].body.toString()).toBe(body); - expect(captured[0].headers['content-length']).toBe( - String(Buffer.byteLength(body)), - ); + expect(captured).toHaveLength(1); + expect(captured[0].headers['content-type']).toBe( + `multipart/form-data; boundary=${boundary}`, + ); + expect(captured[0].body).toEqual(body); + expect(captured[0].headers['content-length']).toBe( + String(Buffer.byteLength(body)), + ); + expect(consoleError).not.toHaveBeenCalled(); + } finally { + consoleError.mockRestore(); + } }); // The browser client ships `log_comment` as a URL query param, and diff --git a/packages/api/src/routers/api/clickhouseProxy.ts b/packages/api/src/routers/api/clickhouseProxy.ts index e41946c1b0..d67e00b118 100644 --- a/packages/api/src/routers/api/clickhouseProxy.ts +++ b/packages/api/src/routers/api/clickhouseProxy.ts @@ -11,6 +11,7 @@ import { getNonNullUserWithTeam } from '@/middleware/auth'; import { validateRequestHeaders } from '@/middleware/validation'; import { recordOperationOutcome } from '@/utils/instrumentation'; import logger from '@/utils/logger'; +import { writeProxyBody } from '@/utils/proxyBody'; import { IPV6_BRACKET_RE, isPrivateIp } from '@/utils/validators'; import { objectIdSchema } from '@/utils/zod'; @@ -261,22 +262,7 @@ const proxyMiddleware: RequestHandler = return res.sendStatus(405); } - let body = _req.body; - if (_req.headers['content-type'] === 'application/json') { - try { - body = JSON.stringify(body); - } catch (e) { - console.error(e); - } - } - - try { - proxyReq.write(body); - } catch { - console.error( - `clickhouseProxy error writing body, body is type ${typeof body}`, - ); - } + writeProxyBody(proxyReq, _req); }, proxyRes: (proxyRes, _req, res) => { const startedAt = (res as Response).locals?.hdxProxyStartedAt; diff --git a/packages/api/src/utils/__tests__/proxyBody.test.ts b/packages/api/src/utils/__tests__/proxyBody.test.ts new file mode 100644 index 0000000000..9f382fec48 --- /dev/null +++ b/packages/api/src/utils/__tests__/proxyBody.test.ts @@ -0,0 +1,132 @@ +import { writeProxyBody } from '@/utils/proxyBody'; + +jest.mock('@/utils/logger', () => ({ + __esModule: true, + default: { error: jest.fn() }, +})); + +const makeProxyRequest = () => ({ + getHeader: jest.fn(), + setHeader: jest.fn(), + removeHeader: jest.fn(), + write: jest.fn().mockReturnValue(true), + destroy: jest.fn().mockReturnThis(), +}); + +describe('parsed proxy request bodies', () => { + it.each([undefined, false])( + 'leaves unconsumed multipart streams alone (_body=%s)', + parsed => { + const proxy = makeProxyRequest(); + writeProxyBody(proxy, { + headers: { 'content-type': 'multipart/form-data; boundary=test' }, + body: {}, + _body: parsed, + }); + expect(proxy.write).not.toHaveBeenCalled(); + expect(proxy.setHeader).not.toHaveBeenCalled(); + expect(proxy.destroy).not.toHaveBeenCalled(); + }, + ); + + it.each([ + 'application/json', + 'application/json; charset=utf-8', + 'Application/JSON; Charset=UTF-8', + ])('serializes parsed %s and frames UTF-8 bytes', contentType => { + const proxy = makeProxyRequest(); + writeProxyBody(proxy, { + headers: { 'content-type': contentType }, + body: { query: "SELECT 'café'" }, + _body: true, + }); + const expected = '{"query":"SELECT \'café\'"}'; + expect(proxy.write).toHaveBeenCalledWith(expected); + expect(proxy.setHeader).toHaveBeenCalledWith( + 'content-length', + Buffer.byteLength(expected), + ); + expect(proxy.destroy).not.toHaveBeenCalled(); + }); + + it('preserves repeated form fields and empty values', () => { + const proxy = makeProxyRequest(); + writeProxyBody(proxy, { + headers: { + 'content-type': 'application/x-www-form-urlencoded; charset=UTF-8', + }, + body: { query: 'SELECT 1', param: ['a+b', 'c d'], empty: '' }, + _body: true, + }); + expect(proxy.write).toHaveBeenCalledWith( + 'query=SELECT+1¶m=a%2Bb¶m=c+d&empty=', + ); + }); + + it.each(['', 'SELECT 1', Buffer.from([0, 255, 10])])( + 'preserves parsed text and buffer payloads (%s)', + body => { + const proxy = makeProxyRequest(); + writeProxyBody(proxy, { headers: {}, body, _body: true }); + expect(proxy.write).toHaveBeenCalledWith(body); + expect(proxy.setHeader).toHaveBeenCalledWith( + 'content-length', + Buffer.byteLength(body), + ); + }, + ); + + it('keeps chunked framing without a conflicting content length', () => { + const proxy = makeProxyRequest(); + proxy.getHeader.mockReturnValue('chunked'); + writeProxyBody(proxy, { + headers: { 'content-type': 'application/json' }, + body: { query: 'SELECT 1' }, + _body: true, + }); + expect(proxy.removeHeader).toHaveBeenCalledWith('content-length'); + expect(proxy.setHeader).not.toHaveBeenCalled(); + expect(proxy.write).toHaveBeenCalledWith('{"query":"SELECT 1"}'); + }); + + it('destroys the upstream request when a body write fails', () => { + const proxy = makeProxyRequest(); + const error = new Error('write failed'); + proxy.write.mockImplementation(() => { + throw error; + }); + writeProxyBody(proxy, { headers: {}, body: 'SELECT 1', _body: true }); + expect(proxy.destroy).toHaveBeenCalledWith(error); + }); + + it('updates encoding metadata after the parser decoded the payload', () => { + const proxy = makeProxyRequest(); + writeProxyBody(proxy, { + headers: { + 'content-type': 'application/json; charset="utf-16le"', + 'content-encoding': 'gzip', + }, + body: { query: 'SELECT 1' }, + _body: true, + }); + expect(proxy.removeHeader).toHaveBeenCalledWith('content-encoding'); + expect(proxy.setHeader).toHaveBeenCalledWith( + 'content-type', + 'application/json; charset=utf-8', + ); + expect(proxy.write).toHaveBeenCalledWith('{"query":"SELECT 1"}'); + }); + + it('destroys the upstream request when JSON serialization fails', () => { + const proxy = makeProxyRequest(); + const body: { self?: unknown } = {}; + body.self = body; + writeProxyBody(proxy, { + headers: { 'content-type': 'application/json' }, + body, + _body: true, + }); + expect(proxy.destroy).toHaveBeenCalledWith(expect.any(TypeError)); + expect(proxy.write).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/api/src/utils/proxyBody.ts b/packages/api/src/utils/proxyBody.ts new file mode 100644 index 0000000000..2fcb6af61c --- /dev/null +++ b/packages/api/src/utils/proxyBody.ts @@ -0,0 +1,73 @@ +import type { ClientRequest, IncomingHttpHeaders } from 'http'; + +import logger from '@/utils/logger'; + +type ParsedRequest = { + headers: IncomingHttpHeaders; + body?: unknown; + _body?: boolean; +}; + +type ProxyRequest = Pick< + ClientRequest, + 'getHeader' | 'setHeader' | 'removeHeader' | 'write' | 'destroy' +>; + +export function writeProxyBody(proxyReq: ProxyRequest, req: ParsedRequest) { + // body-parser sets _body only when it consumes the stream. Otherwise the + // proxy must pipe the original bytes, even when req.body was set to {}. + if (!req._body) return; + + try { + const contentType = req.headers['content-type']; + const mediaType = contentType?.split(';')[0].trim().toLowerCase(); + let body = req.body; + + if (typeof body !== 'string' && !Buffer.isBuffer(body)) { + if (mediaType === 'application/json') { + body = JSON.stringify(body); + } else if ( + mediaType === 'application/x-www-form-urlencoded' && + body != null && + typeof body === 'object' + ) { + const params = new URLSearchParams(); + const entries: [string, unknown][] = Object.entries(body); + for (const [key, value] of entries) { + for (const item of Array.isArray(value) ? value : [value]) { + params.append(key, String(item)); + } + } + body = params.toString(); + } + } + + if (typeof body !== 'string' && !Buffer.isBuffer(body)) { + throw new Error('Unsupported parsed ClickHouse proxy body'); + } + + // The parsers already decompressed and decoded the incoming payload. + proxyReq.removeHeader('content-encoding'); + if (typeof body === 'string' && contentType) { + const utf8ContentType = contentType.replace( + /;\s*charset\s*=\s*(?:"[^"]*"|[^;]*)/i, + '; charset=utf-8', + ); + if (utf8ContentType !== contentType) { + proxyReq.setHeader('content-type', utf8ContentType); + } + } + + if (proxyReq.getHeader('transfer-encoding')) { + proxyReq.removeHeader('content-length'); + } else { + proxyReq.setHeader('content-length', Buffer.byteLength(body)); + } + proxyReq.write(body); + } catch (err) { + logger.error({ err }, 'Unable to forward ClickHouse proxy body'); + proxyReq.destroy( + err instanceof Error ? err : new Error('Unable to forward proxy body'), + ); + } +} From 9002ff8c4cfb04e184967985bfbdbc5f1c761aa3 Mon Sep 17 00:00:00 2001 From: Aditya Reddy <82123567+adityareddy-dev@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:12:20 -0500 Subject: [PATCH 2/3] api: preserve ClickHouse proxy request bytes --- .changeset/proxy-parsed-bodies.md | 6 +- packages/api/src/api-app.ts | 13 +- .../api/__tests__/clickhouseProxy.int.test.ts | 167 ++++++++---- .../api/src/utils/__tests__/proxyBody.test.ts | 239 ++++++++++++------ packages/api/src/utils/proxyBody.ts | 65 ++--- 5 files changed, 302 insertions(+), 188 deletions(-) diff --git a/.changeset/proxy-parsed-bodies.md b/.changeset/proxy-parsed-bodies.md index f1fc139e10..b947b44a10 100644 --- a/.changeset/proxy-parsed-bodies.md +++ b/.changeset/proxy-parsed-bodies.md @@ -2,6 +2,6 @@ '@hyperdx/api': patch --- -Forward parsed JSON and form request bodies through the ClickHouse proxy, -including charset parameters and repeated form fields. Keep request framing in -sync with serialized bytes and preserve unparsed multipart streams. +Forward the bytes consumed by the ClickHouse proxy's body parsers without +changing JSON values, form fields or text encodings. Keep request framing in +sync with inflated bytes and preserve unparsed multipart streams. diff --git a/packages/api/src/api-app.ts b/packages/api/src/api-app.ts index c2d9028a15..cf19282309 100644 --- a/packages/api/src/api-app.ts +++ b/packages/api/src/api-app.ts @@ -22,6 +22,7 @@ import usageStats from './tasks/usageStats'; import { registerAITelemetry } from './utils/aiTelemetry'; import logger, { expressLogger } from './utils/logger'; import passport from './utils/passport'; +import { captureProxyBody } from './utils/proxyBody'; registerAITelemetry(); @@ -56,9 +57,15 @@ if (!config.IS_CI && config.FRONTEND_URL) { app.disable('x-powered-by'); app.use(compression()); -app.use(express.json({ limit: '32mb' })); -app.use(express.text({ limit: '32mb' })); -app.use(express.urlencoded({ extended: false, limit: '32mb' })); +app.use(express.json({ limit: '32mb', verify: captureProxyBody })); +app.use(express.text({ limit: '32mb', verify: captureProxyBody })); +app.use( + express.urlencoded({ + extended: false, + limit: '32mb', + verify: captureProxyBody, + }), +); app.use(session(sess)); if (!config.IS_LOCAL_APP_MODE) { diff --git a/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts b/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts index efb774e218..0ae173b78c 100644 --- a/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts +++ b/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts @@ -17,7 +17,7 @@ type CapturedRequest = { }; /** - * Pins what /clickhouse-proxy forwards upstream: parsed body re-injection with + * Pins what /clickhouse-proxy forwards upstream: captured parser bytes with * valid framing, the raw-stream passthrough of multipart/form-data bodies, which * `@clickhouse/client-web` emits when query params exceed its URL budget * (regression context: ClickHouse support-escalation #8482), and the @@ -101,7 +101,7 @@ describe('clickhouse-proxy forwarding', () => { }); it.each(['application/json', 'application/json; charset=utf-8'])( - 're-serializes %s and updates Content-Length to its UTF-8 byte length', + 'preserves %s bytes and Content-Length', async contentType => { const { agent, team } = await getLoggedInAgent(server); const connectionId = await createConnection(team._id.toString()); @@ -110,11 +110,7 @@ describe('clickhouse-proxy forwarding', () => { param_name: 'café', }; const original = JSON.stringify(body, null, 2); - const serialized = JSON.stringify(body); - expect(Buffer.byteLength(original)).toBeGreaterThan( - Buffer.byteLength(serialized), - ); - expect(Buffer.byteLength(serialized)).toBeGreaterThan(serialized.length); + expect(Buffer.byteLength(original)).toBeGreaterThan(original.length); await agent .post('/clickhouse-proxy/?query_id=test-json') @@ -125,21 +121,20 @@ describe('clickhouse-proxy forwarding', () => { .expect(200); expect(captured).toHaveLength(1); - expect(captured[0].body).toEqual(Buffer.from(serialized)); + expect(captured[0].body).toEqual(Buffer.from(original)); expect(captured[0].headers['content-type']).toBe(contentType); expect(captured[0].headers['content-length']).toBe( - String(Buffer.byteLength(serialized)), + String(Buffer.byteLength(original)), ); expect(captured[0].headers['transfer-encoding']).toBeUndefined(); }, ); - it('forwards parsed UTF-16LE JSON as UTF-8 with matching headers', async () => { + it('preserves parsed UTF-16LE JSON bytes and charset', async () => { const { agent, team } = await getLoggedInAgent(server); const connectionId = await createConnection(team._id.toString()); const body = { query: "SELECT '東京 🍰' FORMAT JSON" }; const original = Buffer.from(JSON.stringify(body, null, 2), 'utf16le'); - const serialized = Buffer.from(JSON.stringify(body)); const proxyRequest = agent .post('/clickhouse-proxy/?query_id=test-utf16-json') @@ -151,13 +146,11 @@ describe('clickhouse-proxy forwarding', () => { await proxyRequest.expect(200); expect(captured).toHaveLength(1); - expect(captured[0].body).toEqual(serialized); - expect(captured[0].headers['content-type']).toMatch( - /^application\/json;\s*charset=utf-8$/i, - ); - expect(captured[0].headers['content-length']).toBe( - String(serialized.length), + expect(captured[0].body).toEqual(original); + expect(captured[0].headers['content-type']).toBe( + 'application/json; charset=utf-16le', ); + expect(captured[0].headers['content-length']).toBe(String(original.length)); expect(captured[0].headers['content-encoding']).toBeUndefined(); }); @@ -165,8 +158,8 @@ describe('clickhouse-proxy forwarding', () => { const { agent, team } = await getLoggedInAgent(server); const connectionId = await createConnection(team._id.toString()); const body = { query: "SELECT 'café' FORMAT JSON" }; - const original = gzipSync(Buffer.from(JSON.stringify(body, null, 2))); - const serialized = Buffer.from(JSON.stringify(body)); + const serialized = Buffer.from(JSON.stringify(body, null, 2)); + const original = gzipSync(serialized); const proxyRequest = agent .post('/clickhouse-proxy/?query_id=test-gzip-json') @@ -192,62 +185,130 @@ describe('clickhouse-proxy forwarding', () => { it.each([ 'application/x-www-form-urlencoded', 'application/x-www-form-urlencoded; charset=utf-8', - ])( - 're-serializes %s without losing repeated form keys', - async contentType => { + ])('preserves %s bytes and repeated form keys', async contentType => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const original = + 'query=SELECT%20%27caf%C3%A9%27¶m=first¶m=second%20value&empty=&literal=%2B%26%3D'; + const entries = [ + ['query', "SELECT 'café'"], + ['param', 'first'], + ['param', 'second value'], + ['empty', ''], + ['literal', '+&='], + ]; + + await agent + .post('/clickhouse-proxy/?query_id=test-urlencoded') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', contentType) + .send(original) + .timeout({ response: 5000, deadline: 10000 }) + .expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body.toString()).toBe(original); + expect([...new URLSearchParams(captured[0].body.toString())]).toEqual( + entries, + ); + expect(captured[0].headers['content-type']).toBe(contentType); + expect(captured[0].headers['content-length']).toBe( + String(Buffer.byteLength(original)), + ); + }); + + it('keeps chunked framing when forwarding captured JSON bytes', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const body = { query: "SELECT 'café' FORMAT JSON" }; + const original = JSON.stringify(body, null, 2); + const proxyRequest = agent + .post('/clickhouse-proxy/?query_id=test-chunked-json') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json; charset=utf-8') + .set('Transfer-Encoding', 'chunked') + .timeout({ response: 5000, deadline: 10000 }); + + proxyRequest.write(original.slice(0, 10)); + proxyRequest.write(original.slice(10)); + await proxyRequest.expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body.toString()).toBe(original); + expect(captured[0].headers['transfer-encoding']).toBe('chunked'); + expect(captured[0].headers['content-length']).toBeUndefined(); + }); + + it.each(['/clickhouse-proxy/', '/CLICKHOUSE-PROXY/'])( + 'preserves JSON number literals and duplicate keys on %s', + async route => { const { agent, team } = await getLoggedInAgent(server); const connectionId = await createConnection(team._id.toString()); const original = - 'query=SELECT%20%27caf%C3%A9%27¶m=first¶m=second%20value&empty=&literal=%2B%26%3D'; - const entries = [ - ['query', "SELECT 'café'"], - ['param', 'first'], - ['param', 'second value'], - ['empty', ''], - ['literal', '+&='], - ]; - const serialized = new URLSearchParams(entries).toString(); + '{"id":9007199254740993,"value":1.0,"value":1e2,"query":"SELECT 1"}'; await agent - .post('/clickhouse-proxy/?query_id=test-urlencoded') + .post(`${route}?query_id=test-json-fidelity`) .set('x-hyperdx-connection-id', connectionId) - .set('Content-Type', contentType) + .set('Content-Type', 'application/json; charset=utf-8') .send(original) .timeout({ response: 5000, deadline: 10000 }) .expect(200); expect(captured).toHaveLength(1); - expect(captured[0].body.toString()).toBe(serialized); - expect([...new URLSearchParams(captured[0].body.toString())]).toEqual( - entries, - ); - expect(captured[0].headers['content-type']).toBe(contentType); - expect(captured[0].headers['content-length']).toBe( - String(Buffer.byteLength(serialized)), - ); + expect(captured[0].body).toEqual(Buffer.from(original)); }, ); - it('keeps chunked framing when re-injecting parsed JSON', async () => { + it('preserves an empty JSON body for query-string SQL', async () => { const { agent, team } = await getLoggedInAgent(server); const connectionId = await createConnection(team._id.toString()); - const body = { query: "SELECT 'café' FORMAT JSON" }; - const original = JSON.stringify(body, null, 2); + + await agent + .post('/clickhouse-proxy/?query=SELECT+1') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json') + .set('Content-Length', '0') + .timeout({ response: 5000, deadline: 10000 }) + .expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body).toEqual(Buffer.alloc(0)); + expect(captured[0].headers['content-length']).toBe('0'); + }); + + it('preserves bytes under duplicate charset parameters', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + const contentType = 'application/json; charset=utf-8; charset=utf-16le'; + const original = Buffer.from('{"query":"SELECT 1"}', 'utf16le'); const proxyRequest = agent - .post('/clickhouse-proxy/?query_id=test-chunked-json') + .post('/clickhouse-proxy/?query_id=test-duplicate-charset') .set('x-hyperdx-connection-id', connectionId) - .set('Content-Type', 'application/json; charset=utf-8') - .set('Transfer-Encoding', 'chunked') + .set('Content-Type', contentType) + .set('Content-Length', String(original.length)) .timeout({ response: 5000, deadline: 10000 }); - - proxyRequest.write(original.slice(0, 10)); - proxyRequest.write(original.slice(10)); + proxyRequest.write(original); await proxyRequest.expect(200); expect(captured).toHaveLength(1); - expect(captured[0].body.toString()).toBe(JSON.stringify(body)); - expect(captured[0].headers['transfer-encoding']).toBe('chunked'); - expect(captured[0].headers['content-length']).toBeUndefined(); + expect(captured[0].body).toEqual(original); + expect(captured[0].headers['content-type']).toBe(contentType); + expect(captured[0].headers['content-length']).toBe(String(original.length)); + }); + + it('still rejects invalid JSON before forwarding', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + + await agent + .post('/clickhouse-proxy/') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json') + .send('{invalid') + .expect(400); + + expect(captured).toHaveLength(0); }); it('forwards a large multipart/form-data body byte-for-byte (unparsed stream passthrough)', async () => { diff --git a/packages/api/src/utils/__tests__/proxyBody.test.ts b/packages/api/src/utils/__tests__/proxyBody.test.ts index 9f382fec48..4fef70bdd0 100644 --- a/packages/api/src/utils/__tests__/proxyBody.test.ts +++ b/packages/api/src/utils/__tests__/proxyBody.test.ts @@ -1,4 +1,7 @@ -import { writeProxyBody } from '@/utils/proxyBody'; +import { IncomingMessage, ServerResponse } from 'http'; +import { Socket } from 'net'; + +import { captureProxyBody, writeProxyBody } from '@/utils/proxyBody'; jest.mock('@/utils/logger', () => ({ __esModule: true, @@ -13,80 +16,178 @@ const makeProxyRequest = () => ({ destroy: jest.fn().mockReturnThis(), }); -describe('parsed proxy request bodies', () => { - it.each([undefined, false])( - 'leaves unconsumed multipart streams alone (_body=%s)', +describe('ClickHouse proxy body capture', () => { + const makeRequest = ( + path?: string, + url?: string, + ): Parameters[0] => + Object.assign(new IncomingMessage(new Socket()), { path, url }); + const response = new ServerResponse(new IncomingMessage(new Socket())); + + it.each([ + { path: '/clickhouse-proxy', url: '/clickhouse-proxy' }, + { path: '/CLICKHOUSE-PROXY', url: '/CLICKHOUSE-PROXY' }, + { path: '/clickhouse-proxy/', url: '/clickhouse-proxy/' }, + { + path: '/clickhouse-proxy', + url: '/clickhouse-proxy?query=SELECT%201', + }, + { + path: '/clickhouse-proxy/database', + url: '/clickhouse-proxy/database?query=SELECT%201', + }, + { + path: '/clickhouse-proxy', + url: 'http://localhost/clickhouse-proxy?query=SELECT%201', + }, + ])('retains the inflated bytes for $url', ({ path, url }) => { + const req = makeRequest(path, url); + const body = Buffer.from('{ "value": 9007199254740993, "scale": 1.000 }'); + + captureProxyBody(req, response, body); + + expect(req._hdxProxyBody).toBe(body); + }); + + it.each([ + undefined, + '', + '/clickhouse-proxy-other', + '/login', + '/api/clickhouse-proxy', + ])('does not retain bodies for unrelated path %s', path => { + const req = makeRequest(path); + + captureProxyBody(req, response, Buffer.from('private body')); + + expect(req._hdxProxyBody).toBeUndefined(); + }); +}); + +describe('captured proxy request bodies', () => { + it('leaves an unconsumed stream alone', () => { + const proxy = makeProxyRequest(); + + writeProxyBody(proxy, { headers: {} }); + + expect(proxy.write).not.toHaveBeenCalled(); + expect(proxy.getHeader).not.toHaveBeenCalled(); + expect(proxy.setHeader).not.toHaveBeenCalled(); + expect(proxy.removeHeader).not.toHaveBeenCalled(); + expect(proxy.destroy).not.toHaveBeenCalled(); + }); + + it.each([undefined, false, true])( + 'does not reconstruct a parsed body without captured bytes (_body=%s)', parsed => { const proxy = makeProxyRequest(); - writeProxyBody(proxy, { - headers: { 'content-type': 'multipart/form-data; boundary=test' }, - body: {}, + const req = { + headers: {}, + _hdxProxyBody: undefined, + body: { query: 'SELECT 1' }, _body: parsed, - }); + }; + + writeProxyBody(proxy, req); + expect(proxy.write).not.toHaveBeenCalled(); expect(proxy.setHeader).not.toHaveBeenCalled(); + expect(proxy.removeHeader).not.toHaveBeenCalled(); expect(proxy.destroy).not.toHaveBeenCalled(); }, ); - it.each([ - 'application/json', - 'application/json; charset=utf-8', - 'Application/JSON; Charset=UTF-8', - ])('serializes parsed %s and frames UTF-8 bytes', contentType => { + it.each<[string, Buffer]>([ + [ + 'JSON number spelling and whitespace', + Buffer.from('{ "value": 9007199254740993, "scale": 1.000 }'), + ], + [ + 'form field spelling and repeated values', + Buffer.from('query=SELECT%201¶m=a%2bb¶m=c+d&empty='), + ], + ['binary bytes', Buffer.from([0, 255, 10])], + ])('preserves %s', (_description, body) => { const proxy = makeProxyRequest(); - writeProxyBody(proxy, { - headers: { 'content-type': contentType }, - body: { query: "SELECT 'café'" }, - _body: true, - }); - const expected = '{"query":"SELECT \'café\'"}'; - expect(proxy.write).toHaveBeenCalledWith(expected); - expect(proxy.setHeader).toHaveBeenCalledWith( - 'content-length', - Buffer.byteLength(expected), - ); + + writeProxyBody(proxy, { headers: {}, _hdxProxyBody: body }); + + expect(proxy.write).toHaveBeenCalledTimes(1); + expect(proxy.write).toHaveBeenCalledWith(body); + expect(proxy.setHeader).toHaveBeenCalledWith('content-length', body.length); expect(proxy.destroy).not.toHaveBeenCalled(); }); - it('preserves repeated form fields and empty values', () => { + it('forwards an empty captured body with a zero content length', () => { const proxy = makeProxyRequest(); - writeProxyBody(proxy, { - headers: { - 'content-type': 'application/x-www-form-urlencoded; charset=UTF-8', - }, - body: { query: 'SELECT 1', param: ['a+b', 'c d'], empty: '' }, + const body = Buffer.alloc(0); + + writeProxyBody(proxy, { headers: {}, _hdxProxyBody: body }); + + expect(proxy.write).toHaveBeenCalledWith(body); + expect(proxy.setHeader).toHaveBeenCalledWith('content-length', 0); + expect(proxy.destroy).not.toHaveBeenCalled(); + }); + + it('uses the captured bytes instead of the parsed body', () => { + const proxy = makeProxyRequest(); + const body = Buffer.from('{ "value": 9007199254740993 }'); + const req = { + headers: {}, + _hdxProxyBody: body, + body: { value: 9007199254740992 }, _body: true, - }); - expect(proxy.write).toHaveBeenCalledWith( - 'query=SELECT+1¶m=a%2Bb¶m=c+d&empty=', - ); + }; + + writeProxyBody(proxy, req); + + expect(proxy.write).toHaveBeenCalledWith(body); }); - it.each(['', 'SELECT 1', Buffer.from([0, 255, 10])])( - 'preserves parsed text and buffer payloads (%s)', - body => { - const proxy = makeProxyRequest(); - writeProxyBody(proxy, { headers: {}, body, _body: true }); - expect(proxy.write).toHaveBeenCalledWith(body); - expect(proxy.setHeader).toHaveBeenCalledWith( - 'content-length', - Buffer.byteLength(body), - ); - }, - ); + it('frames the byte length of a multibyte payload', () => { + const proxy = makeProxyRequest(); + const text = "SELECT 'café'"; + const body = Buffer.from(text); + + writeProxyBody(proxy, { headers: {}, _hdxProxyBody: body }); + + expect(body.length).toBeGreaterThan(text.length); + expect(proxy.setHeader).toHaveBeenCalledWith('content-length', body.length); + expect(proxy.write).toHaveBeenCalledWith(body); + }); - it('keeps chunked framing without a conflicting content length', () => { + it('removes compression metadata and preserves the original charset', () => { const proxy = makeProxyRequest(); + const body = Buffer.from("SELECT 'café'", 'utf16le'); + const req = { + _hdxProxyBody: body, + headers: { + 'content-type': 'text/plain; charset="utf-16le"', + 'content-encoding': 'gzip', + }, + }; + + writeProxyBody(proxy, req); + + expect(proxy.removeHeader).toHaveBeenCalledWith('content-encoding'); + expect(proxy.removeHeader).not.toHaveBeenCalledWith('content-type'); + expect(proxy.setHeader).toHaveBeenCalledTimes(1); + expect(proxy.setHeader).toHaveBeenCalledWith('content-length', body.length); + expect(proxy.write).toHaveBeenCalledWith(body); + }); + + it('keeps transfer encoding without a conflicting content length', () => { + const proxy = makeProxyRequest(); + const body = Buffer.from('SELECT 1'); proxy.getHeader.mockReturnValue('chunked'); - writeProxyBody(proxy, { - headers: { 'content-type': 'application/json' }, - body: { query: 'SELECT 1' }, - _body: true, - }); + + writeProxyBody(proxy, { headers: {}, _hdxProxyBody: body }); + + expect(proxy.getHeader).toHaveBeenCalledWith('transfer-encoding'); expect(proxy.removeHeader).toHaveBeenCalledWith('content-length'); + expect(proxy.removeHeader).toHaveBeenCalledWith('content-encoding'); expect(proxy.setHeader).not.toHaveBeenCalled(); - expect(proxy.write).toHaveBeenCalledWith('{"query":"SELECT 1"}'); + expect(proxy.write).toHaveBeenCalledWith(body); }); it('destroys the upstream request when a body write fails', () => { @@ -95,38 +196,12 @@ describe('parsed proxy request bodies', () => { proxy.write.mockImplementation(() => { throw error; }); - writeProxyBody(proxy, { headers: {}, body: 'SELECT 1', _body: true }); - expect(proxy.destroy).toHaveBeenCalledWith(error); - }); - it('updates encoding metadata after the parser decoded the payload', () => { - const proxy = makeProxyRequest(); writeProxyBody(proxy, { - headers: { - 'content-type': 'application/json; charset="utf-16le"', - 'content-encoding': 'gzip', - }, - body: { query: 'SELECT 1' }, - _body: true, + headers: {}, + _hdxProxyBody: Buffer.from('SELECT 1'), }); - expect(proxy.removeHeader).toHaveBeenCalledWith('content-encoding'); - expect(proxy.setHeader).toHaveBeenCalledWith( - 'content-type', - 'application/json; charset=utf-8', - ); - expect(proxy.write).toHaveBeenCalledWith('{"query":"SELECT 1"}'); - }); - it('destroys the upstream request when JSON serialization fails', () => { - const proxy = makeProxyRequest(); - const body: { self?: unknown } = {}; - body.self = body; - writeProxyBody(proxy, { - headers: { 'content-type': 'application/json' }, - body, - _body: true, - }); - expect(proxy.destroy).toHaveBeenCalledWith(expect.any(TypeError)); - expect(proxy.write).not.toHaveBeenCalled(); + expect(proxy.destroy).toHaveBeenCalledWith(error); }); }); diff --git a/packages/api/src/utils/proxyBody.ts b/packages/api/src/utils/proxyBody.ts index 2fcb6af61c..801f472450 100644 --- a/packages/api/src/utils/proxyBody.ts +++ b/packages/api/src/utils/proxyBody.ts @@ -1,11 +1,9 @@ -import type { ClientRequest, IncomingHttpHeaders } from 'http'; +import type { ClientRequest, IncomingMessage, ServerResponse } from 'http'; import logger from '@/utils/logger'; -type ParsedRequest = { - headers: IncomingHttpHeaders; - body?: unknown; - _body?: boolean; +type ParsedRequest = Pick & { + _hdxProxyBody?: Buffer; }; type ProxyRequest = Pick< @@ -13,55 +11,28 @@ type ProxyRequest = Pick< 'getHeader' | 'setHeader' | 'removeHeader' | 'write' | 'destroy' >; +export function captureProxyBody( + req: IncomingMessage & ParsedRequest & { path?: string }, + _res: ServerResponse, + body: Buffer, +) { + if (/^\/clickhouse-proxy(?:\/|$)/i.test(req.path ?? '')) { + // verify receives inflated bytes before parsing can change their values. + req._hdxProxyBody = body; + } +} + export function writeProxyBody(proxyReq: ProxyRequest, req: ParsedRequest) { - // body-parser sets _body only when it consumes the stream. Otherwise the - // proxy must pipe the original bytes, even when req.body was set to {}. - if (!req._body) return; + const body = req._hdxProxyBody; + // No parser consumed this stream, so let the proxy pipe the original bytes. + if (body === undefined) return; try { - const contentType = req.headers['content-type']; - const mediaType = contentType?.split(';')[0].trim().toLowerCase(); - let body = req.body; - - if (typeof body !== 'string' && !Buffer.isBuffer(body)) { - if (mediaType === 'application/json') { - body = JSON.stringify(body); - } else if ( - mediaType === 'application/x-www-form-urlencoded' && - body != null && - typeof body === 'object' - ) { - const params = new URLSearchParams(); - const entries: [string, unknown][] = Object.entries(body); - for (const [key, value] of entries) { - for (const item of Array.isArray(value) ? value : [value]) { - params.append(key, String(item)); - } - } - body = params.toString(); - } - } - - if (typeof body !== 'string' && !Buffer.isBuffer(body)) { - throw new Error('Unsupported parsed ClickHouse proxy body'); - } - - // The parsers already decompressed and decoded the incoming payload. proxyReq.removeHeader('content-encoding'); - if (typeof body === 'string' && contentType) { - const utf8ContentType = contentType.replace( - /;\s*charset\s*=\s*(?:"[^"]*"|[^;]*)/i, - '; charset=utf-8', - ); - if (utf8ContentType !== contentType) { - proxyReq.setHeader('content-type', utf8ContentType); - } - } - if (proxyReq.getHeader('transfer-encoding')) { proxyReq.removeHeader('content-length'); } else { - proxyReq.setHeader('content-length', Buffer.byteLength(body)); + proxyReq.setHeader('content-length', body.length); } proxyReq.write(body); } catch (err) { From db7b16fb75d0f6c12222ca51142408ac665e8811 Mon Sep 17 00:00:00 2001 From: Aditya Reddy <82123567+adityareddy-dev@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:24:38 -0500 Subject: [PATCH 3/3] api: fail fast when a proxy body was parsed but not captured --- .../api/__tests__/clickhouseProxy.int.test.ts | 7 ++--- .../api/src/utils/__tests__/proxyBody.test.ts | 27 +++++++++++++++++-- packages/api/src/utils/proxyBody.ts | 14 ++++++++-- 3 files changed, 41 insertions(+), 7 deletions(-) diff --git a/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts b/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts index 0ae173b78c..9e09c5406f 100644 --- a/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts +++ b/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts @@ -4,6 +4,7 @@ import { gzipSync } from 'zlib'; import { getLoggedInAgent, getServer } from '@/fixtures'; import Connection from '@/models/connection'; +import logger from '@/utils/logger'; // Use the real proxy instead of the blanket mock in jest.setup.ts; the int // jest config transpiles it (see jest.int.config.js). @@ -342,7 +343,7 @@ describe('clickhouse-proxy forwarding', () => { ]); expect(Buffer.byteLength(body)).toBeGreaterThan(4096); - const consoleError = jest.spyOn(console, 'error'); + const loggerError = jest.spyOn(logger, 'error'); try { await agent .post('/clickhouse-proxy/?query_id=test-multipart') @@ -360,9 +361,9 @@ describe('clickhouse-proxy forwarding', () => { expect(captured[0].headers['content-length']).toBe( String(Buffer.byteLength(body)), ); - expect(consoleError).not.toHaveBeenCalled(); + expect(loggerError).not.toHaveBeenCalled(); } finally { - consoleError.mockRestore(); + loggerError.mockRestore(); } }); diff --git a/packages/api/src/utils/__tests__/proxyBody.test.ts b/packages/api/src/utils/__tests__/proxyBody.test.ts index 4fef70bdd0..7f9fdd3092 100644 --- a/packages/api/src/utils/__tests__/proxyBody.test.ts +++ b/packages/api/src/utils/__tests__/proxyBody.test.ts @@ -1,6 +1,7 @@ import { IncomingMessage, ServerResponse } from 'http'; import { Socket } from 'net'; +import logger from '@/utils/logger'; import { captureProxyBody, writeProxyBody } from '@/utils/proxyBody'; jest.mock('@/utils/logger', () => ({ @@ -77,8 +78,8 @@ describe('captured proxy request bodies', () => { expect(proxy.destroy).not.toHaveBeenCalled(); }); - it.each([undefined, false, true])( - 'does not reconstruct a parsed body without captured bytes (_body=%s)', + it.each([undefined, false])( + 'does not reconstruct a body without captured bytes (_body=%s)', parsed => { const proxy = makeProxyRequest(); const req = { @@ -97,6 +98,28 @@ describe('captured proxy request bodies', () => { }, ); + it('destroys the upstream request when a parsed body was not captured', () => { + const proxy = makeProxyRequest(); + const req = { + headers: {}, + _hdxProxyBody: undefined, + body: { query: 'SELECT 1' }, + _body: true, + }; + + writeProxyBody(proxy, req); + + expect(proxy.write).not.toHaveBeenCalled(); + expect(proxy.setHeader).not.toHaveBeenCalled(); + expect(proxy.destroy).toHaveBeenCalledTimes(1); + const [[error]] = proxy.destroy.mock.calls; + expect(error).toBeInstanceOf(Error); + expect(logger.error).toHaveBeenCalledWith( + { err: error }, + expect.any(String), + ); + }); + it.each<[string, Buffer]>([ [ 'JSON number spelling and whitespace', diff --git a/packages/api/src/utils/proxyBody.ts b/packages/api/src/utils/proxyBody.ts index 801f472450..5f0a50aaff 100644 --- a/packages/api/src/utils/proxyBody.ts +++ b/packages/api/src/utils/proxyBody.ts @@ -4,6 +4,8 @@ import logger from '@/utils/logger'; type ParsedRequest = Pick & { _hdxProxyBody?: Buffer; + // Set by body-parser before it reads the stream. + _body?: boolean; }; type ProxyRequest = Pick< @@ -24,8 +26,16 @@ export function captureProxyBody( export function writeProxyBody(proxyReq: ProxyRequest, req: ParsedRequest) { const body = req._hdxProxyBody; - // No parser consumed this stream, so let the proxy pipe the original bytes. - if (body === undefined) return; + if (body === undefined) { + // No parser consumed this stream, so let the proxy pipe the original bytes. + if (!req._body) return; + + // A drained stream would leave upstream waiting on the original length. + const err = new Error('ClickHouse proxy body was parsed but not captured'); + logger.error({ err }, 'Unable to forward ClickHouse proxy body'); + proxyReq.destroy(err); + return; + } try { proxyReq.removeHeader('content-encoding');