Add blocking for fastify query and body (#5806)

* Add blocking for fastify query and body

* query params channel name

* sort channels

* add fastify appsec test

* fix http header fingerprint

* change rules name

* fix fingerprint test

* add schema validation tests

* linter

Author: IlyasShabi

.github/workflows/appsec.yml

@@ -128,6 +128,19 @@ jobs:
       - run: yarn test:appsec:plugins:ci
       - uses: codecov/codecov-action@ad3126e916f78f00edff4ed0317cf185271ccc2d # v5.4.2
 
+  fastify:
+    runs-on: ubuntu-latest
+    env:
+      PLUGINS: fastify
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - uses: ./.github/actions/node/oldest-maintenance-lts
+      - uses: ./.github/actions/install
+      - run: yarn test:appsec:plugins:ci
+      - uses: ./.github/actions/node/active-lts
+      - run: yarn test:appsec:plugins:ci
+      - uses: codecov/codecov-action@ad3126e916f78f00edff4ed0317cf185271ccc2d # v5.4.2
+
   graphql:
     runs-on: ubuntu-latest
     env:

packages/datadog-instrumentations/src/fastify.js

@@ -6,6 +6,8 @@ const { addHook, channel, AsyncResource } = require('./helpers/instrument')
 const errorChannel = channel('apm:fastify:middleware:error')
 const handleChannel = channel('apm:fastify:request:handle')
 const routeAddedChannel = channel('apm:fastify:route:added')
+const bodyParserReadCh = channel('datadog:fastify:body-parser:finish')
+const queryParamsReadCh = channel('datadog:fastify:query-params:finish')
 
 const parsingResources = new WeakMap()
 
@@ -106,11 +108,37 @@ function preHandler (request, reply, done) {
 
 function preValidation (request, reply, done) {
   const req = getReq(request)
+  const res = getRes(reply)
   const parsingResource = parsingResources.get(req)
 
-  if (!parsingResource) return done()
+  const processInContext = () => {
+    if (queryParamsReadCh.hasSubscribers && request.query) {
+      const abortController = new AbortController()
 
-  parsingResource.runInAsyncScope(() => done())
+      queryParamsReadCh.publish({
+        req,
+        res,
+        abortController,
+        query: request.query
+      })
+
+      if (abortController.signal.aborted) return
+    }
+
+    if (bodyParserReadCh.hasSubscribers && request.body) {
+      const abortController = new AbortController()
+
+      bodyParserReadCh.publish({ req, res, body: request.body, abortController })
+
+      if (abortController.signal.aborted) return
+    }
+
+    done()
+  }
+
+  if (!parsingResource) return processInContext()
+
+  parsingResource.runInAsyncScope(processInContext)
 }
 
 function preParsing (request, reply, payload, done) {

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

@@ -4,34 +4,36 @@ const dc = require('dc-polyfill')
 
 // TODO: use TBD naming convention
 module.exports = {
+  apolloChannel: dc.tracingChannel('datadog:apollo:request'),
+  apolloServerCoreChannel: dc.tracingChannel('datadog:apollo-server-core:request'),
   bodyParser: dc.channel('datadog:body-parser:read:finish'),
+  childProcessExecutionTracingChannel: dc.tracingChannel('datadog:child_process:execution'),
   cookieParser: dc.channel('datadog:cookie-parser:read:finish'),
-  multerParser: dc.channel('datadog:multer:read:finish'),
-  startGraphqlResolve: dc.channel('datadog:graphql:resolver:start'),
+  expressMiddlewareError: dc.channel('apm:express:middleware:error'),
+  expressProcessParams: dc.channel('datadog:express:process_params:start'),
+  expressSession: dc.channel('datadog:express-session:middleware:finish'),
+  fastifyBodyParser: dc.channel('datadog:fastify:body-parser:finish'),
+  fastifyQueryParams: dc.channel('datadog:fastify:query-params:finish'),
+  fsOperationStart: dc.channel('apm:fs:operation:start'),
   graphqlMiddlewareChannel: dc.tracingChannel('datadog:apollo:middleware'),
-  apolloChannel: dc.tracingChannel('datadog:apollo:request'),
-  apolloServerCoreChannel: dc.tracingChannel('datadog:apollo-server-core:request'),
-  incomingHttpRequestStart: dc.channel('dd-trace:incomingHttpRequestStart'),
+  httpClientRequestStart: dc.channel('apm:http:client:request:start'),
   incomingHttpRequestEnd: dc.channel('dd-trace:incomingHttpRequestEnd'),
-  passportVerify: dc.channel('datadog:passport:verify:finish'),
-  passportUser: dc.channel('datadog:passport:deserializeUser:finish'),
-  expressSession: dc.channel('datadog:express-session:middleware:finish'),
-  queryParser: dc.channel('datadog:query:read:finish'),
-  setCookieChannel: dc.channel('datadog:iast:set-cookie'),
+  incomingHttpRequestStart: dc.channel('dd-trace:incomingHttpRequestStart'),
+  multerParser: dc.channel('datadog:multer:read:finish'),
+  mysql2OuterQueryStart: dc.channel('datadog:mysql2:outerquery:start'),
   nextBodyParsed: dc.channel('apm:next:body-parsed'),
   nextQueryParsed: dc.channel('apm:next:query-parsed'),
-  expressProcessParams: dc.channel('datadog:express:process_params:start'),
-  routerParam: dc.channel('datadog:router:param:start'),
+  passportUser: dc.channel('datadog:passport:deserializeUser:finish'),
+  passportVerify: dc.channel('datadog:passport:verify:finish'),
+  pgPoolQueryStart: dc.channel('datadog:pg:pool:query:start'),
+  pgQueryStart: dc.channel('apm:pg:query:start'),
+  queryParser: dc.channel('datadog:query:read:finish'),
   responseBody: dc.channel('datadog:express:response:json:start'),
-  responseWriteHead: dc.channel('apm:http:server:response:writeHead:start'),
-  httpClientRequestStart: dc.channel('apm:http:client:request:start'),
   responseSetHeader: dc.channel('datadog:http:server:response:set-header:start'),
+  responseWriteHead: dc.channel('apm:http:server:response:writeHead:start'),
+  routerParam: dc.channel('datadog:router:param:start'),
+  setCookieChannel: dc.channel('datadog:iast:set-cookie'),
   setUncaughtExceptionCaptureCallbackStart: dc.channel('datadog:process:setUncaughtExceptionCaptureCallback:start'),
-  pgQueryStart: dc.channel('apm:pg:query:start'),
-  pgPoolQueryStart: dc.channel('datadog:pg:pool:query:start'),
-  mysql2OuterQueryStart: dc.channel('datadog:mysql2:outerquery:start'),
-  wafRunFinished: dc.channel('datadog:waf:run:finish'),
-  fsOperationStart: dc.channel('apm:fs:operation:start'),
-  expressMiddlewareError: dc.channel('apm:express:middleware:error'),
-  childProcessExecutionTracingChannel: dc.tracingChannel('datadog:child_process:execution')
+  startGraphqlResolve: dc.channel('datadog:graphql:resolver:start'),
+  wafRunFinished: dc.channel('datadog:waf:run:finish')
 }

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

@@ -7,6 +7,7 @@ const {
   bodyParser,
   cookieParser,
   multerParser,
+  fastifyBodyParser,
   incomingHttpRequestStart,
   incomingHttpRequestEnd,
   passportVerify,
@@ -16,6 +17,7 @@ const {
   nextBodyParsed,
   nextQueryParsed,
   expressProcessParams,
+  fastifyQueryParams,
   responseBody,
   responseWriteHead,
   responseSetHeader,
@@ -67,6 +69,7 @@ function enable (_config) {
 
     bodyParser.subscribe(onRequestBodyParsed)
     multerParser.subscribe(onRequestBodyParsed)
+    fastifyBodyParser.subscribe(onRequestBodyParsed)
     cookieParser.subscribe(onRequestCookieParser)
     incomingHttpRequestStart.subscribe(incomingHttpStartTranslator)
     incomingHttpRequestEnd.subscribe(incomingHttpEndTranslator)
@@ -77,6 +80,7 @@ function enable (_config) {
     nextBodyParsed.subscribe(onRequestBodyParsed)
     nextQueryParsed.subscribe(onRequestQueryParsed)
     expressProcessParams.subscribe(onRequestProcessParams)
+    fastifyQueryParams.subscribe(onRequestQueryParsed)
     routerParam.subscribe(onRequestProcessParams)
     responseBody.subscribe(onResponseBody)
     responseWriteHead.subscribe(onResponseWriteHead)
@@ -357,6 +361,7 @@ function disable () {
   // Channel#unsubscribe() is undefined for non active channels
   if (bodyParser.hasSubscribers) bodyParser.unsubscribe(onRequestBodyParsed)
   if (multerParser.hasSubscribers) multerParser.unsubscribe(onRequestBodyParsed)
+  if (fastifyBodyParser.hasSubscribers) fastifyBodyParser.unsubscribe(onRequestBodyParsed)
   if (cookieParser.hasSubscribers) cookieParser.unsubscribe(onRequestCookieParser)
   if (incomingHttpRequestStart.hasSubscribers) incomingHttpRequestStart.unsubscribe(incomingHttpStartTranslator)
   if (incomingHttpRequestEnd.hasSubscribers) incomingHttpRequestEnd.unsubscribe(incomingHttpEndTranslator)
@@ -367,6 +372,7 @@ function disable () {
   if (nextBodyParsed.hasSubscribers) nextBodyParsed.unsubscribe(onRequestBodyParsed)
   if (nextQueryParsed.hasSubscribers) nextQueryParsed.unsubscribe(onRequestQueryParsed)
   if (expressProcessParams.hasSubscribers) expressProcessParams.unsubscribe(onRequestProcessParams)
+  if (fastifyQueryParams.hasSubscribers) fastifyQueryParams.unsubscribe(onRequestQueryParsed)
   if (routerParam.hasSubscribers) routerParam.unsubscribe(onRequestProcessParams)
   if (responseBody.hasSubscribers) responseBody.unsubscribe(onResponseBody)
   if (responseWriteHead.hasSubscribers) responseWriteHead.unsubscribe(onResponseWriteHead)

packages/dd-trace/test/appsec/attacker-fingerprinting.fastify.plugin.spec.js

@@ -0,0 +1,83 @@
+'use strict'
+
+const Axios = require('axios')
+const { assert } = require('chai')
+const getPort = require('get-port')
+const path = require('path')
+
+const agent = require('../plugins/agent')
+const appsec = require('../../src/appsec')
+const Config = require('../../src/config')
+
+withVersions('fastify', 'fastify', fastifyVersion => {
+  describe('Attacker fingerprinting', () => {
+    let server, axios
+
+    before(() => {
+      return agent.load(['fastify', 'http'], { client: false })
+    })
+
+    before((done) => {
+      const fastify = require(`../../../../versions/fastify@${fastifyVersion}`).get()
+
+      const app = fastify()
+
+      app.post('/', (request, reply) => {
+        reply.send('DONE')
+      })
+
+      getPort().then((port) => {
+        app.listen({ port }, () => {
+          axios = Axios.create({ baseURL: `http://localhost:${port}` })
+          done()
+        })
+        server = app.server
+      })
+    })
+
+    after(() => {
+      server.close()
+      return agent.close({ ritmReset: false })
+    })
+
+    beforeEach(() => {
+      appsec.enable(new Config(
+        {
+          appsec: {
+            enabled: true,
+            rules: path.join(__dirname, 'attacker-fingerprinting-rules.json')
+          }
+        }
+      ))
+    })
+
+    afterEach(() => {
+      appsec.disable()
+    })
+
+    it('should report http fingerprints', async () => {
+      await axios.post('/?key=testattack',
+        {
+          bodyParam: 'bodyValue'
+        },
+        {
+          headers: {
+            'User-Agent': 'test-user-agent',
+            headerName: 'headerValue',
+            'x-real-ip': '255.255.255.255'
+          }
+        }
+      )
+
+      await agent.assertSomeTraces((traces) => {
+        const span = traces[0][0]
+        assert.property(span.meta, '_dd.appsec.fp.http.header')
+        assert.equal(span.meta['_dd.appsec.fp.http.header'], 'hdr-0110000110-74c2908f-5-55682ec1')
+        assert.property(span.meta, '_dd.appsec.fp.http.network')
+        assert.equal(span.meta['_dd.appsec.fp.http.network'], 'net-1-0100000000')
+        assert.property(span.meta, '_dd.appsec.fp.http.endpoint')
+        assert.equal(span.meta['_dd.appsec.fp.http.endpoint'], 'http-post-8a5edab2-2c70e12b-be31090f')
+      })
+    })
+  })
+})

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

@@ -63,7 +63,7 @@ withVersions('express', 'express', version => {
       appsec.enable(new Config({
         appsec: {
           enabled: true,
-          rules: path.join(__dirname, 'express-rules.json')
+          rules: path.join(__dirname, 'rules-example.json')
         }
       }))
     })
@@ -214,7 +214,7 @@ withVersions('express', 'express', version => {
       appsec.enable(new Config({
         appsec: {
           enabled: true,
-          rules: path.join(__dirname, 'express-rules.json')
+          rules: path.join(__dirname, 'rules-example.json')
         }
       }))
     })