Merge pull request #73805 from caesarxuchao/resource-size-limit

Adding a limit on the size of request body the apiserver will decode for write operations

cmd/kube-apiserver/app/options/options_test.go

@@ -128,7 +128,8 @@ func TestAddFlags(t *testing.T) {
 			MaxMutatingRequestsInFlight: 200,
 			RequestTimeout:              time.Duration(2) * time.Minute,
 			MinRequestTimeout:           1800,
-			JSONPatchMaxCopyBytes:       int64(10 * 1024 * 1024),
+			JSONPatchMaxCopyBytes:       int64(100 * 1024 * 1024),
+			MaxRequestBodyBytes:         int64(100 * 1024 * 1024),
 		},
 		Admission: &kubeoptions.AdmissionOptions{
 			GenericAdmission: &apiserveroptions.AdmissionOptions{

staging/src/k8s.io/apimachinery/pkg/api/errors/errors.go

@@ -365,6 +365,17 @@ func NewTooManyRequestsError(message string) *StatusError {
 	}}
 }
 
+// NewRequestEntityTooLargeError returns an error indicating that the request
+// entity was too large.
+func NewRequestEntityTooLargeError(message string) *StatusError {
+	return &StatusError{metav1.Status{
+		Status:  metav1.StatusFailure,
+		Code:    http.StatusRequestEntityTooLarge,
+		Reason:  metav1.StatusReasonRequestEntityTooLarge,
+		Message: fmt.Sprintf("Request entity too large: %s", message),
+	}}
+}
+
 // NewGenericServerResponse returns a new error for server responses that are not in a recognizable form.
 func NewGenericServerResponse(code int, verb string, qualifiedResource schema.GroupResource, name, serverMessage string, retryAfterSeconds int, isUnexpectedResponse bool) *StatusError {
 	reason := metav1.StatusReasonUnknown
@@ -551,6 +562,19 @@ func IsTooManyRequests(err error) bool {
 	return false
 }
 
+// IsRequestEntityTooLargeError determines if err is an error which indicates
+// the request entity is too large.
+func IsRequestEntityTooLargeError(err error) bool {
+	if ReasonForError(err) == metav1.StatusReasonRequestEntityTooLarge {
+		return true
+	}
+	switch t := err.(type) {
+	case APIStatus:
+		return t.Status().Code == http.StatusRequestEntityTooLarge
+	}
+	return false
+}
+
 // IsUnexpectedServerError returns true if the server response was not in the expected API format,
 // and may be the result of another HTTP actor.
 func IsUnexpectedServerError(err error) bool {

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/types.go

@@ -746,6 +746,10 @@ const (
 	// Status code 406
 	StatusReasonNotAcceptable StatusReason = "NotAcceptable"
 
+	// StatusReasonRequestEntityTooLarge means that the request entity is too large.
+	// Status code 413
+	StatusReasonRequestEntityTooLarge StatusReason = "RequestEntityTooLarge"
+
 	// StatusReasonUnsupportedMediaType means that the content type sent by the client is not acceptable
 	// to the server - for instance, attempting to send protobuf for a resource that supports only json and yaml.
 	// API calls that return UnsupportedMediaType can never succeed.

staging/src/k8s.io/apiserver/pkg/endpoints/groupversion.go

@@ -87,6 +87,10 @@ type APIGroupVersion struct {
 
 	// OpenAPIModels exposes the OpenAPI models to each individual handler.
 	OpenAPIModels openapiproto.Models
+
+	// The limit on the request body size that would be accepted and decoded in a write request.
+	// 0 means no limit.
+	MaxRequestBodyBytes int64
 }
 
 // InstallREST registers the REST handlers (storage, watch, proxy and redirect) into a restful Container.

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/create.go

@@ -85,7 +85,7 @@ func createHandler(r rest.NamedCreater, scope RequestScope, admit admission.Inte
 
 		decoder := scope.Serializer.DecoderToVersion(s.Serializer, scope.HubGroupVersion)
 
-		body, err := readBody(req)
+		body, err := limitedReadBody(req, scope.MaxRequestBodyBytes)
 		if err != nil {
 			scope.err(err, w, req)
 			return

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/delete.go

@@ -72,7 +72,7 @@ func DeleteResource(r rest.GracefulDeleter, allowsOptions bool, scope RequestSco
 
 		options := &metav1.DeleteOptions{}
 		if allowsOptions {
-			body, err := readBody(req)
+			body, err := limitedReadBody(req, scope.MaxRequestBodyBytes)
 			if err != nil {
 				scope.err(err, w, req)
 				return
@@ -226,7 +226,7 @@ func DeleteCollection(r rest.CollectionDeleter, checkBody bool, scope RequestSco
 
 		options := &metav1.DeleteOptions{}
 		if checkBody {
-			body, err := readBody(req)
+			body, err := limitedReadBody(req, scope.MaxRequestBodyBytes)
 			if err != nil {
 				scope.err(err, w, req)
 				return

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/patch.go

@@ -96,7 +96,7 @@ func PatchResource(r rest.Patcher, scope RequestScope, admit admission.Interface
 			return
 		}
 
-		patchBytes, err := readBody(req)
+		patchBytes, err := limitedReadBody(req, scope.MaxRequestBodyBytes)
 		if err != nil {
 			scope.err(err, w, req)
 			return

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/rest.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net/http"
 	"net/url"
@@ -70,6 +71,8 @@ type RequestScope struct {
 
 	// HubGroupVersion indicates what version objects read from etcd or incoming requests should be converted to for in-memory handling.
 	HubGroupVersion schema.GroupVersion
+
+	MaxRequestBodyBytes int64
 }
 
 func (scope *RequestScope) err(err error, w http.ResponseWriter, req *http.Request) {
@@ -333,10 +336,24 @@ func summarizeData(data []byte, maxLength int) string {
 	}
 }
 
-func readBody(req *http.Request) ([]byte, error) {
+func limitedReadBody(req *http.Request, limit int64) ([]byte, error) {
 	defer req.Body.Close()
+	if limit <= 0 {
 		return ioutil.ReadAll(req.Body)
 	}
+	lr := &io.LimitedReader{
+		R: req.Body,
+		N: limit + 1,
+	}
+	data, err := ioutil.ReadAll(lr)
+	if err != nil {
+		return nil, err
+	}
+	if lr.N <= 0 {
+		return nil, errors.NewRequestEntityTooLargeError(fmt.Sprintf("limit is %d", limit))
+	}
+	return data, nil
+}
 
 func parseTimeout(str string) time.Duration {
 	if str != "" {

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/update.go

@@ -70,7 +70,7 @@ func UpdateResource(r rest.Updater, scope RequestScope, admit admission.Interfac
 			return
 		}
 
-		body, err := readBody(req)
+		body, err := limitedReadBody(req, scope.MaxRequestBodyBytes)
 		if err != nil {
 			scope.err(err, w, req)
 			return

staging/src/k8s.io/apiserver/pkg/endpoints/installer.go

@@ -511,6 +511,8 @@ func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storag
 		HubGroupVersion: schema.GroupVersion{Group: fqKindToRegister.Group, Version: runtime.APIVersionInternal},
 
 		MetaGroupVersion: metav1.SchemeGroupVersion,
+
+		MaxRequestBodyBytes: a.group.MaxRequestBodyBytes,
 	}
 	if a.group.MetaGroupVersion != nil {
 		reqScope.MetaGroupVersion = *a.group.MetaGroupVersion

staging/src/k8s.io/apiserver/pkg/server/config.go

@@ -159,6 +159,9 @@ type Config struct {
 	// patch may cause.
 	// This affects all places that applies json patch in the binary.
 	JSONPatchMaxCopyBytes int64
+	// The limit on the request body size that would be accepted and decoded in a write request.
+	// 0 means no limit.
+	MaxRequestBodyBytes int64
 	// MaxRequestsInFlight is the maximum number of parallel non-long-running requests. Every further
 	// request has to wait. Applies only to non-mutating requests.
 	MaxRequestsInFlight int
@@ -264,11 +267,21 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
 		RequestTimeout:              time.Duration(60) * time.Second,
 		MinRequestTimeout:           1800,
 		// 10MB is the recommended maximum client request size in bytes
-		// the etcd server should accept. Thus, we set it as the limit
+		// the etcd server should accept. See
-		// on the size increase the "copy" operations in a json patch
-		// can cause.  See
 		// https://github.com/etcd-io/etcd/blob/release-3.3/etcdserver/server.go#L90.
-		JSONPatchMaxCopyBytes:        int64(10 * 1024 * 1024),
+		// A request body might be encoded in json, and is converted to
+		// proto when persisted in etcd. Assuming the upper bound of
+		// the size ratio is 10:1, we set 100MB as the largest size
+		// increase the "copy" operations in a json patch may cause.
+		JSONPatchMaxCopyBytes: int64(100 * 1024 * 1024),
+		// 10MB is the recommended maximum client request size in bytes
+		// the etcd server should accept. See
+		// https://github.com/etcd-io/etcd/blob/release-3.3/etcdserver/server.go#L90.
+		// A request body might be encoded in json, and is converted to
+		// proto when persisted in etcd. Assuming the upper bound of
+		// the size ratio is 10:1, we set 100MB as the largest request
+		// body size to be accepted and decoded in a write request.
+		MaxRequestBodyBytes:          int64(100 * 1024 * 1024),
 		EnableAPIResponseCompression: utilfeature.DefaultFeatureGate.Enabled(features.APIResponseCompression),
 
 		// Default to treating watch as a long-running operation
@@ -461,6 +474,7 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 		DiscoveryGroupManager: discovery.NewRootAPIsHandler(c.DiscoveryAddresses, c.Serializer),
 
 		enableAPIResponseCompression: c.EnableAPIResponseCompression,
+		maxRequestBodyBytes:          c.MaxRequestBodyBytes,
 	}
 
 	for {

staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go

@@ -156,6 +156,10 @@ type GenericAPIServer struct {
 
 	// HandlerChainWaitGroup allows you to wait for all chain handlers finish after the server shutdown.
 	HandlerChainWaitGroup *utilwaitgroup.SafeWaitGroup
+
+	// The limit on the request body size that would be accepted and decoded in a write request.
+	// 0 means no limit.
+	maxRequestBodyBytes int64
 }
 
 // DelegationTarget is an interface which allows for composition of API servers with top level handling that works
@@ -336,6 +340,7 @@ func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *A
 			apiGroupVersion.OptionsExternalVersion = apiGroupInfo.OptionsExternalVersion
 		}
 		apiGroupVersion.OpenAPIModels = openAPIModels
+		apiGroupVersion.MaxRequestBodyBytes = s.maxRequestBodyBytes
 
 		if err := apiGroupVersion.InstallREST(s.Handler.GoRestfulContainer); err != nil {
 			return fmt.Errorf("unable to setup API %v: %v", apiGroupInfo, err)

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options.go

@@ -45,6 +45,11 @@ type ServerRunOptions struct {
 	// We intentionally did not add a flag for this option. Users of the
 	// apiserver library can wire it to a flag.
 	JSONPatchMaxCopyBytes int64
+	// The limit on the request body size that would be accepted and
+	// decoded in a write request. 0 means no limit.
+	// We intentionally did not add a flag for this option. Users of the
+	// apiserver library can wire it to a flag.
+	MaxRequestBodyBytes int64
 	TargetRAMMB         int
 }
 
@@ -56,6 +61,7 @@ func NewServerRunOptions() *ServerRunOptions {
 		RequestTimeout:              defaults.RequestTimeout,
 		MinRequestTimeout:           defaults.MinRequestTimeout,
 		JSONPatchMaxCopyBytes:       defaults.JSONPatchMaxCopyBytes,
+		MaxRequestBodyBytes:         defaults.MaxRequestBodyBytes,
 	}
 }
 
@@ -68,6 +74,7 @@ func (s *ServerRunOptions) ApplyTo(c *server.Config) error {
 	c.RequestTimeout = s.RequestTimeout
 	c.MinRequestTimeout = s.MinRequestTimeout
 	c.JSONPatchMaxCopyBytes = s.JSONPatchMaxCopyBytes
+	c.MaxRequestBodyBytes = s.MaxRequestBodyBytes
 	c.PublicAddress = s.AdvertiseAddress
 
 	return nil
@@ -116,6 +123,10 @@ func (s *ServerRunOptions) Validate() []error {
 		errors = append(errors, fmt.Errorf("--json-patch-max-copy-bytes can not be negative value"))
 	}
 
+	if s.MaxRequestBodyBytes < 0 {
+		errors = append(errors, fmt.Errorf("--max-resource-write-bytes can not be negative value"))
+	}
+
 	return errors
 }
 

staging/src/k8s.io/apiserver/pkg/server/options/server_run_options_test.go

@@ -41,6 +41,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				RequestTimeout:              time.Duration(2) * time.Minute,
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				TargetRAMMB:                 -65536,
 			},
 			expectErr: "--target-ram-mb can not be negative value",
@@ -55,6 +56,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				RequestTimeout:              time.Duration(2) * time.Minute,
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				TargetRAMMB:                 65536,
 			},
 			expectErr: "--max-requests-inflight can not be negative value",
@@ -69,6 +71,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				RequestTimeout:              time.Duration(2) * time.Minute,
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				TargetRAMMB:                 65536,
 			},
 			expectErr: "--max-mutating-requests-inflight can not be negative value",
@@ -83,6 +86,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				RequestTimeout:              -time.Duration(2) * time.Minute,
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				TargetRAMMB:                 65536,
 			},
 			expectErr: "--request-timeout can not be negative value",
@@ -97,6 +101,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				RequestTimeout:              time.Duration(2) * time.Minute,
 				MinRequestTimeout:           -1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				TargetRAMMB:                 65536,
 			},
 			expectErr: "--min-request-timeout can not be negative value",
@@ -111,10 +116,26 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				RequestTimeout:              time.Duration(2) * time.Minute,
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       -10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				TargetRAMMB:                 65536,
 			},
 			expectErr: "--json-patch-max-copy-bytes can not be negative value",
 		},
+		{
+			name: "Test when MaxRequestBodyBytes is negative value",
+			testOptions: &ServerRunOptions{
+				AdvertiseAddress:            net.ParseIP("192.168.10.10"),
+				CorsAllowedOriginList:       []string{"10.10.10.100", "10.10.10.200"},
+				MaxRequestsInFlight:         400,
+				MaxMutatingRequestsInFlight: 200,
+				RequestTimeout:              time.Duration(2) * time.Minute,
+				MinRequestTimeout:           1800,
+				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         -10 * 1024 * 1024,
+				TargetRAMMB:                 65536,
+			},
+			expectErr: "--max-resource-write-bytes can not be negative value",
+		},
 		{
 			name: "Test when ServerRunOptions is valid",
 			testOptions: &ServerRunOptions{
@@ -125,6 +146,7 @@ func TestServerRunOptionsValidate(t *testing.T) {
 				RequestTimeout:              time.Duration(2) * time.Minute,
 				MinRequestTimeout:           1800,
 				JSONPatchMaxCopyBytes:       10 * 1024 * 1024,
+				MaxRequestBodyBytes:         10 * 1024 * 1024,
 				TargetRAMMB:                 65536,
 			},
 		},

test/integration/apiserver/BUILD

@@ -11,6 +11,7 @@ go_test(
     srcs = [
         "apiserver_test.go",
         "main_test.go",
+        "max_request_body_bytes_test.go",
         "patch_test.go",
         "print_test.go",
     ],
@@ -19,6 +20,7 @@ go_test(
         "integration",
     ],
     deps = [
+        "//cmd/kube-apiserver/app/options:go_default_library",
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/kubectl/cmd/util:go_default_library",
         "//pkg/master:go_default_library",

test/integration/apiserver/max_request_body_bytes_test.go

@@ -0,0 +1,101 @@
+/*
+Copyright 2018 The Kubernetes 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 apiserver
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
+	"k8s.io/kubernetes/test/integration/framework"
+)
+
+// Tests that the apiserver limits the resource size in write operations.
+func TestMaxResourceSize(t *testing.T) {
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	clientSet, _ := framework.StartTestServer(t, stopCh, framework.TestServerSetup{
+		ModifyServerRunOptions: func(opts *options.ServerRunOptions) {
+			opts.GenericServerRunOptions.MaxRequestBodyBytes = 1024 * 1024
+		},
+	})
+
+	hugeData := []byte(strings.Repeat("x", 1024*1024+1))
+
+	c := clientSet.CoreV1().RESTClient()
+	t.Run("Create should limit the request body size", func(t *testing.T) {
+		err := c.Post().AbsPath(fmt.Sprintf("/api/v1/namespaces/default/pods")).
+			Body(hugeData).Do().Error()
+		if err == nil {
+			t.Fatalf("unexpected no error")
+		}
+		if !errors.IsRequestEntityTooLargeError(err) {
+			t.Errorf("expected requested entity too large err, got %v", err)
+
+		}
+	})
+
+	// Create a secret so we can update/patch/delete it.
+	secret := &v1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test",
+		},
+	}
+	_, err := clientSet.CoreV1().Secrets("default").Create(secret)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("Update should limit the request body size", func(t *testing.T) {
+		err = c.Put().AbsPath(fmt.Sprintf("/api/v1/namespaces/default/secrets/test")).
+			Body(hugeData).Do().Error()
+		if err == nil {
+			t.Fatalf("unexpected no error")
+		}
+		if !errors.IsRequestEntityTooLargeError(err) {
+			t.Errorf("expected requested entity too large err, got %v", err)
+
+		}
+	})
+	t.Run("Patch should limit the request body size", func(t *testing.T) {
+		err = c.Patch(types.JSONPatchType).AbsPath(fmt.Sprintf("/api/v1/namespaces/default/secrets/test")).
+			Body(hugeData).Do().Error()
+		if err == nil {
+			t.Fatalf("unexpected no error")
+		}
+		if !errors.IsRequestEntityTooLargeError(err) {
+			t.Errorf("expected requested entity too large err, got %v", err)
+
+		}
+	})
+	t.Run("Delete should limit the request body size", func(t *testing.T) {
+		err = c.Delete().AbsPath(fmt.Sprintf("/api/v1/namespaces/default/secrets/test")).
+			Body(hugeData).Do().Error()
+		if err == nil {
+			t.Fatalf("unexpected no error")
+		}
+		if !errors.IsRequestEntityTooLargeError(err) {
+			t.Errorf("expected requested entity too large err, got %v", err)
+
+		}
+	})
+}