Fastify RASP support

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

@@ -14,6 +14,8 @@ let defaultBlockingActionParameters
 
 const responseBlockedSet = new WeakSet()
 
+const blockDelegations = new WeakMap()
+
 const specificBlockingTypes = {
   GRAPHQL: 'graphql'
 }
@@ -99,6 +101,9 @@ function getBlockingData (req, specificType, actionParameters) {
 }
 
 function block (req, res, rootSpan, abortController, actionParameters = defaultBlockingActionParameters) {
+  // synchronous blocking overrides previously created delegations
+  blockDelegations.delete(res)
+
   try {
     if (res.headersSent) {
       log.warn('[ASM] Cannot send blocking response when headers have already been sent')
@@ -127,11 +132,33 @@ function block (req, res, rootSpan, abortController, actionParameters = defaultB
     rootSpan?.setTag('_dd.appsec.block.failed', 1)
     log.error('[ASM] Blocking error', err)
 
+    // TODO: if blocking fails, then the response will never be sent ?
+
     updateBlockFailureMetric(req)
     return false
   }
 }
 
+function registerBlockDelegation (req, res) {
+  const args = arguments
+
+  return new Promise((resolve) => {
+    // ignore subsequent delegations by never calling their resolve()
+    if (blockDelegations.has(res)) return
+
+    blockDelegations.set(res, { args, resolve })
+  })
+}
+
+function callBlockDelegation (res) {
+  const delegation = blockDelegations.get(res)
+  if (delegation) {
+    const result = block.apply(this, delegation.args)
+    delegation.resolve(result)
+    return result
+  }
+}
+
 function getBlockingAction (actions) {
   // waf only returns one action, but it prioritizes redirect over block
   return actions?.redirect_request || actions?.block_request
@@ -158,6 +185,8 @@ function setDefaultBlockingActionParameters (actions) {
 module.exports = {
   addSpecificEndpoint,
   block,
+  registerBlockDelegation,
+  callBlockDelegation,
   specificBlockingTypes,
   getBlockingData,
   getBlockingAction,

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

@@ -14,10 +14,11 @@ module.exports = {
   expressProcessParams: dc.channel('datadog:express:process_params:start'),
   expressSession: dc.channel('datadog:express-session:middleware:finish'),
   fastifyBodyParser: dc.channel('datadog:fastify:body-parser:finish'),
-  fastifyResponseChannel: dc.channel('datadog:fastify:response:finish'),
-  fastifyQueryParams: dc.channel('datadog:fastify:query-params:finish'),
   fastifyCookieParser: dc.channel('datadog:fastify-cookie:read:finish'),
+  fastifyMiddlewareError: dc.channel('apm:fastify:middleware:error'),
   fastifyPathParams: dc.channel('datadog:fastify:path-params:finish'),
+  fastifyQueryParams: dc.channel('datadog:fastify:query-params:finish'),
+  fastifyResponseChannel: dc.channel('datadog:fastify:response:finish'),
   fsOperationStart: dc.channel('apm:fs:operation:start'),
   graphqlMiddlewareChannel: dc.tracingChannel('datadog:apollo:middleware'),
   httpClientRequestStart: dc.channel('apm:http:client:request:start'),

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

@@ -34,7 +34,7 @@ const apiSecuritySampler = require('./api_security_sampler')
 const web = require('../plugins/util/web')
 const { extractIp } = require('../plugins/util/ip_extractor')
 const { HTTP_CLIENT_IP } = require('../../../../ext/tags')
-const { isBlocked, block, setTemplates, getBlockingAction } = require('./blocking')
+const { isBlocked, block, callBlockDelegation, setTemplates, getBlockingAction } = require('./blocking')
 const UserTracking = require('./user_tracking')
 const { storage } = require('../../../datadog-core')
 const graphql = require('./graphql')
@@ -306,8 +306,13 @@ function onResponseWriteHead ({ req, res, abortController, statusCode, responseH
     storedResponseHeaders.set(req, responseHeaders)
   }
 
+  // TODO: do not call waf if inside block()
+  // if (isBlocking()) {
+  //   return
+  // }
+
   // avoid "write after end" error
-  if (isBlocked(res)) {
+  if (isBlocked(res) || callBlockDelegation(res)) {
     abortController?.abort()
     return
   }

packages/dd-trace/src/appsec/rasp/fs-plugin.js

@@ -35,6 +35,7 @@ class AppsecFsPlugin extends Plugin {
     this.addBind('apm:fs:operation:finish', this._onFsOperationFinishOrRenderEnd)
     this.addBind('tracing:datadog:express:response:render:start', this._onResponseRenderStart)
     this.addBind('tracing:datadog:express:response:render:end', this._onFsOperationFinishOrRenderEnd)
+    // We might have to add the same subscribers for fastify later
 
     super.configure(true)
   }

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

@@ -1,8 +1,12 @@
 'use strict'
 
 const web = require('../../plugins/util/web')
-const { setUncaughtExceptionCaptureCallbackStart, expressMiddlewareError } = require('../channels')
-const { block, isBlocked } = require('../blocking')
+const {
+  setUncaughtExceptionCaptureCallbackStart,
+  expressMiddlewareError,
+  fastifyMiddlewareError
+} = require('../channels')
+const { block, registerBlockDelegation, isBlocked } = require('../blocking')
 const ssrf = require('./ssrf')
 const sqli = require('./sql_injection')
 const lfi = require('./lfi')
@@ -39,7 +43,7 @@ function findDatadogRaspAbortError (err, deep = 10) {
 }
 
 function handleUncaughtExceptionMonitor (error) {
-  if (!blockOnDatadogRaspAbortError({ error })) return
+  if (!blockOnDatadogRaspAbortError({ error, isTopLevel: true })) return
 
   if (process.hasUncaughtExceptionCaptureCallback()) {
     // uncaughtException event is not executed when hasUncaughtExceptionCaptureCallback is true
@@ -81,15 +85,21 @@ function handleUncaughtExceptionMonitor (error) {
   }
 }
 
-function blockOnDatadogRaspAbortError ({ error }) {
+function blockOnDatadogRaspAbortError ({ error, isTopLevel }) {
   const abortError = findDatadogRaspAbortError(error)
   if (!abortError) return false
 
   const { req, res, blockingAction, raspRule, ruleTriggered } = abortError
   if (!isBlocked(res)) {
-    const blocked = block(req, res, web.root(req), null, blockingAction)
+    const blockFn = isTopLevel ? block : registerBlockDelegation
+    const blocked = blockFn(req, res, web.root(req), null, blockingAction)
     if (ruleTriggered) {
-      updateRaspRuleMatchMetricTags(req, raspRule, true, blocked)
+      // block() returns a bool, and registerBlockDelegation() returns a promise
+      // we use Promise.resolve() to handle both cases
+      Promise.resolve(blocked).then(blocked => {
+        // TODO: bug: this should probably be called for each match even without block
+        updateRaspRuleMatchMetricTags(req, raspRule, true, blocked)
+      })
     }
   }
 
@@ -103,7 +113,9 @@ function enable (config) {
   cmdi.enable(config)
 
   process.on('uncaughtExceptionMonitor', handleUncaughtExceptionMonitor)
+
   expressMiddlewareError.subscribe(blockOnDatadogRaspAbortError)
+  fastifyMiddlewareError.subscribe(blockOnDatadogRaspAbortError)
 }
 
 function disable () {
@@ -113,7 +125,9 @@ function disable () {
   cmdi.disable()
 
   process.off('uncaughtExceptionMonitor', handleUncaughtExceptionMonitor)
-  if (expressMiddlewareError.hasSubscribers) expressMiddlewareError.unsubscribe(blockOnDatadogRaspAbortError)
+
+  expressMiddlewareError.unsubscribe(blockOnDatadogRaspAbortError)
+  fastifyMiddlewareError.unsubscribe(blockOnDatadogRaspAbortError)
 }
 
 module.exports = {

packages/dd-trace/src/appsec/rasp/utils.js

@@ -19,6 +19,11 @@ const RULE_TYPES = {
   SSRF: 'ssrf'
 }
 
+const ALLOWED_ROOTSPAN_NAMES = new Set([
+  'express.request',
+  'fastify.request'
+])
+
 class DatadogRaspAbortError extends Error {
   constructor (req, res, blockingAction, raspRule, ruleTriggered) {
     super('DatadogRaspAbortError')
@@ -28,6 +33,12 @@ class DatadogRaspAbortError extends Error {
     this.blockingAction = blockingAction
     this.raspRule = raspRule
     this.ruleTriggered = ruleTriggered
+
+    // hide these props to not pollute app logs
+    Object.defineProperties(this, {
+      req: { enumerable: false },
+      res: { enumerable: false }
+    })
   }
 }
 
@@ -52,9 +63,9 @@ function handleResult (result, req, res, abortController, config, raspRule) {
 
   if (abortController && !abortOnUncaughtException) {
     const blockingAction = getBlockingAction(result?.actions)
+    const rootSpanName = rootSpan?.context?.()?._name
 
-    // Should block only in express
-    if (blockingAction && rootSpan?.context()._name === 'express.request') {
+    if (blockingAction && ALLOWED_ROOTSPAN_NAMES.has(rootSpanName)) {
       const abortError = new DatadogRaspAbortError(req, res, blockingAction, raspRule, ruleTriggered)
       abortController.abort(abortError)
 

packages/dd-trace/test/appsec/blocking.spec.js

@@ -16,7 +16,7 @@ describe('blocking', () => {
   }
 
   let log, telemetry
-  let block, setTemplates
+  let block, registerBlockDelegation, callBlockDelegation, setTemplates
   let req, res, rootSpan
 
   beforeEach(() => {
@@ -35,6 +35,8 @@ describe('blocking', () => {
     })
 
     block = blocking.block
+    registerBlockDelegation = blocking.registerBlockDelegation
+    callBlockDelegation = blocking.callBlockDelegation
     setTemplates = blocking.setTemplates
 
     req = {
@@ -151,6 +153,102 @@ describe('blocking', () => {
     })
   })
 
+  describe('block delegation', () => {
+    it('should delegate block', (done) => {
+      setTemplates(config)
+
+      const abortController = new AbortController()
+      const promise = registerBlockDelegation(req, res, rootSpan, abortController)
+
+      expect(rootSpan.setTag).to.not.have.been.called
+      expect(res.writeHead).to.not.have.been.called
+      expect(res.constructor.prototype.end).to.not.have.been.called
+      expect(abortController.signal.aborted).to.be.false
+      expect(telemetry.updateBlockFailureMetric).to.not.have.been.called
+
+      const blocked = callBlockDelegation(res)
+
+      expect(blocked).to.be.true
+      expect(rootSpan.setTag).to.have.been.calledOnceWithExactly('appsec.blocked', 'true')
+      expect(res.writeHead).to.have.been.calledOnceWithExactly(403, {
+        'Content-Type': 'application/json',
+        'Content-Length': 8
+      })
+      expect(res.constructor.prototype.end).to.have.been.calledOnceWithExactly('jsonBody')
+      expect(abortController.signal.aborted).to.be.true
+      expect(telemetry.updateBlockFailureMetric).to.not.have.been.called
+
+      promise.then(blocked => {
+        expect(blocked).to.be.true
+        done()
+      })
+    })
+
+    it('should only resolve the first blocking delegation per request', (done) => {
+      const firstPromise = registerBlockDelegation(req, res, rootSpan)
+      const secondPromise = sinon.stub()
+      const thirdPromise = sinon.stub()
+      registerBlockDelegation(req, res, rootSpan).then(secondPromise)
+      registerBlockDelegation(req, res, rootSpan).then(thirdPromise)
+
+      const blocked = callBlockDelegation(res)
+
+      expect(blocked).to.be.true
+      expect(rootSpan.setTag).to.have.been.calledOnce
+      expect(res.writeHead).to.have.been.calledOnce
+      expect(res.constructor.prototype.end).to.have.been.calledOnce
+      expect(telemetry.updateBlockFailureMetric).to.not.have.been.called
+
+      firstPromise.then((blocked) => {
+        expect(blocked).to.be.true
+
+        setTimeout(() => {
+          expect(secondPromise).to.not.have.been.called
+          expect(thirdPromise).to.not.have.been.called
+          done()
+        }, 100)
+      })
+    })
+
+    it('should do nothing if no blocking delegation exists', () => {
+      const blocked = callBlockDelegation(res)
+
+      expect(blocked).to.not.be.ok
+      expect(log.warn).to.not.have.been.called
+      expect(rootSpan.setTag).to.not.have.been.called
+      expect(res.writeHead).to.not.have.been.called
+      expect(res.constructor.prototype.end).to.not.have.been.called
+      expect(telemetry.updateBlockFailureMetric).to.not.have.been.called
+    })
+
+    it('should cancel block delegations when block is called', (done) => {
+      const promise = sinon.stub()
+
+      registerBlockDelegation(req, res, rootSpan).then(promise)
+
+      const blocked = block(req, res, rootSpan)
+
+      expect(blocked).to.be.true
+      expect(rootSpan.setTag).to.have.been.calledOnce
+      expect(res.writeHead).to.have.been.calledOnce
+      expect(res.constructor.prototype.end).to.have.been.calledOnce
+      expect(telemetry.updateBlockFailureMetric).to.not.have.been.called
+
+      const result = callBlockDelegation(res)
+
+      expect(result).to.not.be.ok
+      expect(rootSpan.setTag).to.have.been.calledOnce
+      expect(res.writeHead).to.have.been.calledOnce
+      expect(res.constructor.prototype.end).to.have.been.calledOnce
+      expect(telemetry.updateBlockFailureMetric).to.not.have.been.called
+
+      setTimeout(() => {
+        expect(promise).to.not.have.been.called
+        done()
+      }, 100)
+    })
+  })
+
   describe('block with default templates', () => {
     const config = {
       appsec: {

packages/dd-trace/test/appsec/index.spec.js

@@ -98,7 +98,8 @@ describe('AppSec Index', function () {
     }
 
     blocking = {
-      setTemplates: sinon.stub()
+      setTemplates: sinon.stub(),
+      callBlockDelegation: sinon.stub()
     }
 
     UserTracking = {
@@ -1122,6 +1123,7 @@ describe('AppSec Index', function () {
         }, req)
         expect(abortController.abort).to.have.been.calledOnce
         expect(res.constructor.prototype.end).to.have.been.calledOnce
+        expect(blocking.callBlockDelegation).to.have.been.calledOnce
 
         abortController.abort.resetHistory()
 
@@ -1130,6 +1132,17 @@ describe('AppSec Index', function () {
         expect(waf.run).to.have.been.calledOnce
         expect(abortController.abort).to.have.been.calledOnce
         expect(res.constructor.prototype.end).to.have.been.calledOnce
+        expect(blocking.callBlockDelegation).to.have.been.calledOnce
+      })
+
+      it('should call abortController if blocking delegate is successful', () => {
+        blocking.callBlockDelegation.returns(true)
+
+        responseWriteHead.publish({ req, res, abortController, statusCode: 404, responseHeaders: {} })
+
+        expect(blocking.callBlockDelegation).to.have.been.calledOnceWithExactly(res)
+        expect(abortController.abort).to.have.been.calledOnce
+        expect(waf.run).to.not.have.been.called
       })
 
       it('should not call the WAF if response was already analyzed', () => {