diff --git a/packages/ui/src/lib/relay/protocol.ts b/packages/ui/src/lib/relay/protocol.ts index 6e765fca..d1569175 100644 --- a/packages/ui/src/lib/relay/protocol.ts +++ b/packages/ui/src/lib/relay/protocol.ts @@ -61,6 +61,11 @@ export interface TunnelHttpRequestPayload { path: string; query: string; headers: Record; + /** True when the client had a request body to send. The host uses this to + * distinguish a genuine bodyless request from one whose body frames were lost + * through the tunnel (which it must abort as an ambiguous transport failure + * instead of forwarding an empty body the loopback server rejects with 400). */ + hasBody?: boolean; } export interface TunnelHttpResponsePayload { diff --git a/packages/ui/src/lib/relay/tunnel-client.ts b/packages/ui/src/lib/relay/tunnel-client.ts index fd12b3cb..ac8bd9b0 100644 --- a/packages/ui/src/lib/relay/tunnel-client.ts +++ b/packages/ui/src/lib/relay/tunnel-client.ts @@ -819,6 +819,7 @@ export const createRelayTunnelClient = (options: RelayTunnelClientOptions): Rela path: request.path, query: request.query, headers: request.headers, + hasBody: request.body !== null, }; channel.send(encodeTunnelFrame(TunnelFrameType.HttpRequest, streamId, encodeJsonPayload(head))); void (async () => { diff --git a/packages/web/server/lib/relay/DOCUMENTATION.md b/packages/web/server/lib/relay/DOCUMENTATION.md index 943000d8..36fa0d01 100644 --- a/packages/web/server/lib/relay/DOCUMENTATION.md +++ b/packages/web/server/lib/relay/DOCUMENTATION.md @@ -45,6 +45,8 @@ Everything a client normally sends to the single OpenChamber origin: The host dispatcher restricts tunneled traffic to explicit path allowlists (one for HTTP, one for WS). +Request bodies crossing the tunnel are buffered on the host and forwarded to loopback only once the client's `StreamEnd` frame arrives (bodies above ~512 KB stream live instead). A body whose frames were lost in transit therefore never reaches the loopback server as an empty/truncated chunked body — the host aborts the stream and the client sees an ambiguous transport failure it can retry, instead of the loopback server's bare `400` ("Failed to send message (400)" from the mobile app). + ## Authentication model - The tunnel is **transport only**. The OpenChamber server still authenticates every tunneled request exactly as it authenticates a direct remote client. The relay path grants reachability, not authorization. diff --git a/packages/web/server/lib/relay/tunnel-host.js b/packages/web/server/lib/relay/tunnel-host.js index 1decdf0d..a3ec6089 100644 --- a/packages/web/server/lib/relay/tunnel-host.js +++ b/packages/web/server/lib/relay/tunnel-host.js @@ -62,6 +62,17 @@ const STRIPPED_RESPONSE_HEADERS = new Set([ const BACKPRESSURE_LIMIT_BYTES = 4 * 1024 * 1024; const BACKPRESSURE_POLL_MS = 20; +// Bodies smaller than this are fully buffered before the loopback request is +// sent, so a tunneled body that lost frames (relay reconnect, dropped HttpBody +// frames) can never reach the loopback server as an empty/truncated chunked +// body — the server rejects those with a bare 400, surfacing as the mobile +// app's "Failed to send message (400)". Larger bodies stream live as before. +const BODY_BUFFER_MAX_BYTES = 512 * 1024; +// While the body is still being buffered, abort the stream if it never +// completes, so a stalled tunnel converts into an ambiguous transport failure +// (which the client already retries) instead of a hung loopback request. +const BODY_DELIVERY_TIMEOUT_MS = 15_000; + const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); const isHttpRequestPayload = (parsed) => @@ -168,39 +179,14 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf await send(encodeTunnelFrame(TunnelFrameType.StreamEnd, streamId, new Uint8Array(0))); }; - const runHttpStream = async (streamId, request) => { - const method = request.method.toUpperCase(); - if (!isAllowedHttpPath(request.path)) { - dropStream(streamId); - await syntheticResponse(streamId, 403, 'Path is not allowed through the relay'); - return; - } - - const stream = streams.get(streamId); - if (!stream || stream.kind !== 'http') return; - - const hasBody = method !== 'GET' && method !== 'HEAD'; - let requestBody; - if (hasBody) { - requestBody = new ReadableStream({ - start(controller) { - stream.body = controller; - }, - }); - } else { - stream.body = null; - stream.noBody = true; - } - - const loopbackOrigin = `http://127.0.0.1:${getLocalPort()}`; - const url = `${loopbackOrigin}${request.path}${request.query ? `?${request.query}` : ''}`; + const forwardRequest = async (streamId, stream, url, method, request, body, loopbackOrigin) => { let response; try { response = await fetch(url, { method, headers: buildRequestHeaders(request.headers, loopbackOrigin), - body: requestBody, - duplex: hasBody ? 'half' : undefined, + body, + duplex: body ? 'half' : undefined, signal: stream.abort.signal, }); } catch (error) { @@ -242,6 +228,139 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf } }; + const runHttpStream = async (streamId, request) => { + const method = request.method.toUpperCase(); + if (!isAllowedHttpPath(request.path)) { + dropStream(streamId); + await syntheticResponse(streamId, 403, 'Path is not allowed through the relay'); + return; + } + + const stream = streams.get(streamId); + if (!stream || stream.kind !== 'http') return; + + const hasBody = method !== 'GET' && method !== 'HEAD'; + const loopbackOrigin = `http://127.0.0.1:${getLocalPort()}`; + const url = `${loopbackOrigin}${request.path}${request.query ? `?${request.query}` : ''}`; + + if (!hasBody) { + stream.noBody = true; + await forwardRequest(streamId, stream, url, method, request, null, loopbackOrigin); + return; + } + + // Body-carrying request. Buffer the tunneled body frames and forward the + // COMPLETE body only once StreamEnd arrives. Forwarding a body that lost + // frames through the tunnel (relay reconnect, dropped HttpBody frames) + // reaches the loopback server as an empty/truncated chunked body, which it + // rejects with a bare 400 (empty response body) — the "Failed to send + // message (400)" seen from the mobile APK. Bodies above BODY_BUFFER_MAX_BYTES + // stream live so large uploads are not fully buffered. + const buffered = []; + let bufferedBytes = 0; + let bodyFrameCount = 0; + let liveStream = null; + let liveController = null; + let completed = false; + let bodyFailure = null; + let resolveBodyEnd; + const bodyEnded = new Promise((resolve) => { resolveBodyEnd = resolve; }); + + const finishBody = (error) => { + if (completed) return; + completed = true; + bodyFailure = error ?? null; + if (liveController) { + try { + if (error) liveController.error(error); + else liveController.close(); + } catch { + // stream already errored/closed + } + } + resolveBodyEnd(); + }; + + let deliveryDeadline = null; + const switchToLive = () => { + liveStream = new ReadableStream({ + start(controller) { + liveController = controller; + stream.body = controller; + }, + }); + for (const chunk of buffered) { + try { liveController.enqueue(chunk); } catch { break; } + } + buffered.length = 0; + // The loopback request is now streaming live; runHttpStream has nothing + // left to do — clear the deadline and let the async body forwarding own + // this stream from here (abort/StreamEnd close the controller). + if (deliveryDeadline) clearTimeout(deliveryDeadline); + resolveBodyEnd(); + void forwardRequest(streamId, stream, url, method, request, liveStream, loopbackOrigin); + }; + + stream.body = { + enqueue(payload) { + if (completed) return; + bodyFrameCount += 1; + if (liveController) { + try { liveController.enqueue(payload); } catch { + // stream already errored/closed + } + return; + } + buffered.push(payload); + bufferedBytes += payload.length; + if (bufferedBytes > BODY_BUFFER_MAX_BYTES) { + switchToLive(); + } + }, + close() { + finishBody(null); + }, + error(error) { + finishBody(error); + }, + }; + + deliveryDeadline = setTimeout(() => { + if (streams.get(streamId) === stream && !completed && !liveStream) { + dropStream(streamId); + void sendAbort(streamId, 'tunnel request body was not delivered in time'); + } + }, BODY_DELIVERY_TIMEOUT_MS); + deliveryDeadline.unref?.(); + + await bodyEnded; + if (deliveryDeadline) clearTimeout(deliveryDeadline); + if (streams.get(streamId) !== stream) return; // aborted or dropped meanwhile + if (bodyFailure) { + dropStream(streamId); + await sendAbort(streamId, bodyFailure.message ?? 'tunnel request body failed'); + return; + } + if (liveStream) return; // already forwarded via the streaming path + + // The client signaled it had a body but no HttpBody frame arrived before + // StreamEnd — the body frames were lost through the tunnel. Forwarding an + // empty body would make the loopback server reject the request with a bare + // 400. Abort instead so the client treats it as an ambiguous transport + // failure (dispatched, outcome unknown) and can safely retry. + if (request.hasBody === true && bodyFrameCount === 0) { + dropStream(streamId); + await sendAbort(streamId, 'tunnel request body frames were lost'); + return; + } + + // Buffered path: forward the complete body as a single buffer so Bun + // frames it with content-length — never as a chunked body that could be + // truncated. Reset the buffered handler so late frames cannot enqueue. + stream.body = null; + await forwardRequest(streamId, stream, url, method, request, Buffer.concat(buffered), loopbackOrigin); + }; + const handleHttpRequest = (streamId, payload) => { if (streams.has(streamId)) { abortLocalStream(streamId, 'duplicate stream id'); @@ -263,9 +382,9 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf const handleHttpBody = (streamId, payload) => { const stream = streams.get(streamId); if (!stream || stream.kind !== 'http' || stream.noBody) return; - // The body controller attaches synchronously in runHttpStream before any - // await, so by the time body frames arrive it is set for body-carrying - // methods; drop stray body bytes otherwise. + // runHttpStream installs a body sink (buffering handler, or the live stream + // controller once the buffer cap is crossed) before any HttpBody frame can + // arrive; drop stray bytes for request bodies already completed/aborted. try { stream.body?.enqueue(payload); } catch { diff --git a/packages/web/server/lib/relay/tunnel-host.test.js b/packages/web/server/lib/relay/tunnel-host.test.js new file mode 100644 index 00000000..cda4da33 --- /dev/null +++ b/packages/web/server/lib/relay/tunnel-host.test.js @@ -0,0 +1,116 @@ +import { describe, test, expect } from 'bun:test'; +import http from 'node:http'; + +import { createTunnelHost } from './tunnel-host.js'; +import { decodeTunnelFrame, encodeTunnelFrame, encodeJsonPayload, TunnelFrameType } from './tunnel-codec.js'; + +const startLoopback = () => + new Promise((resolve) => { + const requests = []; + const server = http.createServer((req, res) => { + const chunks = []; + req.on('data', (c) => chunks.push(c)); + req.on('end', () => { + requests.push({ + method: req.method, + url: req.url, + body: Buffer.concat(chunks).toString('utf8'), + }); + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: true })); + }); + }); + server.listen(0, '127.0.0.1', () => resolve({ + server, + port: server.address().port, + requests, + stop: () => new Promise((r) => server.close(() => r())), + })); + }); + +const createHarness = async () => { + const loopback = await startLoopback(); + const sentFrames = []; + const host = createTunnelHost({ + connectionId: 'conn-test', + getLocalPort: () => loopback.port, + sendFrame: async (frame) => { + sentFrames.push(decodeTunnelFrame(frame)); + }, + getBufferedAmount: () => 0, + }); + return { host, loopback, sentFrames }; +}; + +const httpHead = (overrides = {}) => encodeTunnelFrame(TunnelFrameType.HttpRequest, 1, encodeJsonPayload({ + method: 'POST', + path: '/api/submit', + query: '', + headers: { 'content-type': 'application/json' }, + ...overrides, +})); + +const waitFor = async (predicate, timeoutMs = 2000) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return true; + await new Promise((r) => setTimeout(r, 20)); + } + return predicate(); +}; + +describe('tunnel-host HTTP body forwarding', () => { + test('buffers tunneled body frames and forwards the complete body', async () => { + const { host, loopback, sentFrames } = await createHarness(); + await host.handleFrame(httpHead()); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpBody, 1, new TextEncoder().encode('alpha'))); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpBody, 1, new TextEncoder().encode('beta'))); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0))); + + const received = await waitFor(() => loopback.requests.length === 1); + expect(received).toBe(true); + expect(loopback.requests[0].method).toBe('POST'); + expect(loopback.requests[0].body).toBe('alphabeta'); + await waitFor(() => sentFrames.some((f) => f.frameType === TunnelFrameType.StreamEnd)); + await loopback.stop(); + }); + + test('body-expected request with zero delivered frames is aborted as ambiguous, not forwarded', async () => { + const { host, loopback, sentFrames } = await createHarness(); + await host.handleFrame(httpHead({ hasBody: true })); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0))); + + const aborted = await waitFor(() => sentFrames.some((f) => f.frameType === TunnelFrameType.StreamAbort)); + expect(aborted).toBe(true); + // Loopback must never have seen a request with a lost body. + expect(loopback.requests.length).toBe(0); + await loopback.stop(); + }); + + test('bodyless request (hasBody absent, legacy client) still forwards empty', async () => { + const { host, loopback } = await createHarness(); + await host.handleFrame(httpHead()); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0))); + + const received = await waitFor(() => loopback.requests.length === 1); + expect(received).toBe(true); + expect(loopback.requests[0].body).toBe(''); + await loopback.stop(); + }); + + test('GET forwards immediately with no body wait', async () => { + const { host, loopback } = await createHarness(); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpRequest, 1, encodeJsonPayload({ + method: 'GET', + path: '/api/health', + query: '', + headers: {}, + }))); + await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0))); + + const received = await waitFor(() => loopback.requests.length === 1); + expect(received).toBe(true); + expect(loopback.requests[0].method).toBe('GET'); + await loopback.stop(); + }); +});