envoyproxy: body parsing support (#3671)

Co-authored-by: Eliott Bouhana <[email protected]>

Author: e-n-0

contrib/envoyproxy/go-control-plane/body_buffer.go

@@ -0,0 +1,44 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2024 Datadog, Inc.
+
+package gocontrolplane
+
+// bodyBuffer manages the buffering of request/response bodies with size limits
+type bodyBuffer struct {
+	Buffer    []byte
+	SizeLimit int
+	Truncated bool
+}
+
+// newBodyBuffer creates a new bodyBuffer with the specified size limit
+func newBodyBuffer(sizeLimit int) *bodyBuffer {
+	return &bodyBuffer{
+		SizeLimit: sizeLimit,
+		Truncated: false,
+	}
+}
+
+// Append adds a chunk of data to the buffer, respecting the size limit
+func (b *bodyBuffer) Append(chunk []byte) {
+	if b.Truncated || len(chunk) == 0 {
+		return
+	}
+
+	currentSize := len(b.Buffer)
+	remainingCapacity := b.SizeLimit - currentSize
+
+	bytesToAdd := len(chunk)
+	if bytesToAdd > remainingCapacity {
+		bytesToAdd = remainingCapacity
+		b.Truncated = true
+		instr.Logger().Debug("external_processing: body size limit reached, truncating body to %d bytes", bytesToAdd)
+	}
+
+	if b.Buffer == nil {
+		b.Buffer = make([]byte, 0, bytesToAdd)
+	}
+
+	b.Buffer = append(b.Buffer, chunk[:bytesToAdd]...)
+}

contrib/envoyproxy/go-control-plane/cmd/serviceextensions/main.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc/credentials"
 
 	gocontrolplane "github.com/DataDog/dd-trace-go/contrib/envoyproxy/go-control-plane/v2"
 	"github.com/DataDog/dd-trace-go/v2/ddtrace/tracer"
@@ -27,7 +28,6 @@ import (
 	extproc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/gorilla/mux"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials"
 )
 
 // AppsecCalloutExtensionService defines the struct that follows the ExternalProcessorServer interface.
@@ -36,26 +36,29 @@ type AppsecCalloutExtensionService struct {
 }
 
 type serviceExtensionConfig struct {
-	extensionPort     string
-	extensionHost     string
-	healthcheckPort   string
-	observabilityMode bool
+	extensionPort        string
+	extensionHost        string
+	healthcheckPort      string
+	observabilityMode    bool
+	bodyParsingSizeLimit int
 }
 
 func loadConfig() serviceExtensionConfig {
 	extensionPortInt := intEnv("DD_SERVICE_EXTENSION_PORT", 443)
 	healthcheckPortInt := intEnv("DD_SERVICE_EXTENSION_HEALTHCHECK_PORT", 80)
 	extensionHostStr := ipEnv("DD_SERVICE_EXTENSION_HOST", net.IP{0, 0, 0, 0}).String()
 	observabilityMode := boolEnv("DD_SERVICE_EXTENSION_OBSERVABILITY_MODE", false)
+	bodyParsingSizeLimit := intEnv("DD_APPSEC_BODY_PARSING_SIZE_LIMIT", 0)
 
 	extensionPortStr := strconv.FormatInt(int64(extensionPortInt), 10)
 	healthcheckPortStr := strconv.FormatInt(int64(healthcheckPortInt), 10)
 
 	return serviceExtensionConfig{
-		extensionPort:     extensionPortStr,
-		extensionHost:     extensionHostStr,
-		healthcheckPort:   healthcheckPortStr,
-		observabilityMode: observabilityMode,
+		extensionPort:        extensionPortStr,
+		extensionHost:        extensionHostStr,
+		healthcheckPort:      healthcheckPortStr,
+		observabilityMode:    observabilityMode,
+		bodyParsingSizeLimit: bodyParsingSizeLimit,
 	}
 }
 
@@ -142,25 +145,24 @@ func startHealthCheck(ctx context.Context, config serviceExtensionConfig) error
 }
 
 func startGPRCSsl(ctx context.Context, service extproc.ExternalProcessorServer, config serviceExtensionConfig) error {
-	cert, err := tls.LoadX509KeyPair("localhost.crt", "localhost.key")
-	if err != nil {
-		return fmt.Errorf("failed to load key pair: %v", err)
-	}
-
 	lis, err := net.Listen("tcp", config.extensionHost+":"+config.extensionPort)
 	if err != nil {
 		return fmt.Errorf("gRPC server: %v", err)
 	}
 
-	grpcCredentials := credentials.NewServerTLSFromCert(&cert)
-	grpcServer := grpc.NewServer(grpc.Creds(grpcCredentials))
+	cert, err := tls.LoadX509KeyPair("localhost.crt", "localhost.key")
+	if err != nil {
+		return fmt.Errorf("failed to load key pair: %v", err)
+	}
 
+	grpcServer := grpc.NewServer(grpc.Creds(credentials.NewServerTLSFromCert(&cert)))
 	appsecEnvoyExternalProcessorServer := gocontrolplane.AppsecEnvoyExternalProcessorServer(
 		service,
 		gocontrolplane.AppsecEnvoyConfig{
 			IsGCPServiceExtension: true,
 			BlockingUnavailable:   config.observabilityMode,
 			Context:               ctx,
+			BodyParsingSizeLimit:  config.bodyParsingSizeLimit,
 		})
 
 	go func() {