feat(wip): net interceptor

Author: kettanaito

_http_common.d.ts

@@ -18,26 +18,23 @@ declare var HTTPParser: {
 export interface HTTPParser<ParserType extends number> {
   new (): HTTPParser<ParserType>
 
-  [HTTPParser.kOnMessageBegin]: () => void
-  [HTTPParser.kOnHeaders]: HeadersCallback
-  [HTTPParser.kOnHeadersComplete]: ParserType extends 0
+  [HTTPParser.kOnMessageBegin]?: () => void
+  [HTTPParser.kOnHeaders]?: HeadersCallback
+  [HTTPParser.kOnHeadersComplete]?: ParserType extends 0
     ? RequestHeadersCompleteCallback
     : ResponseHeadersCompleteCallback
-  [HTTPParser.kOnBody]: (chunk: Buffer) => void
-  [HTTPParser.kOnMessageComplete]: () => void
-  [HTTPParser.kOnExecute]: () => void
-  [HTTPParser.kOnTimeout]: () => void
+  [HTTPParser.kOnBody]?: (chunk: Buffer) => void
+  [HTTPParser.kOnMessageComplete]?: () => void
+  [HTTPParser.kOnExecute]?: () => void
+  [HTTPParser.kOnTimeout]?: () => void
 
   initialize(type: ParserType, asyncResource: object): void
   execute(buffer: Buffer): void
   finish(): void
   free(): void
 }
 
-export type HeadersCallback = (
-  rawHeaders: Array<string>,
-  url: string
-) => void
+export type HeadersCallback = (rawHeaders: Array<string>, url: string) => void
 
 export type RequestHeadersCompleteCallback = (
   versionMajor: number,

src/interceptors/net/demo.test.ts

@@ -0,0 +1,26 @@
+// @vitest-environment node
+import { beforeAll, afterAll, it, expect } from 'vitest'
+import { HttpRequestInterceptor } from './http-interceptor'
+import { waitForClientRequest } from '../../../test/helpers'
+
+const interceptor = new HttpRequestInterceptor()
+
+beforeAll(() => {
+  interceptor.apply()
+})
+
+afterAll(() => {
+  interceptor.dispose()
+})
+
+it('???', async () => {
+  interceptor.on('request', ({ request, controller }) => {
+    console.log('REQUEST LISTENER!')
+    controller.respondWith(new Response('hello world'))
+  })
+
+  const http = await import('http')
+  const request = http.get('http://localhost/resource')
+
+  await waitForClientRequest(request)
+})

src/interceptors/net/http-interceptor.ts

@@ -0,0 +1,178 @@
+import { Readable, Writable } from 'node:stream'
+import { invariant } from 'outvariant'
+import { Interceptor, INTERNAL_REQUEST_ID_HEADER_NAME } from '../../Interceptor'
+import { type HttpRequestEventMap } from '../../glossary'
+import { FetchResponse } from '../../utils/fetchUtils'
+import { SocketInterceptor } from './index'
+import { HttpRequestParser } from './parsers'
+import { baseUrlFromConnectionOptions } from '../Socket/utils/baseUrlFromConnectionOptions'
+import { NetworkConnectionOptions } from './utils/normalize-net-connect-args'
+import { createRequestId } from '../../createRequestId'
+import { emitAsync } from 'src/utils/emitAsync'
+import { RequestController } from '../../RequestController'
+import { handleRequest } from '../../utils/handleRequest'
+
+function toBuffer(data: any, encoding?: BufferEncoding): Buffer {
+  return Buffer.isBuffer(data) ? data : Buffer.from(data, encoding)
+}
+
+export class HttpRequestInterceptor extends Interceptor<HttpRequestEventMap> {
+  static symbol = Symbol('HttpRequestInterceptor')
+
+  constructor() {
+    super(HttpRequestInterceptor.symbol)
+  }
+
+  public setup() {
+    /** @fixme Use interceptor as a singleton? */
+    const interceptor = new SocketInterceptor()
+    interceptor.apply()
+
+    interceptor.on('socket', ({ options, socket }) => {
+      socket.once('write', (chunk, encoding) => {
+        const firstFrame = chunk.toString()
+
+        /**
+         * @fixme This is obviously rather naive
+         */
+        if (firstFrame.includes('HTTP/1.1')) {
+          const method = firstFrame.split(' ')[0]
+
+          const requestParser = createHttpRequestParserStream({
+            requestOptions: {
+              method,
+              ...options,
+            },
+            onRequest: async ({ request }) => {
+              console.log(1)
+
+              const requestId = createRequestId()
+              const controller = new RequestController(request)
+
+              const isRequestHandled = await handleRequest({
+                request,
+                requestId,
+                controller,
+                emitter: this.emitter,
+                onResponse(response) {
+                  console.log('MOCKED!', response.status)
+                },
+                onRequestError(response) {
+                  //
+                },
+                onError(error) {},
+              })
+
+              console.log('REQ! handled?', isRequestHandled)
+
+              if (!isRequestHandled) {
+                // return socket.passthrough()
+              }
+            },
+          })
+          requestParser.write(toBuffer(chunk, encoding))
+          socket.pipe(requestParser)
+        }
+      })
+    })
+  }
+}
+
+function createHttpRequestParserStream(options: {
+  requestOptions: NetworkConnectionOptions & {
+    method: string
+  }
+  onRequest: (args: { request: Request }) => void
+}) {
+  const requestRawHeadersBuffer: Array<string> = []
+  const requestWriteBuffer: Array<Buffer> = []
+  let requestBodyStream: Readable | undefined
+
+  const parser = new HttpRequestParser({
+    onHeaders(rawHeaders) {
+      requestRawHeadersBuffer.push(...rawHeaders)
+    },
+    onHeadersComplete(
+      versionMajor,
+      versionMinor,
+      rawHeaders,
+      _,
+      path,
+      __,
+      ___,
+      ____,
+      shouldKeepAlive
+    ) {
+      const method = options.requestOptions.method?.toUpperCase() || 'GET'
+      const baseUrl = baseUrlFromConnectionOptions(options.requestOptions)
+      const url = new URL(path || '', baseUrl)
+
+      // const headers = FetchResponse.parseRawHeaders([
+      //   ...requestRawHeadersBuffer,
+      //   ...(rawHeaders || []),
+      // ])
+
+      const canHaveBody = method !== 'GET' && method !== 'HEAD'
+
+      // Translate the basic authorization to request headers.
+      // Constructing a Request instance with a URL containing auth is no-op.
+      if (url.username || url.password) {
+        if (!headers.has('authorization')) {
+          headers.set('authorization', `Basic ${url.username}:${url.password}`)
+        }
+        url.username = ''
+        url.password = ''
+      }
+
+      requestBodyStream = new Readable({
+        /**
+         * @note Provide the `read()` method so a `Readable` could be
+         * used as the actual request body (the stream calls "read()").
+         */
+        read() {
+          // If the user attempts to read the request body,
+          // flush the write buffer to trigger the callbacks.
+          // This way, if the request stream ends in the write callback,
+          // it will indeed end correctly.
+          // flushWriteBuffer()
+        },
+      })
+
+      const request = new Request(url, {
+        method,
+        // headers,
+        credentials: 'same-origin',
+        // @ts-expect-error Undocumented Fetch property.
+        duplex: canHaveBody ? 'half' : undefined,
+        body: canHaveBody ? (Readable.toWeb(requestBodyStream) as any) : null,
+      })
+
+      options.onRequest({
+        request,
+      })
+    },
+    onBody(chunk) {
+      invariant(
+        requestBodyStream,
+        'Failed to write to a request stream: stream does not exist'
+      )
+
+      requestBodyStream.push(chunk)
+    },
+    onMessageComplete() {
+      requestBodyStream?.push(null)
+    },
+  })
+
+  const parserStream = new Writable({
+    write(chunk, encoding, callback) {
+      const data = toBuffer(chunk, encoding)
+      requestWriteBuffer.push(data)
+      parser.execute(data)
+      callback()
+    },
+  })
+  parserStream.once('finish', () => parser.free())
+
+  return parserStream
+}

src/interceptors/net/index.ts

@@ -0,0 +1,54 @@
+import net from 'node:net'
+import { Interceptor } from '../../Interceptor'
+import { MockSocket } from './mock-socket'
+import {
+  NetConnectArgs,
+  type NetworkConnectionOptions,
+  normalizeNetConnectArgs,
+} from './utils/normalize-net-connect-args'
+
+export interface SocketConnectionEventMap {
+  /**
+   * Outgoing socket connection.
+   */
+  socket: [
+    args: {
+      socket: MockSocket
+      options: NetworkConnectionOptions
+    }
+  ]
+}
+
+export class SocketInterceptor extends Interceptor<SocketConnectionEventMap> {
+  static symbol = Symbol('SocketInterceptor')
+
+  constructor() {
+    super(SocketInterceptor.symbol)
+  }
+
+  protected setup(): void {
+    const {
+      connect: originalNetConnect,
+      createConnection: originalCreateConnection,
+    } = net
+
+    net.createConnection = (...args: Array<unknown>) => {
+      const [options, connectionListener] = normalizeNetConnectArgs(
+        args as NetConnectArgs
+      )
+      const socket = new MockSocket()
+
+      this.emitter.emit('socket', {
+        options,
+        socket,
+      })
+
+      return socket
+    }
+
+    this.subscriptions.push(() => {
+      net.connect = originalNetConnect
+      net.createConnection = originalCreateConnection
+    })
+  }
+}

src/interceptors/net/mock-socket.ts

@@ -0,0 +1,43 @@
+import net from 'node:net'
+import {
+  normalizeSocketWriteArgs,
+  WriteArgs,
+} from '../Socket/utils/normalizeSocketWriteArgs'
+
+export class MockSocket extends net.Socket {
+  public connecting: boolean
+
+  constructor() {
+    super()
+    this.connecting = false
+    this.connect()
+
+    this._final = (callback) => callback(null)
+  }
+
+  public connect() {
+    this.connecting = true
+    return this
+  }
+
+  public write(...args: Array<unknown>): boolean {
+    const [chunk, encoding, callback] = normalizeSocketWriteArgs(
+      args as WriteArgs
+    )
+    this.emit('write', chunk, encoding, callback)
+    return true
+  }
+
+  public push(chunk: any, encoding?: BufferEncoding): boolean {
+    this.emit('push', chunk, encoding)
+    return super.push(chunk, encoding)
+  }
+
+  public end(...args: Array<unknown>) {
+    const [chunk, encoding, callback] = normalizeSocketWriteArgs(
+      args as WriteArgs
+    )
+    this.emit('write', chunk, encoding, callback)
+    return super.end.apply(this, args as any)
+  }
+}

src/interceptors/net/parsers.ts

@@ -0,0 +1,37 @@
+import {
+  HeadersCallback,
+  HTTPParser,
+  RequestHeadersCompleteCallback,
+} from '_http_common'
+
+export class HttpRequestParser {
+  #parser: HTTPParser<typeof HTTPParser.REQUEST>
+
+  constructor(options: {
+    onMessageBegin?: () => void
+    onHeaders?: HeadersCallback
+    onHeadersComplete?: RequestHeadersCompleteCallback
+    onBody?: (chunk: Buffer) => void
+    onMessageComplete?: () => void
+    onExecute?: () => void
+    onTimeout?: () => void
+  }) {
+    this.#parser = new HTTPParser()
+    this.#parser.initialize(HTTPParser.REQUEST, {})
+    this.#parser[HTTPParser.kOnMessageBegin] = options.onMessageBegin
+    this.#parser[HTTPParser.kOnHeaders] = options.onHeaders
+    this.#parser[HTTPParser.kOnHeadersComplete] = options.onHeadersComplete
+    this.#parser[HTTPParser.kOnBody] = options.onBody
+    this.#parser[HTTPParser.kOnMessageComplete] = options.onMessageComplete
+    this.#parser[HTTPParser.kOnExecute] = options.onExecute
+    this.#parser[HTTPParser.kOnTimeout] = options.onTimeout
+  }
+
+  public execute(buffer: Buffer): void {
+    this.#parser.execute(buffer)
+  }
+
+  public free(): void {
+    this.#parser.free()
+  }
+}