feat(http): add size limits to server connections and listen lifecycle hooks

packages/http/src/common/buffer.ts

@@ -1,16 +1,27 @@
 import type { Reader, Writer } from "./types.js"
 
+export type ReadBufferOptions = {
+  /**
+   * @defaultValue 1024 * 128
+   */
+  maxLineLength?: number
+}
+
 export class ReadBuffer {
   #buffers: Uint8Array[]
   #offset: number
   #decoder: TextDecoder
   #reader: Reader
 
-  constructor(reader: Reader) {
+  #maxLineLength: number
+
+  constructor(reader: Reader, { maxLineLength = 1024 * 128 }: ReadBufferOptions = {}) {
     this.#buffers = []
     this.#offset = 0
     this.#decoder = new TextDecoder()
     this.#reader = reader
+
+    this.#maxLineLength = maxLineLength
   }
 
   get len(): number {
@@ -30,7 +41,7 @@ export class ReadBuffer {
     while (this.len < size && !this.#reader.closed) await this.readOnce()
   }
 
-  async #readUntilByte(byte: number, maxLen: number = 1024 * 128): Promise<number> {
+  async #readUntilByte(byte: number, maxLen: number): Promise<number> {
     if (!this.#buffers.length) await this.readOnce()
     let idx = this.#buffers[0].indexOf(byte, this.#offset)
     if (idx !== -1) return idx - this.#offset
@@ -99,9 +110,9 @@ export class ReadBuffer {
     return text + this.#decoder.decode()
   }
 
-  async readLine(maxLen: number = 1024 * 128): Promise<string> {
+  async readLine(): Promise<string> {
-    const crIdx = await this.#readUntilByte(0x0d, maxLen) // CR
+    const crIdx = await this.#readUntilByte(0x0d, this.#maxLineLength) // CR
-    if (crIdx === -1) throw new Error(`Line too long (exceeded ${maxLen} bytes)`)
+    if (crIdx === -1) throw new Error(`Line too long (exceeded ${this.#maxLineLength} bytes)`)
     if (this.len < crIdx + 2) await this.read(crIdx + 2)
     if (this.#at(crIdx + 1) !== 0x0a)
       throw new Error(
@@ -109,7 +120,8 @@ export class ReadBuffer {
           .toString(16)
           .padStart(2, "0")}`,
       )
-    if (crIdx > maxLen) throw new Error(`Line too long (exceeded ${maxLen} bytes)`)
+    if (crIdx > this.#maxLineLength)
+      throw new Error(`Line too long (exceeded ${this.#maxLineLength} bytes)`)
     const line = this.#sliceStr(0, crIdx)
     this.forward(crIdx + 2)
     return line

packages/http/src/common/objects.ts

@@ -1,5 +1,6 @@
 import type { Body, ReadableHttp, Reader, WritableHttp } from "./types.js"
 import { Headers, MutableHeaders } from "./headers.js"
+import type { BodyReaderOptions } from "./reader.js"
 
 export class ReadableHttpImpl implements ReadableHttp {
   #headers: Headers
@@ -23,16 +24,23 @@ export class ReadableHttpImpl implements ReadableHttp {
   get hasBody(): boolean {
     return !!this.#bodyStream
   }
-  stream(encoding?: "binary"): AsyncIterable<Uint8Array>
+  stream(encoding?: "binary", options?: BodyReaderOptions): AsyncIterable<Uint8Array>
-  stream(encoding: "utf8"): AsyncIterable<string>
+  stream(encoding: "utf8", options?: BodyReaderOptions): AsyncIterable<string>
-  async *stream(encoding?: "binary" | "utf8"): AsyncIterable<string | Uint8Array> {
+  async *stream(
+    encoding?: "binary" | "utf8",
+    { maxBodyLength = Infinity }: BodyReaderOptions = {},
+  ): AsyncIterable<string | Uint8Array> {
     const body = this.#assertBody()
     this.#bodyStream = null
     const decoder = encoding === "utf8" ? new TextDecoder() : null
+    let bodyLength = 0
     while (!body.closed) {
       try {
         const chunk = await body.read()
         if (!chunk.length) continue
+        bodyLength += chunk.length
+        if (bodyLength > maxBodyLength)
+          throw new Error(`Body too large: ${bodyLength} (max: ${maxBodyLength})`)
         if (decoder) yield decoder.decode(chunk)
         else yield chunk
       } catch (e) {
@@ -41,14 +49,14 @@ export class ReadableHttpImpl implements ReadableHttp {
       }
     }
   }
-  async text(encoding?: "utf8"): Promise<string> {
+  async text(encoding?: "utf8", options?: BodyReaderOptions): Promise<string> {
     void encoding
-    return new TextDecoder().decode(await this.bytes())
+    return new TextDecoder().decode(await this.bytes(options))
   }
-  async bytes(): Promise<Uint8Array> {
+  async bytes(options?: BodyReaderOptions): Promise<Uint8Array> {
     const chunks: Uint8Array[] = []
     let size = 0
-    for await (const chunk of this.stream()) {
+    for await (const chunk of this.stream("binary", options)) {
       chunks.push(chunk)
       size += chunk.length
     }
@@ -60,8 +68,8 @@ export class ReadableHttpImpl implements ReadableHttp {
     }
     return bytes
   }
-  async json<T>(): Promise<T> {
+  async json<T>(options?: BodyReaderOptions): Promise<T> {
-    return JSON.parse(await this.text())
+    return JSON.parse(await this.text("utf8", options))
   }
 
   #assertBody(): Reader {

packages/http/src/common/reader.ts

@@ -11,6 +11,13 @@ export interface BodyReader extends Reader {
   ): void
 }
 
+export type BodyReaderOptions = {
+  /**
+   * @defaultValue Infinity
+   */
+  maxBodyLength?: number
+}
+
 abstract class BaseBodyReader implements Pick<BodyReader, "onFinish"> {
   #finishOkHandlers: (() => void | Promise<void>)[]
   #finishKoHandlers: ((err: unknown) => void | Promise<void>)[]
@@ -105,11 +112,17 @@ export class ChunkedBodyReader extends BaseBodyReader implements BodyReader {
   #chunks: Uint8Array[]
   #finished: boolean
 
-  constructor(buffer: ReadBuffer) {
+  #bodyLength: number
+  #maxBodyLength: number
+
+  constructor(buffer: ReadBuffer, { maxBodyLength = Infinity }: BodyReaderOptions = {}) {
     super()
     this.#buffer = buffer
     this.#chunks = []
     this.#finished = false
+
+    this.#bodyLength = 0
+    this.#maxBodyLength = maxBodyLength
   }
 
   get closed(): boolean {
@@ -124,6 +137,9 @@ export class ChunkedBodyReader extends BaseBodyReader implements BodyReader {
       const len = Number.parseInt(chunkLine, 16)
       if (!Number.isInteger(len) || len < 0)
         throw new Error(`Invalid chunk size: ${JSON.stringify(chunkLine)}`)
+      this.#bodyLength += len
+      if (this.#bodyLength > this.#maxBodyLength)
+        throw new Error(`Body too large: ${this.#bodyLength} (max: ${this.#maxBodyLength})`)
       if (!len) {
         this.#finished = true
         await this.#buffer.read(2)
@@ -145,13 +161,39 @@ export class ChunkedBodyReader extends BaseBodyReader implements BodyReader {
   }
 }
 
-export async function readHeaders(buffer: ReadBuffer): Promise<Headers> {
+export type ReadHeadersOptions = {
+  /**
+   * @defaultValue 256
+   */
+  maxHeaderCount?: number
+  /**
+   * @defaultValue Infinity
+   */
+  maxTotalHeaderSize?: number
+}
+
+export async function readHeaders(
+  buffer: ReadBuffer,
+  { maxHeaderCount = 256, maxTotalHeaderSize = Infinity }: ReadHeadersOptions = {},
+): Promise<Headers> {
   const rawHeaders: Record<string, string[]> = {}
+  let headerCount = 0
+  let totalHeaderSize = 0
   while (true) {
     const headerLine = await buffer.readLine()
     if (!headerLine.length) {
       break
     }
+
+    headerCount++
+    if (headerCount > maxHeaderCount)
+      throw new Error(`Too many headers: ${headerCount} (max: ${maxHeaderCount})`)
+    totalHeaderSize += headerLine.length
+    if (totalHeaderSize > maxTotalHeaderSize)
+      throw new Error(
+        `Total max header size exceeded: ${totalHeaderSize} (max: ${maxTotalHeaderSize})`,
+      )
+
     const colon = headerLine.indexOf(":")
     if (colon === -1) throw new Error(`Invalid header line (no colon): ${headerLine.slice(0, 80)}`)
     const key = headerLine.slice(0, colon)
@@ -164,13 +206,19 @@ export async function readHeaders(buffer: ReadBuffer): Promise<Headers> {
   )
 }
 
-export function bodyReader(buffer: ReadBuffer, headers: Headers): BodyReader | null {
+export function bodyReader(
+  buffer: ReadBuffer,
+  headers: Headers,
+  { maxBodyLength = Infinity }: BodyReaderOptions = {},
+): BodyReader | null {
   if (headers.get("Transfer-Encoding") === "chunked") {
-    return new ChunkedBodyReader(buffer)
+    return new ChunkedBodyReader(buffer, { maxBodyLength })
   } else if (typeof headers.get("Content-Length") === "string") {
     const len = +(headers.get("Content-Length") as string)
     if (!Number.isInteger(len) || len < 0)
       throw new Error(`Content-Length is invalid: ${headers.get("Content-Length")}`)
+    if (len > maxBodyLength)
+      throw new Error(`Content-Length is too big: ${len} (max: ${maxBodyLength})`)
     return new BasicBodyReader(buffer, len)
   } else {
     return null

packages/http/src/common/types.ts

@@ -1,3 +1,5 @@
+import type { BodyReaderOptions } from "./reader.js"
+
 export interface RawTransport {
   get closed(): boolean
   close(): void | Promise<void>
@@ -35,11 +37,11 @@ export interface ReadableHttp {
 
   // body
   readonly hasBody: boolean
-  stream(encoding?: "binary"): AsyncIterable<Uint8Array>
+  stream(encoding?: "binary", options?: BodyReaderOptions): AsyncIterable<Uint8Array>
-  stream(encoding: "utf8"): AsyncIterable<string>
+  stream(encoding: "utf8", options?: BodyReaderOptions): AsyncIterable<string>
-  text(encoding?: "utf8"): Promise<string>
+  text(encoding?: "utf8", options?: BodyReaderOptions): Promise<string>
-  bytes(): Promise<Uint8Array>
+  bytes(options?: BodyReaderOptions): Promise<Uint8Array>
-  json<T>(): Promise<T>
+  json<T>(options?: BodyReaderOptions): Promise<T>
 }
 
 export interface WritableHttp {

packages/http/src/server/connection.ts

@@ -1,12 +1,26 @@
 import { ServerRequestImpl, ServerResponseImpl } from "./objects.js"
 import type { RawTransport, Reader } from "../common/types.js"
 import { statusCodeProperties } from "../common/spec.js"
-import { ReadBuffer, WriteBuffer } from "../common/buffer.js"
+import { ReadBuffer, WriteBuffer, type ReadBufferOptions } from "../common/buffer.js"
-import { bodyReader, readHeaders } from "../common/reader.js"
+import {
+  bodyReader,
+  readHeaders,
+  type BodyReaderOptions,
+  type ReadHeadersOptions,
+} from "../common/reader.js"
 import { formatBody, sendBody, writeHeaders } from "../common/writer.js"
 import type { Handler, ServerRequest, ServerResponse } from "./types.js"
 import { shouldClose } from "../common/connection.js"
 
+export type ServerConnectionOptions = ReadBufferOptions &
+  ReadHeadersOptions &
+  BodyReaderOptions & {
+    /**
+     * @defaultValue 1024 * 16
+     */
+    maxTargetLength?: number
+  }
+
 export class ServerConnection {
   #transport: RawTransport
   #readBuffer: ReadBuffer
@@ -14,12 +28,20 @@ export class ServerConnection {
   #encoder: TextEncoder
   #prevBody: Reader | null
 
-  constructor(transport: RawTransport) {
+  #options: ServerConnectionOptions
+
+  constructor(transport: RawTransport, options: ServerConnectionOptions = {}) {
     this.#transport = transport
-    this.#readBuffer = new ReadBuffer(transport)
+    this.#readBuffer = new ReadBuffer(transport, options)
     this.#writeBuffer = new WriteBuffer(transport)
     this.#encoder = new TextEncoder()
     this.#prevBody = null
+
+    this.#options = options
+  }
+
+  get #maxTargetLength(): number {
+    return this.#options.maxTargetLength ?? 1024 * 16
   }
 
   async #parse(): Promise<ServerRequestImpl> {
@@ -35,8 +57,13 @@ export class ServerConnection {
       else if (parts[2] !== "HTTP/1.0") throw new Error(`Invalid HTTP version: ${parts[2]}`)
     }
 
-    const headers = await readHeaders(this.#readBuffer)
+    if (target.length >= this.#maxTargetLength)
-    const bodyStream = bodyReader(this.#readBuffer, headers)
+      throw new Error(
+        `Target max length exceeded: ${target.length} (max: ${this.#maxTargetLength})`,
+      )
+
+    const headers = await readHeaders(this.#readBuffer, this.#options)
+    const bodyStream = bodyReader(this.#readBuffer, headers, this.#options)
     this.#prevBody = bodyStream
 
     return new ServerRequestImpl({

packages/http/src/server/server.ts

@@ -1,16 +1,45 @@
-import { ServerConnection } from "./connection.js"
+import type { RawTransport } from "../common/types.js"
+import { ServerConnection, type ServerConnectionOptions } from "./connection.js"
 import { Router } from "./router.js"
 import type { RawListener } from "./types.js"
 
+export type ListenOptions = ServerConnectionOptions & {
+  /**
+   * Called when a new connection is accepted; return false to automatically close and reject it
+   */
+  onConnect?: (transport: RawTransport) => void | boolean
+  /**
+   * Called when an error is thrown in handling the connection
+   */
+  onError?: (connection: ServerConnection, transport: RawTransport, error: unknown) => void
+}
+
+const defaultOnConnect: NonNullable<ListenOptions["onConnect"]> = () => {
+  return true
+}
+const defaultOnError: NonNullable<ListenOptions["onError"]> = (conn, transport, error) => {
+  console.error("[Server.listen] caught error in connection handling", error)
+}
+
 export class Server extends Router {
-  async listen(listener: RawListener): Promise<void> {
+  async listen(
+    listener: RawListener,
+    { onConnect = defaultOnConnect, onError = defaultOnError, ...options }: ListenOptions = {},
+  ): Promise<void> {
+    const handle = async (transport: RawTransport) => {
+      const connection = new ServerConnection(transport, options)
+      try {
+        if (onConnect(transport) === false) return
+        await connection.handle(this.handler)
+      } catch (e) {
+        onError(connection, transport, e)
+      }
+    }
+
     while (!listener.closed) {
       try {
         const transport = await listener.accept()
-        const connection = new ServerConnection(transport)
+        void handle(transport)
-        void connection
-          .handle(this.handler)
-          .catch((e) => console.error("[Server.listen] caught error in connection handling", e))
       } catch (e) {
         if (listener.closed) break
         throw e