feat(appsec): expose uploaded file content as WAF address server.request.body.files_content

Introduces a new AppSec WAF address `server.request.body.files_content`
(`List<String>`) that exposes the content of each uploaded file in a
multipart/form-data request. Entries correspond positionally to the
existing `server.request.body.filenames` address. Content is capped at
4 096 bytes per file (ISO-8859-1) to keep memory usage bounded.

Changes:
- KnownAddresses: add REQUEST_FILES_CONTENT + forName() case
- Events: add requestFilesContent event (ID 31); FILE_WRITTEN bumped to 32
- InstrumentationGateway: register the new BiFunction case
- GatewayBridge: add onRequestFilesContent handler + DATA_DEPENDENCIES entry
- CommonsFileUploadAppSecModule: after firing filenames, fire content
  (skipped when the filenames event already blocked the request)
- Unit tests: GatewayBridgeSpecification, GatewayBridgeIGRegistrationSpecification,
  KnownAddressesSpecificationForkedTest
- Smoke test: 'block request based on malicious file upload content'
  verifies end-to-end blocking via a custom WAF rule on the new address

Closes APPSEC-61875

Author: jandro996

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

@@ -74,6 +74,14 @@ 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. Each entry in the
+   * list corresponds positionally to {@link #REQUEST_FILES_FILENAMES}. Content is truncated to a
+   * maximum size to avoid excessive memory usage. 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 +213,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[]

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

@@ -17,6 +17,9 @@
 import datadog.trace.api.gateway.RequestContext;
 import datadog.trace.api.gateway.RequestContextSlot;
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.BiFunction;
@@ -47,6 +50,9 @@ public void methodAdvice(MethodTransformer transformer) {
 
   @RequiresRequestContext(RequestContextSlot.APPSEC)
   public static class ParseRequestAdvice {
+
+    static final int MAX_FILE_CONTENT_BYTES = 4096;
+
     @Advice.OnMethodExit(suppress = Throwable.class, onThrowable = Throwable.class)
     static void after(
         @Advice.Return final List<FileItem> fileItems,
@@ -57,11 +63,6 @@ static void after(
       }
 
       CallbackProvider cbp = AgentTracer.get().getCallbackProvider(RequestContextSlot.APPSEC);
-      BiFunction<RequestContext, List<String>, Flow<Void>> callback =
-          cbp.getCallback(EVENTS.requestFilesFilenames());
-      if (callback == null) {
-        return;
-      }
 
       List<String> filenames = new ArrayList<>();
       for (FileItem fileItem : fileItems) {
@@ -77,14 +78,65 @@ static void after(
         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
+      BiFunction<RequestContext, List<String>, Flow<Void>> filenamesCallback =
+          cbp.getCallback(EVENTS.requestFilesFilenames());
+      if (filenamesCallback != null) {
+        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
+      BiFunction<RequestContext, List<String>, Flow<Void>> contentCallback =
+          cbp.getCallback(EVENTS.requestFilesContent());
+      if (contentCallback == null) {
+        return;
+      }
+      List<String> filesContent = new ArrayList<>();
+      for (FileItem fileItem : fileItems) {
+        if (fileItem.isFormField()) {
+          continue;
+        }
+        String name = fileItem.getName();
+        if (name == null || name.isEmpty()) {
+          continue;
+        }
+        String content = "";
+        try {
+          InputStream is = fileItem.getInputStream();
+          byte[] buf = new byte[MAX_FILE_CONTENT_BYTES];
+          int total = 0;
+          int n;
+          while (total < MAX_FILE_CONTENT_BYTES
+              && (n = is.read(buf, total, MAX_FILE_CONTENT_BYTES - total)) != -1) {
+            total += n;
+          }
+          content = new String(buf, 0, total, StandardCharsets.ISO_8859_1);
+        } catch (IOException ignored) {
+        }
+        filesContent.add(content);
+      }
+      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-smoke-tests/appsec/springboot/src/test/groovy/datadog/smoketest/appsec/SpringBootSmokeTest.groovy

@@ -230,6 +230,26 @@ class SpringBootSmokeTest extends AbstractAppSecServerSmokeTest {
           transformers: [],
           on_match    : ['block']
         ],
+        [
+          id          : '__test_file_upload_content_block',
+          name        : 'test rule to block on malicious file upload content',
+          tags        : [
+            type      : 'unrestricted-file-upload',
+            category  : 'attack_attempt',
+            confidence: '1',
+          ],
+          conditions  : [
+            [
+              parameters: [
+                inputs: [[address: 'server.request.body.files_content']],
+                regex : 'dd-test-malicious-file-content',
+              ],
+              operator  : 'match_regex',
+            ]
+          ],
+          transformers: [],
+          on_match    : ['block']
+        ],
         [
           id  : "apiA-100-001",
           name: "API 10 tag rule on request headers",
@@ -611,6 +631,38 @@ class SpringBootSmokeTest extends AbstractAppSecServerSmokeTest {
     }
   }
 
+  void 'block request based on malicious file upload content'() {
+    when:
+    String url = "http://localhost:${httpPort}/upload"
+    def requestBody = new okhttp3.MultipartBody.Builder()
+      .setType(okhttp3.MultipartBody.FORM)
+      .addFormDataPart('file', 'safe-document.txt',
+      RequestBody.create(MediaType.parse('application/octet-stream'), 'dd-test-malicious-file-content'))
+      .build()
+    def request = new Request.Builder()
+      .url(url)
+      .post(requestBody)
+      .build()
+    def response = client.newCall(request).execute()
+    def responseBodyStr = response.body().string()
+
+    then:
+    responseBodyStr.contains("blocked")
+    response.code() == 403
+
+    when:
+    waitForTraceCount(1) == 1
+
+    then:
+    rootSpans.size() == 1
+    forEachRootSpanTrigger {
+      assert it['rule']['id'] == '__test_file_upload_content_block'
+    }
+    rootSpans.each {
+      assert it.meta.get('appsec.blocked') != null, 'appsec.blocked is not set'
+    }
+  }
+
   void 'rasp reports stacktrace on sql injection'() {
     when:
     String url = "http://localhost:${httpPort}/sqli/query?id=' OR 1=1 --"

internal-api/src/main/java/datadog/trace/api/gateway/Events.java

@@ -396,7 +396,19 @@ public EventType<BiFunction<RequestContext, List<String>, Flow<Void>>> requestFi
         REQUEST_FILES_FILENAMES;
   }
 
-  static final int FILE_WRITTEN_ID = 31;
+  static final int REQUEST_FILES_CONTENT_ID = 31;
+
+  @SuppressWarnings("rawtypes")
+  private static final EventType REQUEST_FILES_CONTENT =
+      new ET<>("request.body.files.content", REQUEST_FILES_CONTENT_ID);
+
+  /** Contents of files uploaded in a multipart/form-data request */
+  @SuppressWarnings("unchecked")
+  public EventType<BiFunction<RequestContext, List<String>, Flow<Void>>> requestFilesContent() {
+    return (EventType<BiFunction<RequestContext, List<String>, Flow<Void>>>) REQUEST_FILES_CONTENT;
+  }
+
+  static final int FILE_WRITTEN_ID = 32;
 
   @SuppressWarnings("rawtypes")
   private static final EventType FILE_WRITTEN = new ET<>("file.written", FILE_WRITTEN_ID);

internal-api/src/main/java/datadog/trace/api/gateway/InstrumentationGateway.java

@@ -19,6 +19,7 @@
 import static datadog.trace.api.gateway.Events.REQUEST_BODY_START_ID;
 import static datadog.trace.api.gateway.Events.REQUEST_CLIENT_SOCKET_ADDRESS_ID;
 import static datadog.trace.api.gateway.Events.REQUEST_ENDED_ID;
+import static datadog.trace.api.gateway.Events.REQUEST_FILES_CONTENT_ID;
 import static datadog.trace.api.gateway.Events.REQUEST_FILES_FILENAMES_ID;
 import static datadog.trace.api.gateway.Events.REQUEST_HEADER_DONE_ID;
 import static datadog.trace.api.gateway.Events.REQUEST_HEADER_ID;
@@ -363,6 +364,7 @@ public Flow<Void> apply(RequestContext ctx, StoredBodySupplier storedBodySupplie
       case REQUEST_BODY_CONVERTED_ID:
       case RESPONSE_BODY_ID:
       case REQUEST_FILES_FILENAMES_ID:
+      case REQUEST_FILES_CONTENT_ID:
         return (C)
             new BiFunction<RequestContext, Object, Flow<Void>>() {
               @Override