[Prometheus] Remote - Write: Enforce Maxdecoding Length check (#48218) (#48362)

* fix the snappy decompression



* merging new config vars



---------



(cherry picked from commit 38cc702)

Signed-off-by: Andreas Gkizas <andreas.gkizas@elastic.co>
Co-authored-by: Andrew Gizas <andreas.gkizas@elastic.co>
Co-authored-by: Brandon Morelli <brandon.morelli@elastic.co>

Author: 3 people

changelog/fragments/1766493789-prometheus_remotew_maxdecodinglenght.yaml

@@ -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

docs/reference/metricbeat/metricbeat-metricset-prometheus-remote_write.md

@@ -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]
 

metricbeat/module/prometheus/remote_write/_meta/docs.md

@@ -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]
 

metricbeat/module/prometheus/remote_write/config.go

@@ -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,
 	}
 }

metricbeat/module/prometheus/remote_write/remote_write.go

@@ -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)