Upstream versity/versitygw state of the touched files before b237f0cf7a

s3api/controllers/bucket-put.go

@@ -0,0 +1,725 @@
+// Copyright 2023 Versity Software
+// This file is 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 controllers
+
+import (
+	"encoding/xml"
+	"errors"
+	"net/http"
+	"strings"
+
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/gofiber/fiber/v3"
+	"github.com/versity/versitygw/auth"
+	"github.com/versity/versitygw/debuglogger"
+	"github.com/versity/versitygw/s3api/utils"
+	"github.com/versity/versitygw/s3err"
+	"github.com/versity/versitygw/s3response"
+)
+
+func (c S3ApiController) PutBucketTagging(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+	isPublicBucket := utils.ContextKeyPublicBucket.IsSet(ctx)
+
+	err := c.verifyAccess(ctx, auth.AccessOptions{
+		Acl:             parsedAcl,
+		AclPermission:   auth.PermissionWrite,
+		IsRoot:          isRoot,
+		Acc:             acct,
+		Bucket:          bucket,
+		Actions:         []auth.Action{auth.PutBucketTaggingAction},
+		IsPublicRequest: isPublicBucket,
+	})
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	tagging, err := utils.ParseTagging(ctx.BodyRaw(), utils.TagLimitBucket)
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	err = c.be.PutBucketTagging(ctx.RequestCtx(), bucket, tagging)
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+			Status:      http.StatusNoContent,
+		},
+	}, err
+}
+
+func (c S3ApiController) PutBucketOwnershipControls(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+
+	if err := c.verifyAccess(ctx, auth.AccessOptions{
+		Acl:           parsedAcl,
+		AclPermission: auth.PermissionWrite,
+		IsRoot:        isRoot,
+		Acc:           acct,
+		Bucket:        bucket,
+		Actions:       []auth.Action{auth.PutBucketOwnershipControlsAction},
+	}); err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	var ownershipControls s3response.OwnershipControls
+	if err := xml.Unmarshal(ctx.BodyRaw(), &ownershipControls); err != nil {
+		debuglogger.Logf("failed to unmarshal request body: %v", err)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMalformedXML)
+	}
+
+	rulesCount := len(ownershipControls.Rules)
+	if rulesCount != 1 {
+		debuglogger.Logf("ownership control rules should be 1, got %v", rulesCount)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMalformedXML)
+	}
+
+	if !utils.IsValidOwnership(ownershipControls.Rules[0].ObjectOwnership) {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMalformedXML)
+	}
+
+	err := c.be.PutBucketOwnershipControls(ctx.RequestCtx(), bucket, ownershipControls.Rules[0].ObjectOwnership)
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+		},
+	}, err
+}
+
+func (c S3ApiController) PutBucketVersioning(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+	isPublicBucket := utils.ContextKeyPublicBucket.IsSet(ctx)
+
+	err := c.verifyAccess(ctx, auth.AccessOptions{
+		Acl:             parsedAcl,
+		AclPermission:   auth.PermissionWrite,
+		IsRoot:          isRoot,
+		Acc:             acct,
+		Bucket:          bucket,
+		Actions:         []auth.Action{auth.PutBucketVersioningAction},
+		IsPublicRequest: isPublicBucket,
+	})
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	var versioningConf types.VersioningConfiguration
+	err = xml.Unmarshal(ctx.BodyRaw(), &versioningConf)
+	if err != nil {
+		debuglogger.Logf("error unmarshalling versioning configuration: %v", err)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrInvalidRequest)
+	}
+
+	if versioningConf.Status != types.BucketVersioningStatusEnabled &&
+		versioningConf.Status != types.BucketVersioningStatusSuspended {
+		debuglogger.Logf("invalid versioning configuration status: %v", versioningConf.Status)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMalformedXML)
+	}
+
+	err = c.be.PutBucketVersioning(ctx.RequestCtx(), bucket, versioningConf.Status)
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+		},
+	}, err
+}
+
+func (c S3ApiController) PutObjectLockConfiguration(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+	isPublicBucket := utils.ContextKeyPublicBucket.IsSet(ctx)
+
+	if err := c.verifyAccess(ctx, auth.AccessOptions{
+		Acl:             parsedAcl,
+		AclPermission:   auth.PermissionWrite,
+		IsRoot:          isRoot,
+		Acc:             acct,
+		Bucket:          bucket,
+		Actions:         []auth.Action{auth.PutBucketObjectLockConfigurationAction},
+		IsPublicRequest: isPublicBucket,
+	}); err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	config, err := auth.ParseBucketLockConfigurationInput(ctx.BodyRaw())
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	err = c.be.PutObjectLockConfiguration(ctx.RequestCtx(), bucket, config)
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+		},
+	}, err
+}
+
+func (c S3ApiController) PutBucketCors(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+	isPublicBucket := utils.ContextKeyPublicBucket.IsSet(ctx)
+
+	err := c.verifyAccess(ctx, auth.AccessOptions{
+		Acl:             parsedAcl,
+		AclPermission:   auth.PermissionWrite,
+		IsRoot:          isRoot,
+		Acc:             acct,
+		Bucket:          bucket,
+		Actions:         []auth.Action{auth.PutBucketCorsAction},
+		IsPublicRequest: isPublicBucket,
+	})
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	body := ctx.BodyRaw()
+
+	var corsConfig auth.CORSConfiguration
+	err = xml.Unmarshal(body, &corsConfig)
+	if err != nil {
+		debuglogger.Logf("invalid CORS request body: %v", err)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMalformedXML)
+	}
+
+	// validate the CORS configuration rules
+	err = corsConfig.Validate()
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	err = c.be.PutBucketCors(ctx.RequestCtx(), bucket, body)
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+		},
+	}, err
+}
+
+func (c S3ApiController) PutBucketWebsite(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+	isPublicBucket := utils.ContextKeyPublicBucket.IsSet(ctx)
+
+	err := c.verifyAccess(ctx, auth.AccessOptions{
+		Acl:             parsedAcl,
+		AclPermission:   auth.PermissionWrite,
+		IsRoot:          isRoot,
+		Acc:             acct,
+		Bucket:          bucket,
+		Actions:         []auth.Action{auth.PutBucketWebsiteAction},
+		IsPublicRequest: isPublicBucket,
+	})
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	body := ctx.BodyRaw()
+	if len(body) > maxWebsiteConfigurationBytes {
+		debuglogger.Logf("the request size exceeded the 128KB limit: %d", len(body))
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetMaxMessageLengthExceeded(maxWebsiteConfigurationBytes)
+	}
+
+	var websiteConfig s3response.WebsiteConfiguration
+	err = xml.Unmarshal(body, &websiteConfig)
+	if err != nil {
+		debuglogger.Logf("invalid website config request body: %v", err)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMalformedXML)
+	}
+
+	err = websiteConfig.Validate()
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	err = c.be.PutBucketWebsite(ctx.RequestCtx(), bucket, body)
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+		},
+	}, err
+}
+
+func (c S3ApiController) PutBucketPolicy(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+
+	err := c.verifyAccess(ctx, auth.AccessOptions{
+		Acl:           parsedAcl,
+		AclPermission: auth.PermissionWrite,
+		IsRoot:        isRoot,
+		Acc:           acct,
+		Bucket:        bucket,
+		Actions:       []auth.Action{auth.PutBucketPolicyAction},
+	})
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	err = auth.ValidatePolicyDocument(ctx.BodyRaw(), bucket, c.iam)
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	err = c.be.PutBucketPolicy(ctx.RequestCtx(), bucket, ctx.BodyRaw())
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+			Status:      http.StatusNoContent,
+		},
+	}, err
+}
+
+func (c S3ApiController) PutBucketAcl(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	acl := types.BucketCannedACL(ctx.Get("X-Amz-Acl"))
+	grantFullControl := ctx.Get("X-Amz-Grant-Full-Control")
+	grantRead := ctx.Get("X-Amz-Grant-Read")
+	grantReadACP := ctx.Get("X-Amz-Grant-Read-Acp")
+	grantWrite := ctx.Get("X-Amz-Grant-Write")
+	grantWriteACP := ctx.Get("X-Amz-Grant-Write-Acp")
+	// context locals
+	parsedAcl := utils.ContextKeyParsedAcl.Get(ctx).(auth.ACL)
+	acct := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	isRoot := utils.ContextKeyIsRoot.Get(ctx).(bool)
+
+	grants := grantFullControl + grantRead + grantReadACP + grantWrite + grantWriteACP
+	var input *auth.PutBucketAclInput
+
+	err := c.verifyAccess(ctx,
+		auth.AccessOptions{
+			Acl:           parsedAcl,
+			AclPermission: auth.PermissionWriteAcp,
+			IsRoot:        isRoot,
+			Acc:           acct,
+			Bucket:        bucket,
+			Actions:       []auth.Action{auth.PutBucketAclAction},
+		})
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	if c.disableACL {
+		debuglogger.Logf("PutBucketAcl is not available, as ACLs are disabled at gateway level")
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrACLsDisabled)
+	}
+
+	err = auth.ValidateCannedACL(acl)
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	ownership, err := c.be.GetBucketOwnershipControls(ctx.RequestCtx(), bucket)
+	if err != nil && !errors.Is(err, s3err.GetAPIError(s3err.ErrOwnershipControlsNotFound)) {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+	if ownership == types.ObjectOwnershipBucketOwnerEnforced {
+		debuglogger.Logf("bucket acls are disabled")
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrAclNotSupported)
+	}
+
+	if len(ctx.BodyRaw()) > 0 {
+		var accessControlPolicy auth.AccessControlPolicy
+		err := xml.Unmarshal(ctx.BodyRaw(), &accessControlPolicy)
+		if err != nil {
+			debuglogger.Logf("error unmarshalling access control policy: %v", err)
+			return &Response{
+				MetaOpts: &MetaOptions{
+					BucketOwner: parsedAcl.Owner,
+				},
+			}, s3err.GetAPIError(s3err.ErrMalformedACL)
+		}
+
+		err = accessControlPolicy.Validate()
+		if err != nil {
+			debuglogger.Logf("invalid access control policy: %v", err)
+			return &Response{
+				MetaOpts: &MetaOptions{
+					BucketOwner: parsedAcl.Owner,
+				},
+			}, err
+		}
+
+		if *accessControlPolicy.Owner.ID != parsedAcl.Owner {
+			debuglogger.Logf("invalid access control policy owner id: %v, expected %v", *accessControlPolicy.Owner.ID, parsedAcl.Owner)
+			return &Response{
+				MetaOpts: &MetaOptions{
+					BucketOwner: parsedAcl.Owner,
+				},
+			}, s3err.APIError{
+				Code:           "InvalidArgument",
+				Description:    "Invalid id",
+				HTTPStatusCode: http.StatusBadRequest,
+			}
+		}
+
+		if grants+string(acl) != "" {
+			debuglogger.Logf("invalid request: %q (grants) %q (acl)",
+				grants, acl)
+			return &Response{
+				MetaOpts: &MetaOptions{
+					BucketOwner: parsedAcl.Owner,
+				},
+			}, s3err.GetAPIError(s3err.ErrUnexpectedContent)
+		}
+
+		input = &auth.PutBucketAclInput{
+			Bucket:              &bucket,
+			AccessControlPolicy: &accessControlPolicy,
+		}
+	} else if acl != "" {
+		if grants != "" {
+			debuglogger.Logf("invalid request: %q (grants) %q (acl)",
+				grants, acl)
+			return &Response{
+				MetaOpts: &MetaOptions{
+					BucketOwner: parsedAcl.Owner,
+				},
+			}, s3err.GetAPIError(s3err.ErrBothCannedAndHeaderGrants)
+		}
+
+		input = &auth.PutBucketAclInput{
+			Bucket: &bucket,
+			ACL:    types.BucketCannedACL(acl),
+		}
+	} else if grants != "" {
+		input = &auth.PutBucketAclInput{
+			Bucket:           &bucket,
+			GrantFullControl: &grantFullControl,
+			GrantRead:        &grantRead,
+			GrantReadACP:     &grantReadACP,
+			GrantWrite:       &grantWrite,
+			GrantWriteACP:    &grantWriteACP,
+		}
+	} else {
+		debuglogger.Logf("none of the bucket acl options has been specified: canned, req headers, req body")
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMissingSecurityHeader)
+	}
+
+	updAcl, err := auth.UpdateACL(input, parsedAcl, c.iam)
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, err
+	}
+
+	err = c.be.PutBucketAcl(ctx.RequestCtx(), bucket, updAcl)
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: parsedAcl.Owner,
+		},
+	}, err
+}
+
+func (c S3ApiController) CreateBucket(ctx fiber.Ctx) (*Response, error) {
+	bucket := ctx.Params("bucket")
+	acl := types.BucketCannedACL(c.getAclHeaderValue(ctx, "X-Amz-Acl"))
+	grantFullControl := c.getAclHeaderValue(ctx, "X-Amz-Grant-Full-Control")
+	grantRead := c.getAclHeaderValue(ctx, "X-Amz-Grant-Read")
+	grantReadACP := c.getAclHeaderValue(ctx, "X-Amz-Grant-Read-Acp")
+	grantWrite := c.getAclHeaderValue(ctx, "X-Amz-Grant-Write")
+	grantWriteACP := c.getAclHeaderValue(ctx, "X-Amz-Grant-Write-Acp")
+	lockEnabled := strings.EqualFold(ctx.Get("X-Amz-Bucket-Object-Lock-Enabled"), "true")
+	grants := grantFullControl + grantRead + grantReadACP + grantWrite + grantWriteACP
+	objectOwnership := types.ObjectOwnership(ctx.Get("X-Amz-Object-Ownership"))
+
+	if c.readonly {
+		return &Response{
+			MetaOpts: &MetaOptions{},
+		}, s3err.GetAPIError(s3err.ErrAccessDenied)
+	}
+
+	creator := utils.ContextKeyAccount.Get(ctx).(auth.Account)
+	// A backend that fixes bucket ownership picks the owner itself; otherwise
+	// it is whoever the admin API named, defaulting to the creator. The
+	// context is set either way: the storage backend reads the owner back out
+	// of it to chown the new bucket.
+	bucketOwner, fixedOwner := auth.ResolveFixedBucketOwner(c.iam)
+	if !fixedOwner {
+		bucketOwner = creator
+		if utils.ContextKeyBucketOwner.IsSet(ctx) {
+			bucketOwner = utils.ContextKeyBucketOwner.Get(ctx).(auth.Account)
+		}
+	}
+	utils.ContextKeyBucketOwner.Set(ctx, bucketOwner)
+	isRoot, _ := utils.ContextKeyIsRoot.Get(ctx).(bool)
+
+	if err := auth.VerifyCreateBucketAccess(ctx, c.iam, isRoot, creator, bucket); err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{},
+		}, err
+	}
+
+	// validate the bucket name
+	if ok := utils.IsValidBucketName(bucket); !ok {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: bucketOwner.Access,
+			},
+		}, s3err.GetBucketErr(s3err.ErrInvalidBucketName, bucket)
+	}
+
+	// both bucket canned ACL and acl grants is not allowed
+	if acl != "" && grants != "" {
+		debuglogger.Logf("invalid request: %q (grants) %q (acl)", grants, acl)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: bucketOwner.Access,
+			},
+		}, s3err.GetAPIError(s3err.ErrBothCannedAndHeaderGrants)
+	}
+
+	// validate bucket canned acl
+	err := auth.ValidateCannedACL(acl)
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: bucketOwner.Access,
+			},
+		}, err
+	}
+
+	// if bucket acl is 'private', object ownership should default to 'BucketOwnerPreferred'
+	if acl == types.BucketCannedACLPrivate && objectOwnership == "" {
+		objectOwnership = types.ObjectOwnershipBucketOwnerPreferred
+	}
+
+	// if object ownership is so far empty, it should default to BucketOwnerEnforced
+	if objectOwnership == "" {
+		objectOwnership = types.ObjectOwnershipBucketOwnerEnforced
+	}
+
+	// validate the object ownership value
+	if ok := utils.IsValidOwnership(objectOwnership); !ok {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: bucketOwner.Access,
+			},
+		}, s3err.GetInvalidArgObjectOwnership(string(objectOwnership))
+	}
+
+	// any bucket ACL(canned, grants) is not allowed with object ownership 'BucketOwnerEnforced'
+	// but there's an exception for 'private' bucket canned ACL,
+	// which is the effective default for all buckets. In this case
+	// the ACL(private canned) is allowed with 'BucketOwnerEnforced'
+	if acl != types.BucketCannedACLPrivate && string(acl)+grants != "" && objectOwnership == types.ObjectOwnershipBucketOwnerEnforced {
+		debuglogger.Logf("bucket acls are disabled for %v object ownership", objectOwnership)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: bucketOwner.Access,
+			},
+		}, s3err.GetAPIError(s3err.ErrInvalidBucketAclWithObjectOwnership)
+	}
+
+	var body s3response.CreateBucketConfiguration
+	if len(ctx.BodyRaw()) != 0 {
+		// request body is optional for CreateBucket
+		err := xml.Unmarshal(ctx.BodyRaw(), &body)
+		if err != nil {
+			debuglogger.Logf("failed to parse the request body: %v", err)
+			return &Response{
+				MetaOpts: &MetaOptions{
+					BucketOwner: bucketOwner.Access,
+				},
+			}, s3err.GetAPIError(s3err.ErrMalformedXML)
+		}
+
+		if body.LocationConstraint != nil {
+			region := utils.ContextKeyRegion.Get(ctx).(string)
+			if *body.LocationConstraint != region || *body.LocationConstraint == "us-east-1" {
+				debuglogger.Logf("invalid location constraint: %s", *body.LocationConstraint)
+				return &Response{
+					MetaOpts: &MetaOptions{
+						BucketOwner: bucketOwner.Access,
+					},
+				}, s3err.GetInvalidLocationConstraintErr(*body.LocationConstraint)
+			}
+		}
+	}
+
+	defACL := auth.ACL{
+		Owner: bucketOwner.Access,
+	}
+
+	updAcl, err := auth.UpdateACL(&auth.PutBucketAclInput{
+		GrantFullControl: &grantFullControl,
+		GrantRead:        &grantRead,
+		GrantReadACP:     &grantReadACP,
+		GrantWrite:       &grantWrite,
+		GrantWriteACP:    &grantWriteACP,
+		AccessControlPolicy: &auth.AccessControlPolicy{
+			Owner: &types.Owner{
+				ID: &bucketOwner.Access,
+			}},
+		ACL: types.BucketCannedACL(acl),
+	}, defACL, c.iam)
+	if err != nil {
+		debuglogger.Logf("failed to update bucket acl: %v", err)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: bucketOwner.Access,
+			},
+		}, err
+	}
+
+	err = c.be.CreateBucket(ctx.RequestCtx(), &s3.CreateBucketInput{
+		Bucket:                     &bucket,
+		ObjectOwnership:            objectOwnership,
+		ObjectLockEnabledForBucket: &lockEnabled,
+		CreateBucketConfiguration: &types.CreateBucketConfiguration{
+			Tags: body.TagSet,
+		},
+	}, updAcl)
+	if err != nil {
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: bucketOwner.Access,
+			},
+		}, err
+	}
+
+	return &Response{
+		MetaOpts: &MetaOptions{
+			BucketOwner: bucketOwner.Access,
+		},
+		Headers: map[string]*string{
+			"Location":         utils.GetStringPtr("/" + bucket),
+			"x-amz-bucket-arn": utils.GetStringPtr(auth.ResourceArnPrefix + bucket),
+		},
+	}, nil
+}