From ebf47c47340d9f19a10c6f01df2381dc312852ca Mon Sep 17 00:00:00 2001 From: Sujeito Operator Date: Sat, 22 Aug 2026 21:12:20 +0000 Subject: [PATCH] fix(clickhouse-proxy): re-inject parsed bodies by media type, not exact header match MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `/clickhouse-proxy` re-injects `req.body` after express's parsers may have consumed the request stream, but it decided how by comparing `req.headers['content-type']` for string equality against `'application/json'`. Anything carrying a parameter — `application/json; charset=utf-8` — or any other parsed type missed that branch and reached `proxyReq.write()` as an object. `write()` throws on an object, the throw was swallowed by a bare `catch`, and the stream had already been drained, so nothing was ever sent: the upstream sat waiting for a `Content-Length` worth of bytes that never arrived until the client gave up. Three shapes were affected, all reported in #2942: - `application/json; charset=utf-8` and `application/x-www-form-urlencoded` were parsed and then silently dropped. - `multipart/form-data` — what `@clickhouse/client-web` emits once query params exceed its URL budget — hit `proxyReq.write({})` on every request, because body-parser runs `req.body = req.body || {}` *before* its content-type check. That write always threw; forwarding worked only because httpxy went on to pipe the still-unconsumed stream. - `Content-Length` was never resynced, so a payload whose byte length changed under `JSON.stringify` normalization could be truncated or over-read. `planProxyBody` now makes that decision as a pure function: strip parameters off the content type, write strings and buffers as-is, re-encode urlencoded objects through `URLSearchParams` so repeated keys stay repeated, and JSON everything else. Whether the stream was consumed is read from body-parser's own `_body` flag rather than inferred from the shape of `req.body`, since `{}` cannot distinguish "parsed empty object" from "never touched". Unconsumed requests are skipped explicitly so the raw stream pipes by intent rather than by accident, `Content-Length` follows the re-serialized payload, and a write that does fail destroys the proxied request instead of quietly forwarding a body-less one. Closes #2942 --- .../clickhouse-proxy-body-reinjection.md | 24 +++ .../api/__tests__/clickhouseProxy.int.test.ts | 75 ++++++++- .../api/__tests__/clickhouseProxy.test.ts | 148 ++++++++++++++++++ .../api/src/routers/api/clickhouseProxy.ts | 122 +++++++++++++-- 4 files changed, 356 insertions(+), 13 deletions(-) create mode 100644 .changeset/clickhouse-proxy-body-reinjection.md create mode 100644 packages/api/src/routers/api/__tests__/clickhouseProxy.test.ts diff --git a/.changeset/clickhouse-proxy-body-reinjection.md b/.changeset/clickhouse-proxy-body-reinjection.md new file mode 100644 index 0000000000..02c943a852 --- /dev/null +++ b/.changeset/clickhouse-proxy-body-reinjection.md @@ -0,0 +1,24 @@ +--- +'@hyperdx/api': patch +--- + +Fix `/clickhouse-proxy` request-body re-injection, which dropped every body +whose content type was not the exact string `application/json`. The handler +compared `req.headers['content-type']` for equality, so +`application/json; charset=utf-8` and `application/x-www-form-urlencoded` fell +through with `req.body` still an object: `express.json()`/`express.urlencoded()` +had already drained the stream, `proxyReq.write(object)` threw, the throw was +swallowed by a bare `catch`, and nothing was ever written — leaving the upstream +blocked on a `Content-Length` worth of bytes that never arrived, so the request +hung until the client gave up. Serialization now keys off the media type with +parameters stripped, urlencoded bodies are re-encoded through `URLSearchParams` +(repeated keys stay repeated instead of collapsing to `a=1%2C2`), and +`Content-Length` is resynced to the re-serialized payload so normalization can +no longer truncate or over-read it. Requests that no parser consumed — +`multipart/form-data`, which `@clickhouse/client-web` emits once query params +exceed its URL budget — are now explicitly skipped rather than throwing on +`proxyReq.write({})`: body-parser sets `req.body = {}` before its content-type +check, so that write always threw and forwarding only worked because httpxy went +on to pipe the untouched stream. A write that does fail now destroys the proxied +request, surfacing as the proxy's own 500 instead of a body-less request and an +opaque ClickHouse 400. 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 b658edfd85..df5dd61ddc 100644 --- a/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts +++ b/packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts @@ -20,8 +20,9 @@ type CapturedRequest = { * `@clickhouse/client-web` emits when query params exceed its URL budget * (regression context: ClickHouse support-escalation #8482). * - * Known-latent cases (charset-suffixed JSON, urlencoded) are not covered; - * tracked in https://github.com/hyperdxio/hyperdx/issues/2942. + * Charset-suffixed JSON, urlencoded bodies and Content-Length sync are covered + * below; before #2942 each of them hung until the client gave up. The + * serialization decision itself is unit-tested in clickhouseProxy.test.ts. */ describe('clickhouse-proxy body forwarding', () => { const server = getServer(); @@ -139,6 +140,76 @@ describe('clickhouse-proxy body forwarding', () => { ); }); + it('forwards a charset-suffixed application/json body', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + + const payload = { + query: 'SELECT 1 FORMAT JSON', + query_params: { p0: 'x' }, + }; + await agent + .post('/clickhouse-proxy/?query_id=test-json-charset') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json; charset=utf-8') + .send(payload) + .expect(200); + + expect(captured).toHaveLength(1); + expect(JSON.parse(captured[0].body.toString())).toEqual(payload); + // The re-serialized payload, not the client's original framing, must set + // the length the upstream reads. + expect(captured[0].headers['content-length']).toBe( + String(captured[0].body.length), + ); + }); + + it('forwards an application/x-www-form-urlencoded body', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + + await agent + .post('/clickhouse-proxy/?query_id=test-urlencoded') + .set('x-hyperdx-connection-id', connectionId) + .type('form') + .send({ query: 'SELECT 1 FORMAT JSON', param_p0: 'a b&c' }) + .expect(200); + + expect(captured).toHaveLength(1); + const forwarded = new URLSearchParams(captured[0].body.toString()); + expect(forwarded.get('query')).toBe('SELECT 1 FORMAT JSON'); + expect(forwarded.get('param_p0')).toBe('a b&c'); + expect(captured[0].headers['content-length']).toBe( + String(captured[0].body.length), + ); + }); + + it('syncs Content-Length when re-serialization changes the byte length', async () => { + const { agent, team } = await getLoggedInAgent(server); + const connectionId = await createConnection(team._id.toString()); + + // Pretty-printed on the wire, compact after JSON.stringify: the byte + // length the client announced is not the length the upstream will read. + const payload = { query: 'SELECT 1 FORMAT JSON' }; + const pretty = JSON.stringify(payload, null, 4); + expect(Buffer.byteLength(pretty)).toBeGreaterThan( + Buffer.byteLength(JSON.stringify(payload)), + ); + + await agent + .post('/clickhouse-proxy/?query_id=test-json-length') + .set('x-hyperdx-connection-id', connectionId) + .set('Content-Type', 'application/json') + .send(pretty) + .expect(200); + + expect(captured).toHaveLength(1); + expect(captured[0].body.toString()).toBe(JSON.stringify(payload)); + expect(captured[0].headers['content-length']).toBe( + String(Buffer.byteLength(JSON.stringify(payload))), + ); + }); + it('rejects requests without a connection id header', async () => { const { agent } = await getLoggedInAgent(server); diff --git a/packages/api/src/routers/api/__tests__/clickhouseProxy.test.ts b/packages/api/src/routers/api/__tests__/clickhouseProxy.test.ts new file mode 100644 index 0000000000..23493c7782 --- /dev/null +++ b/packages/api/src/routers/api/__tests__/clickhouseProxy.test.ts @@ -0,0 +1,148 @@ +import { planProxyBody } from '@/routers/api/clickhouseProxy'; + +/** + * `bodyWasParsed` mirrors body-parser's `req._body`: true only when a parser + * actually drained the stream. See planProxyBody's docblock for why the shape + * of `req.body` cannot stand in for it. + */ +describe('planProxyBody', () => { + describe('when no parser consumed the stream', () => { + it('skips a multipart body-parser placeholder', () => { + // body-parser sets `req.body = {}` before its content-type check, so an + // untouched multipart request arrives looking like a parsed empty object. + expect( + planProxyBody('multipart/form-data; boundary=abc', {}, false), + ).toEqual({ action: 'skip' }); + }); + + it('skips when there is no content-type at all', () => { + expect(planProxyBody(undefined, {}, false)).toEqual({ action: 'skip' }); + }); + + it('skips even when the content-type looks parseable', () => { + // `express.json()` bails out before reading when there is no body. + expect(planProxyBody('application/json', {}, false)).toEqual({ + action: 'skip', + }); + }); + }); + + describe('string and buffer bodies', () => { + it('writes a text/plain SQL body verbatim', () => { + const sql = 'SELECT count() FROM system.tables FORMAT JSON'; + expect(planProxyBody('text/plain', sql, true)).toEqual({ + action: 'write', + payload: sql, + }); + }); + + it('preserves an empty string body rather than skipping it', () => { + expect(planProxyBody('text/plain', '', true)).toEqual({ + action: 'write', + payload: '', + }); + }); + + it('writes a buffer body unchanged', () => { + const buf = Buffer.from('SELECT 1'); + expect(planProxyBody('application/octet-stream', buf, true)).toEqual({ + action: 'write', + payload: buf, + }); + }); + + it('skips a null body', () => { + expect(planProxyBody('application/json', null, true)).toEqual({ + action: 'skip', + }); + }); + }); + + describe('json bodies', () => { + const payload = { query: 'SELECT 1 FORMAT JSON' }; + + it('serializes a bare application/json body', () => { + expect(planProxyBody('application/json', payload, true)).toEqual({ + action: 'write', + payload: JSON.stringify(payload), + }); + }); + + it.each([ + 'application/json; charset=utf-8', + 'application/json;charset=UTF-8', + 'APPLICATION/JSON; charset=utf-8', + 'application/json ; charset=utf-8', + ])('serializes charset-suffixed %s', contentType => { + // The strict `=== 'application/json'` comparison this replaces missed + // every one of these, leaving an object for `proxyReq.write` to throw on. + expect(planProxyBody(contentType, payload, true)).toEqual({ + action: 'write', + payload: JSON.stringify(payload), + }); + }); + + it('serializes an empty parsed object', () => { + expect(planProxyBody('application/json', {}, true)).toEqual({ + action: 'write', + payload: '{}', + }); + }); + + it('serializes an array body', () => { + expect(planProxyBody('application/json', [1, 2], true)).toEqual({ + action: 'write', + payload: '[1,2]', + }); + }); + }); + + describe('urlencoded bodies', () => { + it('re-serializes a parsed form body', () => { + const plan = planProxyBody( + 'application/x-www-form-urlencoded', + { query: 'SELECT 1 FORMAT JSON', param_p0: 'a b&c' }, + true, + ); + if (plan.action !== 'write') { + throw new Error(`expected a write plan, got ${plan.action}`); + } + + const forwarded = new URLSearchParams(String(plan.payload)); + expect(forwarded.get('query')).toBe('SELECT 1 FORMAT JSON'); + // Re-encoding has to survive a round trip, not just look plausible. + expect(forwarded.get('param_p0')).toBe('a b&c'); + }); + + it('honours a charset suffix on the form content-type', () => { + expect( + planProxyBody( + 'application/x-www-form-urlencoded; charset=utf-8', + { a: '1' }, + true, + ), + ).toEqual({ action: 'write', payload: 'a=1' }); + }); + + it('keeps repeated keys repeated instead of joining them', () => { + // `new URLSearchParams({a: ['1','2']})` would emit `a=1%2C2`. + expect( + planProxyBody( + 'application/x-www-form-urlencoded', + { a: ['1', '2'] }, + true, + ), + ).toEqual({ action: 'write', payload: 'a=1&a=2' }); + }); + + it('drops undefined values', () => { + expect( + planProxyBody( + 'application/x-www-form-urlencoded', + { a: '1', b: undefined }, + true, + ), + ).toEqual({ action: 'write', payload: 'a=1' }); + }); + }); +}); diff --git a/packages/api/src/routers/api/clickhouseProxy.ts b/packages/api/src/routers/api/clickhouseProxy.ts index e41946c1b0..cac5babf1c 100644 --- a/packages/api/src/routers/api/clickhouseProxy.ts +++ b/packages/api/src/routers/api/clickhouseProxy.ts @@ -82,6 +82,90 @@ const router = express.Router(); const CUSTOM_SETTING_KEY_SEP = '_'; const CUSTOM_SETTING_KEY_USER_SUFFIX = 'user'; +declare module 'http' { + interface IncomingMessage { + /** + * Set by body-parser once it has drained the request stream. It is the + * only signal that separates a parsed empty body from an untouched one — + * see planProxyBody. + */ + _body?: boolean; + } +} + +const URLENCODED_MEDIA_TYPE = 'application/x-www-form-urlencoded'; + +/** + * What to do with a request body the express parsers may have already + * consumed. + * + * `skip` means the stream is still intact, so the proxy can pipe the original + * bytes; writing here would corrupt or duplicate them. + */ +export type ProxyBodyPlan = + | { action: 'write'; payload: string | Buffer } + | { action: 'skip' }; + +const mediaTypeOf = (contentType: string | undefined): string => + (contentType ?? '').split(';', 1)[0].trim().toLowerCase(); + +const serializeUrlencoded = (body: object): string => { + const params = new URLSearchParams(); + for (const [key, value] of Object.entries(body)) { + // `extended: false` yields string | string[]; appending each element keeps + // repeated keys repeated rather than collapsing them into "a,b". + if (Array.isArray(value)) { + for (const item of value) { + params.append(key, String(item)); + } + } else if (value !== undefined) { + params.append(key, String(value)); + } + } + return params.toString(); +}; + +/** + * Decide how to re-inject a proxied request body. + * + * `bodyWasParsed` must come from body-parser's own `req._body` flag rather than + * from the shape of `req.body`: every parser runs `req.body = req.body || {}` + * *before* its content-type check, so an untouched `multipart/form-data` + * request is indistinguishable from a parsed empty object by shape alone. + * `_body` is the only signal that the stream was actually drained, and + * body-parser uses it for the same purpose. + */ +export const planProxyBody = ( + contentType: string | undefined, + body: unknown, + bodyWasParsed: boolean, +): ProxyBodyPlan => { + // Nothing consumed the stream — let it pipe through untouched. + if (!bodyWasParsed || body == null) { + return { action: 'skip' }; + } + + // express.text() already hands back exactly what was on the wire. + if (typeof body === 'string' || Buffer.isBuffer(body)) { + return { action: 'write', payload: body }; + } + + // A JSON scalar (number, boolean) — nothing to re-encode as a form. + if (typeof body !== 'object') { + return { action: 'write', payload: JSON.stringify(body) }; + } + + if (mediaTypeOf(contentType) === URLENCODED_MEDIA_TYPE) { + return { action: 'write', payload: serializeUrlencoded(body) }; + } + + // Only the json and urlencoded parsers produce object bodies, so anything + // left is JSON. Matching on the media type rather than the raw header is + // what lets `application/json; charset=utf-8` through — express.json() + // ignores content-type parameters, and so must we. + return { action: 'write', payload: JSON.stringify(body) }; +}; + router.post( '/test', validateRequest({ @@ -261,20 +345,36 @@ 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); - } + const plan = planProxyBody( + _req.headers['content-type'], + _req.body, + _req._body === true, + ); + + if (plan.action === 'skip') { + return; } try { - proxyReq.write(body); - } catch { - console.error( - `clickhouseProxy error writing body, body is type ${typeof body}`, + // Re-serialization can change the byte count (whitespace + // normalization, urlencoded round-trips) and the upstream reads + // exactly Content-Length bytes, so the header has to follow the + // payload we are about to write. + if (!proxyReq.getHeader('transfer-encoding')) { + proxyReq.setHeader( + 'content-length', + Buffer.byteLength(plan.payload), + ); + } + proxyReq.write(plan.payload); + } catch (e) { + // Continuing body-less surfaces to the user as an opaque 400 from + // ClickHouse; fail the hop so the proxy's error handler reports it. + console.error('clickhouseProxy error writing body', e); + proxyReq.destroy( + e instanceof Error + ? e + : new Error('Failed to forward request body to ClickHouse'), ); } },