Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -387,6 +387,11 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {
false
}

/** Override to false when the multipart implementation uses a set with non-deterministic ordering (e.g. HashSet). */
boolean testBodyFilesContentOrdering() {
true
}

boolean testBodyJson() {
false
}
Expand Down Expand Up @@ -1767,7 +1772,7 @@ abstract class HttpServerTest<SERVER> extends WithHttpServer<SERVER> {

def 'test instrumentation gateway file upload content max files limit'() {
setup:
assumeTrue(testBodyFilesContent())
assumeTrue(testBodyFilesContent() && testBodyFilesContentOrdering())
def maxFilesToInspect = Config.get().getAppSecMaxFileContentCount()
def bodyBuilder = new MultipartBody.Builder().setType(MultipartBody.FORM)
(1..maxFilesToInspect + 1).each {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,11 @@ class VertxRxCircuitBreakerHttpServerForkedTest extends VertxHttpServerForkedTes
false
}

@Override
boolean testBodyFilesContent() {
false
}

@Override
boolean testBodyJson() {
false
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
package datadog.trace.instrumentation.vertx_3_4.server;

import datadog.appsec.api.blocking.BlockingException;
import datadog.trace.api.gateway.BlockResponseFunction;
import datadog.trace.api.gateway.Flow;
import datadog.trace.api.gateway.RequestContext;
import datadog.trace.api.http.MultipartContentDecoder;
import io.vertx.ext.web.FileUpload;
import java.io.FileInputStream;
import java.util.List;
import java.util.function.BiFunction;

public class FileUploadHelper {

public static BlockingException commitBlockingResponse(
BiFunction<RequestContext, List<String>, Flow<Void>> cb,
RequestContext reqCtx,
List<String> data,
String reason) {
Flow<Void> flow = cb.apply(reqCtx, data);
Flow.Action action = flow.getAction();
if (action instanceof Flow.Action.RequestBlockingAction) {
BlockResponseFunction brf = reqCtx.getBlockResponseFunction();
if (brf != null) {
brf.tryCommitBlockingResponse(
reqCtx.getTraceSegment(), (Flow.Action.RequestBlockingAction) action);
return new BlockingException(reason);
}
}
return null;
}

public static String readUploadContent(FileUpload upload, int maxBytes) {
try {
String path = upload.uploadedFileName();
if (path == null || path.isEmpty()) {
return "";
}
String charSet = upload.charSet();
String contentType =
charSet != null && !charSet.isEmpty()
? upload.contentType() + "; charset=" + charSet
: upload.contentType();
try (FileInputStream fis = new FileInputStream(path)) {
return MultipartContentDecoder.readInputStream(fis, maxBytes, contentType);
}
} catch (Exception ignored) {
return "";
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,9 @@

import static datadog.trace.api.gateway.Events.EVENTS;

import datadog.appsec.api.blocking.BlockingException;
import datadog.trace.advice.ActiveRequestContext;
import datadog.trace.advice.RequiresRequestContext;
import datadog.trace.api.gateway.BlockResponseFunction;
import datadog.trace.api.Config;
import datadog.trace.api.gateway.CallbackProvider;
import datadog.trace.api.gateway.Flow;
import datadog.trace.api.gateway.RequestContext;
Expand Down Expand Up @@ -38,35 +37,52 @@ static void after(
return;
}

List<String> filenames = new ArrayList<>(uploads.size());
CallbackProvider cbp = AgentTracer.get().getCallbackProvider(RequestContextSlot.APPSEC);
BiFunction<RequestContext, List<String>, Flow<Void>> filenamesCb =
cbp.getCallback(EVENTS.requestFilesFilenames());
BiFunction<RequestContext, List<String>, Flow<Void>> contentCb =
cbp.getCallback(EVENTS.requestFilesContent());
if (filenamesCb == null && contentCb == null) {
return;
}

int maxFiles = Config.get().getAppSecMaxFileContentCount();
int maxBytes = Config.get().getAppSecMaxFileContentBytes();
List<String> filenames = null;
List<String> filesContent = null;

for (FileUpload upload : uploads) {
String name = upload.fileName();
if (name != null && !name.isEmpty()) {
if (filenamesCb != null && name != null && !name.isEmpty()) {
if (filenames == null) {
filenames = new ArrayList<>();
}
filenames.add(name);
}
if (contentCb != null
&& maxFiles > 0
&& (filesContent == null || filesContent.size() < maxFiles)) {
if (filesContent == null) {
filesContent = new ArrayList<>();
}
filesContent.add(FileUploadHelper.readUploadContent(upload, maxBytes));

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

readUploadConent() performs blocking file I/O (FileInputStream.read) on the vertx event loop thread. Since each loop serves many concurrent connections, blocking stops all requests on the thread for time of file reading. Under the high load or large files, this can cause significant latency increase

}
}
if (filenames.isEmpty()) {
return;

if (filenamesCb != null && filenames != null) {
throwable =
FileUploadHelper.commitBlockingResponse(
filenamesCb, reqCtx, filenames, "Blocked request (multipart file upload)");
}

CallbackProvider cbp = AgentTracer.get().getCallbackProvider(RequestContextSlot.APPSEC);
BiFunction<RequestContext, List<String>, Flow<Void>> cb =
cbp.getCallback(EVENTS.requestFilesFilenames());
if (cb == null) {
if (throwable != null) {
return;
}

Flow<Void> flow = cb.apply(reqCtx, filenames);
Flow.Action action = flow.getAction();
if (action instanceof Flow.Action.RequestBlockingAction) {
BlockResponseFunction brf = reqCtx.getBlockResponseFunction();
if (brf != null) {
brf.tryCommitBlockingResponse(
reqCtx.getTraceSegment(), (Flow.Action.RequestBlockingAction) action);
if (throwable == null) {
throwable = new BlockingException("Blocked request (multipart file upload)");
}
}
if (contentCb != null && filesContent != null) {
throwable =
FileUploadHelper.commitBlockingResponse(
contentCb, reqCtx, filesContent, "Blocked request (file content)");
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,20 @@ public class RoutingContextImplInstrumentation extends InstrumenterModule.AppSec
private static final Reference FILE_UPLOAD_REF =
new Reference.Builder("io.vertx.ext.web.FileUpload")
.withMethod(new String[0], 0, "fileName", "Ljava/lang/String;")
.withMethod(new String[0], 0, "uploadedFileName", "Ljava/lang/String;")
.withMethod(new String[0], 0, "contentType", "Ljava/lang/String;")
.withMethod(new String[0], 0, "charSet", "Ljava/lang/String;")
.build();

public RoutingContextImplInstrumentation() {
super("vertx", "vertx-3.4");
}

@Override
public String[] helperClassNames() {
return new String[] {packageName + ".FileUploadHelper"};
}

@Override
public String instrumentedType() {
return "io.vertx.ext.web.impl.RoutingContextImpl";
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package server

import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.BODY_MULTIPART
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.ERROR
import static org.junit.jupiter.api.Assumptions.assumeTrue
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.LOGIN
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.NOT_FOUND
Expand All @@ -10,6 +12,10 @@ import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.SUCCES
import datadog.trace.agent.test.asserts.TraceAssert
import datadog.trace.agent.test.base.HttpServer
import datadog.trace.agent.test.base.HttpServerTest
import datadog.trace.api.Config
import okhttp3.MediaType
import okhttp3.MultipartBody
import okhttp3.RequestBody
import datadog.trace.api.DDSpanTypes
import datadog.trace.bootstrap.instrumentation.api.Tags
import datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator
Expand Down Expand Up @@ -87,6 +93,42 @@ class VertxHttpServerForkedTest extends HttpServerTest<Vertx> {
true
}

@Override
boolean testBodyFilesContent() {
true
}

@Override
boolean testBodyFilesContentOrdering() {
false
}

// fileUploads() returns a HashSet in Vert.x: check count instead of which specific file is excluded
def 'test instrumentation gateway file upload content max files limit count'() {
setup:
assumeTrue(testBodyFilesContent())
def maxFilesToInspect = Config.get().getAppSecMaxFileContentCount()
def bodyBuilder = new MultipartBody.Builder().setType(MultipartBody.FORM)
(1..maxFilesToInspect + 1).each { i ->
bodyBuilder.addFormDataPart("file${i}", "file${i}.bin",
RequestBody.create(MediaType.parse('application/octet-stream'), "content_of_file_${i}"))
}
def httpRequest = request(BODY_MULTIPART, 'POST', bodyBuilder.build()).build()
def response = client.newCall(httpRequest).execute()

when:
TEST_WRITER.waitForTraces(1)

then:
TEST_WRITER.get(0).any { span ->
def tag = span.getTag('request.body.files_content') as String
tag != null && tag.count('content_of_file_') == maxFilesToInspect
}

cleanup:
response.close()
}

@Override
boolean testBodyJson() {
true
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package server

import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.BODY_MULTIPART
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.ERROR
import static org.junit.jupiter.api.Assumptions.assumeTrue
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.LOGIN
import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.NOT_FOUND
Expand All @@ -10,6 +12,10 @@ import static datadog.trace.agent.test.base.HttpServerTest.ServerEndpoint.SUCCES
import datadog.trace.agent.test.asserts.TraceAssert
import datadog.trace.agent.test.base.HttpServer
import datadog.trace.agent.test.base.HttpServerTest
import datadog.trace.api.Config
import okhttp3.MediaType
import okhttp3.MultipartBody
import okhttp3.RequestBody
import datadog.trace.api.DDSpanTypes
import datadog.trace.bootstrap.instrumentation.api.Tags
import datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator
Expand Down Expand Up @@ -82,6 +88,42 @@ class VertxHttpServerForkedTest extends HttpServerTest<Vertx> {
true
}

@Override
boolean testBodyFilesContent() {
true
}

@Override
boolean testBodyFilesContentOrdering() {
false
}

// fileUploads() returns a HashSet in Vert.x: check count instead of which specific file is excluded
def 'test instrumentation gateway file upload content max files limit count'() {
setup:
assumeTrue(testBodyFilesContent())
def maxFilesToInspect = Config.get().getAppSecMaxFileContentCount()
def bodyBuilder = new MultipartBody.Builder().setType(MultipartBody.FORM)
(1..maxFilesToInspect + 1).each { i ->
bodyBuilder.addFormDataPart("file${i}", "file${i}.bin",
RequestBody.create(MediaType.parse('application/octet-stream'), "content_of_file_${i}"))
}
def httpRequest = request(BODY_MULTIPART, 'POST', bodyBuilder.build()).build()
def response = client.newCall(httpRequest).execute()

when:
TEST_WRITER.waitForTraces(1)

then:
TEST_WRITER.get(0).any { span ->
def tag = span.getTag('request.body.files_content') as String
tag != null && tag.count('content_of_file_') == maxFilesToInspect
}

cleanup:
response.close()
}

@Override
boolean testBodyJson() {
true
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
package datadog.trace.instrumentation.vertx_4_0.server;

import datadog.appsec.api.blocking.BlockingException;
import datadog.trace.api.gateway.BlockResponseFunction;
import datadog.trace.api.gateway.Flow;
import datadog.trace.api.gateway.RequestContext;
import datadog.trace.api.http.MultipartContentDecoder;
import io.vertx.ext.web.FileUpload;
import java.io.FileInputStream;
import java.util.List;
import java.util.function.BiFunction;

public class FileUploadHelper {

public static BlockingException commitBlockingResponse(
BiFunction<RequestContext, List<String>, Flow<Void>> cb,
RequestContext reqCtx,
List<String> data,
String reason) {
Flow<Void> flow = cb.apply(reqCtx, data);
Flow.Action action = flow.getAction();
if (action instanceof Flow.Action.RequestBlockingAction) {
BlockResponseFunction brf = reqCtx.getBlockResponseFunction();
if (brf != null) {
brf.tryCommitBlockingResponse(
reqCtx.getTraceSegment(), (Flow.Action.RequestBlockingAction) action);
return new BlockingException(reason);
}
}
return null;
}

public static String readUploadContent(FileUpload upload, int maxBytes) {
try {
String path = upload.uploadedFileName();
if (path == null || path.isEmpty()) {
return "";
}
String charSet = upload.charSet();
String contentType =
charSet != null && !charSet.isEmpty()
? upload.contentType() + "; charset=" + charSet
: upload.contentType();
try (FileInputStream fis = new FileInputStream(path)) {
return MultipartContentDecoder.readInputStream(fis, maxBytes, contentType);
}
} catch (Exception ignored) {
return "";
}
}
}
Loading