TLS->HTTP Proxy: Support outbound request body (#61)

Adds support for sending outbound chunked-encoding POST requests from
PHP. Before this PR, they simply failed.

## Implementation

* Without this PR, `fetchWithCorsProxy()` exhausts the body stream on
the first, non-proxied request attempt, which means it the second,
proxied attempt can't access the body bytes. This PR tee the request
before the first attempt to make sure we still have an unread stream for
the second attempt.
* Without this PR, `TCPOverFetchWebsocket` ignores chunked-encoded
request body. This PR detects chunked encoding and processes the body
stream.
* Without this PR, `TCPOverFetchWebsocket` accumulates chunked encoding
lengths trailers and passes them to `fetch()`, corrupting the request
body. This PR decodes the chunked-encoded stream and passes plain body
bytes over to `fetch()`.
* Without this PR, `createMemoizedFetch()` locked the response stream
when called twice without awaiting, which caused seemingly random
failures. This PR ensures `response.body.tee()` is called before the
caller has a chance to call `createMemoizedFetch()` again after an event
loop tick.

## Testing instructions

Additional unit test is included in this PR.

Author: adamziel

packages/php-wasm/web-service-worker/src/utils.ts

@@ -175,6 +175,26 @@ export async function cloneRequest(
 	});
 }
 
+/**
+ * Tee a request to ensure the body stream is not consumed
+ * when executing or cloning the request.
+ *
+ * @param request
+ * @returns
+ */
+export async function teeRequest(
+	request: Request
+): Promise<[Request, Request]> {
+	if (!request.body) {
+		return [request, request];
+	}
+	const [body1, body2] = request.body.tee();
+	return [
+		await cloneRequest(request, { body: body1, duplex: 'half' }),
+		await cloneRequest(request, { body: body2, duplex: 'half' }),
+	];
+}
+
 /**
  * Extracts headers from a Request as a plain key->value JS object.
  *

packages/php-wasm/web/src/lib/chunked-decoder.ts

@@ -0,0 +1,106 @@
+import { concatUint8Arrays } from './tls/utils';
+
+/**
+ * A TransformStream that decodes HTTP chunked transfer encoding.
+ * Each chunk starts with the chunk size in hex followed by CRLF,
+ * then the chunk data, then CRLF. A chunk size of 0 indicates the end.
+ */
+export class ChunkedDecoderStream extends TransformStream<
+	Uint8Array,
+	Uint8Array
+> {
+	constructor() {
+		let buffer = new Uint8Array(0);
+		let state:
+			| 'SCAN_CHUNK_SIZE'
+			| 'SCAN_CHUNK_DATA'
+			| 'SCAN_CHUNK_TRAILER'
+			| 'SCAN_FINAL_CHUNK' = 'SCAN_CHUNK_SIZE';
+		let chunkRemainingBytes = 0;
+
+		super({
+			transform(chunk, controller) {
+				// Add new chunk to buffer
+				buffer = concatUint8Arrays([buffer, chunk]);
+
+				while (buffer.length > 0) {
+					if (state === 'SCAN_CHUNK_SIZE') {
+						// Need at least "0\r\n" (3 bytes)
+						if (buffer.length < 3) {
+							return;
+						}
+
+						// Find the chunk size hex digits
+						let chunkBytesNb = 0;
+						while (chunkBytesNb < buffer.length) {
+							const byte = buffer[chunkBytesNb];
+							const isHexDigit =
+								(byte >= 48 && byte <= 57) || // 0-9
+								(byte >= 97 && byte <= 102) || // a-f
+								(byte >= 65 && byte <= 70); // A-F
+							if (!isHexDigit) break;
+							chunkBytesNb++;
+						}
+
+						if (chunkBytesNb === 0) {
+							throw new Error('Invalid chunk size format');
+						}
+
+						// Look for CRLF after chunk size
+						if (
+							buffer.length < chunkBytesNb + 2 ||
+							buffer[chunkBytesNb] !== 13 || // \r
+							buffer[chunkBytesNb + 1] !== 10 // \n
+						) {
+							return;
+						}
+
+						// Parse the chunk size
+						const chunkSizeHex = new TextDecoder().decode(
+							buffer.slice(0, chunkBytesNb)
+						);
+						const chunkSize = parseInt(chunkSizeHex, 16);
+
+						// Remove chunk header from buffer
+						buffer = buffer.slice(chunkBytesNb + 2);
+
+						if (chunkSize === 0) {
+							state = 'SCAN_FINAL_CHUNK';
+							controller.terminate();
+							return;
+						}
+
+						chunkRemainingBytes = chunkSize;
+						state = 'SCAN_CHUNK_DATA';
+					} else if (state === 'SCAN_CHUNK_DATA') {
+						const bytesToRead = Math.min(
+							chunkRemainingBytes,
+							buffer.length
+						);
+						const data = buffer.slice(0, bytesToRead);
+						buffer = buffer.slice(bytesToRead);
+						chunkRemainingBytes -= bytesToRead;
+
+						controller.enqueue(data);
+
+						if (chunkRemainingBytes === 0) {
+							state = 'SCAN_CHUNK_TRAILER';
+						}
+					} else if (state === 'SCAN_CHUNK_TRAILER') {
+						if (buffer.length < 2) {
+							return;
+						}
+
+						if (buffer[0] !== 13 || buffer[1] !== 10) {
+							// \r\n
+							throw new Error('Expected CRLF after chunk data');
+						}
+
+						buffer = buffer.slice(2);
+						state = 'SCAN_CHUNK_SIZE';
+					}
+				}
+			},
+		});
+	}
+}

packages/php-wasm/web/src/lib/fetch-with-cors-proxy.ts

@@ -1,29 +1,27 @@
-import { cloneRequest } from '@php-wasm/web-service-worker';
+import { cloneRequest, teeRequest } from '@php-wasm/web-service-worker';
 
 export async function fetchWithCorsProxy(
 	input: RequestInfo,
 	init?: RequestInit,
 	corsProxyUrl?: string
 ): Promise<Response> {
-	const directFetch = fetch(input, init);
+	const requestObject =
+		typeof input === 'string' ? new Request(input, init) : input;
 	if (!corsProxyUrl) {
-		return directFetch;
+		return await fetch(requestObject);
 	}
 
+	// Tee the request to avoid consuming the request body stream on the initial
+	// fetch() so that we can retry through the cors proxy.
+	const [request1, request2] = await teeRequest(requestObject);
+
 	try {
-		return await directFetch;
+		return await fetch(request1);
 	} catch {
-		let newInput: string | Request;
-		if (typeof input === 'string' || input instanceof URL) {
-			newInput = `${corsProxyUrl}${input}`;
-		} else if (input instanceof Request) {
-			newInput = await cloneRequest(input, {
-				url: `${corsProxyUrl}${input.url}`,
-			});
-		} else {
-			throw new Error('Invalid input type for fetch');
-		}
+		const newRequest = await cloneRequest(request2, {
+			url: `${corsProxyUrl}${requestObject.url}`,
+		});
 
-		return fetch(newInput, init);
+		return await fetch(newRequest, init);
 	}
 }

packages/php-wasm/web/src/lib/tcp-over-fetch-websocket.spec.ts

@@ -1,4 +1,7 @@
-import { TCPOverFetchWebsocket } from './tcp-over-fetch-websocket';
+import {
+	TCPOverFetchWebsocket,
+	RawBytesFetch,
+} from './tcp-over-fetch-websocket';
 import express from 'express';
 import http from 'http';
 import { AddressInfo } from 'net';
@@ -344,6 +347,29 @@ describe('TCPOverFetchWebsocket', () => {
 	});
 });
 
+describe('RawBytesFetch', () => {
+	it('parseHttpRequest should handle an transfer-encoding: chunked POST requests', async () => {
+		const encodedBodyBytes = 'abcde';
+		const encodedChunkedBodyBytes = `${encodedBodyBytes.length}\r\n${encodedBodyBytes}\r\n0\r\n\r\n`;
+		const requestBytes = `POST /echo HTTP/1.1\r\nHost: playground.internal\r\ntransfer-encoding: chunked\r\n\r\n${encodedChunkedBodyBytes}`;
+		const request = await RawBytesFetch.parseHttpRequest(
+			new ReadableStream({
+				start(controller) {
+					controller.enqueue(new TextEncoder().encode(requestBytes));
+					controller.close();
+				},
+			}),
+			'playground.internal',
+			'http'
+		);
+		const parsedRequestBody = await request.body?.getReader().read();
+		const decodedRequestBody = new TextDecoder().decode(
+			parsedRequestBody?.value
+		);
+		expect(decodedRequestBody).toEqual(encodedBodyBytes);
+	});
+});
+
 type MakeRequestOptions = {
 	host: string;
 	port: number;

packages/php-wasm/web/src/lib/tcp-over-fetch-websocket.ts

@@ -43,6 +43,7 @@ import { generateCertificate, GeneratedCertificate } from './tls/certificates';
 import { concatUint8Arrays } from './tls/utils';
 import { ContentTypes } from './tls/1_2/types';
 import { fetchWithCorsProxy } from './fetch-with-cors-proxy';
+import { ChunkedDecoderStream } from './chunked-decoder';
 
 export type TCPOverFetchOptions = {
 	CAroot: GeneratedCertificate;
@@ -417,7 +418,7 @@ function guessProtocol(port: number, data: Uint8Array) {
 	return 'other';
 }
 
-class RawBytesFetch {
+export class RawBytesFetch {
 	/**
 	 * Streams a HTTP response including the status line and headers.
 	 */
@@ -578,13 +579,24 @@ class RawBytesFetch {
 
 		const headersBuffer = inputBuffer.slice(0, headersEndIndex);
 		const parsedHeaders = RawBytesFetch.parseRequestHeaders(headersBuffer);
+		const terminationMode =
+			parsedHeaders.headers.get('Transfer-Encoding') !== null
+				? 'chunked'
+				: 'content-length';
+		const contentLength =
+			parsedHeaders.headers.get('Content-Length') !== null
+				? parseInt(parsedHeaders.headers.get('Content-Length')!, 10)
+				: undefined;
 
 		const bodyBytes = inputBuffer.slice(
 			headersEndIndex + 4 /* Skip \r\n\r\n */
 		);
 		let outboundBodyStream: ReadableStream<Uint8Array> | undefined;
 		if (parsedHeaders.method !== 'GET') {
 			const requestBytesReader = requestBytesStream.getReader();
+			let seenBytes = bodyBytes.length;
+			let last5Bytes = bodyBytes.slice(-6);
+			const emptyChunk = new TextEncoder().encode('0\r\n\r\n');
 			outboundBodyStream = new ReadableStream<Uint8Array>({
 				async start(controller) {
 					if (bodyBytes.length > 0) {
@@ -596,16 +608,47 @@ class RawBytesFetch {
 				},
 				async pull(controller) {
 					const { done, value } = await requestBytesReader.read();
-
+					seenBytes += value?.length || 0;
 					if (value) {
 						controller.enqueue(value);
+						last5Bytes = concatUint8Arrays([
+							last5Bytes,
+							value || new Uint8Array(),
+						]).slice(-5);
 					}
-					if (done) {
+					const shouldTerminate =
+						done ||
+						(terminationMode === 'content-length' &&
+							contentLength !== undefined &&
+							seenBytes >= contentLength) ||
+						(terminationMode === 'chunked' &&
+							last5Bytes.every(
+								(byte, index) => byte === emptyChunk[index]
+							));
+					if (shouldTerminate) {
 						controller.close();
 						return;
 					}
 				},
 			});
+
+			if (terminationMode === 'chunked') {
+				// Strip chunked transfer encoding from the request body stream.
+				// PHP may encode the request body with chunked transfer encoding,
+				// giving us a stream of chunks with a size line ending in \r\n,
+				// a body chunk, and a chunk trailer ending in \r\n.
+				//
+				// We must not include the chunk headers and trailers in the
+				// transmitted data. fetch() trusts us to provide the body stream
+				// in its original form and will pass treat the chunked encoding
+				// artifacts as a part of the data to be transmitted to the server.
+				// This, in turn, means sending over a corrupted request body.
+				//
+				// Therefore, let's just strip any chunked encoding-related bytes.
+				outboundBodyStream = outboundBodyStream.pipeThrough(
+					new ChunkedDecoderStream()
+				);
+			}
 		}
 
 		/**

packages/playground/common/src/create-memoized-fetch.ts

@@ -1,6 +1,7 @@
-export interface CachedFetchResponse {
-	body: ReadableStream<Uint8Array>;
-	responseInit: ResponseInit;
+export interface CacheEntry {
+	responsePromise: Promise<Response>;
+	unlockedBodyStream?: ReadableStream<Uint8Array>;
+	nextResponse: () => Promise<Response>;
 }
 
 /**
@@ -12,32 +13,29 @@ export interface CachedFetchResponse {
  * @param originalFetch The fetch function to memoize. Defaults to the global fetch.
  */
 export function createMemoizedFetch(originalFetch = fetch) {
-	const cache: Record<
-		string,
-		Promise<CachedFetchResponse> | CachedFetchResponse
-	> = {};
+	const fetches: Record<string, CacheEntry> = {};
 
 	return async function memoizedFetch(url: string, options?: RequestInit) {
-		if (!cache[url]) {
-			// Write to cache synchronously to avoid duplicate requests.
-			cache[url] = originalFetch(url, options).then((response) => ({
-				body: response.body!,
-				responseInit: {
-					status: response.status,
-					statusText: response.statusText,
-					headers: response.headers,
+		if (!fetches[url]) {
+			fetches[url] = {
+				responsePromise: originalFetch(url, options),
+				async nextResponse() {
+					// Wait for "result" to be set.
+					const response = await fetches[url].responsePromise;
+					const [left, right] =
+						fetches[url].unlockedBodyStream!.tee();
+					fetches[url].unlockedBodyStream = left;
+					return new Response(right, {
+						status: response.status,
+						statusText: response.statusText,
+						headers: response.headers,
+					});
 				},
-			}));
+			};
+			const response = await fetches[url].responsePromise;
+			fetches[url].unlockedBodyStream = response.body!;
 		}
-		const { body, responseInit } = await cache[url];
-		// Split the response stream so that the cached one is not consumed.
-		const [left, right] = body.tee();
-		// Cache the "left" stream and don't use it until the next .tee().
-		cache[url] = {
-			body: left,
-			responseInit,
-		};
-		// Return the "right" stream for consumption.
-		return new Response(right, responseInit);
+
+		return fetches[url].nextResponse();
 	};
 }

packages/playground/common/src/test/create-memoized-fetch.spec.ts

@@ -23,4 +23,25 @@ describe('createMemoizedFetch', () => {
 		expect(await response1.text()).toBe(await response2.text());
 		expect(fetch).toHaveBeenCalledTimes(1);
 	});
+
+	it('should correctly handle teeing the response stream when called twice without awaiting', async () => {
+		const fetch = vitest.fn().mockResolvedValueOnce(
+			new Response('hello', {
+				status: 200,
+				statusText: 'OK',
+				headers: { 'Content-type': 'text/plain' },
+			})
+		);
+		const memoizedFetch = createMemoizedFetch(fetch);
+		const response1Promise = memoizedFetch('https://example.com');
+		const response2Promise = memoizedFetch('https://example.com');
+		const response1 = await response1Promise;
+		const response2 = await response2Promise;
+		expect(response1.status).toBe(response2.status);
+		expect(response1.headers.get('Content-type')).toBe(
+			response2.headers.get('Content-type')
+		);
+		expect(await response1.text()).toBe(await response2.text());
+		expect(fetch).toHaveBeenCalledTimes(1);
+	});
 });