[release/v1.9.x] Unify and harden HTTP server request limits

go.mod

@@ -39,7 +39,7 @@ require (
 	github.com/nats-io/nats.go v1.52.0
 	github.com/nats-io/nkeys v0.4.16
 	github.com/onsi/gomega v1.42.1
-	github.com/sethvargo/go-limiter v1.1.0
+	github.com/sethvargo/go-limiter v1.2.0
 	github.com/slok/go-http-metrics v0.13.0
 	github.com/spf13/pflag v1.0.10
 	gitlab.com/gitlab-org/api/client-go v1.46.0

go.sum

@@ -414,8 +414,8 @@ github.com/santhosh-tekuri/jsonschema/v6 v6.0.1 h1:PKK9DyHxif4LZo+uQSgXNqs0jj5+x
 github.com/santhosh-tekuri/jsonschema/v6 v6.0.1/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU=
 github.com/sergi/go-diff v1.4.0 h1:n/SP9D5ad1fORl+llWyN+D6qoUETXNZARKjyY2/KVCw=
 github.com/sergi/go-diff v1.4.0/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4=
-github.com/sethvargo/go-limiter v1.1.0 h1:eLeZVQ2zqJOiEs03GguqmBVG6/T6lsZB+6PP1t7J6fA=
-github.com/sethvargo/go-limiter v1.1.0/go.mod h1:01b6tW25Ap+MeLYBuD4aHunMrJoNO5PVUFdS9rac3II=
+github.com/sethvargo/go-limiter v1.2.0 h1:XKL1vsaQ2zztVJrnZSzpRWCq/aLQqMllJ/3D/0bt/cw=
+github.com/sethvargo/go-limiter v1.2.0/go.mod h1:RC+qY2R7PAK81mBCrZEJlUlKnXSIqqQ8B7G44UgZ/1E=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w=
 github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g=

internal/server/event_server.go

@@ -17,12 +17,10 @@ limitations under the License.
 package server
 
 import (
-	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
-	"io"
 	"net/http"
 	"os"
 	"strings"
@@ -97,8 +95,11 @@ func (s *EventServer) ListenAndServe(stopCh <-chan struct{}, mdlw middleware.Mid
 	}
 	h := std.Handler(handlerID, mdlw, mux)
 	srv := &http.Server{
-		Addr:    s.port,
-		Handler: h,
+		Addr:              s.port,
+		Handler:           h,
+		ReadTimeout:       readTimeout,
+		ReadHeaderTimeout: readHeaderTimeout,
+		MaxHeaderBytes:    maxHeaderBytes,
 	}
 
 	go func() {
@@ -127,22 +128,13 @@ func (s *EventServer) ListenAndServe(stopCh <-chan struct{}, mdlw middleware.Mid
 // request context.
 func (s *EventServer) eventMiddleware(h http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		body, err := io.ReadAll(r.Body)
-		if err != nil {
-			s.logger.Error(err, "reading the request body failed")
-			w.WriteHeader(http.StatusBadRequest)
-			return
-		}
-		if err := r.Body.Close(); err != nil {
-			s.logger.Error(err, "closing the request body failed")
-			w.WriteHeader(http.StatusBadRequest)
+		body, ok := readRequestBodyWithLimit(s.logger, w, r)
+		if !ok {
 			return
 		}
-		r.Body = io.NopCloser(bytes.NewBuffer(body))
 
 		event := &eventv1.Event{}
-		err = json.Unmarshal(body, event)
-		if err != nil {
+		if err := json.Unmarshal(body, event); err != nil {
 			s.logger.Error(err, "decoding the request body failed")
 			w.WriteHeader(http.StatusBadRequest)
 			return

internal/server/event_server_test.go

@@ -630,6 +630,70 @@ func TestCleanupMetadata(t *testing.T) {
 	}
 }
 
+func TestEventMiddlewareMaxRequestSize(t *testing.T) {
+	// eventPayload returns a valid event JSON payload of exactly size bytes,
+	// padded through the message field.
+	eventPayload := func(g *WithT, size int) []byte {
+		event := &eventv1.Event{Message: "x"}
+		b, err := json.Marshal(event)
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(size).To(BeNumerically(">=", len(b)))
+		event.Message = strings.Repeat("x", size-len(b)+1)
+		b, err = json.Marshal(event)
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(b).To(HaveLen(size))
+		return b
+	}
+
+	tests := []struct {
+		name       string
+		body       func(g *WithT) []byte
+		wantStatus int
+		wantServed bool
+	}{
+		{
+			name:       "body at the limit is accepted",
+			body:       func(g *WithT) []byte { return eventPayload(g, maxRequestSizeBytes) },
+			wantStatus: http.StatusOK,
+			wantServed: true,
+		},
+		{
+			name:       "body over the limit is rejected",
+			body:       func(g *WithT) []byte { return eventPayload(g, maxRequestSizeBytes+1) },
+			wantStatus: http.StatusRequestEntityTooLarge,
+			wantServed: false,
+		},
+		{
+			name: "body over the limit is rejected before decoding",
+			body: func(g *WithT) []byte {
+				return bytes.Repeat([]byte("A"), maxRequestSizeBytes*2)
+			},
+			wantStatus: http.StatusRequestEntityTooLarge,
+			wantServed: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			var served bool
+			s := &EventServer{logger: log.Log}
+			handler := s.eventMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				served = true
+				w.WriteHeader(http.StatusOK)
+			}))
+
+			rr := httptest.NewRecorder()
+			req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(tt.body(g)))
+			handler.ServeHTTP(rr, req)
+
+			g.Expect(rr.Code).To(Equal(tt.wantStatus))
+			g.Expect(served).To(Equal(tt.wantServed))
+		})
+	}
+}
+
 func readManifest(path, namespace string) (*unstructured.Unstructured, error) {
 	data, err := os.ReadFile(path)
 	if err != nil {

internal/server/limits.go

@@ -0,0 +1,72 @@
+/*
+Copyright 2026 The Flux authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package server
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/go-logr/logr"
+)
+
+const (
+	// maxRequestSizeBytes is the maximum size of a request body accepted by
+	// the webhook receiver and the event server, capped to 3 MiB.
+	maxRequestSizeBytes = 3 * 1024 * 1024
+
+	// maxHeaderBytes is the maximum size of the request headers accepted by
+	// the webhook receiver and the event server, capped to 256 KiB.
+	maxHeaderBytes = 256 * 1024
+
+	// readHeaderTimeout is the maximum duration allowed for reading the
+	// request headers, capped to 10 seconds.
+	readHeaderTimeout = 10 * time.Second
+
+	// readTimeout is the maximum duration allowed for reading the entire
+	// request, including the body, capped to 30 seconds.
+	readTimeout = 30 * time.Second
+)
+
+// readRequestBodyWithLimit reads the request body up to maxRequestSizeBytes
+// and replaces r.Body with an in-memory reader over the bytes read, so that
+// handlers can read the body again. On failure it logs the error and
+// writes the 413 HTTP status code when the body exceeds the maximum
+// allowed size, 400 when reading the body fails.
+func readRequestBodyWithLimit(logger logr.Logger, w http.ResponseWriter, r *http.Request) ([]byte, bool) {
+	body, err := io.ReadAll(io.LimitReader(r.Body, maxRequestSizeBytes+1))
+	if err != nil {
+		logger.Error(err, "reading the request body failed")
+		w.WriteHeader(http.StatusBadRequest)
+		return nil, false
+	}
+	if len(body) > maxRequestSizeBytes {
+		logger.Error(fmt.Errorf("request body exceeds the maximum size of %d bytes", maxRequestSizeBytes),
+			"reading the request body failed")
+		w.WriteHeader(http.StatusRequestEntityTooLarge)
+		return nil, false
+	}
+	if err := r.Body.Close(); err != nil {
+		logger.Error(err, "closing the request body failed")
+		w.WriteHeader(http.StatusBadRequest)
+		return nil, false
+	}
+	r.Body = io.NopCloser(bytes.NewReader(body))
+	return body, true
+}

internal/server/receiver_handler_test.go

@@ -700,7 +700,7 @@ func Test_handlePayload(t *testing.T) {
 					Conditions:  []metav1.Condition{{Type: meta.ReadyCondition, Status: metav1.ConditionTrue}},
 				},
 			},
-			expectedResponseCode: http.StatusBadRequest,
+			expectedResponseCode: http.StatusRequestEntityTooLarge,
 		},
 		{
 			name: "resource matchLabels is ignored if name is not *",

internal/server/receiver_handlers.go

@@ -48,12 +48,9 @@ import (
 	apiv1 "github.com/fluxcd/notification-controller/api/v1"
 )
 
-const (
-	WebhookPathIndexKey string = ".metadata.webhookPath"
-
-	// maxRequestSizeBytes is the maximum size of a request to the API server
-	maxRequestSizeBytes = 3 * 1024 * 1024
-)
+// WebhookPathIndexKey is the key used for indexing the receivers
+// by their webhook path.
+const WebhookPathIndexKey string = ".metadata.webhookPath"
 
 // defaultFluxAPIVersions is a map of Flux API kinds to their API versions.
 var defaultFluxAPIVersions = map[string]string{
@@ -82,6 +79,11 @@ func (s *ReceiverServer) handlePayload(w http.ResponseWriter, r *http.Request) {
 
 	s.logger.Info(fmt.Sprintf("handling request: %s", digest))
 
+	// Enforce the request body size limit and return 413 if cap is reached.
+	if _, ok := readRequestBodyWithLimit(s.logger, w, r); !ok {
+		return
+	}
+
 	var allReceivers apiv1.ReceiverList
 	err := s.kubeClient.List(ctx, &allReceivers, client.MatchingFields{
 		WebhookPathIndexKey: r.RequestURI,
@@ -228,22 +230,13 @@ type validationResult struct {
 // validate authenticates the incoming request against the Receiver's
 // configuration. It returns nil on failure and a non-nil result on success.
 func (s *ReceiverServer) validate(ctx context.Context, receiver apiv1.Receiver, r *http.Request) (*validationResult, error) {
-	// Validate payload size before doing anything else in case we are being DDoSed.
-	b, err := io.ReadAll(io.LimitReader(r.Body, maxRequestSizeBytes+1))
-	if err != nil {
-		return nil, fmt.Errorf("failed to read request body: %w", err)
-	}
-	if len(b) > maxRequestSizeBytes {
-		return nil, fmt.Errorf("request body exceeds the maximum size of %d bytes", maxRequestSizeBytes)
-	}
-	r.Body = io.NopCloser(bytes.NewReader(b))
-
 	// Fetch the secret and extract the token, when a secretRef is set. Only
 	// generic-oidc receivers omit secretRef; they authenticate requests using
 	// the OIDC token instead of the webhook token.
 	var secret *corev1.Secret
 	var token string
 	if receiver.Spec.SecretRef != nil {
+		var err error
 		secret, err = s.secret(ctx, receiver)
 		if err != nil {
 			return nil, fmt.Errorf("unable to read secret, error: %w", err)
@@ -481,8 +474,7 @@ func (s *ReceiverServer) validate(ctx context.Context, receiver apiv1.Receiver,
 		raw, _ := base64.StdEncoding.DecodeString(p.Message.Data)
 
 		var d data
-		err = json.Unmarshal(raw, &d)
-		if err != nil {
+		if err := json.Unmarshal(raw, &d); err != nil {
 			return nil, fmt.Errorf("cannot decode GCR webhook body: %w", err)
 		}
 

internal/server/receiver_server.go

@@ -63,8 +63,11 @@ func (s *ReceiverServer) ListenAndServe(stopCh <-chan struct{}, mdlw middleware.
 	}
 	h := std.Handler(handlerID, mdlw, mux)
 	srv := &http.Server{
-		Addr:    s.port,
-		Handler: h,
+		Addr:              s.port,
+		Handler:           h,
+		ReadTimeout:       readTimeout,
+		ReadHeaderTimeout: readHeaderTimeout,
+		MaxHeaderBytes:    maxHeaderBytes,
 	}
 
 	go func() {

main.go

@@ -267,8 +267,14 @@ func main() {
 	// +kubebuilder:scaffold:builder
 
 	ctx := ctrl.SetupSignalHandler()
+	// Sweep the rate limiter store at every interval and purge entries
+	// older than two intervals, to bound the memory usage of the store.
+	// A bucket becomes irrelevant once the rate limit interval has passed,
+	// as the token gets replenished on the next event.
 	store, err := memorystore.New(&memorystore.Config{
-		Interval: rateLimitInterval,
+		Interval:      rateLimitInterval,
+		SweepInterval: rateLimitInterval,
+		SweepMinTTL:   2 * rateLimitInterval,
 	})
 	if err != nil {
 		setupLog.Error(err, "unable to create middleware store")