[ASM] multer instrumentation

.github/workflows/appsec.yml

@@ -122,7 +122,7 @@ jobs:
   express:
     runs-on: ubuntu-latest
     env:
-      PLUGINS: express|body-parser|cookie-parser
+      PLUGINS: express|body-parser|cookie-parser|multer
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/node/setup

.github/workflows/plugins.yml

@@ -366,7 +366,7 @@ jobs:
   express:
     runs-on: ubuntu-latest
     env:
-      PLUGINS: express|body-parser|cookie-parser
+      PLUGINS: express|body-parser|cookie-parser|multer
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/plugins/test

integration-tests/multer.spec.js

@@ -0,0 +1,69 @@
+'use strict'
+
+const { assert } = require('chai')
+const path = require('path')
+const axios = require('axios')
+
+const {
+  createSandbox,
+  FakeAgent,
+  spawnProc
+} = require('./helpers')
+
+describe('Suspicious request blocking - multer', () => {
+  let sandbox, cwd, startupTestFile, agent, proc, env
+
+  ['1.4.4-lts.1', '1.4.5-lts.1'].forEach((version) => {
+    describe(`v${version}`, () => {
+      before(async () => {
+        sandbox = await createSandbox(['express', `multer@${version}`])
+        cwd = sandbox.folder
+        startupTestFile = path.join(cwd, 'multer/index.js')
+      })
+
+      after(async () => {
+        await sandbox.remove()
+      })
+
+      beforeEach(async () => {
+        agent = await new FakeAgent().start()
+
+        env = {
+          AGENT_PORT: agent.port,
+          DD_APPSEC_RULES: path.join(cwd, 'multer/body-parser-rules.json')
+        }
+
+        const execArgv = []
+
+        proc = await spawnProc(startupTestFile, { cwd, env, execArgv })
+      })
+
+      afterEach(async () => {
+        proc.kill()
+        await agent.stop()
+      })
+
+      it('should not block the request without an attack', async () => {
+        const form = new FormData()
+        form.append('key', 'value')
+
+        const res = await axios.post(proc.url, form)
+
+        assert.equal(res.data, 'DONE')
+      })
+
+      it('should block the request when attack is detected', async () => {
+        try {
+          const form = new FormData()
+          form.append('key', 'testattack')
+
+          await axios.post(proc.url, form)
+
+          return Promise.reject(new Error('Request should not return 200'))
+        } catch (e) {
+          assert.equal(e.response.status, 403)
+        }
+      })
+    })
+  })
+})

integration-tests/multer/body-parser-rules.json

@@ -0,0 +1,33 @@
+{
+  "version": "2.2",
+  "metadata": {
+    "rules_version": "1.5.0"
+  },
+  "rules": [
+    {
+      "id": "test-rule-id-1",
+      "name": "test-rule-name-1",
+      "tags": {
+        "type": "security_scanner",
+        "category": "attack_attempt"
+      },
+      "conditions": [
+        {
+          "parameters": {
+            "inputs": [
+              {
+                "address": "server.request.body"
+              }
+            ],
+            "list": [
+              "testattack"
+            ]
+          },
+          "operator": "phrase_match"
+        }
+      ],
+      "transformers": ["lowercase"],
+      "on_match": ["block"]
+    }
+  ]
+}

integration-tests/multer/index.js

@@ -0,0 +1,38 @@
+'use strict'
+
+const options = {
+  appsec: {
+    enabled: true
+  }
+}
+
+if (process.env.AGENT_PORT) {
+  options.port = process.env.AGENT_PORT
+}
+
+if (process.env.AGENT_URL) {
+  options.url = process.env.AGENT_URL
+}
+
+const tracer = require('dd-trace')
+tracer.init(options)
+
+const http = require('http')
+const express = require('express')
+const multer = require('multer')
+const uploadToMemory = multer({ storage: multer.memoryStorage(), limits: { fileSize: 200000 } })
+
+const app = express()
+
+app.post('/', uploadToMemory.single('file'), (req, res) => {
+  res.end('DONE')
+})
+
+app.get('/', (req, res) => {
+  res.status(200).send('hello world')
+})
+
+const server = http.createServer(app).listen(0, () => {
+  const port = server.address().port
+  process.send?.({ port })
+})

packages/datadog-instrumentations/src/helpers/hooks.js

@@ -79,6 +79,7 @@ module.exports = {
   'mongodb-core': () => require('../mongodb-core'),
   mongoose: () => require('../mongoose'),
   mquery: () => require('../mquery'),
+  multer: () => require('../multer'),
   mysql: () => require('../mysql'),
   mysql2: () => require('../mysql2'),
   net: () => require('../net'),

packages/datadog-instrumentations/src/multer.js

@@ -0,0 +1,37 @@
+'use strict'
+
+const shimmer = require('../../datadog-shimmer')
+const { channel, addHook, AsyncResource } = require('./helpers/instrument')
+
+const multerReadCh = channel('datadog:multer:read:finish')
+
+function publishRequestBodyAndNext (req, res, next) {
+  return shimmer.wrapFunction(next, next => function () {
+    if (multerReadCh.hasSubscribers && req) {
+      const abortController = new AbortController()
+      const body = req.body
+
+      multerReadCh.publish({ req, res, body, abortController })
+
+      if (abortController.signal.aborted) return
+    }
+
+    return next.apply(this, arguments)
+  })
+}
+
+addHook({
+  name: 'multer',
+  file: 'lib/make-middleware.js',
+  versions: ['^1.4.4-lts.1']
+}, makeMiddleware => {
+  return shimmer.wrapFunction(makeMiddleware, makeMiddleware => function () {
+    const middleware = makeMiddleware.apply(this, arguments)
+
+    return shimmer.wrapFunction(middleware, middleware => function wrapMulterMiddleware (req, res, next) {
+      const nextResource = new AsyncResource('bound-anonymous-fn')
+      arguments[2] = nextResource.bind(publishRequestBodyAndNext(req, res, next))
+      return middleware.apply(this, arguments)
+    })
+  })
+})

packages/datadog-instrumentations/test/multer.spec.js

@@ -0,0 +1,108 @@
+'use strict'
+
+const dc = require('dc-polyfill')
+const axios = require('axios')
+const agent = require('../../dd-trace/test/plugins/agent')
+const { storage } = require('../../datadog-core')
+
+withVersions('multer', 'multer', version => {
+  describe('multer parser instrumentation', () => {
+    const multerReadCh = dc.channel('datadog:multer:read:finish')
+    let port, server, middlewareProcessBodyStub, formData
+
+    before(() => {
+      return agent.load(['http', 'express', 'multer'], { client: false })
+    })
+
+    before((done) => {
+      const express = require('../../../versions/express').get()
+      const multer = require(`../../../versions/multer@${version}`).get()
+      const uploadToMemory = multer({ storage: multer.memoryStorage(), limits: { fileSize: 200000 } })
+
+      const app = express()
+
+      app.post('/', uploadToMemory.single('file'), (req, res) => {
+        middlewareProcessBodyStub(req.body.key)
+        res.end('DONE')
+      })
+      server = app.listen(0, () => {
+        port = server.address().port
+        done()
+      })
+    })
+
+    beforeEach(async () => {
+      middlewareProcessBodyStub = sinon.stub()
+
+      formData = new FormData()
+      formData.append('key', 'value')
+    })
+
+    after(() => {
+      server.close()
+      return agent.close({ ritmReset: false })
+    })
+
+    it('should not abort the request by default', async () => {
+      const res = await axios.post(`http://localhost:${port}/`, formData)
+
+      expect(middlewareProcessBodyStub).to.be.calledOnceWithExactly(formData.get('key'))
+      expect(res.data).to.be.equal('DONE')
+    })
+
+    it('should not abort the request with non blocker subscription', async () => {
+      function noop () {}
+      multerReadCh.subscribe(noop)
+
+      try {
+        const res = await axios.post(`http://localhost:${port}/`, formData)
+
+        expect(middlewareProcessBodyStub).to.be.calledOnceWithExactly(formData.get('key'))
+        expect(res.data).to.be.equal('DONE')
+      } finally {
+        multerReadCh.unsubscribe(noop)
+      }
+    })
+
+    it('should abort the request when abortController.abort() is called', async () => {
+      function blockRequest ({ res, abortController }) {
+        res.end('BLOCKED')
+        abortController.abort()
+      }
+      multerReadCh.subscribe(blockRequest)
+
+      try {
+        const res = await axios.post(`http://localhost:${port}/`, formData)
+
+        expect(middlewareProcessBodyStub).not.to.be.called
+        expect(res.data).to.be.equal('BLOCKED')
+      } finally {
+        multerReadCh.unsubscribe(blockRequest)
+      }
+    })
+
+    it('should not lose the http async context', async () => {
+      let store
+      let payload
+
+      function handler (data) {
+        store = storage.getStore()
+        payload = data
+      }
+      multerReadCh.subscribe(handler)
+
+      try {
+        const res = await axios.post(`http://localhost:${port}/`, formData)
+
+        expect(store).to.have.property('req', payload.req)
+        expect(store).to.have.property('res', payload.res)
+        expect(store).to.have.property('span')
+
+        expect(middlewareProcessBodyStub).to.be.calledOnceWithExactly(formData.get('key'))
+        expect(res.data).to.be.equal('DONE')
+      } finally {
+        multerReadCh.unsubscribe(handler)
+      }
+    })
+  })
+})

packages/dd-trace/src/appsec/channels.js

@@ -6,6 +6,7 @@ const dc = require('dc-polyfill')
 module.exports = {
   bodyParser: dc.channel('datadog:body-parser:read:finish'),
   cookieParser: dc.channel('datadog:cookie-parser:read:finish'),
+  multerParser: dc.channel('datadog:multer:read:finish'),
   startGraphqlResolve: dc.channel('datadog:graphql:resolver:start'),
   graphqlMiddlewareChannel: dc.tracingChannel('datadog:apollo:middleware'),
   apolloChannel: dc.tracingChannel('datadog:apollo:request'),

packages/dd-trace/src/appsec/iast/taint-tracking/plugin.js

@@ -26,15 +26,22 @@ class TaintTrackingPlugin extends SourceIastPlugin {
   }
 
   onConfigure () {
+    const onRequestBody = ({ req }) => {
+      const iastContext = getIastContext(storage.getStore())
+      if (iastContext && iastContext.body !== req.body) {
+        this._taintTrackingHandler(HTTP_REQUEST_BODY, req, 'body', iastContext)
+        iastContext.body = req.body
+      }
+    }
+
     this.addSub(
       { channelName: 'datadog:body-parser:read:finish', tag: HTTP_REQUEST_BODY },
-      ({ req }) => {
-        const iastContext = getIastContext(storage.getStore())
-        if (iastContext && iastContext.body !== req.body) {
-          this._taintTrackingHandler(HTTP_REQUEST_BODY, req, 'body', iastContext)
-          iastContext.body = req.body
-        }
-      }
+      onRequestBody
+    )
+
+    this.addSub(
+      { channelName: 'datadog:multer:read:finish', tag: HTTP_REQUEST_BODY },
+      onRequestBody
     )
 
     this.addSub(

packages/dd-trace/src/appsec/index.js

@@ -6,6 +6,7 @@ const remoteConfig = require('./remote_config')
 const {
   bodyParser,
   cookieParser,
+  multerParser,
   incomingHttpRequestStart,
   incomingHttpRequestEnd,
   passportVerify,
@@ -58,6 +59,7 @@ function enable (_config) {
     apiSecuritySampler.configure(_config.appsec)
 
     bodyParser.subscribe(onRequestBodyParsed)
+    multerParser.subscribe(onRequestBodyParsed)
     cookieParser.subscribe(onRequestCookieParser)
     incomingHttpRequestStart.subscribe(incomingHttpStartTranslator)
     incomingHttpRequestEnd.subscribe(incomingHttpEndTranslator)
@@ -299,6 +301,7 @@ function disable () {
 
   // Channel#unsubscribe() is undefined for non active channels
   if (bodyParser.hasSubscribers) bodyParser.unsubscribe(onRequestBodyParsed)
+  if (multerParser.hasSubscribers) multerParser.unsubscribe(onRequestBodyParsed)
   if (cookieParser.hasSubscribers) cookieParser.unsubscribe(onRequestCookieParser)
   if (incomingHttpRequestStart.hasSubscribers) incomingHttpRequestStart.unsubscribe(incomingHttpStartTranslator)
   if (incomingHttpRequestEnd.hasSubscribers) incomingHttpRequestEnd.unsubscribe(incomingHttpEndTranslator)

packages/dd-trace/test/appsec/iast/taint-tracking/plugin.spec.js

@@ -42,13 +42,14 @@ describe('IAST Taint tracking plugin', () => {
   })
 
   it('Should subscribe to body parser, qs, cookie and process_params channel', () => {
-    expect(taintTrackingPlugin._subscriptions).to.have.lengthOf(6)
+    expect(taintTrackingPlugin._subscriptions).to.have.lengthOf(7)
     expect(taintTrackingPlugin._subscriptions[0]._channel.name).to.equals('datadog:body-parser:read:finish')
-    expect(taintTrackingPlugin._subscriptions[1]._channel.name).to.equals('datadog:qs:parse:finish')
-    expect(taintTrackingPlugin._subscriptions[2]._channel.name).to.equals('apm:express:middleware:next')
-    expect(taintTrackingPlugin._subscriptions[3]._channel.name).to.equals('datadog:cookie:parse:finish')
-    expect(taintTrackingPlugin._subscriptions[4]._channel.name).to.equals('datadog:express:process_params:start')
-    expect(taintTrackingPlugin._subscriptions[5]._channel.name).to.equals('apm:graphql:resolve:start')
+    expect(taintTrackingPlugin._subscriptions[1]._channel.name).to.equals('datadog:multer:read:finish')
+    expect(taintTrackingPlugin._subscriptions[2]._channel.name).to.equals('datadog:qs:parse:finish')
+    expect(taintTrackingPlugin._subscriptions[3]._channel.name).to.equals('apm:express:middleware:next')
+    expect(taintTrackingPlugin._subscriptions[4]._channel.name).to.equals('datadog:cookie:parse:finish')
+    expect(taintTrackingPlugin._subscriptions[5]._channel.name).to.equals('datadog:express:process_params:start')
+    expect(taintTrackingPlugin._subscriptions[6]._channel.name).to.equals('apm:graphql:resolve:start')
   })
 
   describe('taint sources', () => {