feat: Implement bucket ownership controls

Bucket ACLs are now disabled by default the same as AWS.
By default the object ownership is BucketOwnerEnforced
which means that bucket ACLs are disabled. If one attempts
to set bucket ACL the following error is returned both in
the gateway and on AWS:
	ErrAclNotSupported: {
		Code:           "AccessControlListNotSupported",
		Description:    "The bucket does not allow ACLs",
		HTTPStatusCode: http.StatusBadRequest,
	},

ACls can be enabled with PutBucketOwnershipControls

Changed bucket canned ACL translation

New backend interface methods:
PutBucketOwnershipControls
GetBucketOwnershipControls
DeleteBucketOwnershipControls

Added these to metrics

auth/acl.go

@@ -28,7 +28,6 @@ import (
 )
 
 type ACL struct {
-	ACL      types.BucketCannedACL
 	Owner    string
 	Grantees []Grantee
 }
@@ -94,12 +93,34 @@ func UpdateACL(input *s3.PutBucketAclInput, acl ACL, iam IAMService) ([]byte, er
 		return nil, s3err.GetAPIError(s3err.ErrAccessDenied)
 	}
 
+	defaultGrantees := []Grantee{
+		{
+			Permission: types.PermissionFullControl,
+			Access:     acl.Owner,
+		},
+	}
+
 	// if the ACL is specified, set the ACL, else replace the grantees
 	if input.ACL != "" {
-		acl.ACL = input.ACL
+		switch input.ACL {
-		acl.Grantees = []Grantee{}
+		case types.BucketCannedACLPublicRead:
+			defaultGrantees = append(defaultGrantees, Grantee{
+				Permission: types.PermissionRead,
+				Access:     "all-users",
+			})
+		case types.BucketCannedACLPublicReadWrite:
+			defaultGrantees = append(defaultGrantees, []Grantee{
+				{
+					Permission: types.PermissionRead,
+					Access:     "all-users",
+				},
+				{
+					Permission: types.PermissionWrite,
+					Access:     "all-users",
+				},
+			}...)
+		}
 	} else {
-		grantees := []Grantee{}
 		accs := []string{}
 
 		if input.GrantRead != nil || input.GrantReadACP != nil || input.GrantFullControl != nil || input.GrantWrite != nil || input.GrantWriteACP != nil {
@@ -108,31 +129,31 @@ func UpdateACL(input *s3.PutBucketAclInput, acl ACL, iam IAMService) ([]byte, er
 			if input.GrantFullControl != nil && *input.GrantFullControl != "" {
 				fullControlList = splitUnique(*input.GrantFullControl, ",")
 				for _, str := range fullControlList {
-					grantees = append(grantees, Grantee{Access: str, Permission: "FULL_CONTROL"})
+					defaultGrantees = append(defaultGrantees, Grantee{Access: str, Permission: "FULL_CONTROL"})
 				}
 			}
 			if input.GrantRead != nil && *input.GrantRead != "" {
 				readList = splitUnique(*input.GrantRead, ",")
 				for _, str := range readList {
-					grantees = append(grantees, Grantee{Access: str, Permission: "READ"})
+					defaultGrantees = append(defaultGrantees, Grantee{Access: str, Permission: "READ"})
 				}
 			}
 			if input.GrantReadACP != nil && *input.GrantReadACP != "" {
 				readACPList = splitUnique(*input.GrantReadACP, ",")
 				for _, str := range readACPList {
-					grantees = append(grantees, Grantee{Access: str, Permission: "READ_ACP"})
+					defaultGrantees = append(defaultGrantees, Grantee{Access: str, Permission: "READ_ACP"})
 				}
 			}
 			if input.GrantWrite != nil && *input.GrantWrite != "" {
 				writeList = splitUnique(*input.GrantWrite, ",")
 				for _, str := range writeList {
-					grantees = append(grantees, Grantee{Access: str, Permission: "WRITE"})
+					defaultGrantees = append(defaultGrantees, Grantee{Access: str, Permission: "WRITE"})
 				}
 			}
 			if input.GrantWriteACP != nil && *input.GrantWriteACP != "" {
 				writeACPList = splitUnique(*input.GrantWriteACP, ",")
 				for _, str := range writeACPList {
-					grantees = append(grantees, Grantee{Access: str, Permission: "WRITE_ACP"})
+					defaultGrantees = append(defaultGrantees, Grantee{Access: str, Permission: "WRITE_ACP"})
 				}
 			}
 
@@ -143,7 +164,7 @@ func UpdateACL(input *s3.PutBucketAclInput, acl ACL, iam IAMService) ([]byte, er
 				if grt.Grantee == nil || grt.Grantee.ID == nil || grt.Permission == "" {
 					return nil, s3err.GetAPIError(s3err.ErrInvalidRequest)
 				}
-				grantees = append(grantees, Grantee{Access: *grt.Grantee.ID, Permission: grt.Permission})
+				defaultGrantees = append(defaultGrantees, Grantee{Access: *grt.Grantee.ID, Permission: grt.Permission})
 				if _, ok := cache[*grt.Grantee.ID]; !ok {
 					cache[*grt.Grantee.ID] = true
 					accs = append(accs, *grt.Grantee.ID)
@@ -159,11 +180,10 @@ func UpdateACL(input *s3.PutBucketAclInput, acl ACL, iam IAMService) ([]byte, er
 		if len(accList) > 0 {
 			return nil, fmt.Errorf("accounts does not exist: %s", strings.Join(accList, ", "))
 		}
-
-		acl.Grantees = grantees
-		acl.ACL = ""
 	}
 
+	acl.Grantees = defaultGrantees
+
 	result, err := json.Marshal(acl)
 	if err != nil {
 		return nil, err
@@ -207,34 +227,21 @@ func splitUnique(s, divider string) []string {
 }
 
 func verifyACL(acl ACL, access string, permission types.Permission) error {
-	if acl.ACL != "" {
+	grantee := Grantee{Access: access, Permission: permission}
-		if (permission == "READ" || permission == "READ_ACP") && (acl.ACL != "public-read" && acl.ACL != "public-read-write") {
+	granteeFullCtrl := Grantee{Access: access, Permission: "FULL_CONTROL"}
-			return s3err.GetAPIError(s3err.ErrAccessDenied)
+	granteeAllUsers := Grantee{Access: "all-users", Permission: permission}
-		}
-		if (permission == "WRITE" || permission == "WRITE_ACP") && acl.ACL != "public-read-write" {
-			return s3err.GetAPIError(s3err.ErrAccessDenied)
-		}
 
+	isFound := false
+
+	for _, grt := range acl.Grantees {
+		if grt == grantee || grt == granteeFullCtrl || grt == granteeAllUsers {
+			isFound = true
+			break
+		}
+	}
+
+	if isFound {
 		return nil
-	} else {
-		if len(acl.Grantees) == 0 {
-			return nil
-		}
-		grantee := Grantee{Access: access, Permission: permission}
-		granteeFullCtrl := Grantee{Access: access, Permission: "FULL_CONTROL"}
-
-		isFound := false
-
-		for _, grt := range acl.Grantees {
-			if grt == grantee || grt == granteeFullCtrl {
-				isFound = true
-				break
-			}
-		}
-
-		if isFound {
-			return nil
-		}
 	}
 
 	return s3err.GetAPIError(s3err.ErrAccessDenied)
@@ -295,23 +302,16 @@ func VerifyAccess(ctx context.Context, be backend.Backend, opts AccessOptions) e
 	if opts.Acc.Role == RoleAdmin {
 		return nil
 	}
-	if opts.Acc.Access == opts.Acl.Owner {
-		return nil
-	}
 
 	policy, policyErr := be.GetBucketPolicy(ctx, opts.Bucket)
-	if policyErr != nil && !errors.Is(policyErr, s3err.GetAPIError(s3err.ErrNoSuchBucketPolicy)) {
+	if policyErr != nil {
-		return policyErr
+		if !errors.Is(policyErr, s3err.GetAPIError(s3err.ErrNoSuchBucketPolicy)) {
+			return policyErr
+		}
+	} else {
+		return VerifyBucketPolicy(policy, opts.Acc.Access, opts.Bucket, opts.Object, opts.Action)
 	}
 
-	// If bucket policy is not set and the ACL is default, only the owner has access
-	if errors.Is(policyErr, s3err.GetAPIError(s3err.ErrNoSuchBucketPolicy)) && opts.Acl.ACL == "" && len(opts.Acl.Grantees) == 0 {
-		return s3err.GetAPIError(s3err.ErrAccessDenied)
-	}
-
-	if err := VerifyBucketPolicy(policy, opts.Acc.Access, opts.Bucket, opts.Object, opts.Action); err != nil {
-		return err
-	}
 	if err := verifyACL(opts.Acl, opts.Acc.Access, opts.AclPermission); err != nil {
 		return err
 	}

auth/bucket_policy.go

@@ -134,11 +134,6 @@ func ValidatePolicyDocument(policyBin []byte, bucket string, iam IAMService) err
 }
 
 func VerifyBucketPolicy(policy []byte, access, bucket, object string, action Action) error {
-	// If bucket policy is not set
-	if policy == nil {
-		return nil
-	}
-
 	var bucketPolicy BucketPolicy
 	if err := json.Unmarshal(policy, &bucketPolicy); err != nil {
 		return err

auth/bucket_policy_actions.go

@@ -55,6 +55,8 @@ const (
 	GetObjectRetentionAction               Action = "s3:GetObjectRetention"
 	PutObjectRetentionAction               Action = "s3:PutObjectRetention"
 	BypassGovernanceRetentionAction        Action = "s3:BypassGovernanceRetention"
+	PutBucketOwnershipControlsAction       Action = "s3:PutBucketOwnershipControls"
+	GetBucketOwnershipControlsAction       Action = "s3:GetBucketOwnershipControls"
 	AllActions                             Action = "s3:*"
 )
 
@@ -91,6 +93,8 @@ var supportedActionList = map[Action]struct{}{
 	GetObjectRetentionAction:               {},
 	PutObjectRetentionAction:               {},
 	BypassGovernanceRetentionAction:        {},
+	PutBucketOwnershipControlsAction:       {},
+	GetBucketOwnershipControlsAction:       {},
 	AllActions:                             {},
 }
 

backend/azure/azure.go

@@ -53,6 +53,7 @@ type key string
 const (
 	keyAclCapital   key = "Acl"
 	keyAclLower     key = "acl"
+	keyOwnership    key = "Ownership"
 	keyTags         key = "Tags"
 	keyPolicy       key = "Policy"
 	keyBucketLock   key = "Bucket-Lock"
@@ -127,6 +128,7 @@ func (az *Azure) String() string {
 func (az *Azure) CreateBucket(ctx context.Context, input *s3.CreateBucketInput, acl []byte) error {
 	meta := map[string]*string{
 		string(keyAclCapital): backend.GetStringPtr(string(acl)),
+		string(keyOwnership):  backend.GetStringPtr(string(input.ObjectOwnership)),
 	}
 
 	acct, ok := ctx.Value("account").(auth.Account)
@@ -251,6 +253,67 @@ func (az *Azure) DeleteBucket(ctx context.Context, input *s3.DeleteBucketInput)
 	return azureErrToS3Err(err)
 }
 
+func (az *Azure) PutBucketOwnershipControls(ctx context.Context, bucket string, ownership types.ObjectOwnership) error {
+	client, err := az.getContainerClient(bucket)
+	if err != nil {
+		return err
+	}
+
+	resp, err := client.GetProperties(ctx, &container.GetPropertiesOptions{})
+	if err != nil {
+		return azureErrToS3Err(err)
+	}
+	resp.Metadata[string(keyOwnership)] = backend.GetStringPtr(string(ownership))
+
+	_, err = client.SetMetadata(ctx, &container.SetMetadataOptions{Metadata: resp.Metadata})
+	if err != nil {
+		return azureErrToS3Err(err)
+	}
+
+	return nil
+}
+
+func (az *Azure) GetBucketOwnershipControls(ctx context.Context, bucket string) (types.ObjectOwnership, error) {
+	var ownship types.ObjectOwnership
+	client, err := az.getContainerClient(bucket)
+	if err != nil {
+		return ownship, err
+	}
+
+	resp, err := client.GetProperties(ctx, &container.GetPropertiesOptions{})
+	if err != nil {
+		return ownship, azureErrToS3Err(err)
+	}
+
+	ownership, ok := resp.Metadata[string(keyOwnership)]
+	if !ok {
+		return ownship, s3err.GetAPIError(s3err.ErrOwnershipControlsNotFound)
+	}
+
+	return types.ObjectOwnership(*ownership), nil
+}
+
+func (az *Azure) DeleteBucketOwnershipControls(ctx context.Context, bucket string) error {
+	client, err := az.getContainerClient(bucket)
+	if err != nil {
+		return err
+	}
+
+	resp, err := client.GetProperties(ctx, &container.GetPropertiesOptions{})
+	if err != nil {
+		return azureErrToS3Err(err)
+	}
+
+	delete(resp.Metadata, string(keyOwnership))
+
+	_, err = client.SetMetadata(ctx, &container.SetMetadataOptions{Metadata: resp.Metadata})
+	if err != nil {
+		return azureErrToS3Err(err)
+	}
+
+	return nil
+}
+
 func (az *Azure) PutObject(ctx context.Context, po *s3.PutObjectInput) (string, error) {
 	tags, err := parseTags(po.Tagging)
 	if err != nil {

backend/backend.go

@@ -21,6 +21,7 @@ import (
 	"io"
 
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/versity/versitygw/s3err"
 	"github.com/versity/versitygw/s3response"
 	"github.com/versity/versitygw/s3select"
@@ -43,6 +44,9 @@ type Backend interface {
 	PutBucketPolicy(_ context.Context, bucket string, policy []byte) error
 	GetBucketPolicy(_ context.Context, bucket string) ([]byte, error)
 	DeleteBucketPolicy(_ context.Context, bucket string) error
+	PutBucketOwnershipControls(_ context.Context, bucket string, ownership types.ObjectOwnership) error
+	GetBucketOwnershipControls(_ context.Context, bucket string) (types.ObjectOwnership, error)
+	DeleteBucketOwnershipControls(_ context.Context, bucket string) error
 
 	// multipart operations
 	CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput) (*s3.CreateMultipartUploadOutput, error)
@@ -138,6 +142,15 @@ func (BackendUnsupported) GetBucketPolicy(_ context.Context, bucket string) ([]b
 func (BackendUnsupported) DeleteBucketPolicy(_ context.Context, bucket string) error {
 	return s3err.GetAPIError(s3err.ErrNotImplemented)
 }
+func (BackendUnsupported) PutBucketOwnershipControls(_ context.Context, bucket string, ownership types.ObjectOwnership) error {
+	return s3err.GetAPIError(s3err.ErrNotImplemented)
+}
+func (BackendUnsupported) GetBucketOwnershipControls(_ context.Context, bucket string) (types.ObjectOwnership, error) {
+	return types.ObjectOwnershipBucketOwnerEnforced, s3err.GetAPIError(s3err.ErrNotImplemented)
+}
+func (BackendUnsupported) DeleteBucketOwnershipControls(_ context.Context, bucket string) error {
+	return s3err.GetAPIError(s3err.ErrNotImplemented)
+}
 
 func (BackendUnsupported) CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput) (*s3.CreateMultipartUploadOutput, error) {
 	return nil, s3err.GetAPIError(s3err.ErrNotImplemented)

backend/posix/posix.go

@@ -75,6 +75,7 @@ const (
 	contentEncHdr       = "content-encoding"
 	emptyMD5            = "d41d8cd98f00b204e9800998ecf8427e"
 	aclkey              = "acl"
+	ownershipkey        = "ownership"
 	etagkey             = "etag"
 	policykey           = "policy"
 	bucketLockKey       = "bucket-lock"
@@ -245,6 +246,9 @@ func (p *Posix) CreateBucket(ctx context.Context, input *s3.CreateBucketInput, a
 	if err := p.meta.StoreAttribute(bucket, "", aclkey, acl); err != nil {
 		return fmt.Errorf("set acl: %w", err)
 	}
+	if err := p.meta.StoreAttribute(bucket, "", ownershipkey, []byte(input.ObjectOwnership)); err != nil {
+		return fmt.Errorf("set ownership: %w", err)
+	}
 
 	if input.ObjectLockEnabledForBucket != nil && *input.ObjectLockEnabledForBucket {
 		now := time.Now()
@@ -304,6 +308,61 @@ func (p *Posix) DeleteBucket(_ context.Context, input *s3.DeleteBucketInput) err
 	return nil
 }
 
+func (p *Posix) PutBucketOwnershipControls(_ context.Context, bucket string, ownership types.ObjectOwnership) error {
+	_, err := os.Stat(bucket)
+	if errors.Is(err, fs.ErrNotExist) {
+		return s3err.GetAPIError(s3err.ErrNoSuchBucket)
+	}
+	if err != nil {
+		return fmt.Errorf("stat bucket: %w", err)
+	}
+
+	if err := p.meta.StoreAttribute(bucket, "", ownershipkey, []byte(ownership)); err != nil {
+		return fmt.Errorf("set ownership: %w", err)
+	}
+
+	return nil
+}
+func (p *Posix) GetBucketOwnershipControls(_ context.Context, bucket string) (types.ObjectOwnership, error) {
+	var ownship types.ObjectOwnership
+	_, err := os.Stat(bucket)
+	if errors.Is(err, fs.ErrNotExist) {
+		return ownship, s3err.GetAPIError(s3err.ErrNoSuchBucket)
+	}
+	if err != nil {
+		return ownship, fmt.Errorf("stat bucket: %w", err)
+	}
+
+	ownership, err := p.meta.RetrieveAttribute(bucket, "", ownershipkey)
+	if errors.Is(err, meta.ErrNoSuchKey) {
+		return ownship, s3err.GetAPIError(s3err.ErrOwnershipControlsNotFound)
+	}
+	if err != nil {
+		return ownship, fmt.Errorf("get bucket ownership status: %w", err)
+	}
+
+	return types.ObjectOwnership(ownership), nil
+}
+func (p *Posix) DeleteBucketOwnershipControls(_ context.Context, bucket string) error {
+	_, err := os.Stat(bucket)
+	if errors.Is(err, fs.ErrNotExist) {
+		return s3err.GetAPIError(s3err.ErrNoSuchBucket)
+	}
+	if err != nil {
+		return fmt.Errorf("stat bucket: %w", err)
+	}
+
+	if err := p.meta.DeleteAttribute(bucket, "", ownershipkey); err != nil {
+		if errors.Is(err, meta.ErrNoSuchKey) {
+			return nil
+		}
+
+		return fmt.Errorf("delete ownership: %w", err)
+	}
+
+	return nil
+}
+
 func (p *Posix) CreateMultipartUpload(ctx context.Context, mpu *s3.CreateMultipartUploadInput) (*s3.CreateMultipartUploadOutput, error) {
 	if mpu.Bucket == nil {
 		return nil, s3err.GetAPIError(s3err.ErrInvalidBucketName)

backend/s3proxy/s3.go

@@ -55,6 +55,8 @@ type S3Proxy struct {
 	debug           bool
 }
 
+var _ backend.Backend = &S3Proxy{}
+
 func New(access, secret, endpoint, region string, disableChecksum, sslSkipVerify, debug bool) (*S3Proxy, error) {
 	s := &S3Proxy{
 		access:          access,
@@ -128,6 +130,37 @@ func (s *S3Proxy) DeleteBucket(ctx context.Context, input *s3.DeleteBucketInput)
 	return handleError(err)
 }
 
+func (s *S3Proxy) PutBucketOwnershipControls(ctx context.Context, bucket string, ownership types.ObjectOwnership) error {
+	_, err := s.client.PutBucketOwnershipControls(ctx, &s3.PutBucketOwnershipControlsInput{
+		Bucket: &bucket,
+		OwnershipControls: &types.OwnershipControls{
+			Rules: []types.OwnershipControlsRule{
+				{
+					ObjectOwnership: ownership,
+				},
+			},
+		},
+	})
+	return handleError(err)
+}
+
+func (s *S3Proxy) GetBucketOwnershipControls(ctx context.Context, bucket string) (types.ObjectOwnership, error) {
+	var ownship types.ObjectOwnership
+	resp, err := s.client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+		Bucket: &bucket,
+	})
+	if err != nil {
+		return ownship, handleError(err)
+	}
+	return resp.OwnershipControls.Rules[0].ObjectOwnership, nil
+}
+func (s *S3Proxy) DeleteBucketOwnershipControls(ctx context.Context, bucket string) error {
+	_, err := s.client.DeleteBucketOwnershipControls(ctx, &s3.DeleteBucketOwnershipControlsInput{
+		Bucket: &bucket,
+	})
+	return handleError(err)
+}
+
 func (s *S3Proxy) CreateMultipartUpload(ctx context.Context, input *s3.CreateMultipartUploadInput) (*s3.CreateMultipartUploadOutput, error) {
 	out, err := s.client.CreateMultipartUpload(ctx, input)
 	return out, handleError(err)

metrics/actions.go

@@ -24,51 +24,54 @@ var (
 )
 
 var (
-	ActionUndetected                 = "ActionUnDetected"
+	ActionUndetected                    = "ActionUnDetected"
-	ActionAbortMultipartUpload       = "s3_AbortMultipartUpload"
+	ActionAbortMultipartUpload          = "s3_AbortMultipartUpload"
-	ActionCompleteMultipartUpload    = "s3_CompleteMultipartUpload"
+	ActionCompleteMultipartUpload       = "s3_CompleteMultipartUpload"
-	ActionCopyObject                 = "s3_CopyObject"
+	ActionCopyObject                    = "s3_CopyObject"
-	ActionCreateBucket               = "s3_CreateBucket"
+	ActionCreateBucket                  = "s3_CreateBucket"
-	ActionCreateMultipartUpload      = "s3_CreateMultipartUpload"
+	ActionCreateMultipartUpload         = "s3_CreateMultipartUpload"
-	ActionDeleteBucket               = "s3_DeleteBucket"
+	ActionDeleteBucket                  = "s3_DeleteBucket"
-	ActionDeleteBucketPolicy         = "s3_DeleteBucketPolicy"
+	ActionDeleteBucketPolicy            = "s3_DeleteBucketPolicy"
-	ActionDeleteBucketTagging        = "s3_DeleteBucketTagging"
+	ActionDeleteBucketTagging           = "s3_DeleteBucketTagging"
-	ActionDeleteObject               = "s3_DeleteObject"
+	ActionDeleteObject                  = "s3_DeleteObject"
-	ActionDeleteObjectTagging        = "s3_DeleteObjectTagging"
+	ActionDeleteObjectTagging           = "s3_DeleteObjectTagging"
-	ActionDeleteObjects              = "s3_DeleteObjects"
+	ActionDeleteObjects                 = "s3_DeleteObjects"
-	ActionGetBucketAcl               = "s3_GetBucketAcl"
+	ActionGetBucketAcl                  = "s3_GetBucketAcl"
-	ActionGetBucketPolicy            = "s3_GetBucketPolicy"
+	ActionGetBucketPolicy               = "s3_GetBucketPolicy"
-	ActionGetBucketTagging           = "s3_GetBucketTagging"
+	ActionGetBucketTagging              = "s3_GetBucketTagging"
-	ActionGetBucketVersioning        = "s3_GetBucketVersioning"
+	ActionGetBucketVersioning           = "s3_GetBucketVersioning"
-	ActionGetObject                  = "s3_GetObject"
+	ActionGetObject                     = "s3_GetObject"
-	ActionGetObjectAcl               = "s3_GetObjectAcl"
+	ActionGetObjectAcl                  = "s3_GetObjectAcl"
-	ActionGetObjectAttributes        = "s3_GetObjectAttributes"
+	ActionGetObjectAttributes           = "s3_GetObjectAttributes"
-	ActionGetObjectLegalHold         = "s3_GetObjectLegalHold"
+	ActionGetObjectLegalHold            = "s3_GetObjectLegalHold"
-	ActionGetObjectLockConfiguration = "s3_GetObjectLockConfiguration"
+	ActionGetObjectLockConfiguration    = "s3_GetObjectLockConfiguration"
-	ActionGetObjectRetention         = "s3_GetObjectRetention"
+	ActionGetObjectRetention            = "s3_GetObjectRetention"
-	ActionGetObjectTagging           = "s3_GetObjectTagging"
+	ActionGetObjectTagging              = "s3_GetObjectTagging"
-	ActionHeadBucket                 = "s3_HeadBucket"
+	ActionHeadBucket                    = "s3_HeadBucket"
-	ActionHeadObject                 = "s3_HeadObject"
+	ActionHeadObject                    = "s3_HeadObject"
-	ActionListAllMyBuckets           = "s3_ListAllMyBuckets"
+	ActionListAllMyBuckets              = "s3_ListAllMyBuckets"
-	ActionListMultipartUploads       = "s3_ListMultipartUploads"
+	ActionListMultipartUploads          = "s3_ListMultipartUploads"
-	ActionListObjectVersions         = "s3_ListObjectVersions"
+	ActionListObjectVersions            = "s3_ListObjectVersions"
-	ActionListObjects                = "s3_ListObjects"
+	ActionListObjects                   = "s3_ListObjects"
-	ActionListObjectsV2              = "s3_ListObjectsV2"
+	ActionListObjectsV2                 = "s3_ListObjectsV2"
-	ActionListParts                  = "s3_ListParts"
+	ActionListParts                     = "s3_ListParts"
-	ActionPutBucketAcl               = "s3_PutBucketAcl"
+	ActionPutBucketAcl                  = "s3_PutBucketAcl"
-	ActionPutBucketPolicy            = "s3_PutBucketPolicy"
+	ActionPutBucketPolicy               = "s3_PutBucketPolicy"
-	ActionPutBucketTagging           = "s3_PutBucketTagging"
+	ActionPutBucketTagging              = "s3_PutBucketTagging"
-	ActionPutBucketVersioning        = "s3_PutBucketVersioning"
+	ActionPutBucketVersioning           = "s3_PutBucketVersioning"
-	ActionPutObject                  = "s3_PutObject"
+	ActionPutObject                     = "s3_PutObject"
-	ActionPutObjectAcl               = "s3_PutObjectAcl"
+	ActionPutObjectAcl                  = "s3_PutObjectAcl"
-	ActionPutObjectLegalHold         = "s3_PutObjectLegalHold"
+	ActionPutObjectLegalHold            = "s3_PutObjectLegalHold"
-	ActionPutObjectLockConfiguration = "s3_PutObjectLockConfiguration"
+	ActionPutObjectLockConfiguration    = "s3_PutObjectLockConfiguration"
-	ActionPutObjectRetention         = "s3_PutObjectRetention"
+	ActionPutObjectRetention            = "s3_PutObjectRetention"
-	ActionPutObjectTagging           = "s3_PutObjectTagging"
+	ActionPutObjectTagging              = "s3_PutObjectTagging"
-	ActionRestoreObject              = "s3_RestoreObject"
+	ActionRestoreObject                 = "s3_RestoreObject"
-	ActionSelectObjectContent        = "s3_SelectObjectContent"
+	ActionSelectObjectContent           = "s3_SelectObjectContent"
-	ActionUploadPart                 = "s3_UploadPart"
+	ActionUploadPart                    = "s3_UploadPart"
-	ActionUploadPartCopy             = "s3_UploadPartCopy"
+	ActionUploadPartCopy                = "s3_UploadPartCopy"
+	ActionPutBucketOwnershipControls    = "s3_PutBucketOwnershipControls"
+	ActionGetBucketOwnershipControls    = "s3_GetBucketOwnershipControls"
+	ActionDeleteBucketOwnershipControls = "s3_DeleteBucketOwnershipControls"
 )
 
 func init() {

s3api/controllers/backend_moq_test.go

@@ -7,6 +7,7 @@ import (
 	"bufio"
 	"context"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/versity/versitygw/backend"
 	"github.com/versity/versitygw/s3response"
 	"io"
@@ -44,6 +45,9 @@ var _ backend.Backend = &BackendMock{}
 //			DeleteBucketFunc: func(contextMoqParam context.Context, deleteBucketInput *s3.DeleteBucketInput) error {
 //				panic("mock out the DeleteBucket method")
 //			},
+//			DeleteBucketOwnershipControlsFunc: func(contextMoqParam context.Context, bucket string) error {
+//				panic("mock out the DeleteBucketOwnershipControls method")
+//			},
 //			DeleteBucketPolicyFunc: func(contextMoqParam context.Context, bucket string) error {
 //				panic("mock out the DeleteBucketPolicy method")
 //			},
@@ -62,6 +66,9 @@ var _ backend.Backend = &BackendMock{}
 //			GetBucketAclFunc: func(contextMoqParam context.Context, getBucketAclInput *s3.GetBucketAclInput) ([]byte, error) {
 //				panic("mock out the GetBucketAcl method")
 //			},
+//			GetBucketOwnershipControlsFunc: func(contextMoqParam context.Context, bucket string) (types.ObjectOwnership, error) {
+//				panic("mock out the GetBucketOwnershipControls method")
+//			},
 //			GetBucketPolicyFunc: func(contextMoqParam context.Context, bucket string) ([]byte, error) {
 //				panic("mock out the GetBucketPolicy method")
 //			},
@@ -122,6 +129,9 @@ var _ backend.Backend = &BackendMock{}
 //			PutBucketAclFunc: func(contextMoqParam context.Context, bucket string, data []byte) error {
 //				panic("mock out the PutBucketAcl method")
 //			},
+//			PutBucketOwnershipControlsFunc: func(contextMoqParam context.Context, bucket string, ownership types.ObjectOwnership) error {
+//				panic("mock out the PutBucketOwnershipControls method")
+//			},
 //			PutBucketPolicyFunc: func(contextMoqParam context.Context, bucket string, policy []byte) error {
 //				panic("mock out the PutBucketPolicy method")
 //			},
@@ -195,6 +205,9 @@ type BackendMock struct {
 	// DeleteBucketFunc mocks the DeleteBucket method.
 	DeleteBucketFunc func(contextMoqParam context.Context, deleteBucketInput *s3.DeleteBucketInput) error
 
+	// DeleteBucketOwnershipControlsFunc mocks the DeleteBucketOwnershipControls method.
+	DeleteBucketOwnershipControlsFunc func(contextMoqParam context.Context, bucket string) error
+
 	// DeleteBucketPolicyFunc mocks the DeleteBucketPolicy method.
 	DeleteBucketPolicyFunc func(contextMoqParam context.Context, bucket string) error
 
@@ -213,6 +226,9 @@ type BackendMock struct {
 	// GetBucketAclFunc mocks the GetBucketAcl method.
 	GetBucketAclFunc func(contextMoqParam context.Context, getBucketAclInput *s3.GetBucketAclInput) ([]byte, error)
 
+	// GetBucketOwnershipControlsFunc mocks the GetBucketOwnershipControls method.
+	GetBucketOwnershipControlsFunc func(contextMoqParam context.Context, bucket string) (types.ObjectOwnership, error)
+
 	// GetBucketPolicyFunc mocks the GetBucketPolicy method.
 	GetBucketPolicyFunc func(contextMoqParam context.Context, bucket string) ([]byte, error)
 
@@ -273,6 +289,9 @@ type BackendMock struct {
 	// PutBucketAclFunc mocks the PutBucketAcl method.
 	PutBucketAclFunc func(contextMoqParam context.Context, bucket string, data []byte) error
 
+	// PutBucketOwnershipControlsFunc mocks the PutBucketOwnershipControls method.
+	PutBucketOwnershipControlsFunc func(contextMoqParam context.Context, bucket string, ownership types.ObjectOwnership) error
+
 	// PutBucketPolicyFunc mocks the PutBucketPolicy method.
 	PutBucketPolicyFunc func(contextMoqParam context.Context, bucket string, policy []byte) error
 
@@ -373,6 +392,13 @@ type BackendMock struct {
 			// DeleteBucketInput is the deleteBucketInput argument value.
 			DeleteBucketInput *s3.DeleteBucketInput
 		}
+		// DeleteBucketOwnershipControls holds details about calls to the DeleteBucketOwnershipControls method.
+		DeleteBucketOwnershipControls []struct {
+			// ContextMoqParam is the contextMoqParam argument value.
+			ContextMoqParam context.Context
+			// Bucket is the bucket argument value.
+			Bucket string
+		}
 		// DeleteBucketPolicy holds details about calls to the DeleteBucketPolicy method.
 		DeleteBucketPolicy []struct {
 			// ContextMoqParam is the contextMoqParam argument value.
@@ -417,6 +443,13 @@ type BackendMock struct {
 			// GetBucketAclInput is the getBucketAclInput argument value.
 			GetBucketAclInput *s3.GetBucketAclInput
 		}
+		// GetBucketOwnershipControls holds details about calls to the GetBucketOwnershipControls method.
+		GetBucketOwnershipControls []struct {
+			// ContextMoqParam is the contextMoqParam argument value.
+			ContextMoqParam context.Context
+			// Bucket is the bucket argument value.
+			Bucket string
+		}
 		// GetBucketPolicy holds details about calls to the GetBucketPolicy method.
 		GetBucketPolicy []struct {
 			// ContextMoqParam is the contextMoqParam argument value.
@@ -571,6 +604,15 @@ type BackendMock struct {
 			// Data is the data argument value.
 			Data []byte
 		}
+		// PutBucketOwnershipControls holds details about calls to the PutBucketOwnershipControls method.
+		PutBucketOwnershipControls []struct {
+			// ContextMoqParam is the contextMoqParam argument value.
+			ContextMoqParam context.Context
+			// Bucket is the bucket argument value.
+			Bucket string
+			// Ownership is the ownership argument value.
+			Ownership types.ObjectOwnership
+		}
 		// PutBucketPolicy holds details about calls to the PutBucketPolicy method.
 		PutBucketPolicy []struct {
 			// ContextMoqParam is the contextMoqParam argument value.
@@ -693,54 +735,57 @@ type BackendMock struct {
 			UploadPartCopyInput *s3.UploadPartCopyInput
 		}
 	}
-	lockAbortMultipartUpload       sync.RWMutex
+	lockAbortMultipartUpload          sync.RWMutex
-	lockChangeBucketOwner          sync.RWMutex
+	lockChangeBucketOwner             sync.RWMutex
-	lockCompleteMultipartUpload    sync.RWMutex
+	lockCompleteMultipartUpload       sync.RWMutex
-	lockCopyObject                 sync.RWMutex
+	lockCopyObject                    sync.RWMutex
-	lockCreateBucket               sync.RWMutex
+	lockCreateBucket                  sync.RWMutex
-	lockCreateMultipartUpload      sync.RWMutex
+	lockCreateMultipartUpload         sync.RWMutex
-	lockDeleteBucket               sync.RWMutex
+	lockDeleteBucket                  sync.RWMutex
-	lockDeleteBucketPolicy         sync.RWMutex
+	lockDeleteBucketOwnershipControls sync.RWMutex
-	lockDeleteBucketTagging        sync.RWMutex
+	lockDeleteBucketPolicy            sync.RWMutex
-	lockDeleteObject               sync.RWMutex
+	lockDeleteBucketTagging           sync.RWMutex
-	lockDeleteObjectTagging        sync.RWMutex
+	lockDeleteObject                  sync.RWMutex
-	lockDeleteObjects              sync.RWMutex
+	lockDeleteObjectTagging           sync.RWMutex
-	lockGetBucketAcl               sync.RWMutex
+	lockDeleteObjects                 sync.RWMutex
-	lockGetBucketPolicy            sync.RWMutex
+	lockGetBucketAcl                  sync.RWMutex
-	lockGetBucketTagging           sync.RWMutex
+	lockGetBucketOwnershipControls    sync.RWMutex
-	lockGetBucketVersioning        sync.RWMutex
+	lockGetBucketPolicy               sync.RWMutex
-	lockGetObject                  sync.RWMutex
+	lockGetBucketTagging              sync.RWMutex
-	lockGetObjectAcl               sync.RWMutex
+	lockGetBucketVersioning           sync.RWMutex
-	lockGetObjectAttributes        sync.RWMutex
+	lockGetObject                     sync.RWMutex
-	lockGetObjectLegalHold         sync.RWMutex
+	lockGetObjectAcl                  sync.RWMutex
-	lockGetObjectLockConfiguration sync.RWMutex
+	lockGetObjectAttributes           sync.RWMutex
-	lockGetObjectRetention         sync.RWMutex
+	lockGetObjectLegalHold            sync.RWMutex
-	lockGetObjectTagging           sync.RWMutex
+	lockGetObjectLockConfiguration    sync.RWMutex
-	lockHeadBucket                 sync.RWMutex
+	lockGetObjectRetention            sync.RWMutex
-	lockHeadObject                 sync.RWMutex
+	lockGetObjectTagging              sync.RWMutex
-	lockListBuckets                sync.RWMutex
+	lockHeadBucket                    sync.RWMutex
-	lockListBucketsAndOwners       sync.RWMutex
+	lockHeadObject                    sync.RWMutex
-	lockListMultipartUploads       sync.RWMutex
+	lockListBuckets                   sync.RWMutex
-	lockListObjectVersions         sync.RWMutex
+	lockListBucketsAndOwners          sync.RWMutex
-	lockListObjects                sync.RWMutex
+	lockListMultipartUploads          sync.RWMutex
-	lockListObjectsV2              sync.RWMutex
+	lockListObjectVersions            sync.RWMutex
-	lockListParts                  sync.RWMutex
+	lockListObjects                   sync.RWMutex
-	lockPutBucketAcl               sync.RWMutex
+	lockListObjectsV2                 sync.RWMutex
-	lockPutBucketPolicy            sync.RWMutex
+	lockListParts                     sync.RWMutex
-	lockPutBucketTagging           sync.RWMutex
+	lockPutBucketAcl                  sync.RWMutex
-	lockPutBucketVersioning        sync.RWMutex
+	lockPutBucketOwnershipControls    sync.RWMutex
-	lockPutObject                  sync.RWMutex
+	lockPutBucketPolicy               sync.RWMutex
-	lockPutObjectAcl               sync.RWMutex
+	lockPutBucketTagging              sync.RWMutex
-	lockPutObjectLegalHold         sync.RWMutex
+	lockPutBucketVersioning           sync.RWMutex
-	lockPutObjectLockConfiguration sync.RWMutex
+	lockPutObject                     sync.RWMutex
-	lockPutObjectRetention         sync.RWMutex
+	lockPutObjectAcl                  sync.RWMutex
-	lockPutObjectTagging           sync.RWMutex
+	lockPutObjectLegalHold            sync.RWMutex
-	lockRestoreObject              sync.RWMutex
+	lockPutObjectLockConfiguration    sync.RWMutex
-	lockSelectObjectContent        sync.RWMutex
+	lockPutObjectRetention            sync.RWMutex
-	lockShutdown                   sync.RWMutex
+	lockPutObjectTagging              sync.RWMutex
-	lockString                     sync.RWMutex
+	lockRestoreObject                 sync.RWMutex
-	lockUploadPart                 sync.RWMutex
+	lockSelectObjectContent           sync.RWMutex
-	lockUploadPartCopy             sync.RWMutex
+	lockShutdown                      sync.RWMutex
+	lockString                        sync.RWMutex
+	lockUploadPart                    sync.RWMutex
+	lockUploadPartCopy                sync.RWMutex
 }
 
 // AbortMultipartUpload calls AbortMultipartUploadFunc.
@@ -1003,6 +1048,42 @@ func (mock *BackendMock) DeleteBucketCalls() []struct {
 	return calls
 }
 
+// DeleteBucketOwnershipControls calls DeleteBucketOwnershipControlsFunc.
+func (mock *BackendMock) DeleteBucketOwnershipControls(contextMoqParam context.Context, bucket string) error {
+	if mock.DeleteBucketOwnershipControlsFunc == nil {
+		panic("BackendMock.DeleteBucketOwnershipControlsFunc: method is nil but Backend.DeleteBucketOwnershipControls was just called")
+	}
+	callInfo := struct {
+		ContextMoqParam context.Context
+		Bucket          string
+	}{
+		ContextMoqParam: contextMoqParam,
+		Bucket:          bucket,
+	}
+	mock.lockDeleteBucketOwnershipControls.Lock()
+	mock.calls.DeleteBucketOwnershipControls = append(mock.calls.DeleteBucketOwnershipControls, callInfo)
+	mock.lockDeleteBucketOwnershipControls.Unlock()
+	return mock.DeleteBucketOwnershipControlsFunc(contextMoqParam, bucket)
+}
+
+// DeleteBucketOwnershipControlsCalls gets all the calls that were made to DeleteBucketOwnershipControls.
+// Check the length with:
+//
+//	len(mockedBackend.DeleteBucketOwnershipControlsCalls())
+func (mock *BackendMock) DeleteBucketOwnershipControlsCalls() []struct {
+	ContextMoqParam context.Context
+	Bucket          string
+} {
+	var calls []struct {
+		ContextMoqParam context.Context
+		Bucket          string
+	}
+	mock.lockDeleteBucketOwnershipControls.RLock()
+	calls = mock.calls.DeleteBucketOwnershipControls
+	mock.lockDeleteBucketOwnershipControls.RUnlock()
+	return calls
+}
+
 // DeleteBucketPolicy calls DeleteBucketPolicyFunc.
 func (mock *BackendMock) DeleteBucketPolicy(contextMoqParam context.Context, bucket string) error {
 	if mock.DeleteBucketPolicyFunc == nil {
@@ -1223,6 +1304,42 @@ func (mock *BackendMock) GetBucketAclCalls() []struct {
 	return calls
 }
 
+// GetBucketOwnershipControls calls GetBucketOwnershipControlsFunc.
+func (mock *BackendMock) GetBucketOwnershipControls(contextMoqParam context.Context, bucket string) (types.ObjectOwnership, error) {
+	if mock.GetBucketOwnershipControlsFunc == nil {
+		panic("BackendMock.GetBucketOwnershipControlsFunc: method is nil but Backend.GetBucketOwnershipControls was just called")
+	}
+	callInfo := struct {
+		ContextMoqParam context.Context
+		Bucket          string
+	}{
+		ContextMoqParam: contextMoqParam,
+		Bucket:          bucket,
+	}
+	mock.lockGetBucketOwnershipControls.Lock()
+	mock.calls.GetBucketOwnershipControls = append(mock.calls.GetBucketOwnershipControls, callInfo)
+	mock.lockGetBucketOwnershipControls.Unlock()
+	return mock.GetBucketOwnershipControlsFunc(contextMoqParam, bucket)
+}
+
+// GetBucketOwnershipControlsCalls gets all the calls that were made to GetBucketOwnershipControls.
+// Check the length with:
+//
+//	len(mockedBackend.GetBucketOwnershipControlsCalls())
+func (mock *BackendMock) GetBucketOwnershipControlsCalls() []struct {
+	ContextMoqParam context.Context
+	Bucket          string
+} {
+	var calls []struct {
+		ContextMoqParam context.Context
+		Bucket          string
+	}
+	mock.lockGetBucketOwnershipControls.RLock()
+	calls = mock.calls.GetBucketOwnershipControls
+	mock.lockGetBucketOwnershipControls.RUnlock()
+	return calls
+}
+
 // GetBucketPolicy calls GetBucketPolicyFunc.
 func (mock *BackendMock) GetBucketPolicy(contextMoqParam context.Context, bucket string) ([]byte, error) {
 	if mock.GetBucketPolicyFunc == nil {
@@ -1971,6 +2088,46 @@ func (mock *BackendMock) PutBucketAclCalls() []struct {
 	return calls
 }
 
+// PutBucketOwnershipControls calls PutBucketOwnershipControlsFunc.
+func (mock *BackendMock) PutBucketOwnershipControls(contextMoqParam context.Context, bucket string, ownership types.ObjectOwnership) error {
+	if mock.PutBucketOwnershipControlsFunc == nil {
+		panic("BackendMock.PutBucketOwnershipControlsFunc: method is nil but Backend.PutBucketOwnershipControls was just called")
+	}
+	callInfo := struct {
+		ContextMoqParam context.Context
+		Bucket          string
+		Ownership       types.ObjectOwnership
+	}{
+		ContextMoqParam: contextMoqParam,
+		Bucket:          bucket,
+		Ownership:       ownership,
+	}
+	mock.lockPutBucketOwnershipControls.Lock()
+	mock.calls.PutBucketOwnershipControls = append(mock.calls.PutBucketOwnershipControls, callInfo)
+	mock.lockPutBucketOwnershipControls.Unlock()
+	return mock.PutBucketOwnershipControlsFunc(contextMoqParam, bucket, ownership)
+}
+
+// PutBucketOwnershipControlsCalls gets all the calls that were made to PutBucketOwnershipControls.
+// Check the length with:
+//
+//	len(mockedBackend.PutBucketOwnershipControlsCalls())
+func (mock *BackendMock) PutBucketOwnershipControlsCalls() []struct {
+	ContextMoqParam context.Context
+	Bucket          string
+	Ownership       types.ObjectOwnership
+} {
+	var calls []struct {
+		ContextMoqParam context.Context
+		Bucket          string
+		Ownership       types.ObjectOwnership
+	}
+	mock.lockPutBucketOwnershipControls.RLock()
+	calls = mock.calls.PutBucketOwnershipControls
+	mock.lockPutBucketOwnershipControls.RUnlock()
+	return calls
+}
+
 // PutBucketPolicy calls PutBucketPolicyFunc.
 func (mock *BackendMock) PutBucketPolicy(contextMoqParam context.Context, bucket string, policy []byte) error {
 	if mock.PutBucketPolicyFunc == nil {

s3api/controllers/base.go

@@ -566,6 +566,43 @@ func (c S3ApiController) ListActions(ctx *fiber.Ctx) error {
 			})
 	}
 
+	if ctx.Request().URI().QueryArgs().Has("ownershipControls") {
+		err := auth.VerifyAccess(ctx.Context(), c.be, auth.AccessOptions{
+			Readonly:      c.readonly,
+			Acl:           parsedAcl,
+			AclPermission: types.PermissionRead,
+			IsRoot:        isRoot,
+			Acc:           acct,
+			Bucket:        bucket,
+			Action:        auth.GetBucketOwnershipControlsAction,
+		})
+		if err != nil {
+			return SendXMLResponse(ctx, nil, err,
+				&MetaOpts{
+					Logger:      c.logger,
+					MetricsMng:  c.mm,
+					Action:      metrics.ActionGetBucketOwnershipControls,
+					BucketOwner: parsedAcl.Owner,
+				})
+		}
+
+		data, err := c.be.GetBucketOwnershipControls(ctx.Context(), bucket)
+		return SendXMLResponse(ctx,
+			s3response.OwnershipControls{
+				Rules: []types.OwnershipControlsRule{
+					{
+						ObjectOwnership: data,
+					},
+				},
+			}, err,
+			&MetaOpts{
+				Logger:      c.logger,
+				MetricsMng:  c.mm,
+				Action:      metrics.ActionGetBucketOwnershipControls,
+				BucketOwner: parsedAcl.Owner,
+			})
+	}
+
 	if ctx.Request().URI().QueryArgs().Has("versioning") {
 		err := auth.VerifyAccess(ctx.Context(), c.be, auth.AccessOptions{
 			Readonly:      c.readonly,
@@ -933,6 +970,9 @@ func (c S3ApiController) PutBucketActions(ctx *fiber.Ctx) error {
 	grantReadACP := ctx.Get("X-Amz-Grant-Read-Acp")
 	granWrite := ctx.Get("X-Amz-Grant-Write")
 	grantWriteACP := ctx.Get("X-Amz-Grant-Write-Acp")
+	objectOwnership := types.ObjectOwnership(
+		ctx.Get("X-Amz-Object-Ownership", string(types.ObjectOwnershipBucketOwnerEnforced)),
+	)
 	mfa := ctx.Get("X-Amz-Mfa")
 	contentMD5 := ctx.Get("Content-MD5")
 	acct := ctx.Locals("account").(auth.Account)
@@ -1000,6 +1040,57 @@ func (c S3ApiController) PutBucketActions(ctx *fiber.Ctx) error {
 			})
 	}
 
+	if ctx.Request().URI().QueryArgs().Has("ownershipControls") {
+		parsedAcl := ctx.Locals("parsedAcl").(auth.ACL)
+		var ownershipControls s3response.OwnershipControls
+		if err := xml.Unmarshal(ctx.Body(), &ownershipControls); err != nil {
+			return SendResponse(ctx, s3err.GetAPIError(s3err.ErrMalformedXML),
+				&MetaOpts{
+					Logger:      c.logger,
+					MetricsMng:  c.mm,
+					Action:      metrics.ActionPutBucketOwnershipControls,
+					BucketOwner: parsedAcl.Owner,
+				})
+		}
+
+		if len(ownershipControls.Rules) != 1 || !utils.IsValidOwnership(ownershipControls.Rules[0].ObjectOwnership) {
+			return SendResponse(ctx, s3err.GetAPIError(s3err.ErrMalformedXML),
+				&MetaOpts{
+					Logger:      c.logger,
+					MetricsMng:  c.mm,
+					Action:      metrics.ActionPutBucketOwnershipControls,
+					BucketOwner: parsedAcl.Owner,
+				})
+		}
+
+		if err := auth.VerifyAccess(ctx.Context(), c.be, auth.AccessOptions{
+			Readonly:      c.readonly,
+			Acl:           parsedAcl,
+			AclPermission: types.PermissionWrite,
+			IsRoot:        isRoot,
+			Acc:           acct,
+			Bucket:        bucket,
+			Action:        auth.PutBucketOwnershipControlsAction,
+		}); err != nil {
+			return SendResponse(ctx, err,
+				&MetaOpts{
+					Logger:      c.logger,
+					MetricsMng:  c.mm,
+					Action:      metrics.ActionPutBucketOwnershipControls,
+					BucketOwner: parsedAcl.Owner,
+				})
+		}
+
+		err := c.be.PutBucketOwnershipControls(ctx.Context(), bucket, ownershipControls.Rules[0].ObjectOwnership)
+		return SendResponse(ctx, err,
+			&MetaOpts{
+				Logger:      c.logger,
+				MetricsMng:  c.mm,
+				Action:      metrics.ActionPutBucketOwnershipControls,
+				BucketOwner: parsedAcl.Owner,
+			})
+	}
+
 	if ctx.Request().URI().QueryArgs().Has("versioning") {
 		parsedAcl := ctx.Locals("parsedAcl").(auth.ACL)
 		err := auth.VerifyAccess(ctx.Context(), c.be, auth.AccessOptions{
@@ -1141,10 +1232,33 @@ func (c S3ApiController) PutBucketActions(ctx *fiber.Ctx) error {
 	grants := grantFullControl + grantRead + grantReadACP + granWrite + grantWriteACP
 
 	if ctx.Request().URI().QueryArgs().Has("acl") {
+		parsedAcl := ctx.Locals("parsedAcl").(auth.ACL)
 		var input *s3.PutBucketAclInput
 
-		parsedAcl := ctx.Locals("parsedAcl").(auth.ACL)
+		ownership, err := c.be.GetBucketOwnershipControls(ctx.Context(), bucket)
-		err := auth.VerifyAccess(ctx.Context(), c.be,
+		if err != nil && !errors.Is(err, s3err.GetAPIError(s3err.ErrOwnershipControlsNotFound)) {
+			return SendResponse(ctx, err,
+				&MetaOpts{
+					Logger:      c.logger,
+					MetricsMng:  c.mm,
+					Action:      metrics.ActionPutBucketAcl,
+					BucketOwner: parsedAcl.Owner,
+				})
+		}
+		if ownership == types.ObjectOwnershipBucketOwnerEnforced {
+			if c.debug {
+				log.Println("bucket acls are disabled")
+			}
+			return SendResponse(ctx, s3err.GetAPIError(s3err.ErrAclNotSupported),
+				&MetaOpts{
+					Logger:      c.logger,
+					MetricsMng:  c.mm,
+					Action:      metrics.ActionPutBucketAcl,
+					BucketOwner: parsedAcl.Owner,
+				})
+		}
+
+		err = auth.VerifyAccess(ctx.Context(), c.be,
 			auth.AccessOptions{
 				Readonly:      c.readonly,
 				Acl:           parsedAcl,
@@ -1259,7 +1373,6 @@ func (c S3ApiController) PutBucketActions(ctx *fiber.Ctx) error {
 			}
 		}
 
-		fmt.Println(*input, parsedAcl)
 		updAcl, err := auth.UpdateACL(input, parsedAcl, c.iam)
 		if err != nil {
 			return SendResponse(ctx, err,
@@ -1289,16 +1402,45 @@ func (c S3ApiController) PutBucketActions(ctx *fiber.Ctx) error {
 				Action:     metrics.ActionCreateBucket,
 			})
 	}
+	if ok := utils.IsValidOwnership(objectOwnership); !ok {
+		if c.debug {
+			log.Printf("invalid bucket object ownership: %v", objectOwnership)
+		}
+		return SendResponse(ctx, s3err.APIError{
+			Code:           "InvalidArgument",
+			Description:    fmt.Sprintf("Invalid x-amz-object-ownership header: %v", objectOwnership),
+			HTTPStatusCode: http.StatusBadRequest,
+		},
+			&MetaOpts{
+				Logger:      c.logger,
+				MetricsMng:  c.mm,
+				Action:      metrics.ActionCreateBucket,
+				BucketOwner: acct.Access,
+			})
+	}
+
+	if acl+grants != "" && objectOwnership == types.ObjectOwnershipBucketOwnerEnforced {
+		if c.debug {
+			log.Printf("bucket acls are disabled for %v object ownership", objectOwnership)
+		}
+		return SendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidBucketAclWithObjectOwnership),
+			&MetaOpts{
+				Logger:      c.logger,
+				MetricsMng:  c.mm,
+				Action:      metrics.ActionCreateBucket,
+				BucketOwner: acct.Access,
+			})
+	}
 
 	if acl != "" && grants != "" {
 		if c.debug {
 			log.Printf("invalid request: %q (grants) %q (acl)", grants, acl)
 		}
-		return SendResponse(ctx, s3err.GetAPIError(s3err.ErrInvalidRequest),
+		return SendResponse(ctx, s3err.GetAPIError(s3err.ErrBothCannedAndHeaderGrants),
 			&MetaOpts{
 				Logger:      c.logger,
 				MetricsMng:  c.mm,
-				Action:      metrics.ActionPutBucketAcl,
+				Action:      metrics.ActionCreateBucket,
 				BucketOwner: acct.Access,
 			})
 	}
@@ -1334,7 +1476,7 @@ func (c S3ApiController) PutBucketActions(ctx *fiber.Ctx) error {
 
 	err = c.be.CreateBucket(ctx.Context(), &s3.CreateBucketInput{
 		Bucket:                     &bucket,
-		ObjectOwnership:            types.ObjectOwnership(acct.Access),
+		ObjectOwnership:            objectOwnership,
 		ObjectLockEnabledForBucket: &lockEnabled,
 	}, updAcl)
 	return SendResponse(ctx, err,
@@ -2042,6 +2184,38 @@ func (c S3ApiController) DeleteBucket(ctx *fiber.Ctx) error {
 			})
 	}
 
+	if ctx.Request().URI().QueryArgs().Has("ownershipControls") {
+		err := auth.VerifyAccess(ctx.Context(), c.be,
+			auth.AccessOptions{
+				Readonly:      c.readonly,
+				Acl:           parsedAcl,
+				AclPermission: types.PermissionWrite,
+				IsRoot:        isRoot,
+				Acc:           acct,
+				Bucket:        bucket,
+				Action:        auth.PutBucketOwnershipControlsAction,
+			})
+		if err != nil {
+			return SendResponse(ctx, err,
+				&MetaOpts{
+					Logger:      c.logger,
+					MetricsMng:  c.mm,
+					Action:      metrics.ActionDeleteBucketOwnershipControls,
+					BucketOwner: parsedAcl.Owner,
+				})
+		}
+
+		err = c.be.DeleteBucketOwnershipControls(ctx.Context(), bucket)
+		return SendResponse(ctx, err,
+			&MetaOpts{
+				Logger:      c.logger,
+				MetricsMng:  c.mm,
+				Action:      metrics.ActionDeleteBucketOwnershipControls,
+				BucketOwner: parsedAcl.Owner,
+				Status:      http.StatusNoContent,
+			})
+	}
+
 	if ctx.Request().URI().QueryArgs().Has("policy") {
 		err := auth.VerifyAccess(ctx.Context(), c.be,
 			auth.AccessOptions{

s3api/controllers/base_test.go

@@ -395,6 +395,9 @@ func TestS3ApiController_ListActions(t *testing.T) {
 			GetObjectLockConfigurationFunc: func(contextMoqParam context.Context, bucket string) ([]byte, error) {
 				return objectLockResult, nil
 			},
+			GetBucketOwnershipControlsFunc: func(contextMoqParam context.Context, bucket string) (types.ObjectOwnership, error) {
+				return types.ObjectOwnershipBucketOwnerEnforced, nil
+			},
 		},
 	}
 
@@ -448,6 +451,15 @@ func TestS3ApiController_ListActions(t *testing.T) {
 			wantErr:    false,
 			statusCode: 404,
 		},
+		{
+			name: "Get-bucket-ownership-control-success",
+			app:  app,
+			args: args{
+				req: httptest.NewRequest(http.MethodGet, "/my-bucket?ownershipControls", nil),
+			},
+			wantErr:    false,
+			statusCode: 200,
+		},
 		{
 			name: "Get-bucket-tagging-success",
 			app:  app,
@@ -562,7 +574,7 @@ func TestS3ApiController_PutBucketActions(t *testing.T) {
 	app := fiber.New()
 
 	// Mock valid acl
-	acl := auth.ACL{Owner: "valid access", ACL: "public-read-write"}
+	acl := auth.ACL{Owner: "valid access"}
 	acldata, err := json.Marshal(acl)
 	if err != nil {
 		t.Errorf("Failed to parse the params: %v", err.Error())
@@ -636,6 +648,22 @@ func TestS3ApiController_PutBucketActions(t *testing.T) {
 	</ObjectLockConfiguration>
 	`
 
+	ownershipBody := `
+	<OwnershipControls xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+		<Rule>
+			<ObjectOwnership>BucketOwnerEnforced</ObjectOwnership>
+		</Rule>
+	</OwnershipControls>
+	`
+
+	invalidOwnershipBody := `
+	<OwnershipControls xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+		<Rule>
+			<ObjectOwnership>invalid_value</ObjectOwnership>
+		</Rule>
+	</OwnershipControls>
+	`
+
 	s3ApiController := S3ApiController{
 		be: &BackendMock{
 			GetBucketAclFunc: func(context.Context, *s3.GetBucketAclInput) ([]byte, error) {
@@ -659,6 +687,12 @@ func TestS3ApiController_PutBucketActions(t *testing.T) {
 			PutObjectLockConfigurationFunc: func(contextMoqParam context.Context, bucket string, config []byte) error {
 				return nil
 			},
+			PutBucketOwnershipControlsFunc: func(contextMoqParam context.Context, bucket string, ownership types.ObjectOwnership) error {
+				return nil
+			},
+			GetBucketOwnershipControlsFunc: func(contextMoqParam context.Context, bucket string) (types.ObjectOwnership, error) {
+				return types.ObjectOwnershipBucketOwnerPreferred, nil
+			},
 		},
 	}
 	// Mock ctx.Locals
@@ -691,6 +725,9 @@ func TestS3ApiController_PutBucketActions(t *testing.T) {
 	errAclBodyReq := httptest.NewRequest(http.MethodPut, "/my-bucket?acl", strings.NewReader(body))
 	errAclBodyReq.Header.Set("X-Amz-Grant-Read", "hello")
 
+	invAclOwnershipReq := httptest.NewRequest(http.MethodPut, "/my-bucket", nil)
+	invAclOwnershipReq.Header.Set("X-Amz-Grant-Read", "hello")
+
 	tests := []struct {
 		name       string
 		app        *fiber.App
@@ -716,6 +753,24 @@ func TestS3ApiController_PutBucketActions(t *testing.T) {
 			wantErr:    false,
 			statusCode: 200,
 		},
+		{
+			name: "Put-bucket-ownership-controls-invalid-ownership",
+			app:  app,
+			args: args{
+				req: httptest.NewRequest(http.MethodPut, "/my-bucket?ownershipControls", strings.NewReader(invalidOwnershipBody)),
+			},
+			wantErr:    false,
+			statusCode: 400,
+		},
+		{
+			name: "Put-bucket-ownership-controls-success",
+			app:  app,
+			args: args{
+				req: httptest.NewRequest(http.MethodPut, "/my-bucket?ownershipControls", strings.NewReader(ownershipBody)),
+			},
+			wantErr:    false,
+			statusCode: 200,
+		},
 		{
 			name: "Put-object-lock-configuration-invalid-body",
 			app:  app,
@@ -816,7 +871,16 @@ func TestS3ApiController_PutBucketActions(t *testing.T) {
 			statusCode: 200,
 		},
 		{
-			name: "Put-bucket-invalid-bucket-name",
+			name: "Create-bucket-invalid-acl-ownership-combination",
+			app:  app,
+			args: args{
+				req: invAclOwnershipReq,
+			},
+			wantErr:    false,
+			statusCode: 400,
+		},
+		{
+			name: "Create-bucket-invalid-bucket-name",
 			app:  app,
 			args: args{
 				req: httptest.NewRequest(http.MethodPut, "/aa", nil),
@@ -825,7 +889,7 @@ func TestS3ApiController_PutBucketActions(t *testing.T) {
 			statusCode: 400,
 		},
 		{
-			name: "Put-bucket-success",
+			name: "Create-bucket-success",
 			app:  app,
 			args: args{
 				req: httptest.NewRequest(http.MethodPut, "/my-bucket", nil),
@@ -1160,6 +1224,12 @@ func TestS3ApiController_DeleteBucket(t *testing.T) {
 			DeleteBucketTaggingFunc: func(contextMoqParam context.Context, bucket string) error {
 				return nil
 			},
+			DeleteBucketPolicyFunc: func(contextMoqParam context.Context, bucket string) error {
+				return nil
+			},
+			DeleteBucketOwnershipControlsFunc: func(contextMoqParam context.Context, bucket string) error {
+				return nil
+			},
 		},
 	}
 
@@ -1198,6 +1268,23 @@ func TestS3ApiController_DeleteBucket(t *testing.T) {
 			wantErr:    false,
 			statusCode: 204,
 		},
+		{
+			name: "Delete-bucket-ownership-controls-success",
+			app:  app,
+			args: args{
+				req: httptest.NewRequest(http.MethodDelete, "/my-bucket?ownershipControls", nil),
+			},
+			wantErr:    false,
+			statusCode: 204,
+		}, {
+			name: "Delete-bucket-policy-success",
+			app:  app,
+			args: args{
+				req: httptest.NewRequest(http.MethodDelete, "/my-bucket?policy", nil),
+			},
+			wantErr:    false,
+			statusCode: 204,
+		},
 	}
 	for _, tt := range tests {
 		resp, err := tt.app.Test(tt.args.req)

s3api/middlewares/acl-parser.go

@@ -50,7 +50,8 @@ func AclParser(be backend.Backend, logger s3log.AuditLogger, readonly bool) fibe
 			!ctx.Request().URI().QueryArgs().Has("tagging") &&
 			!ctx.Request().URI().QueryArgs().Has("versioning") &&
 			!ctx.Request().URI().QueryArgs().Has("policy") &&
-			!ctx.Request().URI().QueryArgs().Has("object-lock") {
+			!ctx.Request().URI().QueryArgs().Has("object-lock") &&
+			!ctx.Request().URI().QueryArgs().Has("ownershipControls") {
 			if err := auth.MayCreateBucket(acct, isRoot); err != nil {
 				return controllers.SendXMLResponse(ctx, nil, err, &controllers.MetaOpts{Logger: logger, Action: "CreateBucket"})
 			}

s3api/utils/utils.go

@@ -326,3 +326,16 @@ func ParsObjectLockHdrs(ctx *fiber.Ctx) (*objLockCfg, error) {
 		LegalHoldStatus: legalHold,
 	}, nil
 }
+
+func IsValidOwnership(val types.ObjectOwnership) bool {
+	switch val {
+	case types.ObjectOwnershipBucketOwnerEnforced:
+		return true
+	case types.ObjectOwnershipBucketOwnerPreferred:
+		return true
+	case types.ObjectOwnershipObjectWriter:
+		return true
+	default:
+		return false
+	}
+}

s3api/utils/utils_test.go

@@ -335,3 +335,50 @@ func TestFilterObjectAttributes(t *testing.T) {
 		})
 	}
 }
+
+func TestIsValidOwnership(t *testing.T) {
+	type args struct {
+		val types.ObjectOwnership
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "valid-BucketOwnerEnforced",
+			args: args{
+				val: types.ObjectOwnershipBucketOwnerEnforced,
+			},
+			want: true,
+		},
+		{
+			name: "valid-BucketOwnerPreferred",
+			args: args{
+				val: types.ObjectOwnershipBucketOwnerPreferred,
+			},
+			want: true,
+		},
+		{
+			name: "valid-ObjectWriter",
+			args: args{
+				val: types.ObjectOwnershipObjectWriter,
+			},
+			want: true,
+		},
+		{
+			name: "invalid_value",
+			args: args{
+				val: types.ObjectOwnership("invalid_value"),
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := IsValidOwnership(tt.args.val); got != tt.want {
+				t.Errorf("IsValidOwnership() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

s3err/s3err.go

@@ -123,6 +123,10 @@ const (
 	ErrBucketTaggingNotFound
 	ErrObjectLockInvalidHeaders
 	ErrRequestTimeTooSkewed
+	ErrInvalidBucketAclWithObjectOwnership
+	ErrBothCannedAndHeaderGrants
+	ErrOwnershipControlsNotFound
+	ErrAclNotSupported
 
 	// Non-AWS errors
 	ErrExistingObjectIsDirectory
@@ -472,6 +476,26 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "The difference between the request time and the server's time is too large.",
 		HTTPStatusCode: http.StatusForbidden,
 	},
+	ErrInvalidBucketAclWithObjectOwnership: {
+		Code:           "ErrInvalidBucketAclWithObjectOwnership",
+		Description:    "Bucket cannot have ACLs set with ObjectOwnership's BucketOwnerEnforced setting",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
+	ErrBothCannedAndHeaderGrants: {
+		Code:           "InvalidRequest",
+		Description:    "Specifying both Canned ACLs and Header Grants is not allowed",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
+	ErrOwnershipControlsNotFound: {
+		Code:           "OwnershipControlsNotFoundError",
+		Description:    "The bucket ownership controls were not found",
+		HTTPStatusCode: http.StatusNotFound,
+	},
+	ErrAclNotSupported: {
+		Code:           "AccessControlListNotSupported",
+		Description:    "The bucket does not allow ACLs",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 
 	// non aws errors
 	ErrExistingObjectIsDirectory: {

s3response/s3response.go

@@ -217,3 +217,7 @@ type Grantee struct {
 	ID          string
 	DisplayName string
 }
+
+type OwnershipControls struct {
+	Rules []types.OwnershipControlsRule `xml:"Rule"`
+}

tests/integration/group-tests.go

@@ -65,6 +65,8 @@ func TestCreateBucket(s *S3Conf) {
 	CreateBucket_invalid_bucket_name(s)
 	CreateBucket_existing_bucket(s)
 	CreateBucket_owned_by_you(s)
+	CreateBucket_invalid_ownership(s)
+	CreateBucket_ownership_with_acl(s)
 	CreateBucket_as_user(s)
 	CreateBucket_default_acl(s)
 	CreateBucket_non_default_acl(s)
@@ -89,6 +91,24 @@ func TestDeleteBucket(s *S3Conf) {
 	DeleteBucket_success_status_code(s)
 }
 
+func TestPutBucketOwnershipControls(s *S3Conf) {
+	PutBucketOwnershipControls_non_existing_bucket(s)
+	PutBucketOwnershipControls_multiple_rules(s)
+	PutBucketOwnershipControls_invalid_ownership(s)
+	PutBucketOwnershipControls_success(s)
+}
+
+func TestGetBucketOwnershipControls(s *S3Conf) {
+	GetBucketOwnershipControls_non_existing_bucket(s)
+	GetBucketOwnershipControls_default_ownership(s)
+	GetBucketOwnershipControls_success(s)
+}
+
+func TestDeleteBucketOwnershipControls(s *S3Conf) {
+	DeleteBucketOwnershipControls_non_existing_bucket(s)
+	DeleteBucketOwnershipControls_success(s)
+}
+
 func TestPutBucketTagging(s *S3Conf) {
 	PutBucketTagging_non_existing_bucket(s)
 	PutBucketTagging_long_tags(s)
@@ -267,6 +287,7 @@ func TestCompleteMultipartUpload(s *S3Conf) {
 
 func TestPutBucketAcl(s *S3Conf) {
 	PutBucketAcl_non_existing_bucket(s)
+	PutBucketAcl_disabled(s)
 	PutBucketAcl_invalid_acl_canned_and_acp(s)
 	PutBucketAcl_invalid_acl_canned_and_grants(s)
 	PutBucketAcl_invalid_acl_acp_and_grants(s)
@@ -280,6 +301,9 @@ func TestPutBucketAcl(s *S3Conf) {
 
 func TestGetBucketAcl(s *S3Conf) {
 	GetBucketAcl_non_existing_bucket(s)
+	GetBucketAcl_translation_canned_public_read(s)
+	GetBucketAcl_translation_canned_public_read_write(s)
+	GetBucketAcl_translation_canned_private(s)
 	GetBucketAcl_access_denied(s)
 	GetBucketAcl_success(s)
 }
@@ -395,6 +419,9 @@ func TestFullFlow(s *S3Conf) {
 	TestHeadBucket(s)
 	TestListBuckets(s)
 	TestDeleteBucket(s)
+	TestPutBucketOwnershipControls(s)
+	TestGetBucketOwnershipControls(s)
+	TestDeleteBucketOwnershipControls(s)
 	TestPutBucketTagging(s)
 	TestGetBucketTagging(s)
 	TestDeleteBucketTagging(s)
@@ -505,6 +532,8 @@ func GetIntTests() IntTests {
 		"CreateBucket_invalid_bucket_name":                                    CreateBucket_invalid_bucket_name,
 		"CreateBucket_existing_bucket":                                        CreateBucket_existing_bucket,
 		"CreateBucket_owned_by_you":                                           CreateBucket_owned_by_you,
+		"CreateBucket_invalid_ownership":                                      CreateBucket_invalid_ownership,
+		"CreateBucket_ownership_with_acl":                                     CreateBucket_ownership_with_acl,
 		"CreateBucket_as_user":                                                CreateBucket_as_user,
 		"CreateDeleteBucket_success":                                          CreateDeleteBucket_success,
 		"CreateBucket_default_acl":                                            CreateBucket_default_acl,
@@ -518,6 +547,15 @@ func GetIntTests() IntTests {
 		"DeleteBucket_non_existing_bucket":                                    DeleteBucket_non_existing_bucket,
 		"DeleteBucket_non_empty_bucket":                                       DeleteBucket_non_empty_bucket,
 		"DeleteBucket_success_status_code":                                    DeleteBucket_success_status_code,
+		"PutBucketOwnershipControls_non_existing_bucket":                      PutBucketOwnershipControls_non_existing_bucket,
+		"PutBucketOwnershipControls_multiple_rules":                           PutBucketOwnershipControls_multiple_rules,
+		"PutBucketOwnershipControls_invalid_ownership":                        PutBucketOwnershipControls_invalid_ownership,
+		"PutBucketOwnershipControls_success":                                  PutBucketOwnershipControls_success,
+		"GetBucketOwnershipControls_non_existing_bucket":                      GetBucketOwnershipControls_non_existing_bucket,
+		"GetBucketOwnershipControls_default_ownership":                        GetBucketOwnershipControls_default_ownership,
+		"GetBucketOwnershipControls_success":                                  GetBucketOwnershipControls_success,
+		"DeleteBucketOwnershipControls_non_existing_bucket":                   DeleteBucketOwnershipControls_non_existing_bucket,
+		"DeleteBucketOwnershipControls_success":                               DeleteBucketOwnershipControls_success,
 		"PutBucketTagging_non_existing_bucket":                                PutBucketTagging_non_existing_bucket,
 		"PutBucketTagging_long_tags":                                          PutBucketTagging_long_tags,
 		"PutBucketTagging_success":                                            PutBucketTagging_success,
@@ -636,6 +674,9 @@ func GetIntTests() IntTests {
 		"PutBucketAcl_success_canned_acl":                                     PutBucketAcl_success_canned_acl,
 		"PutBucketAcl_success_acp":                                            PutBucketAcl_success_acp,
 		"GetBucketAcl_non_existing_bucket":                                    GetBucketAcl_non_existing_bucket,
+		"GetBucketAcl_translation_canned_public_read":                         GetBucketAcl_translation_canned_public_read,
+		"GetBucketAcl_translation_canned_public_read_write":                   GetBucketAcl_translation_canned_public_read_write,
+		"GetBucketAcl_translation_canned_private":                             GetBucketAcl_translation_canned_private,
 		"GetBucketAcl_access_denied":                                          GetBucketAcl_access_denied,
 		"GetBucketAcl_success":                                                GetBucketAcl_success,
 		"PutBucketPolicy_non_existing_bucket":                                 PutBucketPolicy_non_existing_bucket,

tests/integration/tests.go

@@ -1720,47 +1720,76 @@ func CreateBucket_owned_by_you(s *S3Conf) error {
 	})
 }
 
-func CreateBucket_default_acl(s *S3Conf) error {
+func CreateBucket_invalid_ownership(s *S3Conf) error {
-	testName := "CreateBucket_default_acl"
+	testName := "CreateBucket_invalid_ownership"
 	runF(testName)
 
-	bucket := getBucketName()
+	invalidOwnership := types.ObjectOwnership("invalid_ownership")
-	client := s3.NewFromConfig(s.Config())
+	err := setup(s, getBucketName(), withOwnership(invalidOwnership))
-
+	if err := checkApiErr(err, s3err.APIError{
-	err := setup(s, bucket)
+		Code:           "InvalidArgument",
-	if err != nil {
+		Description:    fmt.Sprintf("Invalid x-amz-object-ownership header: %v", invalidOwnership),
+		HTTPStatusCode: http.StatusBadRequest,
+	}); err != nil {
 		failF("%v: %v", testName, err)
 		return fmt.Errorf("%v: %w", testName, err)
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-	out, err := client.GetBucketAcl(ctx, &s3.GetBucketAclInput{Bucket: &bucket})
-	cancel()
-	if err != nil {
-		failF("%v: %v", testName, err)
-		return fmt.Errorf("%v: %w", testName, err)
-	}
-
-	if *out.Owner.ID != s.awsID {
-		failF("%v: expected bucket owner to be %v, instead got %v", testName, s.awsID, *out.Owner.ID)
-		return fmt.Errorf("%v: expected bucket owner to be %v, instead got %v", testName, s.awsID, *out.Owner.ID)
-	}
-
-	if len(out.Grants) != 0 {
-		failF("%v: expected grants to be empty instead got %v", testName, len(out.Grants))
-		return fmt.Errorf("%v: expected grants to be empty instead got %v", testName, len(out.Grants))
-	}
-
-	err = teardown(s, bucket)
-	if err != nil {
-		failF("%v: %v", err)
-		return fmt.Errorf("%v: %w", testName, err)
-	}
-
 	passF(testName)
 	return nil
 }
 
+func CreateBucket_ownership_with_acl(s *S3Conf) error {
+	testName := "CreateBucket_ownership_with_acl"
+
+	runF(testName)
+	client := s3.NewFromConfig(s.Config())
+
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	_, err := client.CreateBucket(ctx, &s3.CreateBucketInput{
+		Bucket:          getPtr(getBucketName()),
+		ObjectOwnership: types.ObjectOwnershipBucketOwnerEnforced,
+		ACL:             types.BucketCannedACLPublicRead,
+	})
+	cancel()
+	if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidBucketAclWithObjectOwnership)); err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	passF(testName)
+	return nil
+}
+
+func CreateBucket_default_acl(s *S3Conf) error {
+	testName := "CreateBucket_default_acl"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{Bucket: &bucket})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if *out.Owner.ID != s.awsID {
+			return fmt.Errorf("expected bucket owner to be %v, instead got %v", s.awsID, *out.Owner.ID)
+		}
+		if len(out.Grants) != 1 {
+			return fmt.Errorf("expected grants length to be 1, instead got %v", len(out.Grants))
+		}
+		grt := out.Grants[0]
+		if grt.Permission != types.PermissionFullControl {
+			return fmt.Errorf("expected the grantee to have full-control permission, instead got %v", grt.Permission)
+		}
+		if *grt.Grantee.ID != s.awsID {
+			return fmt.Errorf("expected the grantee id to be %v, instead got %v", s.awsID, *grt.Grantee.ID)
+		}
+
+		return nil
+	})
+}
+
 func CreateBucket_non_default_acl(s *S3Conf) error {
 	testName := "CreateBucket_non_default_acl"
 	runF(testName)
@@ -1776,6 +1805,12 @@ func CreateBucket_non_default_acl(s *S3Conf) error {
 	}
 
 	grants := []types.Grant{
+		{
+			Grantee: &types.Grantee{
+				ID: &s.awsID,
+			},
+			Permission: types.PermissionFullControl,
+		},
 		{
 			Grantee: &types.Grantee{
 				ID: getPtr("grt1"),
@@ -1800,7 +1835,13 @@ func CreateBucket_non_default_acl(s *S3Conf) error {
 	client := s3.NewFromConfig(s.Config())
 
 	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
-	_, err = client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: &bucket, GrantFullControl: getPtr("grt1"), GrantReadACP: getPtr("grt2"), GrantWrite: getPtr("grt3")})
+	_, err = client.CreateBucket(ctx, &s3.CreateBucketInput{
+		Bucket:           &bucket,
+		GrantFullControl: getPtr("grt1"),
+		GrantReadACP:     getPtr("grt2"),
+		GrantWrite:       getPtr("grt3"),
+		ObjectOwnership:  types.ObjectOwnershipBucketOwnerPreferred,
+	})
 	cancel()
 	if err != nil {
 		failF("%v: %v", err)
@@ -2195,6 +2236,220 @@ func DeleteBucket_success_status_code(s *S3Conf) error {
 	return nil
 }
 
+func PutBucketOwnershipControls_non_existing_bucket(s *S3Conf) error {
+	testName := "PutBucketOwnershipControls_non_existing_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketOwnershipControls(ctx, &s3.PutBucketOwnershipControlsInput{
+			Bucket: getPtr(getBucketName()),
+			OwnershipControls: &types.OwnershipControls{
+				Rules: []types.OwnershipControlsRule{
+					{
+						ObjectOwnership: types.ObjectOwnershipBucketOwnerPreferred,
+					},
+				},
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func PutBucketOwnershipControls_multiple_rules(s *S3Conf) error {
+	testName := "PutBucketOwnershipControls_multiple_rules"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketOwnershipControls(ctx, &s3.PutBucketOwnershipControlsInput{
+			Bucket: &bucket,
+			OwnershipControls: &types.OwnershipControls{
+				Rules: []types.OwnershipControlsRule{
+					{
+						ObjectOwnership: types.ObjectOwnershipBucketOwnerPreferred,
+					},
+					{
+						ObjectOwnership: types.ObjectOwnershipObjectWriter,
+					},
+				},
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMalformedXML)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func PutBucketOwnershipControls_invalid_ownership(s *S3Conf) error {
+	testName := "PutBucketOwnershipControls_invalid_ownership"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketOwnershipControls(ctx, &s3.PutBucketOwnershipControlsInput{
+			Bucket: &bucket,
+			OwnershipControls: &types.OwnershipControls{
+				Rules: []types.OwnershipControlsRule{
+					{
+						ObjectOwnership: types.ObjectOwnership("invalid_ownership"),
+					},
+				},
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMalformedXML)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func PutBucketOwnershipControls_success(s *S3Conf) error {
+	testName := "PutBucketOwnershipControls_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketOwnershipControls(ctx, &s3.PutBucketOwnershipControlsInput{
+			Bucket: &bucket,
+			OwnershipControls: &types.OwnershipControls{
+				Rules: []types.OwnershipControlsRule{
+					{
+						ObjectOwnership: types.ObjectOwnershipObjectWriter,
+					},
+				},
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func GetBucketOwnershipControls_non_existing_bucket(s *S3Conf) error {
+	testName := "GetBucketOwnershipControls_non_existing_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+			Bucket: getPtr(getBucketName()),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func GetBucketOwnershipControls_default_ownership(s *S3Conf) error {
+	testName := "GetBucketOwnershipControls_default_ownership"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		resp, err := s3client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(resp.OwnershipControls.Rules) != 1 {
+			return fmt.Errorf("expected ownership control rules length to be 1, instead got %v", len(resp.OwnershipControls.Rules))
+		}
+		if resp.OwnershipControls.Rules[0].ObjectOwnership != types.ObjectOwnershipBucketOwnerEnforced {
+			return fmt.Errorf("expected the bucket ownership to be %v, instead got %v", types.ObjectOwnershipBucketOwnerEnforced, resp.OwnershipControls.Rules[0].ObjectOwnership)
+		}
+
+		return nil
+	})
+}
+
+func GetBucketOwnershipControls_success(s *S3Conf) error {
+	testName := "GetBucketOwnershipControls_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketOwnershipControls(ctx, &s3.PutBucketOwnershipControlsInput{
+			Bucket: &bucket,
+			OwnershipControls: &types.OwnershipControls{
+				Rules: []types.OwnershipControlsRule{
+					{
+						ObjectOwnership: types.ObjectOwnershipObjectWriter,
+					},
+				},
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		resp, err := s3client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(resp.OwnershipControls.Rules) != 1 {
+			return fmt.Errorf("expected ownership control rules length to be 1, instead got %v", len(resp.OwnershipControls.Rules))
+		}
+		if resp.OwnershipControls.Rules[0].ObjectOwnership != types.ObjectOwnershipObjectWriter {
+			return fmt.Errorf("expected the bucket ownership to be %v, instead got %v", types.ObjectOwnershipObjectWriter, resp.OwnershipControls.Rules[0].ObjectOwnership)
+		}
+
+		return nil
+	})
+}
+
+func DeleteBucketOwnershipControls_non_existing_bucket(s *S3Conf) error {
+	testName := "DeleteBucketOwnershipControls_non_existing_bucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.DeleteBucketOwnershipControls(ctx, &s3.DeleteBucketOwnershipControlsInput{
+			Bucket: getPtr(getBucketName()),
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func DeleteBucketOwnershipControls_success(s *S3Conf) error {
+	testName := "DeleteBucketOwnershipControls_success"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.DeleteBucketOwnershipControls(ctx, &s3.DeleteBucketOwnershipControlsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrOwnershipControlsNotFound)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
 func PutBucketTagging_non_existing_bucket(s *S3Conf) error {
 	testName := "PutBucketTagging_non_existing_bucket"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5893,6 +6148,23 @@ func PutBucketAcl_non_existing_bucket(s *S3Conf) error {
 	})
 }
 
+func PutBucketAcl_disabled(s *S3Conf) error {
+	testName := "PutBucketAcl_disabled"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketAcl(ctx, &s3.PutBucketAclInput{
+			Bucket:    &bucket,
+			ACL:       types.BucketCannedACLPublicRead,
+			GrantRead: &s.awsID,
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrAclNotSupported)); err != nil {
+			return err
+		}
+		return nil
+	})
+}
+
 func PutBucketAcl_invalid_acl_canned_and_acp(s *S3Conf) error {
 	testName := "PutBucketAcl_invalid_acl_canned_and_acp"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5908,7 +6180,7 @@ func PutBucketAcl_invalid_acl_canned_and_acp(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_invalid_acl_canned_and_grants(s *S3Conf) error {
@@ -5938,7 +6210,7 @@ func PutBucketAcl_invalid_acl_canned_and_grants(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_invalid_acl_acp_and_grants(s *S3Conf) error {
@@ -5968,7 +6240,7 @@ func PutBucketAcl_invalid_acl_acp_and_grants(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_invalid_owner(s *S3Conf) error {
@@ -5997,7 +6269,7 @@ func PutBucketAcl_invalid_owner(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_invalid_owner_not_in_body(s *S3Conf) error {
@@ -6023,7 +6295,7 @@ func PutBucketAcl_invalid_owner_not_in_body(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_success_access_denied(s *S3Conf) error {
@@ -6068,7 +6340,7 @@ func PutBucketAcl_success_access_denied(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_success_canned_acl(s *S3Conf) error {
@@ -6100,7 +6372,7 @@ func PutBucketAcl_success_canned_acl(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_success_acp(s *S3Conf) error {
@@ -6141,7 +6413,7 @@ func PutBucketAcl_success_acp(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketAcl_success_grants(s *S3Conf) error {
@@ -6186,7 +6458,7 @@ func PutBucketAcl_success_grants(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func GetBucketAcl_non_existing_bucket(s *S3Conf) error {
@@ -6205,6 +6477,156 @@ func GetBucketAcl_non_existing_bucket(s *S3Conf) error {
 	})
 }
 
+func GetBucketAcl_translation_canned_public_read(s *S3Conf) error {
+	testName := "GetBucketAcl_translation_canned_public_read"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		grants := []types.Grant{
+			{
+				Grantee: &types.Grantee{
+					ID:   &s.awsID,
+					Type: types.TypeCanonicalUser,
+				},
+				Permission: types.PermissionFullControl,
+			},
+			{
+				Grantee: &types.Grantee{
+					ID:   getPtr("all-users"),
+					Type: types.TypeCanonicalUser,
+				},
+				Permission: types.PermissionRead,
+			},
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketAcl(ctx, &s3.PutBucketAclInput{
+			Bucket: &bucket,
+			ACL:    types.BucketCannedACLPublicRead,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if ok := compareGrants(out.Grants, grants); !ok {
+			return fmt.Errorf("expected grants to be %v, instead got %v", grants, out.Grants)
+		}
+		if *out.Owner.ID != s.awsID {
+			return fmt.Errorf("expected bucket owner to be %v, instead got %v", s.awsID, *out.Owner.ID)
+		}
+
+		return nil
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
+}
+
+func GetBucketAcl_translation_canned_public_read_write(s *S3Conf) error {
+	testName := "GetBucketAcl_translation_canned_public_read_write"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		grants := []types.Grant{
+			{
+				Grantee: &types.Grantee{
+					ID:   &s.awsID,
+					Type: types.TypeCanonicalUser,
+				},
+				Permission: types.PermissionFullControl,
+			},
+			{
+				Grantee: &types.Grantee{
+					ID:   getPtr("all-users"),
+					Type: types.TypeCanonicalUser,
+				},
+				Permission: types.PermissionRead,
+			},
+			{
+				Grantee: &types.Grantee{
+					ID:   getPtr("all-users"),
+					Type: types.TypeCanonicalUser,
+				},
+				Permission: types.PermissionWrite,
+			},
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketAcl(ctx, &s3.PutBucketAclInput{
+			Bucket: &bucket,
+			ACL:    types.BucketCannedACLPublicReadWrite,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if ok := compareGrants(out.Grants, grants); !ok {
+			return fmt.Errorf("expected grants to be %v, instead got %v", grants, out.Grants)
+		}
+		if *out.Owner.ID != s.awsID {
+			return fmt.Errorf("expected bucket owner to be %v, instead got %v", s.awsID, *out.Owner.ID)
+		}
+
+		return nil
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
+}
+
+func GetBucketAcl_translation_canned_private(s *S3Conf) error {
+	testName := "GetBucketAcl_translation_canned_private"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		grants := []types.Grant{
+			{
+				Grantee: &types.Grantee{
+					ID:   &s.awsID,
+					Type: types.TypeCanonicalUser,
+				},
+				Permission: types.PermissionFullControl,
+			},
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.PutBucketAcl(ctx, &s3.PutBucketAclInput{
+			Bucket: &bucket,
+			ACL:    types.BucketCannedACLPrivate,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if ok := compareGrants(out.Grants, grants); !ok {
+			return fmt.Errorf("expected grants to be %v, instead got %v", grants, out.Grants)
+		}
+		if *out.Owner.ID != s.awsID {
+			return fmt.Errorf("expected bucket owner to be %v, instead got %v", s.awsID, *out.Owner.ID)
+		}
+
+		return nil
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
+}
+
 func GetBucketAcl_access_denied(s *S3Conf) error {
 	testName := "GetBucketAcl_access_denied"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -6291,6 +6713,15 @@ func GetBucketAcl_success(s *S3Conf) error {
 			return err
 		}
 
+		grants = append([]types.Grant{
+			{
+				Grantee: &types.Grantee{
+					ID: &s.awsID,
+				},
+				Permission: types.PermissionFullControl,
+			},
+		}, grants...)
+
 		if ok := compareGrants(out.Grants, grants); !ok {
 			return fmt.Errorf("expected grants to be %v, instead got %v", grants, out.Grants)
 		}
@@ -6299,7 +6730,7 @@ func GetBucketAcl_success(s *S3Conf) error {
 		}
 
 		return nil
-	})
+	}, withOwnership(types.ObjectOwnershipBucketOwnerPreferred))
 }
 
 func PutBucketPolicy_non_existing_bucket(s *S3Conf) error {

tests/integration/utils.go

@@ -66,6 +66,7 @@ func setup(s *S3Conf, bucket string, opts ...setupOpt) error {
 	_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
 		Bucket:                     &bucket,
 		ObjectLockEnabledForBucket: &cfg.LockEnabled,
+		ObjectOwnership:            cfg.Ownership,
 	})
 	cancel()
 	return err
@@ -121,6 +122,7 @@ func teardown(s *S3Conf, bucket string) error {
 
 type setupCfg struct {
 	LockEnabled bool
+	Ownership   types.ObjectOwnership
 }
 
 type setupOpt func(*setupCfg)
@@ -128,6 +130,9 @@ type setupOpt func(*setupCfg)
 func withLock() setupOpt {
 	return func(s *setupCfg) { s.LockEnabled = true }
 }
+func withOwnership(o types.ObjectOwnership) setupOpt {
+	return func(s *setupCfg) { s.Ownership = o }
+}
 
 func actionHandler(s *S3Conf, testName string, handler func(s3client *s3.Client, bucket string) error, opts ...setupOpt) error {
 	runF(testName)