fix(api): forward parsed ClickHouse proxy bodies

.changeset/proxy-parsed-bodies.md

@@ -0,0 +1,7 @@
+---
+'@hyperdx/api': patch
+---
+
+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.

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) {

packages/api/src/routers/api/__tests__/clickhouseProxy.int.test.ts

@@ -1,8 +1,10 @@
 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';
+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).
@@ -16,14 +18,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: 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
  * 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 +65,7 @@ describe('clickhouse-proxy forwarding', () => {
   });
 
   afterAll(async () => {
+    upstream.closeAllConnections();
     await new Promise<void>((resolve, reject) =>
       upstream.close(err => (err ? reject(err) : resolve())),
     );
@@ -101,6 +101,217 @@ describe('clickhouse-proxy forwarding', () => {
     expect(captured[0].headers['x-clickhouse-user']).toBe('default');
   });
 
+  it.each(['application/json', 'application/json; charset=utf-8'])(
+    'preserves %s bytes and Content-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);
+      expect(Buffer.byteLength(original)).toBeGreaterThan(original.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(original));
+      expect(captured[0].headers['content-type']).toBe(contentType);
+      expect(captured[0].headers['content-length']).toBe(
+        String(Buffer.byteLength(original)),
+      );
+      expect(captured[0].headers['transfer-encoding']).toBeUndefined();
+    },
+  );
+
+  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 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(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();
+  });
+
+  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 serialized = Buffer.from(JSON.stringify(body, null, 2));
+    const original = gzipSync(serialized);
+
+    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',
+  ])('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&param=first&param=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 =
+        '{"id":9007199254740993,"value":1.0,"value":1e2,"query":"SELECT 1"}';
+
+      await agent
+        .post(`${route}?query_id=test-json-fidelity`)
+        .set('x-hyperdx-connection-id', connectionId)
+        .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).toEqual(Buffer.from(original));
+    },
+  );
+
+  it('preserves an empty JSON body for query-string SQL', async () => {
+    const { agent, team } = await getLoggedInAgent(server);
+    const connectionId = await createConnection(team._id.toString());
+
+    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-duplicate-charset')
+      .set('x-hyperdx-connection-id', connectionId)
+      .set('Content-Type', contentType)
+      .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(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 () => {
     const { agent, team } = await getLoggedInAgent(server);
     const connectionId = await createConnection(team._id.toString());
@@ -120,25 +331,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 loggerError = jest.spyOn(logger, '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(loggerError).not.toHaveBeenCalled();
+    } finally {
+      loggerError.mockRestore();
+    }
   });
 
   // The browser client ships `log_comment` as a URL query param, and

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;