s3: return BucketAlreadyOwnedByYou when recreating your own bucket (#9822)

* s3: return BucketAlreadyOwnedByYou when recreating your own bucket PutBucket returned BucketAlreadyExists for every existing bucket, even when the caller already owns it, so idempotent re-creation (e.g. a container that creates its bucket on startup) couldn't tell "someone else took the name" from "it's already mine". Recreating a bucket you own now returns BucketAlreadyOwnedByYou, unless the request conflicts with the existing bucket: a different Object Lock setting, or an ACL on the request or the existing bucket. To detect the latter, a requested non-default canned/grant ACL is now persisted on creation instead of being dropped. * s3: fail PutBucket when the existing bucket's config can't be read When a bucket already exists, an unreadable config left the recreate defaulting to BucketAlreadyOwnedByYou, masking the backend error and possibly accepting a conflicting recreate (Object Lock / ACL unknown). Surface the read error instead. * s3: return the stored bucket ACL from GetBucketAcl GetBucketAcl always returned the owner's default full-control grant and ignored any stored ACL, so a bucket created with a canned ACL or one set via PutBucketAcl never read back correctly. Decode the stored grants instead, sharing one grants-to-XML helper with the object ACL handler. The shared helper also emits each grantee's real xsi:type (e.g. Group for public-read) instead of a hardcoded CanonicalUser, so group grants read back correctly for both bucket and object ACLs. * s3: resolve the right already-exists error on the concurrent-create race When two requests create the same bucket at once, the loser's mkdir fails and the handler fell back to a flat BucketAlreadyExists, bypassing the same-owner idempotency check. Route both the pre-check and the race fallback through one existingBucketError helper so a same-owner recreate still gets BucketAlreadyOwnedByYou. * s3: record the bucket owner's account id at creation setBucketOwner only stored the creating identity name, so the canonical account id wasn't available later. Persist it under ExtAmzOwnerKey too, the same field PutBucketAcl writes, so the bucket owner can be reported independently of whoever reads it. * s3: report the bucket owner from GetBucketAcl, not the caller GetBucketAcl built the ACL Owner from the caller's account header, so an admin or cross-account read returned the wrong owner. Use the owner persisted on the bucket, falling back to the caller only when none is recorded.
2026-09-08 15:41:15 +02:00 · 2026-06-04 15:33:03 -07:00
parent a24f4844d3
commit 8d59069a0a
4 changed files with 200 additions and 107 deletions
@@ -308,6 +308,45 @@ func ValidateAndTransferGrants(accountManager AccountManager, grants []*s3.Grant
 	return result, s3err.ErrNone
 }

+// buildAccessControlList converts stored ACP grants into the XML response form.
+// When no grants are stored it falls back to a single full-control grant for the
+// owner, matching AWS's default private ACL.
+func buildAccessControlList(accountManager AccountManager, grants []*s3.Grant, ownerId, ownerDisplayName string) AccessControlList {
+	if len(grants) == 0 {
+		return AccessControlList{Grant: []Grant{{
+			Grantee: Grantee{
+				ID:          ownerId,
+				DisplayName: ownerDisplayName,
+				Type:        "CanonicalUser",
+				XMLXSI:      "CanonicalUser",
+				XMLNS:       "http://www.w3.org/2001/XMLSchema-instance",
+			},
+			Permission: Permission(s3_constants.PermissionFullControl),
+		}}}
+	}
+
+	var acl AccessControlList
+	for _, grant := range grants {
+		localGrant := Grant{Permission: Permission(*grant.Permission)}
+		if grant.Grantee != nil {
+			localGrant.Grantee = Grantee{
+				Type:   *grant.Grantee.Type,
+				XMLXSI: *grant.Grantee.Type,
+				XMLNS:  "http://www.w3.org/2001/XMLSchema-instance",
+			}
+			if grant.Grantee.ID != nil {
+				localGrant.Grantee.ID = *grant.Grantee.ID
+				localGrant.Grantee.DisplayName = accountManager.GetAccountNameById(*grant.Grantee.ID)
+			}
+			if grant.Grantee.URI != nil {
+				localGrant.Grantee.URI = *grant.Grantee.URI
+			}
+		}
+		acl.Grant = append(acl.Grant, localGrant)
+	}
+	return acl
+}
+
 // GetAcpGrants return grants parsed from entry
 func GetAcpGrants(entryExtended map[string][]byte) []*s3.Grant {
 	acpBytes, ok := entryExtended[s3_constants.ExtAmzAclKey]
@@ -165,6 +165,29 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 	// Get authenticated identity from context (secure, cannot be spoofed)
 	currentIdentityId := s3_constants.GetIdentityNameFromContext(r)

+	// Parse any requested bucket ACL (canned ACL or grant headers) up front so it
+	// can be validated, persisted on creation, and factored into the already-exists
+	// response. A "private" canned ACL is the default and counts as no explicit ACL.
+	requestHasACL := hasExplicitBucketACL(r)
+	var aclGrantsBytes []byte
+	if requestHasACL {
+		accountId := getAccountId(r)
+		_, grants, errCode := ParseAndValidateAclHeaders(r, s3a.iam, "", accountId, accountId, false)
+		if errCode != s3err.ErrNone {
+			s3err.WriteErrorResponse(w, r, errCode)
+			return
+		}
+		if len(grants) > 0 {
+			grantsBytes, err := json.Marshal(grants)
+			if err != nil {
+				glog.Errorf("PutBucketHandler: marshal ACL grants for %s: %v", bucket, err)
+				s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+				return
+			}
+			aclGrantsBytes = grantsBytes
+		}
+	}
+
 	// Check collection existence first
 	collectionExists := false
 	if s3a.isTableBucket(bucket) {
@@ -192,54 +215,11 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}

-	// Check bucket directory existence and get metadata
+	// Bucket already exists: report whether the caller already owns it or the
+	// name is taken / the request conflicts.
 	if exist, err := s3a.exists(s3a.option.BucketsPath, bucket, true); err == nil && exist {
-		// Bucket exists, check ownership and settings
-		if entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket); err == nil {
-			// Get existing bucket owner
-			var existingOwnerId string
-			if entry.Extended != nil {
-				if id, ok := entry.Extended[s3_constants.AmzIdentityId]; ok {
-					existingOwnerId = string(id)
-				}
-			}
-
-			// Check ownership
-			if existingOwnerId != "" && existingOwnerId != currentIdentityId {
-				// Different owner - always fail with BucketAlreadyExists
-				glog.V(3).Infof("PutBucketHandler: bucket %s owned by %s, requested by %s", bucket, existingOwnerId, currentIdentityId)
-				s3err.WriteErrorResponse(w, r, s3err.ErrBucketAlreadyExists)
-				return
-			}
-
-			// Same owner or no owner set - check for conflicting settings
-			objectLockRequested := strings.EqualFold(r.Header.Get(s3_constants.AmzBucketObjectLockEnabled), "true")
-
-			// Get current bucket configuration
-			bucketConfig, errCode := s3a.getBucketConfig(bucket)
-			if errCode != s3err.ErrNone {
-				glog.Errorf("PutBucketHandler: failed to get bucket config for %s: %v", bucket, errCode)
-				// If we can't get config, assume no conflict and allow recreation
-			} else {
-				// Check for Object Lock conflict
-				currentObjectLockEnabled := bucketConfig.ObjectLockConfig != nil &&
-					bucketConfig.ObjectLockConfig.ObjectLockEnabled == s3_constants.ObjectLockEnabled
-
-				if objectLockRequested != currentObjectLockEnabled {
-					// Conflicting Object Lock settings - fail with BucketAlreadyExists
-					glog.V(3).Infof("PutBucketHandler: bucket %s has conflicting Object Lock settings (requested: %v, current: %v)",
-						bucket, objectLockRequested, currentObjectLockEnabled)
-					s3err.WriteErrorResponse(w, r, s3err.ErrBucketAlreadyExists)
-					return
-				}
-			}
-
-			// Bucket already exists - always return BucketAlreadyExists per S3 specification
-			// The S3 tests expect BucketAlreadyExists in all cases, not BucketAlreadyOwnedByYou
-			glog.V(3).Infof("PutBucketHandler: bucket %s already exists", bucket)
-			s3err.WriteErrorResponse(w, r, s3err.ErrBucketAlreadyExists)
-			return
-		}
+		s3err.WriteErrorResponse(w, r, s3a.existingBucketError(r, bucket, currentIdentityId, requestHasACL))
+		return
 	}

 	// If collection exists but bucket directory doesn't, this is an inconsistent state
@@ -265,6 +245,15 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		// Set bucket owner
 		setBucketOwner(r)(entry)

+		// Persist a requested non-default ACL so GetBucketAcl and idempotent
+		// recreation observe it (private is the default and is not stored).
+		if len(aclGrantsBytes) > 0 {
+			if entry.Extended == nil {
+				entry.Extended = make(map[string][]byte)
+			}
+			entry.Extended[s3_constants.ExtAmzAclKey] = aclGrantsBytes
+		}
+
 		// Set Object Lock configuration atomically during bucket creation
 		if objectLockEnabled {
 			glog.V(3).Infof("PutBucketHandler: enabling Object Lock and Versioning for bucket %s atomically", bucket)
@@ -290,10 +279,10 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		}
 	}); err != nil {
 		// If mkdir failed because another request created the bucket concurrently,
-		// return BucketAlreadyExists instead of InternalError.
+		// return the appropriate already-exists error instead of InternalError.
 		if exist, checkErr := s3a.exists(s3a.option.BucketsPath, bucket, true); checkErr == nil && exist {
 			glog.V(3).Infof("PutBucketHandler: bucket %s was created concurrently", bucket)
-			s3err.WriteErrorResponse(w, r, s3err.ErrBucketAlreadyExists)
+			s3err.WriteErrorResponse(w, r, s3a.existingBucketError(r, bucket, currentIdentityId, requestHasACL))
 			return
 		}
 		glog.Errorf("PutBucketHandler mkdir: %v", err)
@@ -497,16 +486,93 @@ var ErrAutoCreatePermissionDenied = errors.New("permission denied - requires Adm
 // ErrInvalidBucketName is returned when a bucket name doesn't meet S3 naming requirements
 var ErrInvalidBucketName = errors.New("invalid bucket name")

+// existingBucketError returns the error for a PutBucket whose target bucket
+// already exists: BucketAlreadyOwnedByYou for an idempotent recreate by the
+// owner, or BucketAlreadyExists when the name is owned by someone else or the
+// request conflicts with the existing bucket (a different Object Lock setting,
+// or an ACL on the request or the existing bucket).
+func (s3a *S3ApiServer) existingBucketError(r *http.Request, bucket, currentIdentityId string, requestHasACL bool) s3err.ErrorCode {
+	entry, err := s3a.getEntry(s3a.option.BucketsPath, bucket)
+	if err != nil {
+		// We just observed the bucket exists but can't read it; report it as taken.
+		glog.Errorf("PutBucketHandler: failed to read existing bucket %s: %v", bucket, err)
+		return s3err.ErrBucketAlreadyExists
+	}
+
+	var existingOwnerId string
+	if entry.Extended != nil {
+		if id, ok := entry.Extended[s3_constants.AmzIdentityId]; ok {
+			existingOwnerId = string(id)
+		}
+	}
+
+	// Different owner: the name is taken in the shared namespace.
+	if existingOwnerId != "" && existingOwnerId != currentIdentityId {
+		glog.V(3).Infof("PutBucketHandler: bucket %s owned by %s, requested by %s", bucket, existingOwnerId, currentIdentityId)
+		return s3err.ErrBucketAlreadyExists
+	}
+
+	// Same owner (or an unowned bucket the caller can claim). Recreating your own
+	// bucket is idempotent and returns BucketAlreadyOwnedByYou, unless the request
+	// conflicts with the existing bucket: a different Object Lock setting, or an ACL
+	// on the request or the existing bucket.
+	// (s3-tests: test_bucket_create_exists vs test_bucket_recreate_*_acl.)
+	if requestHasACL {
+		return s3err.ErrBucketAlreadyExists
+	}
+
+	objectLockRequested := strings.EqualFold(r.Header.Get(s3_constants.AmzBucketObjectLockEnabled), "true")
+	bucketConfig, errCode := s3a.getBucketConfig(bucket)
+	if errCode != s3err.ErrNone {
+		// Can't read the existing bucket's settings, so we can't tell whether this
+		// recreate conflicts; surface the failure instead of assuming idempotency.
+		glog.Errorf("PutBucketHandler: failed to get bucket config for %s: %v", bucket, errCode)
+		return errCode
+	}
+	currentObjectLockEnabled := bucketConfig.ObjectLockConfig != nil &&
+		bucketConfig.ObjectLockConfig.ObjectLockEnabled == s3_constants.ObjectLockEnabled
+	if objectLockRequested != currentObjectLockEnabled || len(bucketConfig.ACL) > 0 {
+		glog.V(3).Infof("PutBucketHandler: bucket %s already exists", bucket)
+		return s3err.ErrBucketAlreadyExists
+	}
+
+	glog.V(3).Infof("PutBucketHandler: bucket %s already owned by requester", bucket)
+	return s3err.ErrBucketAlreadyOwnedByYou
+}
+
+// hasExplicitBucketACL reports whether the request carries an explicit, non-default
+// bucket ACL via a canned ACL header (other than "private") or grant headers.
+func hasExplicitBucketACL(r *http.Request) bool {
+	if canned := r.Header.Get(s3_constants.AmzCannedAcl); canned != "" && !strings.EqualFold(canned, s3_constants.CannedAclPrivate) {
+		return true
+	}
+	for _, h := range []string{s3_constants.AmzAclFullControl, s3_constants.AmzAclRead, s3_constants.AmzAclReadAcp, s3_constants.AmzAclWrite, s3_constants.AmzAclWriteAcp} {
+		if r.Header.Get(h) != "" {
+			return true
+		}
+	}
+	return false
+}
+
 // setBucketOwner creates a function that sets the bucket owner from the request context
 func setBucketOwner(r *http.Request) func(entry *filer_pb.Entry) {
 	currentIdentityId := s3_constants.GetIdentityNameFromContext(r)
+	// Record the canonical account id too so GetBucketAcl can report the bucket
+	// owner instead of whoever is reading (e.g. an admin or another account).
+	accountId := r.Header.Get(s3_constants.AmzAccountId)
 	return func(entry *filer_pb.Entry) {
+		if currentIdentityId == "" && accountId == "" {
+			return
+		}
+		if entry.Extended == nil {
+			entry.Extended = make(map[string][]byte)
+		}
 		if currentIdentityId != "" {
-			if entry.Extended == nil {
-				entry.Extended = make(map[string][]byte)
-			}
 			entry.Extended[s3_constants.AmzIdentityId] = []byte(currentIdentityId)
 		}
+		if accountId != "" {
+			entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(accountId)
+		}
 	}
 }

@@ -722,23 +788,25 @@ func (s3a *S3ApiServer) GetBucketAclHandler(w http.ResponseWriter, r *http.Reque
 		return
 	}

-	amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
-	amzDisplayName := s3a.iam.GetAccountNameById(amzAccountId)
+	// Report the bucket's owner (recorded at creation or by PutBucketAcl), not the
+	// caller; fall back to the caller only when no owner was persisted. Likewise
+	// return any stored ACL, defaulting to the owner's full-control grant.
+	ownerId := r.Header.Get(s3_constants.AmzAccountId)
+	var storedGrants []*s3.Grant
+	if bucketConfig, errCode := s3a.getBucketConfig(bucket); errCode == s3err.ErrNone && bucketConfig.Entry != nil {
+		storedGrants = GetAcpGrants(bucketConfig.Entry.Extended)
+		if bucketConfig.Owner != "" {
+			ownerId = bucketConfig.Owner
+		}
+	}
+	ownerDisplayName := s3a.iam.GetAccountNameById(ownerId)
 	response := AccessControlPolicy{
 		Owner: CanonicalUser{
-			ID:          amzAccountId,
-			DisplayName: amzDisplayName,
+			ID:          ownerId,
+			DisplayName: ownerDisplayName,
 		},
+		AccessControlList: buildAccessControlList(s3a.iam, storedGrants, ownerId, ownerDisplayName),
 	}
-	response.AccessControlList.Grant = append(response.AccessControlList.Grant, Grant{
-		Grantee: Grantee{
-			ID:          amzAccountId,
-			DisplayName: amzDisplayName,
-			Type:        "CanonicalUser",
-			XMLXSI:      "CanonicalUser",
-			XMLNS:       "http://www.w3.org/2001/XMLSchema-instance"},
-		Permission: s3.PermissionFullControl,
-	})
 	writeSuccessResponseXML(w, r, response)
 }

@@ -35,6 +35,32 @@ func newBucketRequest(method, bucket, query, body string) *http.Request {
 	return req
 }

+func TestHasExplicitBucketACL(t *testing.T) {
+	cases := []struct {
+		name    string
+		headers map[string]string
+		want    bool
+	}{
+		{name: "none", headers: nil, want: false},
+		{name: "private is default", headers: map[string]string{s3_constants.AmzCannedAcl: "private"}, want: false},
+		{name: "canned public-read", headers: map[string]string{s3_constants.AmzCannedAcl: "public-read"}, want: true},
+		{name: "canned case-insensitive private", headers: map[string]string{s3_constants.AmzCannedAcl: "PRIVATE"}, want: false},
+		{name: "grant read", headers: map[string]string{s3_constants.AmzAclRead: `id="x"`}, want: true},
+		{name: "grant full control", headers: map[string]string{s3_constants.AmzAclFullControl: `id="x"`}, want: true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			req := newBucketRequest(http.MethodPut, "b", "", "")
+			for k, v := range tc.headers {
+				req.Header.Set(k, v)
+			}
+			if got := hasExplicitBucketACL(req); got != tc.want {
+				t.Fatalf("hasExplicitBucketACL = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
 func TestGetBucketPolicyStatusIsPublic(t *testing.T) {
 	cases := []struct {
 		name string
@@ -107,53 +107,13 @@ func (s3a *S3ApiServer) GetObjectAclHandler(w http.ResponseWriter, r *http.Reque

 	objectOwnerDisplayName = s3a.iam.GetAccountNameById(objectOwner)

-	// Build ACL response
+	// Build ACL response from stored ACL metadata (or the owner's default grant).
 	response := AccessControlPolicy{
 		Owner: CanonicalUser{
 			ID:          objectOwner,
 			DisplayName: objectOwnerDisplayName,
 		},
-	}
-
-	// Get grants from stored ACL metadata
-	grants := GetAcpGrants(entry.Extended)
-	if len(grants) > 0 {
-		// Convert AWS SDK grants to local Grant format
-		for _, grant := range grants {
-			localGrant := Grant{
-				Permission: Permission(*grant.Permission),
-			}
-
-			if grant.Grantee != nil {
-				localGrant.Grantee = Grantee{
-					Type:   *grant.Grantee.Type,
-					XMLXSI: "CanonicalUser",
-					XMLNS:  "http://www.w3.org/2001/XMLSchema-instance",
-				}
-
-				if grant.Grantee.ID != nil {
-					localGrant.Grantee.ID = *grant.Grantee.ID
-					localGrant.Grantee.DisplayName = s3a.iam.GetAccountNameById(*grant.Grantee.ID)
-				}
-
-				if grant.Grantee.URI != nil {
-					localGrant.Grantee.URI = *grant.Grantee.URI
-				}
-			}
-
-			response.AccessControlList.Grant = append(response.AccessControlList.Grant, localGrant)
-		}
-	} else {
-		// Fallback to default full control for object owner
-		response.AccessControlList.Grant = append(response.AccessControlList.Grant, Grant{
-			Grantee: Grantee{
-				ID:          objectOwner,
-				DisplayName: objectOwnerDisplayName,
-				Type:        "CanonicalUser",
-				XMLXSI:      "CanonicalUser",
-				XMLNS:       "http://www.w3.org/2001/XMLSchema-instance"},
-			Permission: Permission(s3_constants.PermissionFullControl),
-		})
+		AccessControlList: buildAccessControlList(s3a.iam, GetAcpGrants(entry.Extended), objectOwner, objectOwnerDisplayName),
 	}

 	writeSuccessResponseXML(w, r, response)