diff --git a/auth/acl.go b/auth/acl.go
new file mode 100644
index 0000000..05c9101
--- /dev/null
+++ b/auth/acl.go
@@ -0,0 +1,517 @@
+// 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 auth
+
+import (
+	"context"
+	"encoding/json"
+	"encoding/xml"
+	"fmt"
+	"slices"
+	"strings"
+
+	"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/debuglogger"
+	"github.com/versity/versitygw/s3err"
+)
+
+type ACL struct {
+	Owner    string
+	Grantees []Grantee
+}
+
+// IsPublic specifies if the acl grants public read access
+func (acl *ACL) IsPublic(permission Permission) bool {
+	for _, grt := range acl.Grantees {
+		if grt.Permission == permission && grt.Type == types.TypeGroup && grt.Access == "all-users" {
+			return true
+		}
+	}
+
+	return false
+}
+
+type Grantee struct {
+	Permission Permission
+	Access     string
+	Type       types.Type
+}
+
+type GetBucketAclOutput struct {
+	XMLName           xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ AccessControlPolicy"`
+	Owner             *types.Owner
+	AccessControlList AccessControlList
+}
+
+type PutBucketAclInput struct {
+	Bucket              *string
+	ACL                 types.BucketCannedACL
+	AccessControlPolicy *AccessControlPolicy
+	GrantFullControl    *string
+	GrantRead           *string
+	GrantReadACP        *string
+	GrantWrite          *string
+	GrantWriteACP       *string
+}
+
+type AccessControlPolicy struct {
+	AccessControlList AccessControlList `xml:"AccessControlList"`
+	Owner             *types.Owner
+}
+
+func (acp *AccessControlPolicy) Validate() error {
+	if !acp.AccessControlList.isValid() {
+		return s3err.GetAPIError(s3err.ErrMalformedACL)
+	}
+
+	// The Owner can't be nil
+	if acp.Owner == nil {
+		return s3err.GetAPIError(s3err.ErrMalformedACL)
+	}
+
+	// The Owner ID can't be empty
+	if acp.Owner.ID == nil || *acp.Owner.ID == "" {
+		return s3err.GetAPIError(s3err.ErrMalformedACL)
+	}
+
+	return nil
+}
+
+type AccessControlList struct {
+	Grants []Grant `xml:"Grant"`
+}
+
+// Validates the AccessControlList
+func (acl *AccessControlList) isValid() bool {
+	for _, el := range acl.Grants {
+		if !el.isValid() {
+			return false
+		}
+	}
+
+	return true
+}
+
+type Permission string
+
+const (
+	PermissionFullControl Permission = "FULL_CONTROL"
+	PermissionWrite       Permission = "WRITE"
+	PermissionWriteAcp    Permission = "WRITE_ACP"
+	PermissionRead        Permission = "READ"
+	PermissionReadAcp     Permission = "READ_ACP"
+)
+
+// Check if the permission is valid
+func (p Permission) isValid() bool {
+	return p == PermissionFullControl ||
+		p == PermissionRead ||
+		p == PermissionReadAcp ||
+		p == PermissionWrite ||
+		p == PermissionWriteAcp
+}
+
+type Grant struct {
+	Grantee    *Grt       `xml:"Grantee"`
+	Permission Permission `xml:"Permission"`
+}
+
+// Checks if Grant is valid
+func (g *Grant) isValid() bool {
+	return g.Permission.isValid() && g.Grantee.isValid()
+}
+
+type Grt struct {
+	XMLNS string     `xml:"xmlns:xsi,attr"`
+	Type  types.Type `xml:"xsi:type,attr"`
+	ID    string     `xml:"ID"`
+}
+
+// Custom Unmarshalling for Grt to parse xsi:type properly
+func (g *Grt) UnmarshalXML(d *xml.Decoder, start xml.StartElement) error {
+	// Iterate through the XML tokens to process the attributes
+	for _, attr := range start.Attr {
+		// Check if the attribute is xsi:type and belongs to the xsi namespace
+		if attr.Name.Space == "http://www.w3.org/2001/XMLSchema-instance" && attr.Name.Local == "type" {
+			g.Type = types.Type(attr.Value)
+		}
+		// Handle xmlns:xsi
+		if attr.Name.Local == "xmlns:xsi" {
+			g.XMLNS = attr.Value
+		}
+	}
+
+	// Decode the inner XML elements like ID
+	for {
+		t, err := d.Token()
+		if err != nil {
+			return err
+		}
+
+		switch se := t.(type) {
+		case xml.StartElement:
+			if se.Name.Local == "ID" {
+				if err := d.DecodeElement(&g.ID, &se); err != nil {
+					return err
+				}
+			}
+		case xml.EndElement:
+			if se.Name.Local == start.Name.Local {
+				return nil
+			}
+		}
+	}
+}
+
+// Validates Grt
+func (g *Grt) isValid() bool {
+	// Validate the Type
+	// Only these 2 types are supported in the gateway
+	if g.Type != types.TypeCanonicalUser && g.Type != types.TypeGroup {
+		return false
+	}
+
+	// The ID prop shouldn't be empty
+	if g.ID == "" {
+		return false
+	}
+
+	return true
+}
+
+func ParseACL(data []byte) (ACL, error) {
+	if len(data) == 0 {
+		return ACL{}, nil
+	}
+
+	var acl ACL
+	if err := json.Unmarshal(data, &acl); err != nil {
+		return acl, fmt.Errorf("parse acl: %w", err)
+	}
+	return acl, nil
+}
+
+func ParseACLOutput(data []byte, owner string) (GetBucketAclOutput, error) {
+	grants := []Grant{}
+
+	if len(data) == 0 {
+		return GetBucketAclOutput{
+			Owner: &types.Owner{
+				ID: &owner,
+			},
+			AccessControlList: AccessControlList{
+				Grants: grants,
+			},
+		}, nil
+	}
+
+	var acl ACL
+	if err := json.Unmarshal(data, &acl); err != nil {
+		return GetBucketAclOutput{}, fmt.Errorf("parse acl: %w", err)
+	}
+
+	for _, elem := range acl.Grantees {
+		acs := elem.Access
+		grants = append(grants, Grant{
+			Grantee: &Grt{
+				XMLNS: "http://www.w3.org/2001/XMLSchema-instance",
+				ID:    acs,
+				Type:  elem.Type,
+			},
+			Permission: elem.Permission,
+		})
+	}
+
+	return GetBucketAclOutput{
+		Owner: &types.Owner{
+			ID: &acl.Owner,
+		},
+		AccessControlList: AccessControlList{
+			Grants: grants,
+		},
+	}, nil
+}
+
+func UpdateACL(input *PutBucketAclInput, acl ACL, iam IAMService) ([]byte, error) {
+	if input == nil {
+		return nil, s3err.GetAPIError(s3err.ErrInvalidRequest)
+	}
+
+	defaultGrantees := []Grantee{
+		{
+			Permission: PermissionFullControl,
+			Access:     acl.Owner,
+			Type:       types.TypeCanonicalUser,
+		},
+	}
+
+	// if the ACL is specified, set the ACL, else replace the grantees
+	if input.ACL != "" {
+		switch input.ACL {
+		case types.BucketCannedACLPublicRead:
+			defaultGrantees = append(defaultGrantees, Grantee{
+				Permission: PermissionRead,
+				Access:     "all-users",
+				Type:       types.TypeGroup,
+			})
+		case types.BucketCannedACLPublicReadWrite:
+			defaultGrantees = append(defaultGrantees, []Grantee{
+				{
+					Permission: PermissionRead,
+					Access:     "all-users",
+					Type:       types.TypeGroup,
+				},
+				{
+					Permission: PermissionWrite,
+					Access:     "all-users",
+					Type:       types.TypeGroup,
+				},
+			}...)
+		}
+	} else {
+		accs := []string{}
+
+		if input.GrantRead != nil || input.GrantReadACP != nil || input.GrantFullControl != nil || input.GrantWrite != nil || input.GrantWriteACP != nil {
+			fullControlList, readList, readACPList, writeList, writeACPList := []string{}, []string{}, []string{}, []string{}, []string{}
+
+			if input.GrantFullControl != nil && *input.GrantFullControl != "" {
+				fullControlList = splitUnique(*input.GrantFullControl, ",")
+				for _, str := range fullControlList {
+					defaultGrantees = append(defaultGrantees, Grantee{
+						Access:     str,
+						Permission: PermissionFullControl,
+						Type:       types.TypeCanonicalUser,
+					})
+				}
+			}
+			if input.GrantRead != nil && *input.GrantRead != "" {
+				readList = splitUnique(*input.GrantRead, ",")
+				for _, str := range readList {
+					defaultGrantees = append(defaultGrantees, Grantee{
+						Access:     str,
+						Permission: PermissionRead,
+						Type:       types.TypeCanonicalUser,
+					})
+				}
+			}
+			if input.GrantReadACP != nil && *input.GrantReadACP != "" {
+				readACPList = splitUnique(*input.GrantReadACP, ",")
+				for _, str := range readACPList {
+					defaultGrantees = append(defaultGrantees, Grantee{
+						Access:     str,
+						Permission: PermissionReadAcp,
+						Type:       types.TypeCanonicalUser,
+					})
+				}
+			}
+			if input.GrantWrite != nil && *input.GrantWrite != "" {
+				writeList = splitUnique(*input.GrantWrite, ",")
+				for _, str := range writeList {
+					defaultGrantees = append(defaultGrantees, Grantee{
+						Access:     str,
+						Permission: PermissionWrite,
+						Type:       types.TypeCanonicalUser,
+					})
+				}
+			}
+			if input.GrantWriteACP != nil && *input.GrantWriteACP != "" {
+				writeACPList = splitUnique(*input.GrantWriteACP, ",")
+				for _, str := range writeACPList {
+					defaultGrantees = append(defaultGrantees, Grantee{
+						Access:     str,
+						Permission: PermissionWriteAcp,
+						Type:       types.TypeCanonicalUser,
+					})
+				}
+			}
+
+			accs = append(append(append(append(fullControlList, readList...), writeACPList...), readACPList...), writeList...)
+		} else {
+			cache := make(map[string]bool)
+			for _, grt := range input.AccessControlPolicy.AccessControlList.Grants {
+				if grt.Grantee == nil || grt.Grantee.ID == "" || grt.Permission == "" {
+					return nil, s3err.GetAPIError(s3err.ErrInvalidRequest)
+				}
+
+				access := grt.Grantee.ID
+				defaultGrantees = append(defaultGrantees, Grantee{
+					Access:     access,
+					Permission: grt.Permission,
+					Type:       types.TypeCanonicalUser,
+				})
+				if _, ok := cache[access]; !ok {
+					cache[access] = true
+					accs = append(accs, access)
+				}
+			}
+		}
+
+		// Check if the specified accounts exist
+		accList, err := iam.ResolveAccounts(accs)
+		if err != nil {
+			return nil, err
+		}
+		if len(accList) > 0 {
+			return nil, fmt.Errorf("accounts does not exist: %s", strings.Join(accList, ", "))
+		}
+	}
+
+	acl.Grantees = defaultGrantees
+
+	result, err := json.Marshal(acl)
+	if err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}
+
+func splitUnique(s, divider string) []string {
+	elements := strings.Split(s, divider)
+	uniqueElements := make(map[string]bool)
+	result := make([]string, 0, len(elements))
+
+	for _, element := range elements {
+		if _, ok := uniqueElements[element]; !ok {
+			result = append(result, element)
+			uniqueElements[element] = true
+		}
+	}
+
+	return result
+}
+
+// aclActionPermissions maps every action an ACL grant can authorize to the
+// permission that grants it. The gateway keeps no object ACLs, so the
+// bucket's grants stand in for its objects' as well: READ also reads
+// objects, and READ_ACP and WRITE_ACP also cover object ACLs. A grant
+// authorizes nothing else. Bucket configuration, the bucket policy, tagging
+// and object lock settings are the bucket owner's, or whoever a policy
+// grants them to. s3:DeleteObjectVersion is left out because WRITE grants
+// it to the bucket owner alone. READ reads an object but not its
+// attributes, so neither s3:GetObjectAttributes nor
+// s3:GetObjectVersionAttributes is here.
+var aclActionPermissions = map[Action]Permission{
+	ListBucketAction:                 PermissionRead,
+	ListBucketVersionsAction:         PermissionRead,
+	ListBucketMultipartUploadsAction: PermissionRead,
+	GetObjectAction:                  PermissionRead,
+	GetObjectVersionAction:           PermissionRead,
+	PutObjectAction:                  PermissionWrite,
+	DeleteObjectAction:               PermissionWrite,
+	GetBucketAclAction:               PermissionReadAcp,
+	GetObjectAclAction:               PermissionReadAcp,
+	PutBucketAclAction:               PermissionWriteAcp,
+	PutObjectAclAction:               PermissionWriteAcp,
+}
+
+// aclAllows reports whether acl lets access perform action. The bucket
+// owner may perform any action on its own bucket, as an AWS account may.
+// Any other grantee may perform only the actions its grants map to, and
+// none at all while ACLs are disabled.
+func aclAllows(acl ACL, access string, action Action, disableACL bool) bool {
+	if access == acl.Owner {
+		return true
+	}
+	if disableACL {
+		return false
+	}
+
+	permission, ok := aclActionPermissions[action]
+	if !ok {
+		return false
+	}
+
+	grantee := Grantee{
+		Access:     access,
+		Permission: permission,
+		Type:       types.TypeCanonicalUser,
+	}
+	granteeFullCtrl := Grantee{
+		Access:     access,
+		Permission: PermissionFullControl,
+		Type:       types.TypeCanonicalUser,
+	}
+	granteeAllUsers := Grantee{
+		Access:     "all-users",
+		Permission: permission,
+		Type:       types.TypeGroup,
+	}
+
+	return slices.ContainsFunc(acl.Grantees, func(grt Grantee) bool {
+		return grt == grantee || grt == granteeFullCtrl || grt == granteeAllUsers
+	})
+}
+
+// Verifies if the bucket acl grants public access
+func VerifyPublicBucketACL(ctx context.Context, be backend.Backend, bucket string, permission Permission) error {
+	aclBytes, err := be.GetBucketAcl(ctx, &s3.GetBucketAclInput{
+		Bucket: &bucket,
+	})
+	if err != nil {
+		return err
+	}
+
+	acl, err := ParseACL(aclBytes)
+	if err != nil {
+		return err
+	}
+
+	if !acl.IsPublic(permission) {
+		return errAccessDenied
+	}
+
+	return nil
+}
+
+// UpdateBucketACLOwner sets default ACL with new owner and removes
+// any previous bucket policy that was in place
+func UpdateBucketACLOwner(ctx context.Context, be backend.Backend, bucket, newOwner string) error {
+	acl := ACL{
+		Owner: newOwner,
+		Grantees: []Grantee{
+			{
+				Permission: PermissionFullControl,
+				Access:     newOwner,
+				Type:       types.TypeCanonicalUser,
+			},
+		},
+	}
+
+	result, err := json.Marshal(acl)
+	if err != nil {
+		return fmt.Errorf("marshal ACL: %w", err)
+	}
+
+	err = be.PutBucketAcl(ctx, bucket, result)
+	if err != nil {
+		return err
+	}
+
+	return be.DeleteBucketPolicy(ctx, bucket)
+}
+
+// ValidateCannedACL validates bucket canned acl value
+func ValidateCannedACL(acl types.BucketCannedACL) error {
+	switch types.BucketCannedACL(acl) {
+	case types.BucketCannedACLPrivate, types.BucketCannedACLPublicRead, types.BucketCannedACLPublicReadWrite, "":
+		return nil
+	default:
+		debuglogger.Logf("invalid bucket canned acl: %v", acl)
+		return s3err.GetInvalidArgumentErr(s3err.InvalidArgCannedAcl, string(acl))
+	}
+}
diff --git a/tests/integration/CreateBucket.go b/tests/integration/CreateBucket.go
new file mode 100644
index 0000000..2ebe3da
--- /dev/null
+++ b/tests/integration/CreateBucket.go
@@ -0,0 +1,663 @@
+// 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 integration
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/versity/versitygw/s3err"
+)
+
+func CreateBucket_invalid_bucket_name(s *S3Conf) error {
+	testName := "CreateBucket_invalid_bucket_name"
+	runF(testName)
+	err := setup(s, "aa")
+	if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidBucketName)); err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	err = setup(s, ".gitignore")
+	if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidBucketName)); err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	err = setup(s, "my-bucket.")
+	if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidBucketName)); err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	err = setup(s, "bucket-%")
+	if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidBucketName)); err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+	passF(testName)
+	return nil
+}
+
+func CreateBucket_as_user(s *S3Conf) error {
+	testName := "CreateBucket_as_user"
+	runF(testName)
+
+	testuser := getUser("user")
+	cfg := *s
+	cfg.awsID = testuser.access
+	cfg.awsSecret = testuser.secret
+	err := createUsers(s, []user{testuser})
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	err = setup(&cfg, getBucketName())
+	if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrAccessDenied)); err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	passF(testName)
+	return nil
+}
+
+func CreateBucket_existing_bucket(s *S3Conf) error {
+	testName := "CreateBucket_existing_bucket"
+	runF(testName)
+	bucket := getBucketName()
+	adminUser := getUser("admin")
+	if err := createUsers(s, []user{adminUser}); err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	adminCfg := *s
+	adminCfg.awsID = adminUser.access
+	adminCfg.awsSecret = adminUser.secret
+
+	err := setup(&adminCfg, bucket)
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+	err = setup(s, bucket)
+	var bne *types.BucketAlreadyExists
+	if !errors.As(err, &bne) {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	err = teardown(s, bucket)
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+	passF(testName)
+	return nil
+}
+
+func CreateBucket_owned_by_you(s *S3Conf) error {
+	testName := "CreateBucket_owned_by_you"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		var bErr *types.BucketAlreadyOwnedByYou
+		if !errors.As(err, &bErr) {
+			return fmt.Errorf("expected error to be %w, instead got %w", s3err.GetAPIError(s3err.ErrBucketAlreadyOwnedByYou), err)
+		}
+
+		return nil
+	})
+}
+
+func CreateBucket_invalid_ownership(s *S3Conf) error {
+	testName := "CreateBucket_invalid_ownership"
+	runF(testName)
+
+	invalidOwnership := types.ObjectOwnership("invalid_ownership")
+	err := setup(s, getBucketName(), withOwnership(invalidOwnership))
+	if err := checkApiErr(err, s3err.GetInvalidArgObjectOwnership(string(invalidOwnership))); err != nil {
+		failF("%v: %v", testName, 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 := s.GetClient()
+
+	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 getString(out.Owner.ID) != s.awsID {
+			return fmt.Errorf("expected bucket owner to be %v, instead got %v",
+				s.awsID, getString(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 getString(grt.Grantee.ID) != s.awsID {
+			return fmt.Errorf("expected the grantee id to be %v, instead got %v",
+				s.awsID, getString(grt.Grantee.ID))
+		}
+
+		return nil
+	})
+}
+
+func CreateBucket_non_default_acl(s *S3Conf) error {
+	testName := "CreateBucket_non_default_acl"
+	runF(testName)
+
+	testuser1, testuser2, testuser3 := getUser("user"), getUser("user"), getUser("user")
+	err := createUsers(s, []user{testuser1, testuser2, testuser3})
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	grants := []types.Grant{
+		{
+			Grantee: &types.Grantee{
+				ID:   &s.awsID,
+				Type: types.TypeCanonicalUser,
+			},
+			Permission: types.PermissionFullControl,
+		},
+		{
+			Grantee: &types.Grantee{
+				ID:   &testuser1.access,
+				Type: types.TypeCanonicalUser,
+			},
+			Permission: types.PermissionFullControl,
+		},
+		{
+			Grantee: &types.Grantee{
+				ID:   &testuser2.access,
+				Type: types.TypeCanonicalUser,
+			},
+			Permission: types.PermissionReadAcp,
+		},
+		{
+			Grantee: &types.Grantee{
+				ID:   &testuser3.access,
+				Type: types.TypeCanonicalUser,
+			},
+			Permission: types.PermissionWrite,
+		},
+	}
+
+	bucket := getBucketName()
+	client := s.GetClient()
+
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.CreateBucket(ctx, &s3.CreateBucketInput{
+		Bucket:           &bucket,
+		GrantFullControl: &testuser1.access,
+		GrantReadACP:     &testuser2.access,
+		GrantWrite:       &testuser3.access,
+		ObjectOwnership:  types.ObjectOwnershipBucketOwnerPreferred,
+	})
+	cancel()
+	if 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 !compareGrants(out.Grants, grants) {
+		failF("%v: expected bucket acl grants to be %v, instead got %v", testName, grants, out.Grants)
+		return fmt.Errorf("%v: expected bucket acl grants to be %v, instead got %v", testName, grants, out.Grants)
+	}
+
+	err = teardown(s, bucket)
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	passF(testName)
+	return nil
+}
+
+func CreateBucket_private_canned_acl(s *S3Conf) error {
+	testName := "CreateBucket_private_canned_acl"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+			ACL:    types.BucketCannedACLPrivate,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(res.Owner.ID) != s.awsID {
+			return fmt.Errorf("expected bucket owner to be %v, instead got %v",
+				s.awsID, getString(res.Owner.ID))
+		}
+		if len(res.Grants) != 1 {
+			return fmt.Errorf("expected grants length to be 1, instead got %v",
+				len(res.Grants))
+		}
+		grt := res.Grants[0]
+		if grt.Permission != types.PermissionFullControl {
+			return fmt.Errorf("expected the grantee to have full-control permission, instead got %v",
+				grt.Permission)
+		}
+		if getString(grt.Grantee.ID) != s.awsID {
+			return fmt.Errorf("expected the grantee id to be %v, instead got %v",
+				s.awsID, getString(grt.Grantee.ID))
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if out.OwnershipControls == nil {
+			return fmt.Errorf("unexpected nil OwnershipControls")
+		}
+		if len(out.OwnershipControls.Rules) != 1 {
+			return fmt.Errorf("expected the OwnershipControls rules length to be 1, instead got %v", len(out.OwnershipControls.Rules))
+		}
+		if out.OwnershipControls.Rules[0].ObjectOwnership != types.ObjectOwnershipBucketOwnerPreferred {
+			return fmt.Errorf("expected the ObjectOwnership to be %s, instead got %s", types.ObjectOwnershipBucketOwnerPreferred, out.OwnershipControls.Rules[0].ObjectOwnership)
+		}
+
+		return teardown(s, bucket)
+	})
+}
+
+func CreateBucket_private_canned_acl_bucket_owner_enforced_ownership(s *S3Conf) error {
+	testName := "CreateBucket_private_canned_acl_bucket_owner_enforced_ownership"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket:          &bucket,
+			ACL:             types.BucketCannedACLPrivate,
+			ObjectOwnership: types.ObjectOwnershipBucketOwnerEnforced,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(res.Owner.ID) != s.awsID {
+			return fmt.Errorf("expected bucket owner to be %v, instead got %v",
+				s.awsID, getString(res.Owner.ID))
+		}
+		if len(res.Grants) != 1 {
+			return fmt.Errorf("expected grants length to be 1, instead got %v",
+				len(res.Grants))
+		}
+		grt := res.Grants[0]
+		if grt.Permission != types.PermissionFullControl {
+			return fmt.Errorf("expected the grantee to have full-control permission, instead got %v",
+				grt.Permission)
+		}
+		if getString(grt.Grantee.ID) != s.awsID {
+			return fmt.Errorf("expected the grantee id to be %v, instead got %v",
+				s.awsID, getString(grt.Grantee.ID))
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if out.OwnershipControls == nil {
+			return fmt.Errorf("unexpected nil OwnershipControls")
+		}
+		if len(out.OwnershipControls.Rules) != 1 {
+			return fmt.Errorf("expected the OwnershipControls rules length to be 1, instead got %v", len(out.OwnershipControls.Rules))
+		}
+		if out.OwnershipControls.Rules[0].ObjectOwnership != types.ObjectOwnershipBucketOwnerEnforced {
+			return fmt.Errorf("expected the ObjectOwnership to be %s, instead got %s", types.ObjectOwnershipBucketOwnerEnforced, out.OwnershipControls.Rules[0].ObjectOwnership)
+		}
+
+		return teardown(s, bucket)
+	})
+}
+
+func CreateBucket_default_object_lock(s *S3Conf) error {
+	testName := "CreateBucket_default_object_lock"
+	runF(testName)
+
+	bucket := getBucketName()
+	lockEnabled := true
+
+	client := s.GetClient()
+
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	_, err := client.CreateBucket(ctx, &s3.CreateBucketInput{
+		Bucket:                     &bucket,
+		ObjectLockEnabledForBucket: &lockEnabled,
+	})
+	cancel()
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	resp, err := client.GetObjectLockConfiguration(ctx, &s3.GetObjectLockConfigurationInput{
+		Bucket: &bucket,
+	})
+	cancel()
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	if resp.ObjectLockConfiguration.ObjectLockEnabled != types.ObjectLockEnabledEnabled {
+		failF("%v: expected object lock to be enabled", testName)
+		return fmt.Errorf("%v: expected object lock to be enabled", testName)
+	}
+
+	err = teardown(s, bucket)
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	passF(testName)
+	return nil
+}
+
+func CreateBucket_invalid_location_constraint(s *S3Conf) error {
+	testName := "CreateBucket_invalid_location_constraint"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		var region types.BucketLocationConstraint
+		if types.BucketLocationConstraint(s.awsID) == types.BucketLocationConstraintUsWest1 {
+			region = types.BucketLocationConstraintUsWest2
+		} else {
+			region = types.BucketLocationConstraintUsWest1
+		}
+
+		createBucket := func(region types.BucketLocationConstraint) error {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+				Bucket: &bucket,
+				CreateBucketConfiguration: &types.CreateBucketConfiguration{
+					LocationConstraint: region,
+				},
+			})
+			cancel()
+
+			return checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidLocationConstraint))
+		}
+
+		for _, lConstraint := range []types.BucketLocationConstraint{region, "us-east-1"} {
+			err := createBucket(lConstraint)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+func CreateBucket_long_tags(s *S3Conf) error {
+	testName := "CreateBucket_long_tags"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		tagging := []types.Tag{{Key: getPtr(genRandString(200)), Value: getPtr("val")}}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+			CreateBucketConfiguration: &types.CreateBucketConfiguration{
+				Tags: tagging,
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidTagKey)); err != nil {
+			return err
+		}
+
+		tagging = []types.Tag{{Key: getPtr("key"), Value: getPtr(genRandString(300))}}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+			CreateBucketConfiguration: &types.CreateBucketConfiguration{
+				Tags: tagging,
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidTagValue)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func CreateBucket_invalid_tags(s *S3Conf) error {
+	testName := "CreateBucket_invalid_tags"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		for i, test := range []struct {
+			tags []types.Tag
+			err  s3err.APIError
+		}{
+			// invalid tag key tests
+			{[]types.Tag{{Key: getPtr("user!name"), Value: getPtr("value")}}, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
+			{[]types.Tag{{Key: getPtr("foo#bar"), Value: getPtr("value")}}, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
+			{[]types.Tag{
+				{Key: getPtr("validkey"), Value: getPtr("validvalue")},
+				{Key: getPtr("data%20"), Value: getPtr("value")},
+			}, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
+			{[]types.Tag{
+				{Key: getPtr("abcd"), Value: getPtr("xyz123")},
+				{Key: getPtr("a*b"), Value: getPtr("value")},
+			}, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
+			// invalid tag value tests
+			{[]types.Tag{
+				{Key: getPtr("hello"), Value: getPtr("world")},
+				{Key: getPtr("key"), Value: getPtr("name?test")},
+			}, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
+			{[]types.Tag{
+				{Key: getPtr("foo"), Value: getPtr("bar")},
+				{Key: getPtr("key"), Value: getPtr("`path")},
+			}, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
+			{[]types.Tag{{Key: getPtr("valid"), Value: getPtr("comma,separated")}}, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
+			{[]types.Tag{{Key: getPtr("valid"), Value: getPtr("semicolon;test")}}, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
+			{[]types.Tag{{Key: getPtr("valid"), Value: getPtr("(parentheses)")}}, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
+		} {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+				Bucket: &bucket,
+				CreateBucketConfiguration: &types.CreateBucketConfiguration{
+					Tags: test.tags,
+				},
+			})
+			cancel()
+			if err == nil {
+				return fmt.Errorf("test %v failed: expected err %w, instead got nil", i+1, test.err)
+			}
+
+			if err := checkApiErr(err, test.err); err != nil {
+				return fmt.Errorf("test %v failed: %w", i+1, err)
+			}
+		}
+
+		return nil
+	})
+}
+
+func CreateBucket_duplicate_keys(s *S3Conf) error {
+	testName := "CreateBucket_duplicate_keys"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		tagging := []types.Tag{
+			{Key: getPtr("key"), Value: getPtr("value")},
+			{Key: getPtr("key"), Value: getPtr("value-1")},
+			{Key: getPtr("key-1"), Value: getPtr("value-2")},
+			{Key: getPtr("key-2"), Value: getPtr("value-3")},
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+			CreateBucketConfiguration: &types.CreateBucketConfiguration{
+				Tags: tagging,
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrDuplicateTagKey)); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func CreateBucket_tag_count_limit(s *S3Conf) error {
+	testName := "CreateBucket_tag_count_limit"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		tagSet := []types.Tag{}
+
+		for i := range 51 {
+			tagSet = append(tagSet, types.Tag{
+				Key:   getPtr(fmt.Sprintf("key-%v", i)),
+				Value: getPtr(genRandString(10)),
+			})
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+			CreateBucketConfiguration: &types.CreateBucketConfiguration{
+				Tags: tagSet,
+			},
+		})
+		cancel()
+		return checkApiErr(err, s3err.GetAPIError(s3err.ErrBucketTaggingLimited))
+	})
+}
+
+func CreateBucket_invalid_canned_acl(s *S3Conf) error {
+	testName := "CreateBucket_invalid_canned_acl"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+			ACL:    types.BucketCannedACL("invalid_acl"),
+		})
+		cancel()
+		return checkSdkApiErr(err, "InvalidArgument")
+	})
+}
+
+func CreateBucket_success(s *S3Conf) error {
+	testName := "CreateBucket_success"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		expectedLocation := "/" + bucket
+		if getString(res.Location) != expectedLocation {
+			return fmt.Errorf("expected the bucket Location to be %s, instead got %s", expectedLocation, getString(res.Location))
+		}
+		expectedArn := "arn:aws:s3:::" + bucket
+		if getString(res.BucketArn) != expectedArn {
+			return fmt.Errorf("expected the bucket arn to be %s, instead got %s", expectedArn, getString(res.BucketArn))
+		}
+
+		return teardown(s, bucket)
+	})
+}
diff --git a/tests/integration/iam.go b/tests/integration/iam.go
new file mode 100644
index 0000000..28aae11
--- /dev/null
+++ b/tests/integration/iam.go
@@ -0,0 +1,367 @@
+// 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 integration
+
+import (
+	"context"
+	"encoding/xml"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"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"
+)
+
+func IAM_user_access_denied(s *S3Conf) error {
+	testName := "IAM_user_access_denied"
+	runF(testName)
+
+	testuser := getUser("user")
+	err := createUsers(s, []user{testuser})
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	out, err := execCommand(s.getAdminCommand("-a", testuser.access, "-s", testuser.secret, "-er", s.endpoint, "delete-user", "-a", "random_access")...)
+	if err == nil {
+		failF("%v: expected cmd error", testName)
+		return fmt.Errorf("%v: expected cmd error", testName)
+	}
+	if !strings.Contains(string(out), s3err.GetAPIError(s3err.ErrAdminAccessDenied).Code) {
+		failF("%v: expected response error message to be %v, instead got %s",
+			testName, s3err.GetAPIError(s3err.ErrAdminAccessDenied).Error(), out)
+		return fmt.Errorf("%v: expected response error message to be %v, instead got %s",
+			testName, s3err.GetAPIError(s3err.ErrAdminAccessDenied).Error(), out)
+	}
+
+	passF(testName)
+
+	return nil
+}
+
+func IAM_userplus_access_denied(s *S3Conf) error {
+	testName := "IAM_userplus_access_denied"
+	runF(testName)
+
+	testuser := getUser("userplus")
+	err := createUsers(s, []user{testuser})
+	if err != nil {
+		failF("%v: %v", testName, err)
+		return fmt.Errorf("%v: %w", testName, err)
+	}
+
+	out, err := execCommand(s.getAdminCommand("-a", testuser.access, "-s", testuser.secret, "-er", s.endpoint, "delete-user", "-a", "random_access")...)
+	if err == nil {
+		failF("%v: expected cmd error", testName)
+		return fmt.Errorf("%v: expected cmd error", testName)
+	}
+	if !strings.Contains(string(out), s3err.GetAPIError(s3err.ErrAdminAccessDenied).Code) {
+		failF("%v: expected response error message to be %v, instead got %s",
+			testName, s3err.GetAPIError(s3err.ErrAdminAccessDenied).Error(), out)
+		return fmt.Errorf("%v: expected response error message to be %v, instead got %s",
+			testName, s3err.GetAPIError(s3err.ErrAdminAccessDenied).Error(), out)
+	}
+
+	passF(testName)
+
+	return nil
+}
+
+func IAM_userplus_CreateBucket(s *S3Conf) error {
+	testName := "IAM_userplus_CreateBucket"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		testuser := getUser("userplus")
+		err := createUsers(s, []user{testuser})
+		if err != nil {
+			return err
+		}
+
+		cfg := *s
+		cfg.awsID = testuser.access
+		cfg.awsSecret = testuser.secret
+
+		bckt := getBucketName()
+		err = setup(&cfg, bckt)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: &bckt})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		err = teardown(&cfg, bckt)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func IAM_admin_ChangeBucketOwner(s *S3Conf) error {
+	testName := "IAM_admin_ChangeBucketOwner"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		testuser, adminuser := getUser("user"), getUser("admin")
+		err := createUsers(s, []user{adminuser, testuser})
+		if err != nil {
+			return err
+		}
+
+		err = changeBucketsOwner(s, []string{bucket}, testuser.access)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		resp, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{Bucket: &bucket})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if getString(resp.Owner.ID) != testuser.access {
+			return fmt.Errorf("expected the bucket owner to be %v, instead got %v",
+				testuser.access, getString(resp.Owner.ID))
+		}
+
+		return nil
+	})
+}
+
+func IAM_ChangeBucketOwner_back_to_root(s *S3Conf) error {
+	testName := "IAM_ChangeBucketOwner_back_to_root"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		testuser := getUser("user")
+		if err := createUsers(s, []user{testuser}); err != nil {
+			return err
+		}
+
+		// Change the bucket ownership to a random user
+		if err := changeBucketsOwner(s, []string{bucket}, testuser.access); err != nil {
+			return err
+		}
+
+		// Change the bucket ownership back to the root user
+		if err := changeBucketsOwner(s, []string{bucket}, s.awsID); err != nil {
+			return err
+		}
+
+		return nil
+	})
+}
+
+func IAM_ListBuckets(s *S3Conf) error {
+	testName := "IAM_ListBuckets"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		return listBuckets(s)
+	})
+}
+
+func IAM_CreateBucket_empty_owner_header(s *S3Conf) error {
+	testName := "IAM_CreateBucket_empty_owner_header"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		req, err := createSignedReq(
+			http.MethodPatch,
+			s.endpoint,
+			fmt.Sprintf("%s/create", bucket),
+			s.awsID,
+			s.awsSecret,
+			"s3",
+			s.awsRegion,
+			"",
+			nil,
+			time.Now(),
+			nil,
+		)
+		if err != nil {
+			return err
+		}
+
+		resp, err := s.httpClient.Do(req)
+		if err != nil {
+			return err
+		}
+
+		return checkHTTPResponseApiErr(resp, s3err.GetAPIError(s3err.ErrAdminEmptyBucketOwnerHeader))
+	})
+}
+
+func IAM_CreateBucket_non_existing_user(s *S3Conf) error {
+	testName := "IAM_CreateBucket_non_existing_user"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		req, err := createSignedReq(
+			http.MethodPatch,
+			s.endpoint,
+			fmt.Sprintf("%s/create", bucket),
+			s.awsID,
+			s.awsSecret,
+			"s3",
+			s.awsRegion,
+			"",
+			nil,
+			time.Now(),
+			map[string]string{
+				"x-vgw-owner": "non-existing-user",
+			},
+		)
+		if err != nil {
+			return err
+		}
+
+		resp, err := s.httpClient.Do(req)
+		if err != nil {
+			return err
+		}
+
+		return checkHTTPResponseApiErr(resp, s3err.GetAPIError(s3err.ErrAdminUserNotFound))
+	})
+}
+
+func IAM_CreateBucket_success(s *S3Conf) error {
+	testName := "IAM_CreateBucket_success"
+	return actionHandlerNoSetup(s, testName, func(s3client *s3.Client, bucket string) error {
+		tagSet := []types.Tag{
+			{Key: getPtr("key"), Value: getPtr("value")},
+		}
+		body, err := xml.Marshal(s3response.CreateBucketConfiguration{
+			TagSet: tagSet,
+		})
+		if err != nil {
+			return err
+		}
+
+		testUser1, testUser2 := getUser("user"), getUser("user")
+		err = createUsers(s, []user{testUser1, testUser2})
+		if err != nil {
+			return err
+		}
+
+		req, err := createSignedReq(
+			http.MethodPatch,
+			s.endpoint,
+			fmt.Sprintf("%s/create", bucket),
+			s.awsID,
+			s.awsSecret,
+			"s3",
+			s.awsRegion,
+			"",
+			body,
+			time.Now(),
+			map[string]string{
+				"x-amz-bucket-object-lock-enabled": "true",
+				"x-amz-object-ownership":           string(types.ObjectOwnershipBucketOwnerPreferred),
+				"x-amz-grant-read":                 testUser2.access,
+				"x-vgw-owner":                      testUser1.access,
+			},
+		)
+		if err != nil {
+			return err
+		}
+
+		resp, err := s.httpClient.Do(req)
+		if err != nil {
+			return err
+		}
+
+		if resp.StatusCode != http.StatusCreated {
+			return fmt.Errorf("expected the response status code to be %v, instead got %v", http.StatusCreated, resp.StatusCode)
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		tagging, err := s3client.GetBucketTagging(ctx, &s3.GetBucketTaggingInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !areTagsSame(tagSet, tagging.TagSet) {
+			return fmt.Errorf("expected the bucket tagging to be %v, instead got %v", tagSet, tagging.TagSet)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		ownership, err := s3client.GetBucketOwnershipControls(ctx, &s3.GetBucketOwnershipControlsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		var ownershipControls types.ObjectOwnership
+		if ownership.OwnershipControls != nil && len(ownership.OwnershipControls.Rules) == 1 {
+			ownershipControls = ownership.OwnershipControls.Rules[0].ObjectOwnership
+		}
+
+		if ownershipControls != types.ObjectOwnershipBucketOwnerPreferred {
+			return fmt.Errorf("expected the bucket ownership controls to be %s, instaed got %s", types.ObjectOwnershipBucketOwnerPreferred, ownershipControls)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		acl, err := s3client.GetBucketAcl(ctx, &s3.GetBucketAclInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if len(acl.Grants) != 2 {
+			return fmt.Errorf("expected the length of acl grants to be 2, instead got %v", len(acl.Grants))
+		}
+
+		var granteeChecked bool
+		var ownerChecked bool
+		for _, grant := range acl.Grants {
+			// owner
+			if getString(grant.Grantee.ID) == testUser1.access {
+				ownerChecked = true
+				if grant.Permission != types.PermissionFullControl {
+					return fmt.Errorf("expected the owner '%s' to have %s permission, instead got %s", testUser1.access, types.PermissionFullControl, grant.Permission)
+				}
+
+				continue
+			}
+
+			if getString(grant.Grantee.ID) != testUser2.access {
+				return fmt.Errorf("expected the grantee ID to be %v, instaed got %v", testUser2.access, getString(grant.Grantee.ID))
+			}
+			if grant.Permission != types.PermissionRead {
+				return fmt.Errorf("expected the %v user permission to be %s, instead got %s", testUser2.access, types.PermissionRead, grant.Permission)
+			}
+
+			granteeChecked = true
+		}
+
+		if !ownerChecked {
+			return fmt.Errorf("missing the owner '%s' full control acl", testUser1.access)
+		}
+		if !granteeChecked {
+			return fmt.Errorf("missing the user %s in read grantees acl", testUser2.access)
+		}
+
+		return teardown(s, bucket)
+	})
+}
diff --git a/tests/test_rest_acl.sh b/tests/test_rest_acl.sh
new file mode 100644
index 0000000..019b137
--- /dev/null
+++ b/tests/test_rest_acl.sh
@@ -0,0 +1,240 @@
+#!/usr/bin/env bats
+
+# Copyright 2024 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.
+
+load ./bats-support/load
+load ./bats-assert/load
+
+source ./tests/commands/put_object.sh
+source ./tests/drivers/file.sh
+source ./tests/drivers/create_bucket/create_bucket_rest.sh
+source ./tests/drivers/get_bucket_acl/get_bucket_acl_rest.sh
+source ./tests/logger.sh
+source ./tests/setup.sh
+source ./tests/util/util_object.sh
+source ./tests/util/util_public_access_block.sh
+
+export RUN_USERS=true
+
+if [ "$SKIP_ACL_TESTING" == "true" ] || [ "$SKIP_USERS_TESTS" == "true" ]; then
+  skip "Skipping ACL tests"
+  exit 0
+fi
+
+# tags: curl,minimal-request,GetBucketAcl
+@test "REST - get ACL" {
+  run get_bucket_name "$BUCKET_ONE_NAME"
+  assert_success
+  bucket_name="$output"
+
+  run setup_bucket_v2 "$bucket_name"
+  assert_success
+
+  run get_and_check_acl_rest "$bucket_name"
+  assert_success
+}
+
+# tags: curl,minimal-request,PutBucketAcl
+@test "REST - put ACL" {
+  run get_bucket_name "$BUCKET_ONE_NAME"
+  assert_success
+  bucket_name="$output"
+
+  test_file="test_file"
+  run setup_bucket_and_file_v2 "$bucket_name" "$test_file"
+  assert_success
+
+  run put_bucket_ownership_controls "$bucket_name" "BucketOwnerPreferred"
+  assert_success
+
+  run put_object "s3api" "$TEST_FILE_FOLDER/$test_file" "$bucket_name" "$test_file"
+  assert_success
+
+  run create_versitygw_acl_user_or_get_direct_user "$USERNAME_ONE" "$PASSWORD_ONE"
+  assert_success
+  canonical_id=${lines[0]}
+  user_canonical_id=${lines[1]}
+  username=${lines[2]}
+  password=${lines[3]}
+
+  run setup_acl "$TEST_FILE_FOLDER/acl-file.txt" "CanonicalUser" "$user_canonical_id" "READ" "$canonical_id"
+  assert_success
+
+  run list_objects_with_user_rest_verify_access_denied "$bucket_name" "$username" "$password"
+  assert_success
+
+  run put_bucket_acl_rest "$bucket_name" "$TEST_FILE_FOLDER/acl-file.txt"
+  assert_success
+
+  if [ "$DIRECT" == "true" ]; then
+    sleep 5
+  fi
+
+  run list_objects_with_user_rest_verify_success "$bucket_name" "$username" "$password" "$test_file"
+  assert_success
+}
+
+# tags: curl,PutBucketAcl,x-amz-acl
+@test "REST - put public-read canned acl" {
+  run get_bucket_name "$BUCKET_ONE_NAME"
+  assert_success
+  bucket_name="$output"
+
+  test_file="test_file"
+  run setup_bucket_and_file_v2 "$bucket_name" "$test_file"
+  assert_success
+
+  run put_bucket_ownership_controls "$bucket_name" "BucketOwnerPreferred"
+  assert_success
+
+  run put_object "s3api" "$TEST_FILE_FOLDER/$test_file" "$bucket_name" "$test_file"
+  assert_success
+
+  run create_versitygw_acl_user_or_get_direct_user "$USERNAME_ONE" "$PASSWORD_ONE"
+  assert_success
+  canonical_id=${lines[0]}
+  user_canonical_id=${lines[1]}
+  username=${lines[2]}
+  password=${lines[3]}
+
+  run list_objects_with_user_rest_verify_access_denied "$bucket_name" "$username" "$password"
+  assert_success
+
+  if [ "$DIRECT" == "true" ]; then
+    run allow_public_access "$bucket_name"
+    assert_success
+  fi
+  run put_canned_acl_rest "$bucket_name" "public-read"
+  assert_success
+
+  run list_objects_with_user_rest_verify_success "$bucket_name" "$username" "$password" "$test_file"
+  assert_success
+}
+
+# tags: curl,minimal-request,PutBucketAcl
+@test "REST - put invalid ACL" {
+  run get_bucket_name "$BUCKET_ONE_NAME"
+  assert_success
+  bucket_name="$output"
+
+  run setup_bucket_v2 "$bucket_name"
+  assert_success
+
+   run put_bucket_ownership_controls "$bucket_name" "BucketOwnerPreferred"
+  assert_success
+
+  run create_versitygw_acl_user_or_get_direct_user "$USERNAME_ONE" "$PASSWORD_ONE"
+  assert_success
+  canonical_id=${lines[0]}
+  user_canonical_id=${lines[1]}
+  username=${lines[2]}
+  password=${lines[3]}
+
+  run setup_acl "$TEST_FILE_FOLDER/acl-file.txt" "CanonicalUser" "$user_canonical_id" "READD" "$canonical_id"
+  assert_success
+
+  if [ "$DIRECT" == "true" ]; then
+    run allow_public_access "$bucket_name"
+    assert_success
+  fi
+  run put_invalid_acl_rest_verify_failure "$bucket_name" "$TEST_FILE_FOLDER/acl-file.txt"
+  assert_success
+}
+
+# tags: curl,PutBucketAcl,x-amz-acl
+@test "REST - put public-read-write canned acl" {
+  run get_bucket_name "$BUCKET_ONE_NAME"
+  assert_success
+  bucket_name="$output"
+
+  test_file="test_file"
+  run setup_bucket_and_file_v2 "$bucket_name" "$test_file"
+  assert_success
+
+  run put_bucket_ownership_controls "$bucket_name" "BucketOwnerPreferred"
+  assert_success
+
+  run create_versitygw_acl_user_or_get_direct_user "$USERNAME_ONE" "$PASSWORD_ONE"
+  assert_success
+  canonical_id=${lines[0]}
+  user_canonical_id=${lines[1]}
+  username=${lines[2]}
+  password=${lines[3]}
+
+  run put_object_with_user "s3api" "$TEST_FILE_FOLDER/$test_file" "$bucket_name" "$test_file" "$username" "$password"
+  assert_failure
+
+  if [ "$DIRECT" == "true" ]; then
+    run allow_public_access "$bucket_name"
+    assert_success
+  fi
+  run put_canned_acl_rest "$bucket_name" "public-read-write"
+  assert_success
+
+  run put_object_with_user "s3api" "$TEST_FILE_FOLDER/$test_file" "$bucket_name" "$test_file" "$username" "$password"
+  assert_success
+}
+
+# tags: curl,PutBucketAcl,x-amz-acl
+@test "REST - invalid canned acl" {
+  if [ "$DIRECT" != "true" ]; then
+    skip "https://github.com/versity/versitygw/issues/1367"
+  fi
+  run get_bucket_name "$BUCKET_ONE_NAME"
+  assert_success
+  bucket_name="$output"
+
+  test_file="test_file"
+  run setup_bucket_and_file_v2 "$bucket_name" "$test_file"
+  assert_success
+
+  run put_bucket_ownership_controls "$bucket_name" "BucketOwnerPreferred"
+  assert_success
+
+  run put_bucket_acl_rest_canned_invalid "$bucket_name" "privatee"
+  assert_success
+}
+
+# tags: curl,GetBucketAcl,x-amz-grant-read-acp
+@test "REST - FULL_CONTROL permission not returned for owner after CreateBucket with GRANT_READ_ACP" {
+  if [ "$DIRECT" != "true" ]; then
+    skip "https://github.com/versity/versitygw/issues/1407"
+  fi
+  run get_bucket_name "$BUCKET_ONE_NAME"
+  assert_success
+  bucket_name="$output"
+
+  run create_versitygw_acl_user_or_get_direct_user "$USERNAME_ONE" "$PASSWORD_ONE"
+  assert_success
+  canonical_id=${lines[0]}
+  user_canonical_id=${lines[1]}
+  username=${lines[2]}
+  password=${lines[3]}
+
+  run bucket_cleanup_if_bucket_exists "$bucket_name"
+  assert_success
+  if [ "$DIRECT" == "true" ]; then
+    id="id=$user_canonical_id"
+  else
+    id="$user_canonical_id"
+  fi
+
+  envs="GRANT_READ_ACP=$id OBJECT_OWNERSHIP=BucketOwnerPreferred"
+  run create_bucket_rest_expect_success "$bucket_name" "$envs"
+  assert_success
+
+  run get_bucket_acl_rest "$bucket_name" "" "check_that_acl_xml_does_not_have_owner_permission"
+  assert_success
+}