s3: honor a bucket owner recorded as an identity (#10567)

* s3: resolve a bucket owner recorded as an identity

The admin UI and weed shell record a bucket's owner as an identity name in
s3-identity-id and never write the account id the S3 API stores alongside it,
so such a bucket looked unowned: its ACL owner fell back to the default admin
account, and under the default BucketOwnerEnforced ownership every object
uploaded to it was stamped with that account instead of the bucket owner.

Resolve the identity to its account when no account id is recorded, in the one
place both the bucket metadata and the bucket config derive the owner from.

* s3: drop the recorded account when the bucket owner is reassigned

Changing the owner of a bucket created through the S3 API left its old account
id behind, and that outranks the identity when the owner is resolved, so the
new owner never took effect for object ownership or the bucket ACL.

weed/admin/dash/bucket_management.go

@@ -310,7 +310,10 @@ func (s *AdminServer) SetBucketOwner(bucketName string, owner string) error {
 			bucketEntry.Extended = make(map[string][]byte)
 		}
 
-		// Set or remove the owner
+		// Set or remove the owner. The account id recorded by the S3 API goes with
+		// it: the S3 API derives it from the owning identity when it is absent, so
+		// leaving the old one behind would keep the previous owner of the objects.
+		delete(bucketEntry.Extended, s3_constants.ExtAmzOwnerKey)
 		if owner == "" {
 			delete(bucketEntry.Extended, s3_constants.AmzIdentityId)
 		} else {

weed/s3api/auth_credentials.go

@@ -1443,6 +1443,16 @@ func (iam *IdentityAccessManagement) GetAccountNameById(canonicalId string) stri
 	return ""
 }
 
+// GetAccountIdByIdentityName resolves an identity name to the account its
+// resources are owned under. Bucket owners recorded outside the S3 API (the
+// admin UI, weed shell) name an identity, not an account.
+func (iam *IdentityAccessManagement) GetAccountIdByIdentityName(name string) string {
+	if identity := iam.lookupByIdentityName(name); identity != nil && identity.Account != nil {
+		return identity.Account.Id
+	}
+	return ""
+}
+
 func (iam *IdentityAccessManagement) GetAccountIdByEmail(email string) string {
 	iam.m.RLock()
 	defer iam.m.RUnlock()

weed/s3api/bucket_metadata.go

@@ -77,6 +77,22 @@ func (r *BucketRegistry) LoadBucketMetadata(entry *filer_pb.Entry) {
 	r.unMarkNotFound(entry.Name)
 }
 
+// bucketOwnerAccountId returns the account id owning the bucket entry. A bucket
+// created outside the S3 API (the admin UI, weed shell) records only its owning
+// identity, so that identity is resolved to its account. Without this such a
+// bucket looks unowned: it reports the default admin account as its ACL owner,
+// and under BucketOwnerEnforced every object written to it is stamped with that
+// account instead of the bucket owner.
+func bucketOwnerAccountId(accountManager AccountManager, entry *filer_pb.Entry) string {
+	if ownerAccountId := string(entry.Extended[s3_constants.ExtAmzOwnerKey]); ownerAccountId != "" {
+		return ownerAccountId
+	}
+	if identityName := string(entry.Extended[s3_constants.AmzIdentityId]); identityName != "" {
+		return accountManager.GetAccountIdByIdentityName(identityName)
+	}
+	return ""
+}
+
 func buildBucketMetadata(accountManager AccountManager, entry *filer_pb.Entry) *BucketMetaData {
 	entryJson, _ := json.Marshal(entry)
 	glog.V(3).Infof("build bucket metadata,entry=%s", entryJson)
@@ -108,9 +124,8 @@ func buildBucketMetadata(accountManager AccountManager, entry *filer_pb.Entry) *
 
 		//access control policy
 		//owner
-		acpOwnerBytes, ok := entry.Extended[s3_constants.ExtAmzOwnerKey]
-		if ok && len(acpOwnerBytes) > 0 {
-			ownerAccountId := string(acpOwnerBytes)
+		ownerAccountId := bucketOwnerAccountId(accountManager, entry)
+		if ownerAccountId != "" {
 			ownerAccountName := accountManager.GetAccountNameById(ownerAccountId)
 			if ownerAccountName == "" {
 				glog.Warningf("owner[id=%s] is invalid, bucket: %s", ownerAccountId, bucketMetadata.Name)

weed/s3api/bucket_owner_identity_test.go

@@ -0,0 +1,86 @@
+package s3api
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// ownerIdentityIAM loads an account-less identity and one scoped to an explicit
+// account, the two shapes a bucket owner name can resolve through.
+func ownerIdentityIAM(t *testing.T) *IdentityAccessManagement {
+	t.Helper()
+	iam := &IdentityAccessManagement{}
+	require.NoError(t, iam.loadS3ApiConfiguration(&iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{Name: "test", Actions: []string{"Read", "Write"}},
+			{
+				Name:    "scoped",
+				Actions: []string{"Read", "Write"},
+				Account: &iam_pb.Account{Id: "100000000001", DisplayName: "Scoped"},
+			},
+		},
+	}))
+	return iam
+}
+
+// A bucket created outside the S3 API — the admin UI or weed shell, which record
+// the owner as an identity and never write the account id the S3 API stores — is
+// owned by that identity, not by the default admin account. Getting this wrong
+// stamps every object uploaded to the bucket with the admin account, since the
+// default BucketOwnerEnforced ownership hands objects to the bucket owner.
+func TestBucketOwnerFromIdentityId(t *testing.T) {
+	iam := ownerIdentityIAM(t)
+
+	for _, tc := range []struct {
+		name          string
+		identityId    string
+		expectOwnerId string
+	}{
+		{"account-less identity", "test", "test"},
+		{"identity scoped to an account", "scoped", "100000000001"},
+		{"unknown identity", "ghost", AccountAdmin.Id},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			entry := &filer_pb.Entry{
+				Name:     "bucket",
+				Extended: map[string][]byte{s3_constants.AmzIdentityId: []byte(tc.identityId)},
+			}
+
+			metadata := buildBucketMetadata(iam, entry)
+			require.NotNil(t, metadata.Owner)
+			assert.Equal(t, tc.expectOwnerId, *metadata.Owner.ID)
+
+			s3a := &S3ApiServer{iam: iam}
+			config := s3a.newBucketConfigFromEntry("bucket", entry)
+			assert.Equal(t, tc.identityId, config.IdentityId)
+			if tc.expectOwnerId != AccountAdmin.Id {
+				assert.Equal(t, tc.expectOwnerId, config.Owner,
+					"the bucket config owner must match the metadata owner")
+			} else {
+				assert.Empty(t, config.Owner, "an unresolvable owner leaves the config unowned")
+			}
+		})
+	}
+}
+
+// The account id recorded by the S3 API stays authoritative when both are present.
+func TestBucketOwnerPrefersAccountId(t *testing.T) {
+	iam := ownerIdentityIAM(t)
+
+	entry := &filer_pb.Entry{
+		Name: "bucket",
+		Extended: map[string][]byte{
+			s3_constants.AmzIdentityId:  []byte("scoped"),
+			s3_constants.ExtAmzOwnerKey: []byte("test"),
+		},
+	}
+
+	metadata := buildBucketMetadata(iam, entry)
+	require.NotNil(t, metadata.Owner)
+	assert.Equal(t, "test", *metadata.Owner.ID)
+}

weed/s3api/s3api_acl_helper.go

@@ -18,6 +18,7 @@ import (
 type AccountManager interface {
 	GetAccountNameById(canonicalId string) string
 	GetAccountIdByEmail(email string) string
+	GetAccountIdByIdentityName(name string) string
 }
 
 // ExtractAcl extracts the acl from the request body, or from the header if request body is empty

weed/s3api/s3api_bucket_config.go

@@ -420,9 +420,7 @@ func (s3a *S3ApiServer) newBucketConfigFromEntry(bucket string, entry *filer_pb.
 			// Parse ACL once and cache public-read status.
 			config.IsPublicRead = parseAndCachePublicReadStatus(acl)
 		}
-		if owner, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists {
-			config.Owner = string(owner)
-		}
+		config.Owner = bucketOwnerAccountId(s3a.iam, entry)
 		if identityId, exists := entry.Extended[s3_constants.AmzIdentityId]; exists {
 			config.IdentityId = string(identityId)
 		}

weed/shell/command_s3_bucket_owner.go

@@ -96,6 +96,9 @@ func (c *commandS3BucketOwner) Do(args []string, commandEnv *CommandEnv, writer
 				entry.Extended = make(map[string][]byte)
 			}
 			entry.Extended[s3_constants.AmzIdentityId] = []byte(owner)
+			// The S3 API derives the owning account from this identity when no
+			// account id is recorded; a leftover one would keep the old owner.
+			delete(entry.Extended, s3_constants.ExtAmzOwnerKey)
 			fmt.Fprintf(writer, "Setting owner of bucket %s to: %s\n", *bucketName, owner)
 
 			// Update the entry
@@ -114,6 +117,7 @@ func (c *commandS3BucketOwner) Do(args []string, commandEnv *CommandEnv, writer
 		if *deleteOwner {
 			if entry.Extended != nil {
 				delete(entry.Extended, s3_constants.AmzIdentityId)
+				delete(entry.Extended, s3_constants.ExtAmzOwnerKey)
 			}
 			fmt.Fprintf(writer, "Removing owner from bucket %s\n", *bucketName)