s3: allow PutObjectAcl when auth is off (#12155)

* s3: allow PutObjectAcl when auth is off

* s3: ownerless objects claim the admin account when auth is off

An object written through the filer carries no owner metadata, so
objectOwner fell back to an empty account id and ExtractAcl rejected the
canned ACL's canonical grantee. Extend TestPutObjectAclWithoutAuth to
cover the ownerless case.

* s3: GetObjectAcl reports the admin owner for ownerless objects when auth is off

The same empty-owner fallback produced an ACL response with a blank
Owner ID and a blank canonical grantee.

* s3: ignore the client-supplied account id when auth is off

Auth does not strip AmzAccountId on the iam-disabled path, so the
ownerless fallback could persist a forged owner. requestAccountId
substitutes the admin account, matching the idiom already used in the
upload ACL path.

---------

Co-authored-by: user.email <chrislusf@users.noreply.github.com>

weed/s3api/s3api_acp.go

@@ -16,6 +16,15 @@ func getAccountId(r *http.Request) string {
 	}
 }
 
+// requestAccountId returns the authenticated caller's account; with auth off
+// the header is client-supplied, so the admin account stands in instead.
+func (s3a *S3ApiServer) requestAccountId(r *http.Request) string {
+	if s3a.iam != nil && !s3a.iam.isEnabled() {
+		return AccountAdmin.Id
+	}
+	return r.Header.Get(s3_constants.AmzAccountId)
+}
+
 func (s3a *S3ApiServer) checkAccessByOwnership(r *http.Request, bucket string) s3err.ErrorCode {
 	metadata, errCode := s3a.bucketRegistry.GetBucketMetadata(bucket)
 	if errCode != s3err.ErrNone {

weed/s3api/s3api_object_acl_permissions_test.go

@@ -184,3 +184,50 @@ func TestPutObjectAclPermissions(t *testing.T) {
 		})
 	}
 }
+
+// With auth off, objects are owned by the admin account but requests carry no
+// account, so the owner check denied PutObjectAcl to every caller.
+func TestPutObjectAclWithoutAuth(t *testing.T) {
+	const bucket, object = "acl-bucket", "image.png"
+	for _, tt := range []struct {
+		name      string
+		ownerless bool
+		forgedId  string
+	}{
+		{name: "stored owner"},
+		{name: "ownerless", ownerless: true},
+		{name: "forged account header", ownerless: true, forgedId: "forged-account"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			entry := &filer_pb.Entry{Name: object}
+			if !tt.ownerless {
+				entry.Extended = map[string][]byte{s3_constants.ExtAmzOwnerKey: []byte(AccountAdmin.Id)}
+			}
+			filer := &objectACLUpdateFiler{fakeLookupFiler: fakeLookupFiler{entry: entry}}
+			s3a := newHeadBucketTestServer(t, filer)
+			s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+			s3a.bucketConfigCache.Set(bucket, &BucketConfig{Name: bucket, Ownership: s3_constants.OwnershipObjectWriter})
+			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+"/"+object+"?acl", http.NoBody)
+			req = mux.SetURLVars(req, map[string]string{"bucket": bucket, "object": object})
+			req.Header.Set(s3_constants.AmzCannedAcl, "public-read")
+			if tt.forgedId != "" {
+				req.Header.Set(s3_constants.AmzAccountId, tt.forgedId)
+			}
+			rr := httptest.NewRecorder()
+			s3a.iam.Auth(s3a.PutObjectAclHandler, s3_constants.ACTION_WRITE_ACP)(rr, req)
+			require.Equal(t, http.StatusOK, rr.Code, rr.Body.String())
+
+			filer.mu.Lock()
+			defer filer.mu.Unlock()
+			require.NotNil(t, filer.update)
+			require.Equal(t, AccountAdmin.Id, string(filer.update.Entry.Extended[s3_constants.ExtAmzOwnerKey]))
+			grants := GetAcpGrants(filer.update.Entry.Extended)
+			require.Len(t, grants, 2)
+			require.Equal(t, s3_constants.GranteeGroupAllUsers, aws.StringValue(grants[1].Grantee.URI))
+		})
+	}
+}

weed/s3api/s3api_object_handlers_acl.go

@@ -105,7 +105,7 @@ func (s3a *S3ApiServer) GetObjectAclHandler(w http.ResponseWriter, r *http.Reque
 	// Get object owner from metadata, fallback to request account
 	var objectOwner string
 	var objectOwnerDisplayName string
-	amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
+	amzAccountId := s3a.requestAccountId(r)
 
 	if entry.Extended != nil {
 		if ownerBytes, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists {
@@ -224,7 +224,7 @@ func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Reque
 
 	// Get current object owner from metadata
 	var objectOwner string
-	amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
+	amzAccountId := s3a.requestAccountId(r)
 
 	if entry.Extended != nil {
 		if ownerBytes, exists := entry.Extended[s3_constants.ExtAmzOwnerKey]; exists {
@@ -234,7 +234,8 @@ func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Reque
 
 	// **PERMISSION CHECKS**
 
-	isAdmin := s3a.isUserAdmin(r)
+	// Without auth there is no caller to check, as in checkAccessByOwnership.
+	isAdmin := s3a.isUserAdmin(r) || (s3a.iam != nil && !s3a.iam.isEnabled())
 
 	// Without stored owner metadata there is no way to prove a non-admin
 	// caller owns this object, so they must not modify its ACL or claim it.

weed/s3api/s3api_object_upload_acl.go

@@ -151,10 +151,8 @@ func (s3a *S3ApiServer) preparePutObjectACL(r *http.Request, bucket string) (*ht
 	}
 	aclRequest.Header.Set(s3_constants.AmzCannedAcl, canned)
 	explicit := canned != "" || custom
-	accountID := r.Header.Get(s3_constants.AmzAccountId)
-	if !s3a.iam.isEnabled() {
-		accountID = AccountAdmin.Id
-	} else if explicit {
+	accountID := s3a.requestAccountId(r)
+	if explicit && s3a.iam.isEnabled() {
 		// Setting an ACL during upload or copy also requires s3:PutObjectAcl. Use the
 		// unified authorization path so bucket-policy allows and explicit denies
 		// retain the same semantics as standalone ACL requests.