Adding a limit on the maximum bytes accepted to be decoded in a resource write request.

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

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

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

@@ -341,6 +341,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
@@ -527,6 +538,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

@@ -713,6 +713,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

@@ -79,7 +79,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

@@ -66,7 +66,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
@@ -227,7 +227,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

@@ -88,7 +88,7 @@ func PatchResource(r rest.Patcher, scope RequestScope, admit admission.Interface
 		ctx := req.Context()
 		ctx = request.WithNamespace(ctx, namespace)
 
-		patchJS, err := readBody(req)
+		patchJS, 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"
@@ -71,6 +72,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) {
@@ -324,9 +327,23 @@ 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()
-	return ioutil.ReadAll(req.Body)
+	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 {

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

@@ -64,7 +64,7 @@ func UpdateResource(r rest.Updater, scope RequestScope, admit admission.Interfac
 		ctx := req.Context()
 		ctx = request.WithNamespace(ctx, namespace)
 
-		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

@@ -505,6 +505,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

@@ -163,6 +163,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
@@ -272,7 +275,13 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
 		// 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),
+		JSONPatchMaxCopyBytes: int64(10 * 1024 * 1024),
+		// 10MB is the recommended maximum client request size in bytes
+		// the etcd server should accept. Thus, we set it as the
+		// maximum bytes accepted to be decoded in a resource write
+		// request.  See
+		// https://github.com/etcd-io/etcd/blob/release-3.3/etcdserver/server.go#L90.
+		MaxRequestBodyBytes:          int64(10 * 1024 * 1024),
 		EnableAPIResponseCompression: utilfeature.DefaultFeatureGate.Enabled(features.APIResponseCompression),
 
 		// Default to treating watch as a long-running operation
@@ -490,6 +499,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

@@ -158,6 +158,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
@@ -340,6 +344,7 @@ func (s *GenericAPIServer) installAPIResources(apiPrefix string, apiGroupInfo *A
 			apiGroupVersion.OptionsExternalVersion = apiGroupInfo.OptionsExternalVersion
 		}
 		apiGroupVersion.OpenAPIModels = openAPIGroupModels
+		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,7 +45,12 @@ 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
-	TargetRAMMB           int
+	// 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
 }
 
 func NewServerRunOptions() *ServerRunOptions {
@@ -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,11 +116,27 @@ 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{
 				AdvertiseAddress:            net.ParseIP("192.168.10.10"),
@@ -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,
 			},
 		},