diff --git a/go.mod b/go.mod index 3b7c9100a..a19e1c3ad 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index eb11d8147..2d3e40cbe 100644 --- a/go.sum +++ b/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= diff --git a/internal/server/event_server.go b/internal/server/event_server.go index 18a02b3b7..7d812d66b 100644 --- a/internal/server/event_server.go +++ b/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 diff --git a/internal/server/event_server_test.go b/internal/server/event_server_test.go index 25d1e060c..bcc0c0954 100644 --- a/internal/server/event_server_test.go +++ b/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 { diff --git a/internal/server/limits.go b/internal/server/limits.go new file mode 100644 index 000000000..ec565e2bc --- /dev/null +++ b/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 +} diff --git a/internal/server/receiver_handler_test.go b/internal/server/receiver_handler_test.go index afdcd8d30..0b946fe01 100644 --- a/internal/server/receiver_handler_test.go +++ b/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 *", diff --git a/internal/server/receiver_handlers.go b/internal/server/receiver_handlers.go index 323716efc..ef33a11c0 100644 --- a/internal/server/receiver_handlers.go +++ b/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) } diff --git a/internal/server/receiver_server.go b/internal/server/receiver_server.go index e0e705883..cda1aebb1 100644 --- a/internal/server/receiver_server.go +++ b/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() { diff --git a/main.go b/main.go index a9a6e3176..dda5fd2f8 100644 --- a/main.go +++ b/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")