feat(appsec): API10: Instrumentation & analysis of downstream requests

Sends all the outgoing request and response data to the waf to analyze.

Author: IlyasShabi

integration-tests/helpers/fake-agent.js

@@ -225,7 +225,7 @@ module.exports = class FakeAgent extends EventEmitter {
    *     the function `fn` has finished running. If `fn` throws an error, the promise will be rejected once `timeout`
    *     is reached.
    */
-  assertTelemetryReceived (fn, requestType, timeout = 30_000, expectedMessageCount = 1) {
+  assertTelemetryReceived (fn, requestType, timeout = 30_000, expectedMessageCount = 1, resolveAtFirstSuccess = false) {
     if (typeof fn !== 'function') {
       expectedMessageCount = timeout
       timeout = requestType
@@ -259,7 +259,7 @@ module.exports = class FakeAgent extends EventEmitter {
       msgCount += 1
       try {
         fn(msg)
-        if (msgCount === expectedMessageCount) {
+        if (resolveAtFirstSuccess || msgCount === expectedMessageCount) {
           resultResolve()
         }
       } catch (e) {

packages/datadog-instrumentations/src/http/client.js

@@ -14,6 +14,7 @@ const finishChannel = channel('apm:http:client:request:finish')
 const endChannel = channel('apm:http:client:request:end')
 const asyncStartChannel = channel('apm:http:client:request:asyncStart')
 const errorChannel = channel('apm:http:client:request:error')
+const responseFinishChannel = channel('apm:http:client:response:finish')
 
 const names = ['http', 'https', 'node:http', 'node:https']
 
@@ -39,6 +40,112 @@ function normalizeCallback (inputOptions, callback, inputURL) {
   return typeof inputOptions === 'function' ? [inputOptions, inputURL || {}] : [callback, inputOptions]
 }
 
+/**
+ * Wires the downstream response so we can observe when the customer consumes
+ * the body and when the stream finishes
+ *
+ * @param {object} ctx - Instrumentation context
+ * @param {import('http').IncomingMessage} res - The downstream response object.
+ * @returns {{ finalizeIfNeeded: () => void }|null} Cleanup helper used for drain.
+ */
+function setupResponseInstrumentation (ctx, res) {
+  const shouldInstrumentFinish = responseFinishChannel.hasSubscribers
+
+  if (!shouldInstrumentFinish) {
+    return null
+  }
+
+  let bodyConsumed = false
+  let finishCalled = false
+  let originalRead = null
+  let dataListenerAdded = false
+  let dataReadStarted = false
+
+  const { shouldCollectBody } = ctx
+  const bodyChunks = shouldCollectBody ? [] : null
+
+  const collectChunk = chunk => {
+    if (!shouldCollectBody || !chunk) return
+
+    if (typeof chunk === 'string') {
+      bodyChunks.push(chunk)
+    } else if (Buffer.isBuffer(chunk)) {
+      bodyChunks.push(chunk)
+    } else {
+      // Handle Uint8Array or other array-like types
+      bodyChunks.push(Buffer.from(chunk))
+    }
+  }
+
+  // Listen for body consumption
+  const onNewListener = (eventName) => {
+    if (eventName === 'data' || eventName === 'readable') {
+      bodyConsumed = true
+
+      // For 'data' events, add our own listener to collect chunks
+      if (eventName === 'data' && !dataListenerAdded && !dataReadStarted) {
+        dataListenerAdded = true
+        res.on('data', collectChunk)
+      }
+
+      // For 'readable' events, wrap the read() method
+      if (eventName === 'readable' && !originalRead && !dataListenerAdded && typeof res.read === 'function') {
+        originalRead = res.read
+        res.read = function () {
+          const chunk = originalRead.apply(this, arguments)
+          if (!dataListenerAdded) {
+            dataReadStarted = true
+            collectChunk(chunk)
+          }
+          return chunk
+        }
+      }
+    }
+  }
+
+  res.on('newListener', onNewListener)
+
+  // Cleanup function to restore original behavior
+  const cleanup = () => {
+    res.off('newListener', onNewListener)
+    res.off('data', collectChunk)
+
+    if (originalRead) {
+      res.read = originalRead
+      originalRead = null
+    }
+  }
+
+  const notifyFinish = () => {
+    if (finishCalled) return
+    finishCalled = true
+
+    // Combine collected chunks into a single body
+    let body = null
+    if (bodyChunks?.length) {
+      const firstChunk = bodyChunks[0]
+      body = typeof firstChunk === 'string'
+        ? bodyChunks.join('')
+        : Buffer.concat(bodyChunks)
+    }
+
+    responseFinishChannel.publish({ ctx, res, body })
+    cleanup()
+  }
+
+  res.once('end', notifyFinish)
+  res.once('close', notifyFinish)
+
+  return {
+    finalizeIfNeeded () {
+      if (!bodyConsumed) {
+        // Body not consumed, resume to complete the response
+        notifyFinish()
+      }
+    },
+  }
+}
+
 function patch (http, methodName) {
   shimmer.wrap(http, methodName, instrumentRequest)
 
@@ -103,7 +210,18 @@ function patch (http, methodName) {
                 ctx.res = res
                 res.once('end', finish)
                 res.once(errorMonitor, finish)
-                break
+
+                const instrumentation = setupResponseInstrumentation(ctx, res)
+
+                if (!instrumentation) {
+                  break
+                }
+
+                const result = emit.apply(this, arguments)
+
+                instrumentation.finalizeIfNeeded()
+
+                return result
               }
               case 'connect':
               case 'upgrade':