fix: drop the owner's FULL_CONTROL grant when grant headers set the bucket ACL

CreateBucket and PutBucketAcl with x-amz-grant-* headers stored the
requested grants plus a FULL_CONTROL grant for the bucket owner, so
GetBucketAcl returned a grant AWS doesn't. On AWS the headers replace the
whole ACL and the owner gets FULL_CONTROL only when a header names it.

UpdateACL now keeps just the header grants when there are any. A canned
ACL and a bucket created without grants still get the owner's FULL_CONTROL
grant. The owner keeps full access either way, since aclAllows lets the
bucket owner perform any action regardless of grants.

Fixes #1407

auth/acl.go

@@ -287,11 +287,12 @@ func UpdateACL(input *PutBucketAclInput, acl ACL, iam IAMService) ([]byte, error
 
 		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{}
+			headerGrantees := []Grantee{}
 
 			if input.GrantFullControl != nil && *input.GrantFullControl != "" {
 				fullControlList = splitUnique(*input.GrantFullControl, ",")
 				for _, str := range fullControlList {
-					defaultGrantees = append(defaultGrantees, Grantee{
+					headerGrantees = append(headerGrantees, Grantee{
 						Access:     str,
 						Permission: PermissionFullControl,
 						Type:       types.TypeCanonicalUser,
@@ -301,7 +302,7 @@ func UpdateACL(input *PutBucketAclInput, acl ACL, iam IAMService) ([]byte, error
 			if input.GrantRead != nil && *input.GrantRead != "" {
 				readList = splitUnique(*input.GrantRead, ",")
 				for _, str := range readList {
-					defaultGrantees = append(defaultGrantees, Grantee{
+					headerGrantees = append(headerGrantees, Grantee{
 						Access:     str,
 						Permission: PermissionRead,
 						Type:       types.TypeCanonicalUser,
@@ -311,7 +312,7 @@ func UpdateACL(input *PutBucketAclInput, acl ACL, iam IAMService) ([]byte, error
 			if input.GrantReadACP != nil && *input.GrantReadACP != "" {
 				readACPList = splitUnique(*input.GrantReadACP, ",")
 				for _, str := range readACPList {
-					defaultGrantees = append(defaultGrantees, Grantee{
+					headerGrantees = append(headerGrantees, Grantee{
 						Access:     str,
 						Permission: PermissionReadAcp,
 						Type:       types.TypeCanonicalUser,
@@ -321,7 +322,7 @@ func UpdateACL(input *PutBucketAclInput, acl ACL, iam IAMService) ([]byte, error
 			if input.GrantWrite != nil && *input.GrantWrite != "" {
 				writeList = splitUnique(*input.GrantWrite, ",")
 				for _, str := range writeList {
-					defaultGrantees = append(defaultGrantees, Grantee{
+					headerGrantees = append(headerGrantees, Grantee{
 						Access:     str,
 						Permission: PermissionWrite,
 						Type:       types.TypeCanonicalUser,
@@ -331,7 +332,7 @@ func UpdateACL(input *PutBucketAclInput, acl ACL, iam IAMService) ([]byte, error
 			if input.GrantWriteACP != nil && *input.GrantWriteACP != "" {
 				writeACPList = splitUnique(*input.GrantWriteACP, ",")
 				for _, str := range writeACPList {
-					defaultGrantees = append(defaultGrantees, Grantee{
+					headerGrantees = append(headerGrantees, Grantee{
 						Access:     str,
 						Permission: PermissionWriteAcp,
 						Type:       types.TypeCanonicalUser,
@@ -340,6 +341,13 @@ func UpdateACL(input *PutBucketAclInput, acl ACL, iam IAMService) ([]byte, error
 			}
 
 			accs = append(append(append(append(fullControlList, readList...), writeACPList...), readACPList...), writeList...)
+
+			// Grant headers replace the whole ACL, as on AWS, so the owner
+			// gets no FULL_CONTROL grant unless a header names it. The owner
+			// keeps full access to its bucket anyway, see aclAllows.
+			if len(headerGrantees) > 0 {
+				defaultGrantees = headerGrantees
+			}
 		} else {
 			cache := make(map[string]bool)
 			for _, grt := range input.AccessControlPolicy.AccessControlList.Grants {

auth/acl_test.go

@@ -0,0 +1,117 @@
+// Copyright 2026 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 (
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/stretchr/testify/assert"
+)
+
+// existingAccountsIAM resolves every access key id it is asked about, so
+// UpdateACL accepts any grantee.
+type existingAccountsIAM struct {
+	IAMService
+}
+
+func (existingAccountsIAM) ResolveAccounts([]string) ([]string, error) {
+	return nil, nil
+}
+
+// TestUpdateACL checks which grants end up in the bucket ACL. Grant headers
+// replace the ACL, as on AWS, so the owner gets a FULL_CONTROL grant only
+// from a canned ACL, from the default ACL of a bucket created without
+// grants, or when a header names it.
+func TestUpdateACL(t *testing.T) {
+	empty := ""
+	readAcpGrantee := "user1"
+	ownerGrantee := "owner"
+
+	ownerFullControl := Grantee{
+		Permission: PermissionFullControl,
+		Access:     "owner",
+		Type:       types.TypeCanonicalUser,
+	}
+
+	tests := []struct {
+		name  string
+		input *PutBucketAclInput
+		want  []Grantee
+	}{
+		{
+			name: "grant header",
+			input: &PutBucketAclInput{
+				GrantReadACP: &readAcpGrantee,
+			},
+			want: []Grantee{
+				{
+					Permission: PermissionReadAcp,
+					Access:     "user1",
+					Type:       types.TypeCanonicalUser,
+				},
+			},
+		},
+		{
+			name: "grant header naming the owner",
+			input: &PutBucketAclInput{
+				GrantFullControl: &ownerGrantee,
+			},
+			want: []Grantee{ownerFullControl},
+		},
+		{
+			name: "create bucket without grants",
+			input: &PutBucketAclInput{
+				GrantFullControl:    &empty,
+				GrantRead:           &empty,
+				GrantReadACP:        &empty,
+				GrantWrite:          &empty,
+				GrantWriteACP:       &empty,
+				AccessControlPolicy: &AccessControlPolicy{},
+			},
+			want: []Grantee{ownerFullControl},
+		},
+		{
+			name: "canned acl",
+			input: &PutBucketAclInput{
+				ACL: types.BucketCannedACLPublicRead,
+			},
+			want: []Grantee{
+				ownerFullControl,
+				{
+					Permission: PermissionRead,
+					Access:     "all-users",
+					Type:       types.TypeGroup,
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			aclJSON, err := UpdateACL(tt.input, ACL{Owner: "owner"}, existingAccountsIAM{})
+			assert.NoError(t, err)
+
+			acl, err := ParseACL(aclJSON)
+			assert.NoError(t, err)
+			assert.Equal(t, "owner", acl.Owner)
+			assert.Equal(t, tt.want, acl.Grantees)
+
+			for _, action := range []Action{GetBucketAclAction, PutBucketAclAction, ListBucketAction, PutObjectAction, PutBucketPolicyAction} {
+				assert.True(t, aclAllows(acl, "owner", action, false), "owner denied %s", action)
+			}
+		})
+	}
+}

tests/integration/CreateBucket.go

@@ -212,13 +212,6 @@ func CreateBucket_non_default_acl(s *S3Conf) error {
 	}
 
 	grants := []types.Grant{
-		{
-			Grantee: &types.Grantee{
-				ID:   &s.awsID,
-				Type: types.TypeCanonicalUser,
-			},
-			Permission: types.PermissionFullControl,
-		},
 		{
 			Grantee: &types.Grantee{
 				ID:   &testuser1.access,

tests/integration/iam.go

@@ -328,38 +328,21 @@ func IAM_CreateBucket_success(s *S3Conf) error {
 			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))
+		if getString(acl.Owner.ID) != testUser1.access {
+			return fmt.Errorf("expected the bucket owner to be %v, instead got %v", testUser1.access, getString(acl.Owner.ID))
 		}
 
-		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
+		// the grant header replaces the acl, so the owner gets no FULL_CONTROL grant
+		if len(acl.Grants) != 1 {
+			return fmt.Errorf("expected the length of acl grants to be 1, instead got %v", len(acl.Grants))
 		}
 
-		if !ownerChecked {
-			return fmt.Errorf("missing the owner '%s' full control acl", testUser1.access)
+		grant := acl.Grants[0]
+		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 !granteeChecked {
-			return fmt.Errorf("missing the user %s in read grantees acl", testUser2.access)
+		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)
 		}
 
 		return teardown(s, bucket)

tests/test_rest_acl.sh

@@ -209,8 +209,8 @@ fi
 
 # 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"
+  if [ "$RECREATE_BUCKETS" == "false" ]; then
+    skip "skip bucket create tests for static buckets"
   fi
   run get_bucket_name "$BUCKET_ONE_NAME"
   assert_success