feat(appsec): expose uploaded file content as new WAF address server.request.body.files_content for commons-fileupload (#11137)

## What Does This Do

  - Introduces the `server.request.body.files_content` address and `requestFilesContent()` event in the gateway API, wired through `GatewayBridge` and`InstrumentationGateway`
  - Extends `ServletFileUpload.parseRequest()` instrumentation (commons-fileupload) to read up to 4096 bytes of each uploaded file's content and fire the new WAF callback; blocks with a `BlockingException` on `RequestBlockingAction`; content event is skipped when the filenames event has already blocked the request

  ## Additional Info

-   Content is capped at 4096 bytes per file to keep memory usage bounded
-  Number of files is capped at 25 files
-   This PR covers the gateway wiring and the commons-fileupload entry point. Coverage for other multipart stacks (Tomcat `request.getParts()`, Jetty, Liberty) will follow in successive PRs

# Contributor Checklist

- Format the title according to [the contribution guidelines](https://github.com/DataDog/dd-trace-java/blob/master/CONTRIBUTING.md#title-format)
- Assign the `type:` and (`comp:` or `inst:`) labels in addition to [any other useful labels](https://github.com/DataDog/dd-trace-java/blob/master/CONTRIBUTING.md#labels)
- Avoid using `close`, `fix`, or [any linking keywords](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword) when referencing an issue  
  Use `solves` instead, and assign the PR [milestone](https://github.com/DataDog/dd-trace-java/milestones) to the issue
- Update the [CODEOWNERS](https://github.com/DataDog/dd-trace-java/blob/master/.github/CODEOWNERS) file on source file addition, migration, or deletion
- Update [public documentation](https://docs.datadoghq.com/tracing/trace_collection/library_config/java/) with any new configuration flags or behaviors

Jira ticket: [APPSEC-61875]

***Note:*** **Once your PR is ready to merge, add it to the merge queue by commenting `/merge`.** `/merge -c` cancels the queue request. `/merge -f --reason "reason"` skips all merge queue checks; please use this judiciously, as some checks do not run at the PR-level. For more information, see [this doc](https://datadoghq.atlassian.net/wiki/spaces/DEVX/pages/3121612126/MergeQueue).

<!--
# Opening vs Drafting a PR:
When opening a pull request, please open it as a draft to not auto assign reviewers before you feel the pull request is in a reviewable state.

# Linking a JIRA ticket:
Please link your JIRA ticket by adding its identifier between brackets (ex [PROJ-IDENT]) in the PR description, not the title.
This requirement only applies to Datadog employees.
-->

[APPSEC-61875]: https://datadoghq.atlassian.net/browse/APPSEC-61875?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Co-authored-by: devflow.devflow-routing-intake <devflow.devflow-routing-intake@kubernetes.us1.ddbuild.io>

Author: jandro996

dd-java-agent/appsec/src/main/java/com/datadog/appsec/event/data/KnownAddresses.java

@@ -74,6 +74,13 @@ public interface KnownAddresses {
   Address<Long> REQUEST_COMBINED_FILE_SIZE =
       new Address<>("server.request.body.combined_file_size");
 
+  /**
+   * Contains the content of each uploaded file in a multipart/form-data request. Content is
+   * truncated per file and only the first files are inspected. Available only on inspected
+   * multipart/form-data requests.
+   */
+  Address<List<String>> REQUEST_FILES_CONTENT = new Address<>("server.request.body.files_content");
+
   /**
    * The parsed query string.
    *
@@ -205,6 +212,8 @@ static Address<?> forName(String name) {
         return REQUEST_FILES_FILENAMES;
       case "server.request.body.combined_file_size":
         return REQUEST_COMBINED_FILE_SIZE;
+      case "server.request.body.files_content":
+        return REQUEST_FILES_CONTENT;
       case "server.request.query":
         return REQUEST_QUERY;
       case "server.request.headers.no_cookies":

dd-java-agent/appsec/src/main/java/com/datadog/appsec/gateway/GatewayBridge.java

@@ -131,6 +131,7 @@ public class GatewayBridge {
   private volatile DataSubscriberInfo execCmdSubInfo;
   private volatile DataSubscriberInfo shellCmdSubInfo;
   private volatile DataSubscriberInfo requestFilesFilenamesSubInfo;
+  private volatile DataSubscriberInfo requestFilesContentSubInfo;
 
   public GatewayBridge(
       SubscriptionService subscriptionService,
@@ -208,6 +209,10 @@ public void init() {
       subscriptionService.registerCallback(
           EVENTS.requestFilesFilenames(), this::onRequestFilesFilenames);
     }
+    if (additionalIGEvents.contains(EVENTS.requestFilesContent())) {
+      subscriptionService.registerCallback(
+          EVENTS.requestFilesContent(), this::onRequestFilesContent);
+    }
   }
 
   /**
@@ -235,6 +240,7 @@ public void reset() {
     execCmdSubInfo = null;
     shellCmdSubInfo = null;
     requestFilesFilenamesSubInfo = null;
+    requestFilesContentSubInfo = null;
   }
 
   private Flow<Void> onUser(final RequestContext ctx_, final String user) {
@@ -605,6 +611,31 @@ private Flow<Void> onRequestFilesFilenames(RequestContext ctx_, List<String> fil
     }
   }
 
+  private Flow<Void> onRequestFilesContent(RequestContext ctx_, List<String> filesContent) {
+    AppSecRequestContext ctx = ctx_.getData(RequestContextSlot.APPSEC);
+    if (ctx == null || filesContent == null || filesContent.isEmpty()) {
+      return NoopFlow.INSTANCE;
+    }
+    while (true) {
+      DataSubscriberInfo subInfo = requestFilesContentSubInfo;
+      if (subInfo == null) {
+        subInfo = producerService.getDataSubscribers(KnownAddresses.REQUEST_FILES_CONTENT);
+        requestFilesContentSubInfo = subInfo;
+      }
+      if (subInfo == null || subInfo.isEmpty()) {
+        return NoopFlow.INSTANCE;
+      }
+      DataBundle bundle =
+          new SingletonDataBundle<>(KnownAddresses.REQUEST_FILES_CONTENT, filesContent);
+      try {
+        GatewayContext gwCtx = new GatewayContext(false);
+        return producerService.publishDataEvent(subInfo, ctx, bundle, gwCtx);
+      } catch (ExpiredSubscriberInfoException e) {
+        requestFilesContentSubInfo = null;
+      }
+    }
+  }
+
   private Flow<Void> onDatabaseSqlQuery(RequestContext ctx_, String sql) {
     AppSecRequestContext ctx = ctx_.getData(RequestContextSlot.APPSEC);
     if (ctx == null) {
@@ -1464,6 +1495,7 @@ private static class IGAppSecEventDependencies {
       DATA_DEPENDENCIES.put(KnownAddresses.REQUEST_BODY_OBJECT, l(EVENTS.requestBodyProcessed()));
       DATA_DEPENDENCIES.put(
           KnownAddresses.REQUEST_FILES_FILENAMES, l(EVENTS.requestFilesFilenames()));
+      DATA_DEPENDENCIES.put(KnownAddresses.REQUEST_FILES_CONTENT, l(EVENTS.requestFilesContent()));
     }
 
     private static Collection<datadog.trace.api.gateway.EventType<?>> l(

dd-java-agent/appsec/src/test/groovy/com/datadog/appsec/event/data/KnownAddressesSpecificationForkedTest.groovy

@@ -26,6 +26,7 @@ class KnownAddressesSpecificationForkedTest extends Specification {
       'server.request.body.files_field_names',
       'server.request.body.filenames',
       'server.request.body.combined_file_size',
+      'server.request.body.files_content',
       'server.request.query',
       'server.request.headers.no_cookies',
       'grpc.server.method',
@@ -58,7 +59,7 @@ class KnownAddressesSpecificationForkedTest extends Specification {
 
   void 'number of known addresses is expected number'() {
     expect:
-    Address.instanceCount() == 46
+    Address.instanceCount() == 47
     KnownAddresses.WAF_CONTEXT_PROCESSOR.serial == Address.instanceCount() - 1
   }
 }

dd-java-agent/appsec/src/test/groovy/com/datadog/appsec/gateway/GatewayBridgeIGRegistrationSpecification.groovy

@@ -34,4 +34,15 @@ class GatewayBridgeIGRegistrationSpecification extends DDSpecification {
     then:
     1 * ig.registerCallback(Events.REQUEST_BODY_DONE, _)
   }
+
+  void 'requestFilesContent is registered via data address'() {
+    given:
+    1 * eventDispatcher.allSubscribedDataAddresses() >> [KnownAddresses.REQUEST_FILES_CONTENT]
+
+    when:
+    bridge.init()
+
+    then:
+    1 * ig.registerCallback(Events.get().requestFilesContent(), _)
+  }
 }

dd-java-agent/appsec/src/test/groovy/com/datadog/appsec/gateway/GatewayBridgeSpecification.groovy

@@ -123,6 +123,7 @@ class GatewayBridgeSpecification extends DDSpecification {
   BiFunction<RequestContext, String, Flow<Void>> fileLoadedCB
   BiFunction<RequestContext, String, Flow<Void>> fileWrittenCB
   BiFunction<RequestContext, List<String>, Flow<Void>> requestFilesFilenamesCB
+  BiFunction<RequestContext, List<String>, Flow<Void>> requestFilesContentCB
   BiFunction<RequestContext, String, Flow<Void>> requestSessionCB
   BiFunction<RequestContext, String[], Flow<Void>> execCmdCB
   BiFunction<RequestContext, String, Flow<Void>> shellCmdCB
@@ -463,7 +464,7 @@ class GatewayBridgeSpecification extends DDSpecification {
 
   void callInitAndCaptureCBs() {
     // force all callbacks to be registered
-    _ * eventDispatcher.allSubscribedDataAddresses() >> [KnownAddresses.REQUEST_PATH_PARAMS, KnownAddresses.REQUEST_BODY_OBJECT, KnownAddresses.REQUEST_FILES_FILENAMES]
+    _ * eventDispatcher.allSubscribedDataAddresses() >> [KnownAddresses.REQUEST_PATH_PARAMS, KnownAddresses.REQUEST_BODY_OBJECT, KnownAddresses.REQUEST_FILES_FILENAMES, KnownAddresses.REQUEST_FILES_CONTENT]
 
     1 * ig.registerCallback(EVENTS.requestStarted(), _) >> {
       requestStartedCB = it[1]; null
@@ -561,6 +562,9 @@ class GatewayBridgeSpecification extends DDSpecification {
     1 * ig.registerCallback(EVENTS.requestFilesFilenames(), _) >> {
       requestFilesFilenamesCB = it[1]; null
     }
+    1 * ig.registerCallback(EVENTS.requestFilesContent(), _) >> {
+      requestFilesContentCB = it[1]; null
+    }
     0 * ig.registerCallback(_, _)
 
     bridge.init()
@@ -1142,6 +1146,38 @@ class GatewayBridgeSpecification extends DDSpecification {
     0 * eventDispatcher.publishDataEvent(*_)
   }
 
+  void 'process request files content'() {
+    setup:
+    final filesContent = ['%PDF-1.4 malicious content', '#!/bin/bash\nrm -rf /']
+    eventDispatcher.getDataSubscribers({
+      KnownAddresses.REQUEST_FILES_CONTENT in it
+    }) >> nonEmptyDsInfo
+    DataBundle bundle
+    GatewayContext gatewayContext
+
+    when:
+    Flow<?> flow = requestFilesContentCB.apply(ctx, filesContent)
+
+    then:
+    1 * eventDispatcher.publishDataEvent(nonEmptyDsInfo, ctx.data, _ as DataBundle, _ as GatewayContext) >> {
+      a, b, db, gw -> bundle = db; gatewayContext = gw; NoopFlow.INSTANCE
+    }
+    bundle.get(KnownAddresses.REQUEST_FILES_CONTENT) == filesContent
+    flow.result == null
+    flow.action == Flow.Action.Noop.INSTANCE
+    gatewayContext.isTransient == false
+    gatewayContext.isRasp == false
+  }
+
+  void 'process request files content with empty list returns noop'() {
+    when:
+    Flow<?> flow = requestFilesContentCB.apply(ctx, [])
+
+    then:
+    flow == NoopFlow.INSTANCE
+    0 * eventDispatcher.publishDataEvent(*_)
+  }
+
   void 'process exec cmd'() {
     setup:
     final cmd = ['/bin/../usr/bin/reboot', '-f'] as String[]

…pload/CommonsFileUploadAppSecModule.java …monsFileUploadAppSecInstrumentation.javadd-java-agent/instrumentation/commons-fileupload-1.5/src/main/java/datadog/trace/instrumentation/commons/fileupload/CommonsFileUploadAppSecModule.java renamed to dd-java-agent/instrumentation/commons-fileupload-1.5/src/main/java/datadog/trace/instrumentation/commons/fileupload/CommonsFileUploadAppSecInstrumentation.java dd-java-agent/instrumentation/commons-fileupload-1.5/src/main/java/datadog/trace/instrumentation/commons/fileupload/CommonsFileUploadAppSecModule.java renamed to dd-java-agent/instrumentation/commons-fileupload-1.5/src/main/java/datadog/trace/instrumentation/commons/fileupload/CommonsFileUploadAppSecInstrumentation.java

@@ -24,10 +24,10 @@
 import org.apache.commons.fileupload.FileItem;
 
 @AutoService(InstrumenterModule.class)
-public class CommonsFileUploadAppSecModule extends InstrumenterModule.AppSec
+public class CommonsFileUploadAppSecInstrumentation extends InstrumenterModule.AppSec
     implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
 
-  public CommonsFileUploadAppSecModule() {
+  public CommonsFileUploadAppSecInstrumentation() {
     super("commons-fileupload");
   }
 
@@ -36,6 +36,13 @@ public String instrumentedType() {
     return "org.apache.commons.fileupload.servlet.ServletFileUpload";
   }
 
+  @Override
+  public String[] helperClassNames() {
+    return new String[] {
+      "datadog.trace.instrumentation.commons.fileupload.FileItemContentReader",
+    };
+  }
+
   @Override
   public void methodAdvice(MethodTransformer transformer) {
     transformer.applyAdvice(
@@ -47,6 +54,7 @@ public void methodAdvice(MethodTransformer transformer) {
 
   @RequiresRequestContext(RequestContextSlot.APPSEC)
   public static class ParseRequestAdvice {
+
     @Advice.OnMethodExit(suppress = Throwable.class, onThrowable = Throwable.class)
     static void after(
         @Advice.Return final List<FileItem> fileItems,
@@ -57,9 +65,11 @@ static void after(
       }
 
       CallbackProvider cbp = AgentTracer.get().getCallbackProvider(RequestContextSlot.APPSEC);
-      BiFunction<RequestContext, List<String>, Flow<Void>> callback =
+      BiFunction<RequestContext, List<String>, Flow<Void>> filenamesCallback =
           cbp.getCallback(EVENTS.requestFilesFilenames());
-      if (callback == null) {
+      BiFunction<RequestContext, List<String>, Flow<Void>> contentCallback =
+          cbp.getCallback(EVENTS.requestFilesContent());
+      if (filenamesCallback == null && contentCallback == null) {
         return;
       }
 
@@ -73,18 +83,42 @@ static void after(
           filenames.add(name);
         }
       }
-      if (filenames.isEmpty()) {
+      if (filenames.isEmpty() && contentCallback == null) {
         return;
       }
 
-      Flow<Void> flow = callback.apply(reqCtx, filenames);
-      Flow.Action action = flow.getAction();
-      if (action instanceof Flow.Action.RequestBlockingAction) {
-        Flow.Action.RequestBlockingAction rba = (Flow.Action.RequestBlockingAction) action;
+      // Fire filenames event
+      if (filenamesCallback != null && !filenames.isEmpty()) {
+        Flow<Void> flow = filenamesCallback.apply(reqCtx, filenames);
+        Flow.Action action = flow.getAction();
+        if (action instanceof Flow.Action.RequestBlockingAction) {
+          Flow.Action.RequestBlockingAction rba = (Flow.Action.RequestBlockingAction) action;
+          BlockResponseFunction brf = reqCtx.getBlockResponseFunction();
+          if (brf != null) {
+            brf.tryCommitBlockingResponse(reqCtx.getTraceSegment(), rba);
+            t = new BlockingException("Blocked request (multipart file upload)");
+            reqCtx.getTraceSegment().effectivelyBlocked();
+            return;
+          }
+        }
+      }
+
+      // Fire content event only if not blocked
+      if (contentCallback == null) {
+        return;
+      }
+      List<String> filesContent = FileItemContentReader.readContents(fileItems);
+      if (filesContent.isEmpty()) {
+        return;
+      }
+      Flow<Void> contentFlow = contentCallback.apply(reqCtx, filesContent);
+      Flow.Action contentAction = contentFlow.getAction();
+      if (contentAction instanceof Flow.Action.RequestBlockingAction) {
+        Flow.Action.RequestBlockingAction rba = (Flow.Action.RequestBlockingAction) contentAction;
         BlockResponseFunction brf = reqCtx.getBlockResponseFunction();
         if (brf != null) {
           brf.tryCommitBlockingResponse(reqCtx.getTraceSegment(), rba);
-          t = new BlockingException("Blocked request (multipart file upload)");
+          t = new BlockingException("Blocked request (multipart file upload content)");
           reqCtx.getTraceSegment().effectivelyBlocked();
         }
       }

dd-java-agent/instrumentation/commons-fileupload-1.5/src/main/java/datadog/trace/instrumentation/commons/fileupload/FileItemContentReader.java

@@ -0,0 +1,45 @@
+package datadog.trace.instrumentation.commons.fileupload;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.commons.fileupload.FileItem;
+
+/** Reads uploaded file content for WAF inspection. */
+public final class FileItemContentReader {
+  public static final int MAX_CONTENT_BYTES = 4096;
+  public static final int MAX_FILES_TO_INSPECT = 25;
+
+  public static List<String> readContents(List<FileItem> fileItems) {
+    List<String> result = new ArrayList<>();
+    for (FileItem fileItem : fileItems) {
+      if (result.size() >= MAX_FILES_TO_INSPECT) {
+        break;
+      }
+      if (fileItem.isFormField()) {
+        continue;
+      }
+      result.add(readContent(fileItem));
+    }
+    return result;
+  }
+
+  public static String readContent(FileItem fileItem) {
+    try (InputStream is = fileItem.getInputStream()) {
+      byte[] buf = new byte[MAX_CONTENT_BYTES];
+      int total = 0;
+      int n;
+      while (total < MAX_CONTENT_BYTES
+          && (n = is.read(buf, total, MAX_CONTENT_BYTES - total)) != -1) {
+        total += n;
+      }
+      return new String(buf, 0, total, StandardCharsets.ISO_8859_1);
+    } catch (IOException ignored) {
+      return "";
+    }
+  }
+
+  private FileItemContentReader() {}
+}