fix(http): limit buffered request bodies

docs/3.http-handling.md

@@ -153,7 +153,7 @@ The `RequestContext` interface provides access to all request-related informatio
 
 ## Read request bodies
 
-The `ReadBody` function reads the raw request body as a `Buffer`.
+The `ReadBody` function reads the raw request body as a `Buffer`. Buffered request bodies are limited to 1 MiB by default. Requests with a larger `Content-Length`, or whose received data exceeds the limit, receive HTTP 413 Payload Too Large.
 
 ```typescript
 import { Controller, Post, RawBody } from "@antelopejs/interface-api";
@@ -179,3 +179,18 @@ class UserController extends Controller("/users") {
   }
 }
 ```
+
+Pass a byte limit to `@RawBody`, `@JSONBody`, or `ReadBody` when an endpoint intentionally accepts a larger buffered body:
+
+```typescript
+const TEN_MIBIBYTES = 10 * 1024 * 1024;
+
+class UploadController extends Controller("/uploads") {
+  @Post()
+  async upload(@RawBody(TEN_MIBIBYTES) body: Buffer) {
+    return { bytes: body.length };
+  }
+}
+```
+
+For bodies that should not be buffered in memory, use `@Context()` and consume `context.rawRequest` as a stream instead.

src/index.ts

@@ -50,6 +50,22 @@ export type ControllerClass<T = Record<string, any>> = Class<T> & {
 };
 
 const SERVER_ERROR_BODY_LOG_LIMIT = 2048;
+const PAYLOAD_TOO_LARGE_STATUS = 413;
+const PAYLOAD_TOO_LARGE_MESSAGE = "Payload Too Large";
+
+/**
+ * Default maximum buffered request body size in bytes.
+ */
+export const DEFAULT_REQUEST_BODY_LIMIT = 1024 * 1024;
+
+interface RequestBodyState {
+  buffers: Buffer[];
+  didExceed: boolean;
+  length: number;
+  limit: number;
+}
+
+const requestBodyStates = new WeakMap<RequestContext, RequestBodyState>();
 
 /**
  * Result object of an API call.
@@ -1113,30 +1129,97 @@ export const WebsocketHandler = MakeMethodDecorator(
  * Get the body from a RequestContext object.
  *
  * @param context Request context
+ * @param limit Maximum body size in bytes
  * @returns Body buffer
  */
-export function ReadBody(context: RequestContext): Promise<Buffer> {
-  if (context.body === undefined) {
-    context.body = new Promise((resolve, reject) => {
-      const buffers: Buffer[] = [];
-      context.rawRequest.on("readable", () => {
-        while (true) {
-          const chunk = context.rawRequest.read() as Buffer | null;
-          if (!chunk) {
-            break;
-          }
-          buffers.push(chunk);
-        }
-      });
+export function ReadBody(
+  context: RequestContext,
+  limit = DEFAULT_REQUEST_BODY_LIMIT,
+): Promise<Buffer> {
+  let state = requestBodyStates.get(context);
+  let shouldRestart = false;
+  if (state) {
+    const previousLimit = state.limit;
+    state.limit = Math.max(state.limit, limit);
+    shouldRestart =
+      state.didExceed &&
+      state.limit > previousLimit &&
+      state.length <= state.limit;
+  } else {
+    state = { buffers: [], didExceed: false, length: 0, limit };
+    requestBodyStates.set(context, state);
+  }
+  if (context.body === undefined || shouldRestart) {
+    state.didExceed = false;
+    context.body = Promise.resolve().then(() =>
+      readRequestBody(context.rawRequest, state),
+    );
+  }
+  return (context.body as Promise<Buffer>).then((body) =>
+    enforceBodyLimit(body, limit),
+  );
+}
 
-      context.rawRequest.on("end", () => {
-        resolve(Buffer.concat(buffers));
-      });
+function readRequestBody(
+  request: IncomingMessage,
+  state: RequestBodyState,
+): Promise<Buffer> {
+  const contentLength = Number(request.headers["content-length"]);
+  if (contentLength > state.limit) {
+    state.didExceed = true;
+    request.pause();
+    return Promise.reject(createPayloadTooLargeResult());
+  }
 
-      context.rawRequest.on("error", reject);
-    });
+  return new Promise((resolve, reject) => {
+    const cleanup = () => {
+      request.off("data", onData);
+      request.off("end", onEnd);
+      request.off("error", onError);
+    };
+    const onData = (chunk: Buffer | string) => {
+      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+      state.length += buffer.length;
+      state.buffers.push(buffer);
+      if (state.length <= state.limit) {
+        return;
+      }
+      state.didExceed = true;
+      cleanup();
+      request.pause();
+      reject(createPayloadTooLargeResult());
+    };
+    const onEnd = () => {
+      cleanup();
+      resolve(joinRequestBody(state));
+    };
+    const onError = (error: Error) => {
+      cleanup();
+      reject(error);
+    };
+
+    request.on("data", onData);
+    request.once("end", onEnd);
+    request.once("error", onError);
+    request.resume();
+  });
+}
+
+function joinRequestBody(state: RequestBodyState): Buffer {
+  const body = Buffer.concat(state.buffers, state.length);
+  state.buffers = [];
+  return body;
+}
+
+function enforceBodyLimit(body: Buffer, limit: number): Buffer {
+  if (body.length > limit) {
+    throw createPayloadTooLargeResult();
   }
-  return context.body as Promise<Buffer>;
+  return body;
+}
+
+function createPayloadTooLargeResult(): HTTPResult {
+  return new HTTPResult(PAYLOAD_TOO_LARGE_STATUS, PAYLOAD_TOO_LARGE_MESSAGE);
 }
 
 /**
@@ -1213,6 +1296,8 @@ export function AddParameterModifier(
  * This is useful for processing raw data from the client, such as file uploads
  * or custom data formats.
  *
+ * @param limit Maximum body size in bytes
+ *
  * Example:
  * ```ts
  * @Post()
@@ -1222,8 +1307,11 @@ export function AddParameterModifier(
  * }
  * ```
  */
-export const RawBody = MakeParameterAndPropertyDecorator((target, key, param) =>
-  SetParameterProvider(target, key, param, ReadBody),
+export const RawBody = MakeParameterAndPropertyDecorator(
+  (target, key, param, limit: number = DEFAULT_REQUEST_BODY_LIMIT) =>
+    SetParameterProvider(target, key, param, (context) =>
+      ReadBody(context, limit),
+    ),
 );
 
 /**
@@ -1233,6 +1321,8 @@ export const RawBody = MakeParameterAndPropertyDecorator((target, key, param) =>
  * This is useful for handling JSON payloads in POST, PUT, and other methods
  * that accept request bodies.
  *
+ * @param limit Maximum body size in bytes
+ *
  * Example:
  * ```ts
  * @Post()
@@ -1244,9 +1334,9 @@ export const RawBody = MakeParameterAndPropertyDecorator((target, key, param) =>
  * ```
  */
 export const JSONBody = MakeParameterAndPropertyDecorator(
-  (target, key, index) => {
+  (target, key, index, limit: number = DEFAULT_REQUEST_BODY_LIMIT) => {
     SetParameterProvider(target, key, index, (ctx: RequestContext) =>
-      ReadBody(ctx).then((body: unknown) => {
+      ReadBody(ctx, limit).then((body: unknown) => {
         if (!body || (body instanceof Buffer && body.length === 0)) {
           return undefined;
         }

src/tests/index.test.ts

@@ -1,5 +1,5 @@
 import assert from "node:assert";
-import type { ServerResponse } from "node:http";
+import type { IncomingMessage, ServerResponse } from "node:http";
 import { PassThrough } from "node:stream";
 import {
   Connection,
@@ -37,7 +37,11 @@ import logListener, {
 } from "@antelopejs/interface-core/logging/listener";
 import sinon, { type SinonSpy } from "sinon";
 import WebSocket from "ws";
-import { HTTPResult as LocalHTTPResult } from "../index";
+import {
+  DEFAULT_REQUEST_BODY_LIMIT,
+  HTTPResult as LocalHTTPResult,
+  ReadBody,
+} from "../index";
 
 const SpyMethod = MakeMethodDecorator((_target, _key, descriptor) => {
   descriptor.value = sinon.spy(descriptor.value);
@@ -1218,6 +1222,118 @@ describe("WebSocket", () => {
   });
 });
 
+type BodyRequest = PassThrough & IncomingMessage;
+
+interface BodyTestContext {
+  context: RequestContext;
+  request: BodyRequest;
+}
+
+function createBodyTestContext(contentLength?: number): BodyTestContext {
+  const request = new PassThrough() as BodyRequest;
+  request.headers = {};
+  if (contentLength !== undefined) {
+    request.headers["content-length"] = contentLength.toString();
+  }
+  return {
+    request,
+    context: {
+      rawRequest: request,
+      rawResponse: {} as ServerResponse,
+      url: new URL(URL_BASE),
+      routeParameters: {},
+      response: new LocalHTTPResult(),
+    },
+  };
+}
+
+function assertBodyListenersRemoved(request: BodyRequest): void {
+  assert.equal(request.listenerCount("data"), 0);
+  assert.equal(request.listenerCount("end"), 0);
+  assert.equal(request.listenerCount("error"), 0);
+}
+
+async function assertPayloadTooLarge(body: Promise<Buffer>): Promise<void> {
+  await assert.rejects(body, (error) => {
+    assert(error instanceof LocalHTTPResult);
+    assert.equal(error.getStatus(), 413);
+    assert.equal(error.getBody(), "Payload Too Large");
+    return true;
+  });
+}
+
+describe("ReadBody limits", () => {
+  it("Accepts a normal body with the default limit", async () => {
+    const test = createBodyTestContext();
+    const body = ReadBody(test.context);
+    test.request.end("normal body");
+
+    assert.equal((await body).toString(), "normal body");
+    assertBodyListenersRemoved(test.request);
+  });
+
+  it("Accepts a body exactly at the configured limit", async () => {
+    const test = createBodyTestContext(4);
+    const body = ReadBody(test.context, 4);
+    test.request.end("test");
+
+    assert.equal((await body).toString(), "test");
+    assertBodyListenersRemoved(test.request);
+  });
+
+  it("Rejects an oversized Content-Length before adding listeners", async () => {
+    const test = createBodyTestContext(DEFAULT_REQUEST_BODY_LIMIT + 1);
+
+    await assertPayloadTooLarge(ReadBody(test.context));
+    assert.equal(test.request.isPaused(), true);
+    assertBodyListenersRemoved(test.request);
+  });
+
+  it("Rejects and stops an oversized chunked body", async () => {
+    const test = createBodyTestContext();
+    const body = ReadBody(test.context, 4);
+    test.request.write("test");
+    test.request.write("!");
+
+    await assertPayloadTooLarge(body);
+    assert.equal(test.request.isPaused(), true);
+    assertBodyListenersRemoved(test.request);
+  });
+
+  it("Applies each limit across concurrent consumers", async () => {
+    const test = createBodyTestContext(5);
+    const permissiveBody = ReadBody(test.context, 8);
+    const strictBody = ReadBody(test.context, 4);
+    test.request.end("12345");
+
+    assert.equal((await permissiveBody).toString(), "12345");
+    await assertPayloadTooLarge(strictBody);
+    assertBodyListenersRemoved(test.request);
+  });
+
+  it("Applies a stricter limit to an already cached body", async () => {
+    const test = createBodyTestContext(5);
+    const body = ReadBody(test.context, 8);
+    test.request.end("12345");
+    assert.equal((await body).toString(), "12345");
+
+    await assertPayloadTooLarge(ReadBody(test.context, 4));
+    assertBodyListenersRemoved(test.request);
+  });
+
+  it("Resumes for a permissive consumer after an earlier rejection", async () => {
+    const test = createBodyTestContext();
+    const strictBody = ReadBody(test.context, 4);
+    test.request.write("12345");
+    await assertPayloadTooLarge(strictBody);
+
+    const body = ReadBody(test.context, 8);
+    test.request.end();
+    assert.equal((await body).toString(), "12345");
+    assertBodyListenersRemoved(test.request);
+  });
+});
+
 // Keep in sync with SERVER_ERROR_BODY_LOG_LIMIT in src/index.ts
 const BODY_LOG_LIMIT = 2048;