http-transport: safe cors defaults, request body size cap

packages/http-transport/src/constants.ts

@@ -2,6 +2,9 @@ import type { MetadataKind } from '@nmtjs/application'
 import { createMeta } from '@nmtjs/application'
 import { ErrorCode } from '@nmtjs/protocol'
 
+// Matches Bun's maxRequestBodySize default so all runtimes behave the same
+export const DEFAULT_MAX_REQUEST_BODY_SIZE = 128 * 1024 * 1024
+
 export enum HttpStatus {
   Continue = 100,
   SwitchingProtocols = 101,

packages/http-transport/src/runtimes/bun.ts

@@ -26,6 +26,9 @@ function adapterFactory(params: HttpAdapterParams<'bun'>): HttpAdapterServer {
 
     return globalThis.Bun.serve({
       ...params.runtime,
+      // Bun's own default (128MiB) applies when neither option is set
+      maxRequestBodySize:
+        params.runtime?.maxRequestBodySize ?? params.maxRequestBodySize,
       unix: params.listen.unix as string,
       port: params.listen.port ?? 0,
       hostname: params.listen.hostname,

packages/http-transport/src/runtimes/node.ts

@@ -10,12 +10,14 @@ import type {
   HttpAdapterServer,
   HttpTransportOptions,
 } from '../types.ts'
+import { DEFAULT_MAX_REQUEST_BODY_SIZE } from '../constants.ts'
 import * as injectables from '../injectables.ts'
 import { createHTTPTransportWorker } from '../server.ts'
 import {
   InternalServerErrorHttpResponse,
   NotFoundHttpResponse,
   OkResponse,
+  PayloadTooLargeError,
 } from '../utils.ts'
 
 const statusResponse = OkResponse()
@@ -26,6 +28,7 @@ type UwsResponse = Parameters<
 >[0] & { aborted?: boolean }
 
 function adapterFactory(params: HttpAdapterParams<'node'>): HttpAdapterServer {
+  const maxBodySize = params.maxRequestBodySize ?? DEFAULT_MAX_REQUEST_BODY_SIZE
   const server = params.tls
     ? SSLApp({
         passphrase: params.tls.passphrase,
@@ -70,12 +73,22 @@ function adapterFactory(params: HttpAdapterParams<'node'>): HttpAdapterServer {
       const url = new URL(req.getUrl(), `${proto}://${host}`)
       url.search = req.getQuery() ? `?${req.getQuery()}` : ''
       try {
+        // uWS delivers chunks without backpressure, so cap what gets copied
+        // into memory before the whole body arrives
+        let received = 0
+        let capped = false
         const body = new ReadableStream<Buffer>({
           start(controller) {
             bodyController = controller
             res.onDataV2((chunk, maxRemainingBodyLength) => {
-              if (aborted) return
+              if (aborted || capped) return
               if (chunk) {
+                received += chunk.byteLength
+                if (received > maxBodySize) {
+                  capped = true
+                  controller.error(new PayloadTooLargeError())
+                  return
+                }
                 const copy = Buffer.allocUnsafe(chunk.byteLength)
                 copy.set(new Uint8Array(chunk))
                 controller.enqueue(copy)

packages/http-transport/src/server.ts

@@ -1,5 +1,6 @@
 import { Buffer } from 'node:buffer'
-import { Duplex, Readable } from 'node:stream'
+import { Duplex, Readable, Transform } from 'node:stream'
+import { pipeline } from 'node:stream/promises'
 
 import type { ApplicationResolvedProcedure } from '@nmtjs/application'
 import type { TransportWorker, TransportWorkerParams } from '@nmtjs/gateway'
@@ -28,16 +29,19 @@ import type {
 } from './types.ts'
 import {
   AllowedHttpMethod,
+  DEFAULT_MAX_REQUEST_BODY_SIZE,
   HttpCodeMap,
   HttpStatus,
   HttpStatusText,
 } from './constants.ts'
 import * as injections from './injectables.ts'
+import { PayloadTooLargeError } from './utils.ts'
 
 const NEEMATA_BLOB_HEADER = 'X-Neemata-Blob'
 const DEFAULT_ALLOWED_METHODS = Object.freeze(['post']) as ('get' | 'post')[]
+// No allowCredentials here: reflecting arbitrary origins with credentials
+// would let any website make cookie-authed requests
 const DEFAULT_CORS_PARAMS = Object.freeze({
-  allowCredentials: 'true',
   allowMethods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'],
   allowHeaders: [
     'Content-Type',
@@ -52,6 +56,11 @@ const DEFAULT_CORS_PARAMS = Object.freeze({
   exposeHeaders: [],
   requestHeaders: [],
 }) satisfies Omit<HttpTransportCorsCustomParams, 'origin'>
+// Credentials are safe to allow when the user explicitly vetted the origin
+const EXPLICIT_ORIGIN_CORS_PARAMS = Object.freeze({
+  ...DEFAULT_CORS_PARAMS,
+  allowCredentials: 'true',
+}) satisfies Omit<HttpTransportCorsCustomParams, 'origin'>
 const CORS_HEADERS_MAP: Record<
   keyof HttpTransportCorsCustomParams | 'origin',
   string
@@ -82,6 +91,7 @@ export class HttpTransportServer implements TransportWorker<
 > {
   #server: HttpAdapterServer
   #corsOptions?: HttpTransportOptions['cors']
+  #maxRequestBodySize: number
 
   params!: TransportWorkerParams<
     ConnectionType.Unidirectional,
@@ -94,6 +104,8 @@ export class HttpTransportServer implements TransportWorker<
   ) {
     this.#server = this.createServer()
     this.#corsOptions = this.options.cors
+    this.#maxRequestBodySize =
+      this.options.maxRequestBodySize ?? DEFAULT_MAX_REQUEST_BODY_SIZE
   }
 
   async start(
@@ -120,6 +132,9 @@ export class HttpTransportServer implements TransportWorker<
     const method = request.method.toLowerCase()
     const origin = request.headers.get('origin')
     const responseHeaders = new Headers()
+    // CORS makes responses origin-dependent (even denials), so shared caches
+    // must key on Origin to avoid serving them across origins
+    if (this.#corsOptions) responseHeaders.append('Vary', 'Origin')
     if (origin) this.applyCors(origin, request, responseHeaders)
 
     // Handle preflight requests
@@ -158,7 +173,6 @@ export class HttpTransportServer implements TransportWorker<
       let payload: any
 
       if (canHaveBody && body) {
-        const bodyStream = Readable.fromWeb(body as any)
         const cannotDecode =
           !contentType || !this.params.formats.supportsDecoder(contentType)
         if (isBlob || cannotDecode) {
@@ -167,10 +181,34 @@ export class HttpTransportServer implements TransportWorker<
           const size = contentLength
             ? Number.parseInt(contentLength, 10)
             : undefined
-          payload = new ProtocolClientStream(-1, { size, type })
-          bodyStream.pipe(payload)
+          // Declared size over the cap: reject before reading anything
+          if (size !== undefined && size > this.#maxRequestBodySize) {
+            throw new PayloadTooLargeError()
+          }
+          const clientStream = new ProtocolClientStream(-1, { size, type })
+          // The rpc may never read the payload; without a handler a capped
+          // upload would crash the process with an unhandled 'error'
+          clientStream.on('error', () => {})
+          payload = clientStream
+          // pipeline (unlike pipe) propagates source errors; the cap error is
+          // re-surfaced on the payload stream so its consumer rejects with it
+          pipeline(
+            Readable.fromWeb(body as any),
+            this.createBodySizeGuard(),
+            clientStream,
+          ).catch((error) => clientStream.destroy(error))
         } else {
-          const buffer = Buffer.concat(await bodyStream.toArray())
+          const chunks: Buffer[] = []
+          let received = 0
+          for await (const chunk of Readable.fromWeb(body as any)) {
+            received += chunk.byteLength
+            // Reject mid-stream to avoid buffering unbounded payloads
+            if (received > this.#maxRequestBodySize) {
+              throw new PayloadTooLargeError()
+            }
+            chunks.push(chunk)
+          }
+          const buffer = Buffer.concat(chunks)
           if (buffer.byteLength > 0) {
             payload = connection.decoder.decode(buffer)
           }
@@ -196,7 +234,9 @@ export class HttpTransportServer implements TransportWorker<
       if (result instanceof Response) {
         const { status, statusText, headers, body } = result
         headers.forEach((value, key) => {
-          responseHeaders.set(key, value)
+          // Merge Vary so the cors Origin entry isn't lost to shared caches
+          if (key.toLowerCase() === 'vary') responseHeaders.append(key, value)
+          else responseHeaders.set(key, value)
         })
 
         return new Response(body, {
@@ -282,6 +322,17 @@ export class HttpTransportServer implements TransportWorker<
         })
       }
     } catch (error) {
+      if (error instanceof PayloadTooLargeError) {
+        const status = HttpStatus.PayloadTooLarge
+        const text = HttpStatusText[status]
+
+        return new Response(text, {
+          status,
+          statusText: text,
+          headers: responseHeaders,
+        })
+      }
+
       if (error instanceof UnsupportedFormatError) {
         const status =
           error instanceof UnsupportedContentTypeError
@@ -334,6 +385,19 @@ export class HttpTransportServer implements TransportWorker<
     }
   }
 
+  private createBodySizeGuard() {
+    const maxSize = this.#maxRequestBodySize
+    let received = 0
+    return new Transform({
+      transform(chunk: Buffer, _encoding, callback) {
+        received += chunk.byteLength
+        // Enforce the cap even when the declared content-length lies
+        if (received > maxSize) callback(new PayloadTooLargeError())
+        else callback(null, chunk)
+      },
+    })
+  }
+
   private applyCors(
     origin: string,
     request: HttpTransportServerRequest,
@@ -347,15 +411,19 @@ export class HttpTransportServer implements TransportWorker<
       params = { ...DEFAULT_CORS_PARAMS }
     } else if (Array.isArray(this.#corsOptions)) {
       if (this.#corsOptions.includes(origin)) {
-        params = { ...DEFAULT_CORS_PARAMS }
+        params = { ...EXPLICIT_ORIGIN_CORS_PARAMS }
       }
     } else if (typeof this.#corsOptions === 'object') {
       if (
         this.#corsOptions.origin === true ||
         this.#corsOptions.origin.includes(origin)
       ) {
-        params = { ...DEFAULT_CORS_PARAMS }
-        for (const key in DEFAULT_CORS_PARAMS) {
+        params =
+          this.#corsOptions.origin === true
+            ? { ...DEFAULT_CORS_PARAMS }
+            : { ...EXPLICIT_ORIGIN_CORS_PARAMS }
+        // Iterating base params also drops allowCredentials for origin: true
+        for (const key in params) {
           const value = this.#corsOptions[key]
           if (value !== undefined) {
             params[key] = value
@@ -366,14 +434,21 @@ export class HttpTransportServer implements TransportWorker<
       const result = this.#corsOptions(origin, request)
       if (typeof result === 'boolean') {
         if (result) {
-          params = { ...DEFAULT_CORS_PARAMS }
+          params = { ...EXPLICIT_ORIGIN_CORS_PARAMS }
         }
       } else if (typeof result === 'object') {
-        params = { ...DEFAULT_CORS_PARAMS }
-        for (const key in DEFAULT_CORS_PARAMS) {
-          const value = result[key]
-          if (value !== undefined) {
-            params[key] = value
+        // Returned params must still match the requesting origin, otherwise
+        // any origin would get reflected (with credentials for allowlists)
+        if (result.origin === true || result.origin.includes(origin)) {
+          params =
+            result.origin === true
+              ? { ...DEFAULT_CORS_PARAMS }
+              : { ...EXPLICIT_ORIGIN_CORS_PARAMS }
+          for (const key in params) {
+            const value = result[key]
+            if (value !== undefined) {
+              params[key] = value
+            }
           }
         }
       }

packages/http-transport/src/types.ts

@@ -12,19 +12,41 @@ export type HttpTransportOptions<
   listen: HttpTransportListenOptions
   cors?: HttpTransportCorsOptions
   tls?: HttpTransportTlsOptions
+  /**
+   * Maximum request body size in bytes. Requests exceeding it are rejected
+   * with 413 Payload Too Large. Defaults to 128MiB (Bun's own default, kept
+   * consistent across runtimes).
+   */
+  maxRequestBodySize?: number
   runtime?: HttpTransportRuntimes[R]
 }
 
 export type HttpTransportCorsCustomParams = {
-  origin: true | string[]
   allowMethods?: string[]
   allowHeaders?: string[]
-  allowCredentials?: string
   maxAge?: string
   exposeHeaders?: string[]
   requestHeaders?: string[]
   requestMethod?: string
-}
+} & (
+  | {
+      /**
+       * `true` reflects any request origin, an array is an explicit
+       * allowlist. Credentials default on for allowlisted origins only.
+       */
+      origin: true | string[]
+      allowCredentials?: never
+    }
+  | {
+      /**
+       * Explicit `allowCredentials` requires an origin allowlist: combining
+       * credentials with a reflected origin (`origin: true`) would let any
+       * website make credentialed (cookie-authed) requests.
+       */
+      origin: string[]
+      allowCredentials?: string
+    }
+)
 
 export type HttpTransportCorsOptions =
   | true
@@ -83,6 +105,7 @@ export type HttpAdapterParams<
   ) => MaybePromise<Response>
   cors?: HttpTransportCorsOptions
   tls?: HttpTransportTlsOptions
+  maxRequestBodySize?: number
   runtime?: HttpTransportRuntimes[R]
 }
 

packages/http-transport/src/utils.ts

@@ -26,3 +26,9 @@ export const InternalServerErrorHttpResponse = () =>
   })
 
 export const OkResponse = () => new Response('OK', { status: 200 })
+
+export class PayloadTooLargeError extends Error {
+  constructor(message = 'Payload Too Large') {
+    super(message)
+  }
+}