Memory Allocation with Excessive Size Value in Metricbeat Leading to Denial of Service

@@ -0,0 +1,32 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Enforces configurable size limits on incoming requests for remote_write metricset (max_compressed_body_bytes, max_decoded_body_bytes)
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+#description:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: metricbeat
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+#pr: https://github.com/owner/repo/1234
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+#issue: https://github.com/owner/repo/1234

@@ -68,6 +68,25 @@ remote_write:
         #insecure_skip_verify: true
 ```
 
+## Request size limits [_request_size_limits]
+
+```{applies_to}
+stack: ga 9.2.5, ga 9.3.1, ga 9.4
+```
+
+To protect against resource exhaustion from malicious or oversized payloads, the remote_write metricset enforces configurable size limits on incoming requests:
+
+* `max_compressed_body_bytes`: Maximum size of the compressed (snappy-encoded) request body in bytes. Requests exceeding this limit are rejected with HTTP 413 before being read into memory. Default: 2 MB (2097152 bytes).
+* `max_decoded_body_bytes`: Maximum size of the decompressed request body in bytes. The server checks the declared decoded size in the snappy header before allocating memory for decompression, preventing decompression bomb attacks. Default: 10 MB (10485760 bytes).
+
+```yaml
+- module: prometheus
+  metricsets: ["remote_write"]
+  host: "localhost"
+  port: "9201"
+  max_compressed_body_bytes: 2097152    # 2 MB (default)
+  max_decoded_body_bytes: 10485760      # 10 MB (default)
+```
 
 ## Histograms and types [_histograms_and_types_2]
 

@@ -19,16 +19,27 @@ package remote_write
 
 import "github.com/elastic/elastic-agent-libs/transport/tlscommon"
 
+const (
+	// DefaultMaxCompressedBodyBytes is the default maximum size of compressed request body (2MB)
+	DefaultMaxCompressedBodyBytes int64 = 2 * 1024 * 1024
+	// DefaultMaxDecodedBodyBytes is the default maximum size of decoded request body (10MB)
+	DefaultMaxDecodedBodyBytes int64 = 10 * 1024 * 1024
+)
+
 type Config struct {
-	MetricsCount bool                    `config:"metrics_count"`
-	Host         string                  `config:"host"`
-	Port         int                     `config:"port"`
-	TLS          *tlscommon.ServerConfig `config:"ssl"`
+	MetricsCount           bool                    `config:"metrics_count"`
+	Host                   string                  `config:"host"`
+	Port                   int                     `config:"port"`
+	TLS                    *tlscommon.ServerConfig `config:"ssl"`
+	MaxCompressedBodyBytes int64                   `config:"max_compressed_body_bytes"`
+	MaxDecodedBodyBytes    int64                   `config:"max_decoded_body_bytes"`
 }
 
 func defaultConfig() Config {
 	return Config{
-		Host: "localhost",
-		Port: 9201,
+		Host:                   "localhost",
+		Port:                   9201,
+		MaxCompressedBodyBytes: DefaultMaxCompressedBodyBytes,
+		MaxDecodedBodyBytes:    DefaultMaxDecodedBodyBytes,
 	}
 }

@@ -57,6 +57,25 @@ remote_write:
         #insecure_skip_verify: true
 ```
 
+## Request size limits [_request_size_limits]
+
+```{applies_to}
+stack: ga 9.2.5, ga 9.3.1, ga 9.4
+```
+
+To protect against resource exhaustion from malicious or oversized payloads, the remote_write metricset enforces configurable size limits on incoming requests:
+
+* `max_compressed_body_bytes`: Maximum size of the compressed (snappy-encoded) request body in bytes. Requests exceeding this limit are rejected with HTTP 413 before being read into memory. Default: 2 MB (2097152 bytes).
+* `max_decoded_body_bytes`: Maximum size of the decompressed request body in bytes. The server checks the declared decoded size in the snappy header before allocating memory for decompression, preventing decompression bomb attacks. Default: 10 MB (10485760 bytes).
+
+```yaml
+- module: prometheus
+  metricsets: ["remote_write"]
+  host: "localhost"
+  port: "9201"
+  max_compressed_body_bytes: 2097152    # 2 MB (default)
+  max_decoded_body_bytes: 10485760      # 10 MB (default)
+```
 
 ## Histograms and types [_histograms_and_types_2]
 

@@ -18,6 +18,8 @@
 package remote_write
 
 import (
+	"errors"
+	"fmt"
 	"io"
 	"net/http"
 
@@ -56,10 +58,12 @@ type RemoteWriteEventsGeneratorFactory func(ms mb.BaseMetricSet, opts ...RemoteW
 
 type MetricSet struct {
 	mb.BaseMetricSet
-	server          serverhelper.Server
-	events          chan mb.Event
-	promEventsGen   RemoteWriteEventsGenerator
-	eventGenStarted bool
+	server                 serverhelper.Server
+	events                 chan mb.Event
+	promEventsGen          RemoteWriteEventsGenerator
+	eventGenStarted        bool
+	maxCompressedBodyBytes int64
+	maxDecodedBodyBytes    int64
 }
 
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
@@ -92,8 +96,14 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // MetricSetBuilder returns a builder function for a new Prometheus remote_write metricset using
 // the given namespace and event generator
 func MetricSetBuilder(genFactory RemoteWriteEventsGeneratorFactory) func(base mb.BaseMetricSet) (mb.MetricSet, error) {
+	return MetricSetBuilderWithConfig(genFactory, defaultConfig())
+}
+
+// MetricSetBuilderWithConfig returns a builder function for a new Prometheus remote_write metricset using
+// the given namespace, event generator, and a base config that will be merged with module config
+func MetricSetBuilderWithConfig(genFactory RemoteWriteEventsGeneratorFactory, baseConfig Config) func(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return func(base mb.BaseMetricSet) (mb.MetricSet, error) {
-		config := defaultConfig()
+		config := baseConfig
 		err := base.Module().UnpackConfig(&config)
 		if err != nil {
 			return nil, err
@@ -105,10 +115,12 @@ func MetricSetBuilder(genFactory RemoteWriteEventsGeneratorFactory) func(base mb
 		}
 
 		m := &MetricSet{
-			BaseMetricSet:   base,
-			events:          make(chan mb.Event),
-			promEventsGen:   promEventsGen,
-			eventGenStarted: false,
+			BaseMetricSet:          base,
+			events:                 make(chan mb.Event),
+			promEventsGen:          promEventsGen,
+			eventGenStarted:        false,
+			maxCompressedBodyBytes: config.MaxCompressedBodyBytes,
+			maxDecodedBodyBytes:    config.MaxDecodedBodyBytes,
 		}
 
 		svc, err := httpserver.NewHttpServerWithHandler(base, m.handleFunc)
@@ -150,13 +162,35 @@ func (m *MetricSet) handleFunc(writer http.ResponseWriter, req *http.Request) {
 		m.eventGenStarted = true
 	}
 
+	// Limit the size of the compressed request body to prevent resource exhaustion
+	req.Body = http.MaxBytesReader(writer, req.Body, m.maxCompressedBodyBytes)
+
 	compressed, err := io.ReadAll(req.Body)
 	if err != nil {
+		var maxBytesError *http.MaxBytesError
+		if errors.As(err, &maxBytesError) {
+			m.Logger().Warnf("Request body too large: exceeds %d bytes limit", m.maxCompressedBodyBytes)
+			http.Error(writer, fmt.Sprintf("request body too large: exceeds %d bytes limit", m.maxCompressedBodyBytes), http.StatusRequestEntityTooLarge)
+			return
+		}
 		m.Logger().Errorf("Read error %v", err)
 		http.Error(writer, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
+	// Check decoded length before allocating memory to prevent
+	decodedLen, err := snappy.DecodedLen(compressed)
+	if err != nil {
+		m.Logger().Errorf("Decoded length error: %v", err)
+		http.Error(writer, "Decoded length error", http.StatusBadRequest)
+		return
+	}
+	if int64(decodedLen) > m.maxDecodedBodyBytes {
+		m.Logger().Warnf("Decoded length too large: %d bytes exceeds %d max decoded bytes limit (maxDecodedBodyBytes)", decodedLen, m.maxDecodedBodyBytes)
+		http.Error(writer, fmt.Sprintf("decoded length too large: %d bytes exceeds %d max decoded bytes limit (maxDecodedBodyBytes)", decodedLen, m.maxDecodedBodyBytes), http.StatusRequestEntityTooLarge)
+		return
+	}
+
 	reqBuf, err := snappy.Decode(nil, compressed)
 	if err != nil {
 		m.Logger().Errorf("Decode error %v", err)

@@ -18,11 +18,21 @@
 package remote_write
 
 import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
+	"github.com/gogo/protobuf/proto"
+	"github.com/golang/snappy"
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
+	"github.com/elastic/beats/v7/metricbeat/mb"
+	mbtest "github.com/elastic/beats/v7/metricbeat/mb/testing"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
@@ -263,3 +273,186 @@ func TestMetricsCount(t *testing.T) {
 		})
 	}
 }
+
+// createTestWriteRequest creates a prompb.WriteRequest with the given number of samples
+func createTestWriteRequest(numSamples int) *prompb.WriteRequest {
+	samples := make([]prompb.Sample, numSamples)
+	for i := 0; i < numSamples; i++ {
+		samples[i] = prompb.Sample{
+			Value:     float64(i),
+			Timestamp: int64(i * 1000),
+		}
+	}
+
+	return &prompb.WriteRequest{
+		Timeseries: []prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "__name__", Value: "test_metric"},
+					{Name: "instance", Value: "localhost:9090"},
+				},
+				Samples: samples,
+			},
+		},
+	}
+}
+
+// encodeWriteRequest encodes a WriteRequest to snappy-compressed protobuf
+func encodeWriteRequest(req *prompb.WriteRequest) ([]byte, error) {
+	data, err := proto.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+	return snappy.Encode(nil, data), nil
+}
+
+// newTestMetricSet creates a MetricSet for testing using the mbtest infrastructure
+// to ensure proper initialization (including logger)
+func newTestMetricSet(t *testing.T, maxCompressedBodyBytes, maxDecodedBodyBytes int64) *MetricSet {
+	config := map[string]interface{}{
+		"module":     "prometheus",
+		"metricsets": []string{"remote_write"},
+	}
+
+	ms := mbtest.NewMetricSet(t, config)
+	m, ok := ms.(*MetricSet)
+	require.True(t, ok, "expected *MetricSet, got %T", ms)
+
+	// Override the size limits for testing
+	m.maxCompressedBodyBytes = maxCompressedBodyBytes
+	m.maxDecodedBodyBytes = maxDecodedBodyBytes
+	// Ensure events channel exists for the handler
+	m.events = make(chan mb.Event, 100)
+
+	return m
+}
+
+func TestHandleFuncDecodedSizeLimit(t *testing.T) {
+	tests := []struct {
+		name                   string
+		maxDecodedBodyBytes    int64
+		maxCompressedBodyBytes int64
+		numSamples             int
+		expectedStatus         int
+		expectedBodyContains   string
+	}{
+		{
+			name:                   "request within decoded size limit succeeds",
+			maxDecodedBodyBytes:    1024 * 1024, // 1MB
+			maxCompressedBodyBytes: 1024 * 1024, // 1MB
+			numSamples:             10,
+			expectedStatus:         http.StatusAccepted,
+		},
+		{
+			name:                   "request exceeding decoded size limit rejected",
+			maxDecodedBodyBytes:    100, // Very small limit
+			maxCompressedBodyBytes: 1024 * 1024,
+			numSamples:             100, // Will decode to more than 100 bytes
+			expectedStatus:         http.StatusRequestEntityTooLarge,
+			expectedBodyContains:   "decoded length too large",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := newTestMetricSet(t, tt.maxCompressedBodyBytes, tt.maxDecodedBodyBytes)
+
+			// Create a test write request
+			writeReq := createTestWriteRequest(tt.numSamples)
+			body, err := encodeWriteRequest(writeReq)
+			require.NoError(t, err)
+
+			// Create HTTP request
+			req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
+			rec := httptest.NewRecorder()
+
+			// Call the handler
+			m.handleFunc(rec, req)
+
+			// Check the response
+			assert.Equal(t, tt.expectedStatus, rec.Code)
+			if tt.expectedBodyContains != "" {
+				assert.True(t, strings.Contains(rec.Body.String(), tt.expectedBodyContains),
+					"expected body to contain %q, got %q", tt.expectedBodyContains, rec.Body.String())
+			}
+		})
+	}
+}
+
+func TestHandleFuncCompressedSizeLimit(t *testing.T) {
+	tests := []struct {
+		name                   string
+		maxCompressedBodyBytes int64
+		maxDecodedBodyBytes    int64
+		bodySize               int
+		expectedStatus         int
+		expectedBodyContains   string
+	}{
+		{
+			name:                   "compressed body within limit succeeds",
+			maxCompressedBodyBytes: 1024 * 1024, // 1MB
+			maxDecodedBodyBytes:    10 * 1024 * 1024,
+			bodySize:               100,
+			expectedStatus:         http.StatusAccepted,
+		},
+		{
+			name:                   "compressed body exceeding limit rejected",
+			maxCompressedBodyBytes: 50,
+			maxDecodedBodyBytes:    10 * 1024 * 1024,
+			bodySize:               100, // More than 50 bytes
+			expectedStatus:         http.StatusRequestEntityTooLarge,
+			expectedBodyContains:   "request body too large",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := newTestMetricSet(t, tt.maxCompressedBodyBytes, tt.maxDecodedBodyBytes)
+
+			var body []byte
+			if tt.bodySize <= 100 {
+				// For small sizes, use a valid request
+				writeReq := createTestWriteRequest(tt.bodySize)
+				var err error
+				body, err = encodeWriteRequest(writeReq)
+				require.NoError(t, err)
+			} else {
+				// For larger sizes, create arbitrary data
+				body = make([]byte, tt.bodySize)
+				for i := range body {
+					body[i] = byte(i % 256)
+				}
+			}
+
+			// Create HTTP request
+			req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
+			rec := httptest.NewRecorder()
+
+			// Call the handler
+			m.handleFunc(rec, req)
+
+			// Check the response
+			assert.Equal(t, tt.expectedStatus, rec.Code)
+			if tt.expectedBodyContains != "" {
+				assert.True(t, strings.Contains(rec.Body.String(), tt.expectedBodyContains),
+					"expected body to contain %q, got %q", tt.expectedBodyContains, rec.Body.String())
+			}
+		})
+	}
+}
+
+func TestHandleFuncInvalidSnappyData(t *testing.T) {
+	m := newTestMetricSet(t, 1024*1024, 10*1024*1024)
+
+	// Send data with an invalid truncated varint header that will fail at snappy.DecodedLen. We simulate only one sample scenario.
+	// A byte with high bit set (0x80+) indicates continuation, but with no following byte it's invalid
+	invalidData := []byte{0x80, 0x80, 0x80, 0x80, 0x80, 0x80}
+	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(invalidData))
+	rec := httptest.NewRecorder()
+
+	m.handleFunc(rec, req)
+
+	assert.Equal(t, http.StatusBadRequest, rec.Code)
+	assert.True(t, strings.Contains(rec.Body.String(), "Decoded length error"),
+		"expected 'Decoded length error' error, got %q", rec.Body.String())
+}

@@ -7,14 +7,18 @@ package remote_write
 import (
 	"errors"
 	"time"
+
+	rw "github.com/elastic/beats/v7/metricbeat/module/prometheus/remote_write"
 )
 
 type config struct {
-	MetricsCount  bool          `config:"metrics_count"`
-	UseTypes      bool          `config:"use_types"`
-	RateCounters  bool          `config:"rate_counters"`
-	TypesPatterns TypesPatterns `config:"types_patterns" yaml:"types_patterns,omitempty"`
-	Period        time.Duration `config:"period"     validate:"positive"`
+	MetricsCount           bool          `config:"metrics_count"`
+	UseTypes               bool          `config:"use_types"`
+	RateCounters           bool          `config:"rate_counters"`
+	TypesPatterns          TypesPatterns `config:"types_patterns" yaml:"types_patterns,omitempty"`
+	Period                 time.Duration `config:"period"     validate:"positive"`
+	MaxCompressedBodyBytes int64         `config:"max_compressed_body_bytes"`
+	MaxDecodedBodyBytes    int64         `config:"max_decoded_body_bytes"`
 }
 
 type TypesPatterns struct {
@@ -26,7 +30,9 @@ var defaultConfig = config{
 	TypesPatterns: TypesPatterns{
 		CounterPatterns:   nil,
 		HistogramPatterns: nil},
-	Period: time.Second * 60,
+	Period:                 time.Second * 60,
+	MaxCompressedBodyBytes: rw.DefaultMaxCompressedBodyBytes,
+	MaxDecodedBodyBytes:    rw.DefaultMaxDecodedBodyBytes,
 }
 
 func (c *config) Validate() error {

@@ -57,6 +57,25 @@ remote_write:
         #insecure_skip_verify: true
 ```
 
+## Request size limits [_request_size_limits]
+
+```{applies_to}
+stack: ga 9.2.5, ga 9.3.1, ga 9.4
+```
+
+To protect against resource exhaustion from malicious or oversized payloads, the remote_write metricset enforces configurable size limits on incoming requests:
+
+* `max_compressed_body_bytes`: Maximum size of the compressed (snappy-encoded) request body in bytes. Requests exceeding this limit are rejected with HTTP 413 before being read into memory. Default: 2 MB (2097152 bytes).
+* `max_decoded_body_bytes`: Maximum size of the decompressed request body in bytes. The server checks the declared decoded size in the snappy header before allocating memory for decompression, preventing decompression bomb attacks. Default: 10 MB (10485760 bytes).
+
+```yaml
+- module: prometheus
+  metricsets: ["remote_write"]
+  host: "localhost"
+  port: "9201"
+  max_compressed_body_bytes: 2097152    # 2 MB (default)
+  max_decoded_body_bytes: 10485760      # 10 MB (default)
+```
 
 ## Histograms and types [_histograms_and_types_2]
 

@@ -11,8 +11,16 @@ import (
 )
 
 func init() {
+	// Create base config with size limits from x-pack defaults
+	baseConfig := rw.Config{
+		Host:                   "localhost",
+		Port:                   9201,
+		MaxCompressedBodyBytes: defaultConfig.MaxCompressedBodyBytes,
+		MaxDecodedBodyBytes:    defaultConfig.MaxDecodedBodyBytes,
+	}
+
 	mb.Registry.MustAddMetricSet("prometheus", "remote_write",
-		rw.MetricSetBuilder(remoteWriteEventsGeneratorFactory),
+		rw.MetricSetBuilderWithConfig(remoteWriteEventsGeneratorFactory, baseConfig),
 		mb.WithHostParser(parse.EmptyHostParser),
 
 		// must replace ensures that we are replacing the oss implementation with this one