diff --git a/CHANGELOG.md b/CHANGELOG.md index b53819eb496..ed80654a0d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539 * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889 +* [BUGFIX] Distributor: Reject OTLP request bodies whose decompressed size exceeds `-distributor.otlp-max-recv-msg-size` instead of silently truncating them, and decode gzip-compressed OTLP protobuf bodies received over httpgrpc instead of treating them as empty. #7902 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880 diff --git a/pkg/util/http.go b/pkg/util/http.go index 0e7500716ea..ce7cab79992 100644 --- a/pkg/util/http.go +++ b/pkg/util/http.go @@ -181,8 +181,12 @@ func ParseProtoReader(ctx context.Context, reader io.Reader, expectedSize, maxSi func decompressRequest(reader io.Reader, expectedSize, maxSize int, compression CompressionType, sp opentracing.Span) (body []byte, err error) { defer func() { - if err != nil && len(body) > maxSize { + // The readers are capped at maxSize+1 bytes, so a body longer than + // maxSize was truncated and must be rejected even when reading it + // did not fail. + if len(body) > maxSize { err = fmt.Errorf(messageSizeLargerErrFmt, len(body), maxSize) + body = nil } }() if expectedSize > maxSize { @@ -254,6 +258,14 @@ func decompressFromBuffer(buffer *bytes.Buffer, maxSize int, compression Compres return nil, err } return body, nil + case Gzip: + gzReader, err := gzip.NewReader(bytes.NewReader(buffer.Bytes())) + if err != nil { + return nil, err + } + var buf bytes.Buffer + _, err = buf.ReadFrom(io.LimitReader(gzReader, int64(maxSize)+1)) + return buf.Bytes(), err } return nil, nil } diff --git a/pkg/util/http_test.go b/pkg/util/http_test.go index b45e0a8f0bd..c2c10b20f41 100644 --- a/pkg/util/http_test.go +++ b/pkg/util/http_test.go @@ -237,8 +237,68 @@ func TestParseProtoReader_GzipDecompressionBomb(t *testing.T) { var fromWire cortexpb.PreallocWriteRequest err = util.ParseProtoReader(context.Background(), io.NopCloser(&compressed), 0, maxSize, &fromWire, util.Gzip) - // The decompressed output should be limited to maxSize+1 bytes, causing a - // proto unmarshal error (not an OOM). The key assertion is that we don't - // allocate 1 MB of memory. - assert.NotNil(t, err) + // The decompressed output should be limited to maxSize+1 bytes (not an + // OOM) and the request rejected as too large. + require.Error(t, err) + assert.Contains(t, err.Error(), "received message larger than max") +} + +// streamReader hides the underlying reader's type, so ParseProtoReader reads it +// like an HTTP request body instead of using the BytesBuffer() fast path. +type streamReader struct{ r io.Reader } + +func (s streamReader) Read(p []byte) (int, error) { return s.r.Read(p) } + +func TestParseProtoReader_OversizeBodyIsRejected(t *testing.T) { + req := cortexpb.PreallocWriteRequest{} + for i := range 100 { + req.Timeseries = append(req.Timeseries, cortexpb.PreallocTimeseries{TimeSeries: &cortexpb.TimeSeries{ + Labels: []cortexpb.LabelAdapter{{Name: "__name__", Value: "metric_" + strconv.Itoa(i)}, {Name: "job", Value: "probe"}}, + Samples: []cortexpb.Sample{{Value: 1, TimestampMs: 1}}, + }}) + } + body, err := req.Marshal() + require.NoError(t, err) + + var gz bytes.Buffer + gzw := gzip.NewWriter(&gz) + _, err = gzw.Write(body) + require.NoError(t, err) + require.NoError(t, gzw.Close()) + require.Less(t, gz.Len(), len(body)) + + for _, tc := range []struct { + name string + compression util.CompressionType + data []byte + useBytesBuffer bool + }{ + {"noCompression", util.NoCompression, body, false}, + {"gzip", util.Gzip, gz.Bytes(), false}, + {"bytesbuffer gzip", util.Gzip, gz.Bytes(), true}, + } { + newReader := func() io.Reader { + if tc.useBytesBuffer { + return bytesBuffered{Buffer: bytes.NewBuffer(append([]byte(nil), tc.data...))} + } + return streamReader{bytes.NewReader(tc.data)} + } + + t.Run(tc.name+" within limit", func(t *testing.T) { + var fromWire cortexpb.PreallocWriteRequest + require.NoError(t, util.ParseProtoReader(context.Background(), newReader(), 0, len(body), &fromWire, tc.compression)) + require.Len(t, fromWire.Timeseries, 100) + }) + + t.Run(tc.name+" over limit", func(t *testing.T) { + // Every limit below the decoded size must be reported as a size error, + // whether or not the cut lands on a message boundary. + for maxSize := gz.Len(); maxSize < len(body); maxSize++ { + var fromWire cortexpb.PreallocWriteRequest + err := util.ParseProtoReader(context.Background(), newReader(), 0, maxSize, &fromWire, tc.compression) + require.Error(t, err, "maxSize=%d", maxSize) + require.Contains(t, err.Error(), "received message larger than max", "maxSize=%d", maxSize) + } + }) + } } diff --git a/pkg/util/push/otlp.go b/pkg/util/push/otlp.go index adb7b324219..0a4af52eca0 100644 --- a/pkg/util/push/otlp.go +++ b/pkg/util/push/otlp.go @@ -186,6 +186,10 @@ func decodeOTLPWriteRequest(ctx context.Context, r *http.Request, maxSize int) ( if err != nil { return req, err } + // The readers are capped at maxSize+1 bytes, so a longer body was truncated. + if buf.Len() > maxSize { + return req, fmt.Errorf("received message larger than max (%d vs %d)", buf.Len(), maxSize) + } return req, req.UnmarshalJSON(buf.Bytes()) } diff --git a/pkg/util/push/otlp_test.go b/pkg/util/push/otlp_test.go index 68bd7fbead8..4cce310bebe 100644 --- a/pkg/util/push/otlp_test.go +++ b/pkg/util/push/otlp_test.go @@ -1188,6 +1188,47 @@ func TestOTLPWriteHandler(t *testing.T) { } } +func TestOTLPWriteHandler_OversizeBodyIsRejected(t *testing.T) { + cfg := distributor.OTLPConfig{} + exportRequest := generateOTLPWriteRequest() + + for _, contentType := range []string{pbContentType, jsonContentType} { + for _, encodingType := range []string{"", "gzip"} { + t.Run(contentType+" "+encodingType, func(t *testing.T) { + plain, err := getOTLPHttpRequest(&exportRequest, contentType, "") + require.NoError(t, err) + decodedSize := int(plain.ContentLength) + + // The body is either sent without a Content-Length (uncompressed) + // or compressed below the limit, so only its decoded size is too large. + maxRecvMsgSize := decodedSize - 1 + req, err := getOTLPHttpRequest(&exportRequest, contentType, encodingType) + require.NoError(t, err) + if encodingType == "" { + req.ContentLength = -1 + } + require.Less(t, int(req.ContentLength), maxRecvMsgSize) + + push := func(context.Context, *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) { + t.Error("an oversize request must not be pushed") + return &cortexpb.WriteResponse{}, nil + } + overrides := validation.NewOverrides(querier.DefaultLimitsConfig(), nil) + handler := OTLPHandler(maxRecvMsgSize, overrides, cfg, nil, push, nil) + + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, req) + + resp := recorder.Result() + require.Equal(t, http.StatusBadRequest, resp.StatusCode) + b, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Contains(t, string(b), "received message larger than max") + }) + } + } +} + func generateOTLPWriteRequestWithSeries(numSeries, samplesPerSeries, numHistogram int) pmetricotlp.ExportRequest { d := pmetric.NewMetrics()