s3: allow PutBucketAcl when auth is off (#12180)

* s3: allow PutBucketAcl when auth is off

* s3: resolve the PutBucketAcl owner against the recorded owner

The unauthenticated path now resolves the account as anonymous, but XML
ACLs were still checked against it while GetBucketAcl reported an empty
owner for ownerless buckets, so a read-modify-write ACL update failed
with AccessDenied. GetBucketAcl and the XML check now both use the
recorded owner when the bucket has one and the resolved account
otherwise. The canned-ACL grantee moves to the same owner, and a request
carrying no account no longer erases a stored owner.

* s3: cover PutBucketAcl without auth for XML bodies and stored owners

The XML case posts the anonymous owner GetBucketAcl reports for an
ownerless bucket; the canned case asserts a recorded owner survives an
unauthenticated update and receives the full-control grant. Ambient AWS
credentials are cleared so the memory store never turns auth on.

* s3: keep the recorded owner through a PutBucketAcl update

The grantee and the stored owner were resolved differently: grants
named the recorded owner while config.Owner still took the caller's
account, so an admin updating another account's bucket rewrote the
owner and broke an XML retry naming the original one. An ownerless
bucket adopts the ACL's owner; an owned bucket keeps it.

* s3: fall back to the admin owner for an ownerless bucket in ACL handlers

An unauthenticated PutBucketAcl resolved the owner as anonymous and stored
it on the bucket, but the registry reports an ownerless bucket as
admin-owned: BucketOwnerEnforced uploads then stamped anonymous instead of
admin, and ownership checks after enabling auth compared against the wrong
owner. Both ACL handlers now resolve an anonymous fallback to the admin
account, so GetBucketAcl reports the same effective owner the registry
does and the stored owner matches it.

* s3: resolve the admin owner for bucket grants when the caller is anonymous or untrusted

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: skip recording anonymous identity or account as bucket owner

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: report stored bucket ACL grants verbatim

Explicitly saved anonymous full-control grants must survive a
GET-to-PUT round trip unchanged, so the read path no longer rewrites
them to the reported owner.

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: expect the stored anonymous grant verbatim in GetBucketAcl

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: user.email <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/s3api/s3api_acp.go

@@ -25,6 +25,17 @@ func (s3a *S3ApiServer) requestAccountId(r *http.Request) string {
 	return r.Header.Get(s3_constants.AmzAccountId)
 }
 
+// ownerAccountId resolves the account that owns grants and ownerless buckets:
+// the requester's account when one is trusted, else the admin account — with
+// auth off the account header is client-supplied and cannot claim ownership.
+func (s3a *S3ApiServer) ownerAccountId(r *http.Request) string {
+	accountId := getAccountId(r)
+	if accountId == AccountAnonymous.Id || (s3a.iam != nil && !s3a.iam.isEnabled()) {
+		return AccountAdmin.Id
+	}
+	return accountId
+}
+
 func (s3a *S3ApiServer) checkAccessByOwnership(r *http.Request, bucket string) s3err.ErrorCode {
 	metadata, errCode := s3a.bucketRegistry.GetBucketMetadata(bucket)
 	if errCode != s3err.ErrNone {

weed/s3api/s3api_bucket_handlers.go

@@ -257,7 +257,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 	requestHasACL := hasExplicitBucketACL(r)
 	var aclGrantsBytes []byte
 	if requestHasACL {
-		accountId := getAccountId(r)
+		accountId := s3a.ownerAccountId(r)
 		_, grants, errCode := ParseAndValidateAclHeaders(r, s3a.iam, "", accountId, accountId, false)
 		if errCode != s3err.ErrNone {
 			s3err.WriteErrorResponse(w, r, errCode)
@@ -340,7 +340,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		bucketCrtime = entry.Attributes.Crtime
 
 		// Set bucket owner
-		setBucketOwner(r)(entry)
+		s3a.setBucketOwner(r)(entry)
 
 		// Persist a requested non-default ACL so GetBucketAcl and idempotent
 		// recreation observe it (private is the default and is not stored).
@@ -720,11 +720,22 @@ func hasExplicitBucketACL(r *http.Request) bool {
 }
 
 // setBucketOwner creates a function that sets the bucket owner from the request context
-func setBucketOwner(r *http.Request) func(entry *filer_pb.Entry) {
+func (s3a *S3ApiServer) setBucketOwner(r *http.Request) func(entry *filer_pb.Entry) {
 	currentIdentityId := s3_constants.GetIdentityNameFromContext(r)
+	if identityObj := s3_constants.GetIdentityFromContext(r); identityObj != nil {
+		if identity, ok := identityObj.(*Identity); ok && s3a.iam != nil &&
+			s3a.iam.identityAnonymous != nil && identity.Name == s3a.iam.identityAnonymous.Name {
+			currentIdentityId = ""
+		}
+	}
 	// 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).
+	// Nothing is recorded when auth is off (client-supplied header, untrusted)
+	// or for the anonymous account: those buckets report the admin fallback.
 	accountId := r.Header.Get(s3_constants.AmzAccountId)
+	if accountId == AccountAnonymous.Id || (s3a.iam != nil && !s3a.iam.isEnabled()) {
+		accountId = ""
+	}
 	return func(entry *filer_pb.Entry) {
 		if currentIdentityId == "" && accountId == "" {
 			return
@@ -762,7 +773,7 @@ func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error {
 	var bucketCrtime int64
 	if err := s3a.mkdirExclusive(s3a.option.BucketsPath, bucket, func(entry *filer_pb.Entry) {
 		bucketCrtime = entry.Attributes.Crtime
-		setBucketOwner(r)(entry)
+		s3a.setBucketOwner(r)(entry)
 	}); err != nil {
 		// In case of a race condition where another request created the bucket
 		// in the meantime, check for existence before returning an error.
@@ -788,7 +799,7 @@ func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error {
 			// we'll set it here to ensure consistency.
 			if entry.Extended == nil || len(entry.Extended[s3_constants.AmzIdentityId]) == 0 {
 				// No owner set, assign current admin as owner
-				setBucketOwner(r)(entry)
+				s3a.setBucketOwner(r)(entry)
 				if updateErr := s3a.updateEntry(s3a.option.BucketsPath, entry); updateErr != nil {
 					glog.Warningf("autoCreateBucket: failed to set owner for existing bucket %s: %v", bucket, updateErr)
 				} else {
@@ -1017,9 +1028,10 @@ func (s3a *S3ApiServer) GetBucketAclHandler(w http.ResponseWriter, r *http.Reque
 	}
 
 	// 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)
+	// caller; an ownerless bucket reports the registry's default owner, the admin
+	// account, rather than the anonymous caller. Likewise return any stored ACL,
+	// defaulting to the owner's full-control grant.
+	ownerId := s3a.ownerAccountId(r)
 	var storedGrants []*s3.Grant
 	if bucketConfig, errCode := s3a.getBucketConfig(bucket); errCode == s3err.ErrNone {
 		storedGrants = parseAclGrants(bucketConfig.ACL)
@@ -1027,6 +1039,8 @@ func (s3a *S3ApiServer) GetBucketAclHandler(w http.ResponseWriter, r *http.Reque
 			ownerId = bucketConfig.Owner
 		}
 	}
+	// Stored grants are reported verbatim: an explicitly saved anonymous
+	// full-control grant must survive a GET-to-PUT round trip unchanged.
 	ownerDisplayName := s3a.iam.GetAccountNameById(ownerId)
 	response := AccessControlPolicy{
 		Owner: CanonicalUser{
@@ -1050,15 +1064,23 @@ func (s3a *S3ApiServer) PutBucketAclHandler(w http.ResponseWriter, r *http.Reque
 		return
 	}
 
-	// Get account information for ACL processing
-	amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
+	// Without auth the request carries no account; resolve it as PutBucketHandler does.
+	aclAccountId := s3a.ownerAccountId(r)
+
+	// XML ACLs must name the owner GetBucketAcl reports: the stored owner when
+	// the bucket has one, else the request's account — an anonymous request
+	// resolves to the admin account, the registry's default bucket owner.
+	ownerId := aclAccountId
+	if config, errCode := s3a.getBucketConfig(bucket); errCode == s3err.ErrNone && config.Owner != "" {
+		ownerId = config.Owner
+	}
 
 	// Get bucket ownership settings (these would be used for ownership validation in a full implementation)
-	bucketOwnership := ""         // Default/simplified for now - in a full implementation this would be retrieved from bucket config
-	bucketOwnerId := amzAccountId // Simplified - bucket owner is current account
+	bucketOwnership := ""    // Default/simplified for now - in a full implementation this would be retrieved from bucket config
+	bucketOwnerId := ownerId // Simplified - bucket owner is current account
 
 	// Use the existing ACL parsing logic to handle both canned ACLs and XML body
-	grants, errCode := ExtractAcl(r, s3a.iam, bucketOwnership, bucketOwnerId, amzAccountId, amzAccountId)
+	grants, errCode := ExtractAcl(r, s3a.iam, bucketOwnership, bucketOwnerId, ownerId, aclAccountId)
 	if errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
 		return
@@ -1084,7 +1106,11 @@ func (s3a *S3ApiServer) PutBucketAclHandler(w http.ResponseWriter, r *http.Reque
 			config.ACL = nil
 			config.IsPublicRead = false
 		}
-		config.Owner = amzAccountId
+		// An ACL update never takes over ownership: an ownerless bucket adopts
+		// the ACL's owner so the stored owner and full-control grantee agree.
+		if config.Owner == "" {
+			config.Owner = ownerId
+		}
 		return nil
 	})
 

weed/s3api/s3api_bucket_handlers_misc_test.go

@@ -10,8 +10,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/stretchr/testify/require"
@@ -363,3 +365,130 @@ func TestAuthWithPublicReadHonorsPolicyDeny(t *testing.T) {
 	s3a.AuthWithPublicRead(func(w http.ResponseWriter, r *http.Request) { called = true }, s3_constants.ACTION_READ)(rr, objectReq)
 	require.True(t, called, "anonymous object request on a public bucket must reach the handler's tag-aware recheck")
 }
+
+// With auth off the request has no account, so the canned ACL's owner grant
+// named an empty account and PutBucketAcl answered InvalidRequest.
+func TestPutBucketAclWithoutAuth(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "")
+	const bucket = "acl-bucket"
+	filer := &fakeLookupFiler{entry: &filer_pb.Entry{Name: bucket, IsDirectory: true}}
+	s3a := newHeadBucketTestServer(t, filer)
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set(bucket, &BucketConfig{Name: bucket})
+	s3a.iam = NewIdentityAccessManagementWithStore(s3a.option, nil, "memory")
+	t.Cleanup(s3a.iam.Shutdown)
+	require.False(t, s3a.iam.isEnabled())
+
+	req := httptest.NewRequest(http.MethodPut, "http://s3/"+bucket+"?acl", http.NoBody)
+	req = mux.SetURLVars(req, map[string]string{"bucket": bucket})
+	req.Header.Set(s3_constants.AmzCannedAcl, "public-read")
+	rr := httptest.NewRecorder()
+	s3a.iam.Auth(s3a.PutBucketAclHandler, s3_constants.ACTION_WRITE_ACP)(rr, req)
+	require.Equal(t, http.StatusOK, rr.Code, rr.Body.String())
+
+	require.Len(t, filer.txnSeen, 1)
+	var grants []*s3.Grant
+	require.NoError(t, json.Unmarshal(filer.txnSeen[0].Mutations[0].SetExtended[s3_constants.ExtAmzAclKey], &grants))
+	require.Len(t, grants, 2)
+	require.Equal(t, s3_constants.AccountAdminId, aws.StringValue(grants[0].Grantee.ID))
+	require.Equal(t, s3_constants.GranteeGroupAllUsers, aws.StringValue(grants[1].Grantee.URI))
+	require.Equal(t, s3_constants.AccountAdminId, string(filer.txnSeen[0].Mutations[0].SetExtended[s3_constants.ExtAmzOwnerKey]))
+}
+
+// GetBucketAcl reports the admin account for an ownerless bucket without
+// auth, so an XML ACL written back with that owner must validate.
+func TestPutBucketAclXmlWithoutAuth(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "")
+	const bucket = "acl-xml-bucket"
+	filer := &fakeLookupFiler{entry: &filer_pb.Entry{Name: bucket, IsDirectory: true}}
+	s3a := newHeadBucketTestServer(t, filer)
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set(bucket, &BucketConfig{Name: bucket})
+	s3a.iam = NewIdentityAccessManagementWithStore(s3a.option, nil, "memory")
+	t.Cleanup(s3a.iam.Shutdown)
+
+	body := `<AccessControlPolicy xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Owner><ID>admin</ID></Owner><AccessControlList>` +
+		`<Grant><Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="CanonicalUser"><ID>admin</ID></Grantee><Permission>FULL_CONTROL</Permission></Grant>` +
+		`</AccessControlList></AccessControlPolicy>`
+	req := httptest.NewRequest(http.MethodPut, "http://s3/"+bucket+"?acl", strings.NewReader(body))
+	req = mux.SetURLVars(req, map[string]string{"bucket": bucket})
+	rr := httptest.NewRecorder()
+	s3a.iam.Auth(s3a.PutBucketAclHandler, s3_constants.ACTION_WRITE_ACP)(rr, req)
+	require.Equal(t, http.StatusOK, rr.Code, rr.Body.String())
+}
+
+// An ownerless bucket created before the admin fallback stored an anonymous
+// full-control grantee; GetBucketAcl reports it verbatim so a GET-to-PUT
+// round trip keeps the stored grant. A client-supplied account header with
+// auth off cannot claim ownership either.
+func TestGetBucketAclReportsAnonymousGrantVerbatim(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "")
+	const bucket = "acl-legacy-bucket"
+	filer := &fakeLookupFiler{entry: &filer_pb.Entry{Name: bucket, IsDirectory: true}}
+	s3a := newHeadBucketTestServer(t, filer)
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.iam = NewIdentityAccessManagementWithStore(s3a.option, nil, "memory")
+	t.Cleanup(s3a.iam.Shutdown)
+	require.False(t, s3a.iam.isEnabled())
+
+	fullControl := s3_constants.PermissionFullControl
+	grants := []*s3.Grant{{
+		Grantee: &s3.Grantee{
+			Type: aws.String("CanonicalUser"),
+			ID:   aws.String(s3_constants.AccountAnonymousId),
+		},
+		Permission: &fullControl,
+	}}
+	raw, err := json.Marshal(grants)
+	require.NoError(t, err)
+	s3a.bucketConfigCache.Set(bucket, &BucketConfig{Name: bucket, ACL: raw})
+
+	req := httptest.NewRequest(http.MethodGet, "http://s3/"+bucket+"?acl", http.NoBody)
+	req = mux.SetURLVars(req, map[string]string{"bucket": bucket})
+	req.Header.Set(s3_constants.AmzAccountId, "spoofed-owner")
+	rr := httptest.NewRecorder()
+	s3a.iam.Auth(s3a.GetBucketAclHandler, s3_constants.ACTION_READ_ACP)(rr, req)
+	require.Equal(t, http.StatusOK, rr.Code, rr.Body.String())
+	require.Contains(t, rr.Body.String(), "<ID>"+s3_constants.AccountAdminId+"</ID>")
+	require.NotContains(t, rr.Body.String(), "spoofed-owner")
+	require.Contains(t, rr.Body.String(), s3_constants.AccountAnonymousId)
+}
+
+// A bucket with a recorded owner keeps it through an ACL update: the stored
+// owner and the full-control grantee stay the same account, whether the
+// request carries an account or none.
+func TestPutBucketAclKeepsOwner(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "")
+	const bucket = "acl-owned-bucket"
+	for _, account := range []string{"", "admin"} {
+		filer := &fakeLookupFiler{entry: &filer_pb.Entry{
+			Name:        bucket,
+			IsDirectory: true,
+			Extended:    map[string][]byte{s3_constants.ExtAmzOwnerKey: []byte("team-a")},
+		}}
+		s3a := newHeadBucketTestServer(t, filer)
+		s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+		s3a.iam = NewIdentityAccessManagementWithStore(s3a.option, nil, "memory")
+
+		req := httptest.NewRequest(http.MethodPut, "http://s3/"+bucket+"?acl", http.NoBody)
+		req = mux.SetURLVars(req, map[string]string{"bucket": bucket})
+		req.Header.Set(s3_constants.AmzCannedAcl, "private")
+		if account != "" {
+			req.Header.Set(s3_constants.AmzAccountId, account)
+		}
+		rr := httptest.NewRecorder()
+		s3a.iam.Auth(s3a.PutBucketAclHandler, s3_constants.ACTION_WRITE_ACP)(rr, req)
+		require.Equal(t, http.StatusOK, rr.Code, rr.Body.String())
+		s3a.iam.Shutdown()
+
+		require.Len(t, filer.txnSeen, 1)
+		require.NotContains(t, filer.txnSeen[0].Mutations[0].DeleteExtended, s3_constants.ExtAmzOwnerKey)
+		var grants []*s3.Grant
+		require.NoError(t, json.Unmarshal(filer.txnSeen[0].Mutations[0].SetExtended[s3_constants.ExtAmzAclKey], &grants))
+		require.Equal(t, "team-a", aws.StringValue(grants[0].Grantee.ID))
+	}
+}