fix: handle invalid multipart streams

package.json

@@ -36,7 +36,6 @@
     "@japa/spec-reporter": "^1.3.1",
     "@poppinss/dev-utils": "^2.0.3",
     "@types/bytes": "^3.1.1",
-    "@types/end-of-stream": "^1.4.1",
     "@types/fs-extra": "^9.0.13",
     "@types/media-typer": "^1.1.1",
     "@types/node": "^18.8.3",
@@ -52,6 +51,7 @@
     "github-label-sync": "^2.2.0",
     "husky": "^8.0.1",
     "mrm": "^4.1.6",
+    "node-fetch": "^2.6.11",
     "np": "^7.6.2",
     "prettier": "^2.7.1",
     "reflect-metadata": "^0.1.13",
@@ -91,7 +91,6 @@
     "@poppinss/multiparty": "^2.0.1",
     "@poppinss/utils": "^5.0.0",
     "bytes": "^3.1.2",
-    "end-of-stream": "^1.4.4",
     "file-type": "^16.5.4",
     "fs-extra": "^10.1.0",
     "media-typer": "^1.1.0",

src/Multipart/index.ts

@@ -322,7 +322,9 @@ export class Multipart implements MultipartContract {
             this.ctx.request.request.resume()
           }
 
-          if (error.message.match(/maxFields [0-9]+ exceeded/)) {
+          if (error.message.match(/stream ended unexpectedly/)) {
+            reject(new Exception('Invalid multipart request', 400, 'E_INVALID_MULTIPART_REQUEST'))
+          } else if (error.message.match(/maxFields [0-9]+ exceeded/)) {
             reject(new Exception('Fields length limit exceeded', 413, 'E_REQUEST_ENTITY_TOO_LARGE'))
           } else if (error.message.match(/maxFieldsSize [0-9]+ exceeded/)) {
             reject(

src/Multipart/streamFile.ts

@@ -7,63 +7,31 @@
  * file that was distributed with this source code.
  */
 
-import eos from 'end-of-stream'
 import { Readable } from 'stream'
-import { open, close, createWriteStream, unlink } from 'fs-extra'
+import { unlink } from 'fs-extra'
+import { createWriteStream } from 'fs'
+import { pipeline } from 'stream/promises'
 
 /**
  * Writes readable stream to the given location by properly cleaning up readable
  * and writable streams in case of any errors. Also an optional data listener
  * can listen for the `data` event.
  */
-export function streamFile(
+export async function streamFile(
   readStream: Readable,
   location: string,
   dataListener?: (line: Buffer) => void
 ): Promise<void> {
-  return new Promise((resolve, reject) => {
-    open(location, 'w')
-      .then((fd) => {
-        /**
-         * Create write stream and reject promise on error
-         * event
-         */
-        const writeStream = createWriteStream(location)
-        writeStream.on('error', reject)
-
-        /**
-         * Handle closing of read stream from multiple sources
-         */
-        eos(readStream, (error: Error) => {
-          close(fd)
-
-          /**
-           * Resolve when their are no errors in
-           * streaming
-           */
-          if (!error) {
-            resolve()
-            return
-          }
-
-          /**
-           * Otherwise cleanup write stream
-           */
-          reject(error)
-
-          process.nextTick(() => {
-            writeStream.end()
-            unlink(writeStream.path).catch(() => {})
-          })
-        })
-
-        if (typeof dataListener === 'function') {
-          readStream.pause()
-          readStream.on('data', dataListener)
-        }
-
-        readStream.pipe(writeStream)
-      })
-      .catch(reject)
-  })
+  if (typeof dataListener === 'function') {
+    readStream.pause()
+    readStream.on('data', dataListener)
+  }
+
+  const writeStream = createWriteStream(location)
+  try {
+    await pipeline(readStream, writeStream)
+  } catch (error) {
+    unlink(writeStream.path).catch(() => {})
+    throw error
+  }
 }

test/body-parser.spec.ts

@@ -13,6 +13,7 @@
 import 'reflect-metadata'
 import { join } from 'path'
 import { tmpdir } from 'os'
+import fetch from 'node-fetch'
 import supertest from 'supertest'
 import { test } from '@japa/runner'
 import { createServer } from 'http'
@@ -232,6 +233,68 @@ test.group('BodyParser Middleware | form data', (group) => {
       name: null,
     })
   })
+
+  test('abort when multipart body is invalid', async ({ assert, cleanup }) => {
+    const server = createServer(async (req, res) => {
+      const ctx = app.container.use('Adonis/Core/HttpContext').create('/', {}, req, res)
+
+      const middleware = new BodyParserMiddleware(
+        app.container.use('Adonis/Core/Config'),
+        app.container.use('Adonis/Core/Drive')
+      )
+
+      try {
+        await middleware.handle(ctx, async () => {})
+      } catch (error) {
+        res.writeHead(error.status)
+        res.end(error.message)
+      }
+    })
+
+    cleanup(() => server.close())
+    await new Promise<void>((resolve) => server.listen(3333, 'localhost', () => resolve()))
+
+    const response = await fetch('http://localhost:3333', {
+      method: 'POST',
+      headers: {
+        'Content-type': `multipart/form-data; boundary=9d01a3fb93deedb4d0a81389271d097f28fd67e2fcbff2932befc0458ad7`,
+      },
+      body: '--9d01a3fb93deedb4d0a81389271d097f28fd67e2fcbff2932befc0458ad7\x0d\x0aContent-Disposition: form-data; name="test"; filename="csv_files/test.csv"\x0d\x0aContent-Type: application/octet-stream\x0d\x0a\x0d\x0atest123',
+    })
+
+    assert.equal(await response.text(), 'E_INVALID_MULTIPART_REQUEST: Invalid multipart request')
+  })
+
+  test('abort when multipart body is invalid newline characters', async ({ assert, cleanup }) => {
+    const server = createServer(async (req, res) => {
+      const ctx = app.container.use('Adonis/Core/HttpContext').create('/', {}, req, res)
+
+      const middleware = new BodyParserMiddleware(
+        app.container.use('Adonis/Core/Config'),
+        app.container.use('Adonis/Core/Drive')
+      )
+
+      try {
+        await middleware.handle(ctx, async () => {})
+      } catch (error) {
+        res.writeHead(error.status)
+        res.end(error.message)
+      }
+    })
+
+    cleanup(() => server.close())
+    await new Promise<void>((resolve) => server.listen(3333, 'localhost', () => resolve()))
+
+    const response = await fetch('http://localhost:3333', {
+      method: 'POST',
+      headers: {
+        'Content-type': `multipart/form-data; boundary=XXX`,
+      },
+      body: '--XXX\nContent-Disposition: form-data; name="file"; filename="filename.csv"\nContent-Type: text/csv\n\nA,B,C\n1,1.1,name1\n2,2.2,name2\n\n--XXX--',
+    })
+
+    assert.equal(await response.text(), 'Expected CR Received 10')
+  })
 })
 
 test.group('BodyParser Middleware | json', (group) => {