s3api: fix PutObjectAcl permission scoping and owner grants (#11587)

PutObjectAcl had four authorization and ownership bugs:

- The handler embedded the resource path into the action
  (WriteAcp:bucket/object), and authRequest/CanDo then scoped it to
  the request's bucket/object again. A bucket-wide WriteAcp:bucket
  grant could never match, so legitimate owners got 403.
- After authRequest succeeded via an IAM or bucket policy, a leftover
  identity.CanDo gate re-checked only the legacy Actions list, denying
  identities authorized purely by policies.
- For canned and default ACLs, ExtractAcl generated the FULL_CONTROL
  grant for the requesting account instead of the object owner. An
  admin setting private/public-read on another account's object left
  the owner metadata intact but reassigned full control to the admin.
- Objects without stored owner metadata (e.g. written via the filer
  outside S3) fell back to treating the requester as the owner, so any
  user with a WriteAcp grant could take them over. Non-admins are now
  denied; admins keep the takeover fallback.

Grantee validation now also accepts the object's stored owner even when
that account has been removed from the registry, so canned/XML ACLs for
retired owners keep working.

weed/s3api/s3api_acl_helper.go

@@ -21,8 +21,25 @@ type AccountManager interface {
 	GetAccountIdByIdentityName(name string) string
 }
 
+// aclOwnerAccountManager recognizes the stored resource owner as a valid
+// grantee even if that account is no longer registered, while every other
+// unknown account id is still rejected by the wrapped registry lookup.
+type aclOwnerAccountManager struct {
+	AccountManager
+	ownerId string
+}
+
+func (m aclOwnerAccountManager) GetAccountNameById(canonicalId string) string {
+	name := m.AccountManager.GetAccountNameById(canonicalId)
+	if name == "" && canonicalId != "" && canonicalId == m.ownerId {
+		return canonicalId
+	}
+	return name
+}
+
 // ExtractAcl extracts the acl from the request body, or from the header if request body is empty
 func ExtractAcl(r *http.Request, accountManager AccountManager, ownership, bucketOwnerId, ownerId, accountId string) (grants []*s3.Grant, errCode s3err.ErrorCode) {
+	accountManager = aclOwnerAccountManager{AccountManager: accountManager, ownerId: ownerId}
 	if r.Body != nil && r.Body != http.NoBody {
 		defer util_http.CloseRequest(r)
 
@@ -40,7 +57,10 @@ func ExtractAcl(r *http.Request, accountManager AccountManager, ownership, bucke
 
 		return ValidateAndTransferGrants(accountManager, acp.Grants)
 	} else {
-		_, grants, errCode = ParseAndValidateAclHeadersOrElseDefault(r, accountManager, ownership, bucketOwnerId, accountId, true)
+		// Canned and default ACLs grant FULL_CONTROL to the resource owner,
+		// not the requesting account: an admin updating another account's
+		// object must not take over its full-control grant.
+		_, grants, errCode = ParseAndValidateAclHeadersOrElseDefault(r, accountManager, ownership, bucketOwnerId, ownerId, true)
 		return grants, errCode
 	}
 }

weed/s3api/s3api_object_acl_permissions_test.go

@@ -0,0 +1,186 @@
+package s3api
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/require"
+)
+
+// objectACLUpdateFiler records the UpdateEntry request the handler issues so
+// tests can assert that denied requests never write object metadata.
+type objectACLUpdateFiler struct {
+	fakeLookupFiler
+	mu     sync.Mutex
+	update *filer_pb.UpdateEntryRequest
+}
+
+func (f *objectACLUpdateFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.update = req
+	return &filer_pb.UpdateEntryResponse{}, nil
+}
+
+// TestPutObjectAclPermissions exercises PutObjectAcl authorization end to end
+// with signed requests: legacy scoped actions, IAM and bucket policies, canned
+// and XML ACLs, retired owners, and objects without stored owner metadata.
+func TestPutObjectAclPermissions(t *testing.T) {
+	const bucket, object, owner = "acl-bucket", "allowed/image.png", "object-owner"
+	const bucketOwner = "bucket-owner"
+	xmlACL := func(ownerID, granteeID string) string {
+		return fmt.Sprintf(`<AccessControlPolicy xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Owner><ID>%s</ID></Owner><AccessControlList><Grant><Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="CanonicalUser"><ID>%s</ID></Grantee><Permission>FULL_CONTROL</Permission></Grant></AccessControlList></AccessControlPolicy>`, ownerID, granteeID)
+	}
+	tests := []struct {
+		name          string
+		account       string
+		action        Action
+		acl           string
+		policy        string
+		retiredOwner  bool
+		ownerMetadata string
+		body          string
+		status        int
+	}{
+		{name: "bucket-wide write-acp grant", account: owner, action: "WriteAcp:acl-bucket", acl: "private", status: http.StatusOK},
+		{name: "prefix write-acp grant", account: owner, action: "WriteAcp:acl-bucket/allowed/*", acl: "public-read", status: http.StatusOK},
+		{name: "grant outside prefix denied", account: owner, action: "WriteAcp:acl-bucket/other/*", acl: "private", status: http.StatusForbidden},
+		{name: "grant on other bucket denied", account: owner, action: "WriteAcp:other-bucket", acl: "private", status: http.StatusForbidden},
+		{name: "read-only grant denied", account: owner, action: "Read:acl-bucket", acl: "private", status: http.StatusForbidden},
+		{name: "non-owner denied", account: "other-owner", action: "WriteAcp:acl-bucket", acl: "private", status: http.StatusForbidden},
+		{name: "admin sets private", account: AccountAdmin.Id, action: "Admin", acl: "private", status: http.StatusOK},
+		{name: "admin sets public-read", account: AccountAdmin.Id, action: "Admin", acl: "public-read", status: http.StatusOK},
+		{name: "iam policy allow", account: owner, policy: "iam-allow", acl: "private", status: http.StatusOK},
+		{name: "bucket policy allow", account: owner, policy: "bucket-allow", acl: "private", status: http.StatusOK},
+		{name: "iam policy explicit deny", account: owner, action: "WriteAcp:acl-bucket", policy: "iam-deny", acl: "private", status: http.StatusForbidden},
+		{name: "bucket policy explicit deny", account: owner, action: "WriteAcp:acl-bucket", policy: "bucket-deny", acl: "private", status: http.StatusForbidden},
+		{name: "retired owner private", account: AccountAdmin.Id, action: "Admin", acl: "private", retiredOwner: true, status: http.StatusOK},
+		{name: "retired owner public-read", account: AccountAdmin.Id, action: "Admin", acl: "public-read", retiredOwner: true, status: http.StatusOK},
+		{name: "bucket-owner-read with distinct bucket owner", account: AccountAdmin.Id, action: "Admin", acl: "bucket-owner-read", status: http.StatusOK},
+		{name: "bucket-owner-full-control with distinct bucket owner", account: AccountAdmin.Id, action: "Admin", acl: "bucket-owner-full-control", status: http.StatusOK},
+		{name: "xml acl preserves retired owner", account: AccountAdmin.Id, action: "Admin", body: xmlACL(owner, owner), retiredOwner: true, status: http.StatusOK},
+		{name: "xml acl unknown grantee denied", account: AccountAdmin.Id, action: "Admin", body: xmlACL(owner, "unknown-grantee"), retiredOwner: true, status: http.StatusBadRequest},
+		{name: "xml acl owner change denied", account: AccountAdmin.Id, action: "Admin", body: xmlACL("other-owner", owner), status: http.StatusForbidden},
+		{name: "nil extended metadata denied", account: owner, action: "WriteAcp:acl-bucket", acl: "private", ownerMetadata: "nil", status: http.StatusForbidden},
+		{name: "absent owner denied", account: owner, action: "WriteAcp:acl-bucket", acl: "private", ownerMetadata: "absent", status: http.StatusForbidden},
+		{name: "empty owner denied", account: owner, action: "WriteAcp:acl-bucket", acl: "private", ownerMetadata: "empty", status: http.StatusForbidden},
+		{name: "ownerless iam policy denied", account: owner, policy: "iam-allow", acl: "private", ownerMetadata: "absent", status: http.StatusForbidden},
+		{name: "ownerless bucket policy denied", account: owner, policy: "bucket-allow", acl: "private", ownerMetadata: "absent", status: http.StatusForbidden},
+		{name: "admin takes ownerless object private", account: AccountAdmin.Id, action: "Admin", acl: "private", ownerMetadata: "absent", status: http.StatusOK},
+		{name: "admin takes ownerless object public-read", account: AccountAdmin.Id, action: "Admin", acl: "public-read", ownerMetadata: "nil", status: http.StatusOK},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			filer := &objectACLUpdateFiler{fakeLookupFiler: fakeLookupFiler{entry: &filer_pb.Entry{
+				Name: "image.png", Extended: map[string][]byte{s3_constants.ExtAmzOwnerKey: []byte(owner)},
+			}}}
+			// Model the ways an object written outside the S3 path (or by an
+			// older version) can lack owner metadata; the requester must not
+			// be treated as the owner in any of them.
+			switch tt.ownerMetadata {
+			case "nil":
+				filer.entry.Extended = nil
+			case "absent":
+				delete(filer.entry.Extended, s3_constants.ExtAmzOwnerKey)
+			case "empty":
+				filer.entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte{}
+			}
+			s3a := newHeadBucketTestServer(t, filer)
+			s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+			s3a.bucketConfigCache.Set(bucket, &BucketConfig{Name: bucket, Ownership: s3_constants.OwnershipObjectWriter, Owner: bucketOwner})
+			s3a.iam = NewIdentityAccessManagementWithStore(s3a.option, nil, "memory")
+			t.Cleanup(s3a.iam.Shutdown)
+			s3a.iam.isAuthEnabled = true
+			account := &Account{Id: tt.account, DisplayName: tt.account}
+			identity := &Identity{
+				Name: "acl-test-user", Account: account, Actions: []Action{tt.action}, IsStatic: true,
+				Credentials: []*Credential{{AccessKey: routingTestAccessKey, SecretKey: routingTestSecretKey}},
+			}
+			if tt.action == "" {
+				identity.Actions = nil
+			}
+			s3a.iam.accessKeyIdent[routingTestAccessKey] = identity
+			s3a.iam.nameToIdentity[identity.Name] = identity
+			if !tt.retiredOwner {
+				s3a.iam.accounts[owner] = &Account{Id: owner, DisplayName: owner}
+			}
+			s3a.iam.accounts[bucketOwner] = &Account{Id: bucketOwner, DisplayName: bucketOwner}
+			s3a.iam.accounts[account.Id] = account
+			if tt.policy != "" {
+				effect := "Allow"
+				if strings.HasSuffix(tt.policy, "deny") {
+					effect = "Deny"
+				}
+				statement := fmt.Sprintf(`{"Effect":%q,"Action":"s3:PutObjectAcl","Resource":"arn:aws:s3:::acl-bucket/allowed/*"}`, effect)
+				if strings.HasPrefix(tt.policy, "iam") {
+					require.NoError(t, s3a.iam.PutPolicy("acl-policy", `{"Version":"2012-10-17","Statement":[`+statement+`]}`))
+					identity.PolicyNames = []string{"acl-policy"}
+				} else {
+					statement = strings.Replace(statement, `{"Effect":`, `{"Principal":"*","Effect":`, 1)
+					s3a.iam.policyEngine = NewBucketPolicyEngine()
+					require.NoError(t, s3a.iam.policyEngine.engine.SetBucketPolicy(bucket, `{"Version":"2012-10-17","Statement":[`+statement+`]}`))
+				}
+			}
+
+			req := httptest.NewRequest(http.MethodPut, "http://s3/"+bucket+"/"+object+"?acl", strings.NewReader(tt.body))
+			req = mux.SetURLVars(req, map[string]string{"bucket": bucket, "object": object})
+			req.Header.Set(s3_constants.AmzCannedAcl, tt.acl)
+			signRoutingTestRequest(t, req, tt.body, "s3")
+			// The signing helper replaces the body; restore the bodyless shape
+			// a real HTTP request has so it is not mistaken for an empty XML ACL.
+			if tt.body == "" {
+				req.Body = http.NoBody
+			}
+			rr := httptest.NewRecorder()
+			s3a.iam.Auth(s3a.PutObjectAclHandler, s3_constants.ACTION_WRITE_ACP)(rr, req)
+			require.Equal(t, tt.status, rr.Code, rr.Body.String())
+
+			filer.mu.Lock()
+			update := filer.update
+			filer.mu.Unlock()
+			if tt.status != http.StatusOK {
+				require.Nil(t, update, "denied requests must not write metadata")
+				return
+			}
+			require.NotNil(t, update)
+			require.Equal(t, "/buckets/acl-bucket/allowed", update.Directory)
+			wantOwner := owner
+			if tt.ownerMetadata != "" {
+				// Admins keep the existing fallback that establishes them as
+				// owner of objects without stored owner metadata.
+				wantOwner = tt.account
+			}
+			require.Equal(t, wantOwner, string(update.Entry.Extended[s3_constants.ExtAmzOwnerKey]))
+			grants := GetAcpGrants(update.Entry.Extended)
+			wantGrants := 1
+			if tt.acl == "public-read" || strings.HasPrefix(tt.acl, "bucket-owner-") {
+				wantGrants = 2
+			}
+			require.Len(t, grants, wantGrants)
+			require.Equal(t, wantOwner, aws.StringValue(grants[0].Grantee.ID), "full-control grant must belong to the object owner")
+			require.Equal(t, s3_constants.PermissionFullControl, aws.StringValue(grants[0].Permission))
+			if tt.acl == "public-read" {
+				require.Equal(t, s3_constants.GranteeGroupAllUsers, aws.StringValue(grants[1].Grantee.URI))
+				require.Equal(t, s3_constants.PermissionRead, aws.StringValue(grants[1].Permission))
+			}
+			if strings.HasPrefix(tt.acl, "bucket-owner-") {
+				require.Equal(t, bucketOwner, aws.StringValue(grants[1].Grantee.ID))
+				permission := s3_constants.PermissionRead
+				if tt.acl == "bucket-owner-full-control" {
+					permission = s3_constants.PermissionFullControl
+				}
+				require.Equal(t, permission, aws.StringValue(grants[1].Permission))
+			}
+		})
+	}
+}

weed/s3api/s3api_object_handlers_acl.go

@@ -3,7 +3,6 @@ package s3api
 import (
 	"context"
 	"errors"
-	"fmt"
 	"net/http"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -207,15 +206,27 @@ func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Reque
 		}
 	}
 
+	// **PERMISSION CHECKS**
+
+	isAdmin := s3a.isUserAdmin(r)
+
+	// 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.
+	// Admins keep the fallback below to take over ownerless objects, which
+	// can be produced by non-S3 filer writes.
+	if objectOwner == "" && !isAdmin {
+		glog.V(3).Infof("PutObjectAclHandler: Access denied - object %s/%s has no owner metadata", bucket, object)
+		s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
+		return
+	}
+
 	// Fallback to current account if no owner stored
 	if objectOwner == "" {
 		objectOwner = amzAccountId
 	}
 
-	// **PERMISSION CHECKS**
-
 	// 1. Check if user is admin (admins can modify any ACL)
-	if !s3a.isUserAdmin(r) {
+	if !isAdmin {
 		// 2. Check object ownership - only object owner can modify ACL (unless admin)
 		if objectOwner != amzAccountId {
 			glog.V(3).Infof("PutObjectAclHandler: Access denied - user %s is not owner of object %s/%s (owner: %s)",
@@ -224,22 +235,16 @@ func (s3a *S3ApiServer) PutObjectAclHandler(w http.ResponseWriter, r *http.Reque
 			return
 		}
 
-		// 3. Check object-level WRITE_ACP permission
-		// Create the specific action for this object
-		writeAcpAction := Action(fmt.Sprintf("WriteAcp:%s/%s", bucket, object))
-		identity, errCode := s3a.iam.authRequest(r, writeAcpAction)
+		// 3. Check object-level WRITE_ACP permission. The unified auth path
+		// scopes the base action to the request's bucket/object for both
+		// legacy Actions and IAM/bucket policies; embedding the resource path
+		// in the action here would scope it twice.
+		_, errCode := s3a.iam.authRequest(r, Action(s3_constants.ACTION_WRITE_ACP))
 		if errCode != s3err.ErrNone {
 			glog.V(3).Infof("PutObjectAclHandler: Auth failed for WriteAcp action on %s/%s: %v", bucket, object, errCode)
 			s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
 			return
 		}
-
-		// 4. Verify the authenticated identity can perform WriteAcp on this specific object
-		if identity == nil || !identity.CanDo(writeAcpAction, bucket, object) {
-			glog.V(3).Infof("PutObjectAclHandler: Identity %v cannot perform WriteAcp on %s/%s", identity, bucket, object)
-			s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
-			return
-		}
 	} else {
 		glog.V(3).Infof("PutObjectAclHandler: Admin user %s granted ACL modification permission for %s/%s", amzAccountId, bucket, object)
 	}