fix(node): reject overlapping body reads

packages/fastify/src/request.test.ts

@@ -113,6 +113,42 @@ describe('toStandardLazyRequest', () => {
     expect(standardBody).toEqual(expectedBody)
   })
 
+  it('fallback to standard body parser rejects overlapping body reads', async ({
+    onTestFinished,
+  }) => {
+    let results: PromiseSettledResult<unknown>[] = []
+
+    const fastify = Fastify()
+    onTestFinished(() => fastify.close())
+
+    // allow any content type, and leave the body unparsed
+    fastify.addContentTypeParser('*', (req, payload, done) => {
+      done(null, undefined)
+    })
+
+    fastify.all('/', async (req, reply) => {
+      const standardRequest = toStandardLazyRequest(req, reply)
+      results = await Promise.allSettled([
+        standardRequest.resolveBody('json'),
+        standardRequest.resolveBody('json'),
+      ])
+    })
+
+    await fastify.ready()
+    await request(fastify.server)
+      .post('/')
+      .set('content-type', 'application/octet-stream')
+      .send('{"amount":100}')
+
+    expect(results).toEqual([
+      { status: 'fulfilled', value: { amount: 100 } },
+      {
+        status: 'rejected',
+        reason: new TypeError('Failed to read body: body stream already read or destroyed'),
+      },
+    ])
+  })
+
   it('forwards the body hint to the standard body parser', async ({ onTestFinished }) => {
     let fastifyReq: any
     let standardBody: any

packages/node/src/body.test.ts

@@ -457,6 +457,89 @@ describe('toStandardBody', () => {
       expect(await toStandardBody(req!, { hint: 'none' })).toBe(undefined)
     })
 
+    const alreadyReadError = new TypeError(
+      'Failed to read body: body stream already read or destroyed',
+    )
+
+    it.each([
+      ['json', '{"amount":100}', { amount: 100 }],
+      ['file', '{"amount":100}', '{"amount":100}'],
+      ['url-search-params', 'amount=100', new URLSearchParams('amount=100')],
+    ] as const)('throw on overlapping body reads (hint=%s)', async (hint, body, expected) => {
+      let results: PromiseSettledResult<unknown>[] = []
+
+      await request(async (req: IncomingMessage, res: ServerResponse) => {
+        results = await Promise.allSettled([
+          // a File only compares by its metadata, so compare its text instead
+          toStandardBody(req, { hint }).then((b) => (b instanceof File ? b.text() : b)),
+          toStandardBody(req, { hint }),
+        ])
+        res.end()
+      })
+        .post('/')
+        .send(body)
+
+      expect(results).toEqual([
+        { status: 'fulfilled', value: expected },
+        { status: 'rejected', reason: alreadyReadError },
+      ])
+    })
+
+    it('throw on read body again before the octet-stream is drained', async () => {
+      let secondError: unknown
+      let streamed: string | undefined
+
+      await request(async (req: IncomingMessage, res: ServerResponse) => {
+        const stream = (await toStandardBody(req, { hint: 'octet-stream' })) as ReadableStream
+        secondError = await toStandardBody(req, { hint: 'octet-stream' }).catch((e) => e)
+        streamed = await text(stream)
+        res.end()
+      })
+        .post('/')
+        .send('hello')
+
+      expect(streamed).toBe('hello')
+      expect(secondError).toEqual(alreadyReadError)
+    })
+
+    it('throw on read body again before the event-stream is drained', async () => {
+      let events: IteratorResult<unknown>[] = []
+      let secondError: unknown
+
+      await request(async (req: IncomingMessage, res: ServerResponse) => {
+        const iterator = (await toStandardBody(req)) as AsyncIterator<unknown>
+        secondError = await toStandardBody(req).catch((e) => e)
+        events = [await iterator.next(), await iterator.next(), await iterator.next()]
+        res.end()
+      })
+        .post('/')
+        .type('text/event-stream')
+        .send('event: message\ndata: 1\n\nevent: message\ndata: 2\n\nevent: close\ndata: 3\n\n')
+
+      expect(events).toEqual([
+        { done: false, value: 1 },
+        { done: false, value: 2 },
+        { done: true, value: 3 },
+      ])
+      expect(secondError).toEqual(alreadyReadError)
+    })
+
+    it('hint=none neither reads nor marks the body', async () => {
+      let none: StandardBody = {} as any
+      let standardBody: StandardBody
+
+      await request(async (req: IncomingMessage, res: ServerResponse) => {
+        none = await toStandardBody(req, { hint: 'none' })
+        standardBody = await toStandardBody(req)
+        res.end()
+      })
+        .post('/')
+        .send({ foo: 'bar' })
+
+      expect(none).toBe(undefined)
+      expect(standardBody).toEqual({ foo: 'bar' })
+    })
+
     it('prefers user defined body hint over standard-server header', async () => {
       let standardBody: any
 

packages/node/src/body.ts

@@ -20,6 +20,11 @@ export interface ToStandardBodyOptions {
   hint?: StandardBodyHint | undefined
 }
 
+/**
+ * Requests whose body stream a parser has already been attached to.
+ */
+const consumedRequests = new WeakSet<NodeHttpRequest>()
+
 /**
  * Parses the body of a node http request.
  */
@@ -45,11 +50,15 @@ export async function toStandardBody(
     return undefined
   }
 
-  if (!req.readable) {
+  // `readable` stays true until the stream ends, so it misses overlapping or lazy
+  // (stream/iterator) reads; mark the request before the first await to catch those.
+  if (!req.readable || consumedRequests.has(req)) {
     // native fetch error use TypeError
     throw new TypeError('Failed to read body: body stream already read or destroyed')
   }
 
+  consumedRequests.add(req)
+
   if (hint === 'json') {
     const text = await _streamToString(req)
     return parseEmptyableJSON(text)