test(appsec): add Netty integration tests for multipart filenames and file content

Also fix HttpPostRequestDecoderInstrumentation to fire on PREEPILOGUE+isLastChunk
so that multipart requests delivered as a FullHttpRequest (single shot) trigger the
requestBodyProcessed/requestFilesFilenames/requestFilesContent callbacks. The
PREEPILOGUE→EPILOGUE transition requires a second parseBody() call which never
comes when the full request arrives in one shot; PREEPILOGUE+isLastChunk is
equivalent to EPILOGUE in that scenario.

Author: jandro996

dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/base/HttpServerTest.groovy

@@ -136,6 +136,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     ss.registerCallback(events.requestBodyDone(), callbacks.requestBodyEndCb)
     ss.registerCallback(events.requestBodyProcessed(), callbacks.requestBodyObjectCb)
     ss.registerCallback(events.requestFilesFilenames(), callbacks.requestFilesFilenamesCb)
+    ss.registerCallback(events.requestFilesContent(), callbacks.requestFilesContentCb)
     ss.registerCallback(events.responseBody(), callbacks.responseBodyObjectCb)
     ss.registerCallback(events.responseStarted(), callbacks.responseStartedCb)
     ss.registerCallback(events.responseHeader(), callbacks.responseHeaderCb)
@@ -372,6 +373,10 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     false
   }
 
+  boolean testBodyFilesContent() {
+    false
+  }
+
   boolean testBodyJson() {
     false
   }
@@ -1646,6 +1651,29 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
     response.close()
   }
 
+  def 'test instrumentation gateway file upload content'() {
+    setup:
+    assumeTrue(testBodyFilesContent())
+    RequestBody fileBody = RequestBody.create(MediaType.parse('application/octet-stream'), 'file content')
+    def body = new MultipartBody.Builder()
+    .setType(MultipartBody.FORM)
+    .addFormDataPart('file', 'test.bin', fileBody)
+    .build()
+    def httpRequest = request(BODY_MULTIPART, 'POST', body).build()
+    def response = client.newCall(httpRequest).execute()
+
+    when:
+    TEST_WRITER.waitForTraces(1)
+
+    then:
+    TEST_WRITER.get(0).any {
+      it.getTag('request.body.files_content') == '[file content]'
+    }
+
+    cleanup:
+    response.close()
+  }
+
   def 'test instrumentation gateway json request body'() {
     setup:
     assumeTrue(testBodyJson())
@@ -2581,6 +2609,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
       boolean responseBodyTag
       Object responseBody
       List<String> uploadedFilenames
+      List<String> uploadedFilesContent
     }
 
     static final String stringOrEmpty(String string) {
@@ -2757,6 +2786,15 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
       Flow.ResultFlow.empty()
     } as BiFunction<RequestContext, List<String>, Flow<Void>>)
 
+    final BiFunction<RequestContext, List<String>, Flow<Void>> requestFilesContentCb =
+    ({
+      RequestContext rqCtxt, List<String> contents ->
+      rqCtxt.traceSegment.setTagTop('request.body.files_content', contents as String)
+      Context context = rqCtxt.getData(RequestContextSlot.APPSEC)
+      context.uploadedFilesContent = contents
+      Flow.ResultFlow.empty()
+    } as BiFunction<RequestContext, List<String>, Flow<Void>>)
+
     final BiFunction<RequestContext, Object, Flow<Void>> responseBodyObjectCb =
     ({
       RequestContext rqCtxt, Object obj ->

dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/HttpPostRequestDecoderInstrumentation.java

@@ -61,13 +61,15 @@ public Reference[] additionalMuzzleReferences() {
               Reference.EXPECTS_NON_STATIC,
               "currentStatus",
               "Lio/netty/handler/codec/http/multipart/HttpPostRequestDecoder$MultiPartStatus;")
+          .withField(new String[0], Reference.EXPECTS_NON_STATIC, "isLastChunk", "Z")
           .build(),
       new Reference.Builder("io.netty.handler.codec.http.multipart.HttpPostStandardRequestDecoder")
           .withField(
               new String[0],
               Reference.EXPECTS_NON_STATIC,
               "currentStatus",
               "Lio/netty/handler/codec/http/multipart/HttpPostRequestDecoder$MultiPartStatus;")
+          .withField(new String[0], Reference.EXPECTS_NON_STATIC, "isLastChunk", "Z")
           .build()
     };
   }
@@ -88,10 +90,17 @@ static class ParseBodyAdvice {
     static void after(
         @Advice.This InterfaceHttpPostRequestDecoder thiz,
         @Advice.FieldValue("currentStatus") Enum currentStatus,
+        @Advice.FieldValue("isLastChunk") boolean isLastChunk,
         @ActiveRequestContext RequestContext requestContext,
         @Advice.Thrown(readOnly = false) Throwable thr) {
-      if (!currentStatus.name().equals("EPILOGUE")) {
-        return;
+      String statusName = currentStatus.name();
+      if (!statusName.equals("EPILOGUE")) {
+        // For multipart decoders, the PREEPILOGUE→EPILOGUE transition requires a second
+        // parseBody() call that never comes when the full request arrives in one shot.
+        // Fire on PREEPILOGUE + isLastChunk to handle that case.
+        if (!statusName.equals("PREEPILOGUE") || !isLastChunk) {
+          return;
+        }
       }
 
       CallbackProvider cbp = AgentTracer.get().getCallbackProvider(RequestContextSlot.APPSEC);
@@ -160,6 +169,8 @@ static void after(
         BlockResponseFunction brf = requestContext.getBlockResponseFunction();
         if (brf != null) {
           brf.tryCommitBlockingResponse(requestContext.getTraceSegment(), rba);
+          // effectivelyBlocked() is intentionally absent: tryCommitBlockingResponse finishes
+          // the span synchronously in this Netty path; calling it on a finished span throws.
           thr = new BlockingException("Blocked request (multipart/urlencoded post data)");
         }
       }

dd-java-agent/instrumentation/netty/netty-4.1/src/test/groovy/Netty41ServerTest.groovy

@@ -27,6 +27,7 @@ import io.netty.handler.codec.http.HttpResponseStatus
 import io.netty.handler.codec.http.HttpServerCodec
 import io.netty.handler.codec.http.multipart.Attribute
 import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder
+import io.netty.handler.codec.http.multipart.InterfaceHttpData
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame
 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame
 import io.netty.handler.codec.http.websocketx.ContinuationWebSocketFrame
@@ -36,6 +37,7 @@ import io.netty.handler.logging.LogLevel
 import io.netty.handler.logging.LoggingHandler
 import io.netty.util.CharsetUtil
 
+import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.BODY_MULTIPART
 import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.BODY_URLENCODED
 import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.ERROR
 import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
@@ -133,6 +135,31 @@ abstract class Netty41ServerTest extends HttpServerTest<EventLoopGroup> {
                           response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status), content)
                         }
                         break
+                      case BODY_MULTIPART:
+                        if (msg instanceof FullHttpRequest) {
+                          HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(
+                            new HttpRequest() {
+                              @Delegate
+                              HttpRequest delegate = request
+                            })
+
+                          Map m
+                          try {
+                            decoder.offer(msg)
+
+                            m = decoder.bodyHttpDatas
+                              .findAll { it.httpDataType == InterfaceHttpData.HttpDataType.Attribute }
+                              .collectEntries { d -> [d.name, [((Attribute) d).value]] }
+                          } finally {
+                            try {
+                              decoder.destroy()
+                            } catch (Exception ignored) {}
+                          }
+
+                          content = Unpooled.copiedBuffer(m as String, CharsetUtil.UTF_8)
+                          response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status), content)
+                        }
+                        break
                       case REDIRECT:
                         response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status))
                         response.headers().set(HttpHeaderNames.LOCATION, endpoint.body)
@@ -270,6 +297,21 @@ abstract class Netty41ServerTest extends HttpServerTest<EventLoopGroup> {
     true
   }
 
+  @Override
+  boolean testBodyMultipart() {
+    true
+  }
+
+  @Override
+  boolean testBodyFilenames() {
+    true
+  }
+
+  @Override
+  boolean testBodyFilesContent() {
+    true
+  }
+
   @Override
   boolean testBlocking() {
     true