fix: destroy disk write stream on aborted uploads to prevent fd leak

Ref: GHSA-qfvm-cv95-jqjf

Co-authored-by: dkoazw <dkoazw@users.noreply.github.com>

Author: UlisesGascon

lib/make-middleware.js

@@ -234,10 +234,14 @@ function makeMiddleware (setup) {
     busboy.on('file', function (fieldname, fileStream, { filename, encoding, mimeType }) {
       var pendingWritesIncremented = false
 
+      function decrementPendingWrites () {
+        if (!pendingWritesIncremented) return
+        pendingWritesIncremented = false
+        pendingWrites.decrement()
+      }
+
       fileStream.on('error', function (err) {
-        if (pendingWritesIncremented) {
-          pendingWrites.decrement()
-        }
+        decrementPendingWrites()
         abortWithError(err)
       })
 
@@ -300,20 +304,20 @@ function makeMiddleware (setup) {
           if (aborting) {
             appender.removePlaceholder(placeholder)
             uploadedFiles.push({ ...file, ...info })
-            return pendingWrites.decrement()
+            return decrementPendingWrites()
           }
 
           if (err) {
             appender.removePlaceholder(placeholder)
-            pendingWrites.decrement()
+            decrementPendingWrites()
             return abortWithError(err)
           }
 
           var fileInfo = { ...file, ...info }
 
           appender.replacePlaceholder(placeholder, fileInfo)
           uploadedFiles.push(fileInfo)
-          pendingWrites.decrement()
+          decrementPendingWrites()
           indicateDone()
         })
       })

lib/multer-error.js

@@ -10,7 +10,8 @@ var errorMessages = {
   LIMIT_UNEXPECTED_FILE: 'Unexpected field',
   MISSING_FIELD_NAME: 'Field name missing',
   LIMIT_FIELD_NESTING: 'Field name nesting too deep',
-  LIMIT_FIELD_ARRAY_INDEX: 'Field name array index too large'
+  LIMIT_FIELD_ARRAY_INDEX: 'Field name array index too large',
+  STREAM_DESTROYED: 'File stream was destroyed'
 }
 
 function MulterError (code, field) {

storage/disk.js

@@ -2,6 +2,12 @@ var fs = require('fs')
 var os = require('os')
 var path = require('path')
 var crypto = require('crypto')
+var pipeline = require('stream').pipeline
+var MulterError = require('../lib/multer-error')
+
+// Write streams still open for a file, so _removeFile can wait for the
+// descriptor to be closed before unlinking (Windows refuses to unlink open files).
+var openStreams = new WeakMap()
 
 function getFilename (req, file, cb) {
   crypto.randomBytes(16, function (err, raw) {
@@ -35,15 +41,17 @@ DiskStorage.prototype._handleFile = function _handleFile (req, file, cb) {
 
       var finalPath = path.join(destination, filename)
 
-      if (file.stream.destroyed) return
+      if (file.stream.destroyed) return cb(new MulterError('STREAM_DESTROYED'))
 
       var outStream = fs.createWriteStream(finalPath)
 
       file.path = finalPath
+      openStreams.set(file, outStream)
+      outStream.once('close', function () { openStreams.delete(file) })
+
+      pipeline(file.stream, outStream, function (err) {
+        if (err) return cb(err)
 
-      file.stream.pipe(outStream)
-      outStream.on('error', cb)
-      outStream.on('finish', function () {
         cb(null, {
           destination: destination,
           filename: filename,
@@ -62,7 +70,13 @@ DiskStorage.prototype._removeFile = function _removeFile (req, file, cb) {
   delete file.filename
   delete file.path
 
-  fs.unlink(path, cb)
+  var outStream = openStreams.get(file)
+  if (!outStream) return fs.unlink(path, cb)
+
+  // Wait for the descriptor to be released before unlinking; destroy() is a
+  // no-op if the stream is already being torn down.
+  outStream.once('close', function () { fs.unlink(path, cb) })
+  outStream.destroy()
 }
 
 module.exports = function (opts) {

test/abort-destroyed-stream-callback.js

@@ -0,0 +1,47 @@
+/* eslint-env mocha */
+
+var stream = require('stream')
+
+var multer = require('../')
+var rimraf = require('rimraf')
+var temp = require('fs-temp')
+
+// @see https://github.com/expressjs/multer/security/advisories/GHSA-qfvm-cv95-jqjf
+
+describe('disk storage _handleFile with an already-destroyed source stream', function () {
+  var uploadDir
+
+  beforeEach(function (done) {
+    temp.mkdir(function (err, dir) {
+      if (err) return done(err)
+      uploadDir = dir
+      done()
+    })
+  })
+
+  afterEach(function (done) {
+    rimraf(uploadDir, done)
+  })
+
+  it('should always invoke the callback', function (done) {
+    this.timeout(2000)
+
+    var storage = multer.diskStorage({ destination: uploadDir })
+
+    var source = new stream.PassThrough()
+    source.destroy()
+
+    var file = {
+      fieldname: 'file',
+      originalname: 'destroyed.bin',
+      stream: source
+    }
+
+    // If the callback is never invoked (the pre-write guard returning without
+    // settling), mocha fails this test with a timeout. A double invocation is
+    // caught by mocha's "done() called multiple times".
+    storage._handleFile({}, file, function () {
+      done()
+    })
+  })
+})

test/abort-fd-leak.js

@@ -0,0 +1,133 @@
+/* eslint-env mocha */
+
+var assert = require('assert')
+var fs = require('fs')
+var path = require('path')
+var http = require('http')
+
+var express = require('express')
+var multer = require('../')
+var rimraf = require('rimraf')
+var temp = require('fs-temp')
+
+// @see https://github.com/expressjs/multer/security/advisories/GHSA-qfvm-cv95-jqjf
+
+// Count open file descriptors that point at a now-deleted file inside `dir`.
+// Linux exposes these via /proc/self/fd as symlinks whose target ends with
+// " (deleted)". Returns -1 when /proc/self/fd is unavailable (non-Linux).
+function countLeakedFds (dir) {
+  var fdDir = '/proc/self/fd'
+  var entries
+
+  try {
+    entries = fs.readdirSync(fdDir)
+  } catch (e) {
+    return -1
+  }
+
+  var leaked = 0
+
+  for (var i = 0; i < entries.length; i++) {
+    var target
+
+    try {
+      target = fs.readlinkSync(path.join(fdDir, entries[i]))
+    } catch (e) {
+      continue
+    }
+
+    if (target.indexOf(dir) === 0 && / \(deleted\)$/.test(target)) {
+      leaked++
+    }
+  }
+
+  return leaked
+}
+
+describe('file descriptor leak on aborted uploads', function () {
+  var uploadDir, server, port
+
+  beforeEach(function (done) {
+    temp.mkdir(function (err, dir) {
+      if (err) return done(err)
+
+      uploadDir = dir
+      var upload = multer({ dest: dir })
+      var app = express()
+
+      app.post('/upload', upload.single('file'), function (req, res) {
+        res.json({ success: true })
+      })
+
+      app.use(function (err, req, res, next) {
+        res.status(400).json({ error: err.message || err.code })
+      })
+
+      server = app.listen(0, function () {
+        port = server.address().port
+        done()
+      })
+    })
+  })
+
+  afterEach(function (done) {
+    server.close(function () {
+      rimraf(uploadDir, done)
+    })
+  })
+
+  it('should not leak file descriptors when uploads abort mid-write', function (done) {
+    this.timeout(20000)
+
+    if (countLeakedFds(uploadDir) === -1) return this.skip()
+
+    var attempts = 15
+
+    function abortOnce (next) {
+      var boundary = 'FdLeakBound' + Date.now() + Math.random().toString(16).slice(2)
+      var preamble =
+        '--' + boundary + '\r\n' +
+        'Content-Disposition: form-data; name="file"; filename="leak.bin"\r\n' +
+        'Content-Type: application/octet-stream\r\n\r\n'
+      var chunk = Buffer.alloc(64 * 1024, 0x5a)
+
+      var req = http.request({
+        hostname: 'localhost',
+        port: port,
+        path: '/upload',
+        method: 'POST',
+        headers: {
+          'Content-Type': 'multipart/form-data; boundary=' + boundary,
+          'Content-Length': Buffer.byteLength(preamble) + (chunk.length * 10)
+        }
+      })
+
+      req.on('error', function () {})
+      req.write(preamble)
+      req.write(chunk)
+
+      setTimeout(function () {
+        req.destroy()
+        setTimeout(next, 100)
+      }, 30)
+    }
+
+    var i = 0
+
+    function loop () {
+      if (i++ >= attempts) {
+        // give multer's abort cleanup time to run
+        setTimeout(function () {
+          var leaked = countLeakedFds(uploadDir)
+          assert.strictEqual(leaked, 0, 'leaked ' + leaked + ' deleted-file descriptor(s) after ' + attempts + ' aborted uploads')
+          done()
+        }, 500)
+        return
+      }
+
+      abortOnce(loop)
+    }
+
+    loop()
+  })
+})