s3: let bucket-scoped Admin grants auto-create matching buckets (#12268)

* s3: let bucket-scoped Admin grants auto-create matching buckets

autoCreateBucket required a global Admin identity, so an identity
scoped to a prefix like "Admin:outline*" could explicitly CreateBucket
but was denied when an upload auto-created the same bucket. Gate on
CanDo(ACTION_ADMIN, bucket) — the same check explicit CreateBucket
enforces — so a prefix-scoped key lazily creates only matching buckets.

Generated with [Devin](https://devin.ai)

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

* s3: run auto-create bucket admin through the policy-aware path

A scoped Admin identity with an attached IAM policy that denied
s3:CreateBucket could still auto-create a matching bucket, because
userCanAdminBucket only consulted legacy Actions. Evaluate the same
VerifyActionPermission explicit CreateBucket gets, over a synthesized
PUT /bucket request so the attached-policy check resolves to
s3:CreateBucket on the bucket ARN regardless of the upload's shape.

* s3: record verified identity as owner of auto-created buckets

An admin authenticated outside the middleware (IAM integration, post
policy signature) left no identity in the request context, so an
auto-created bucket stored no owner and never reached the owner index.

---------

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

weed/s3api/s3api_bucket_handlers.go

@@ -340,7 +340,7 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
 		bucketCrtime = entry.Attributes.Crtime
 
 		// Set bucket owner
-		s3a.setBucketOwner(r)(entry)
+		s3a.setBucketOwner(r, nil)(entry)
 
 		// Persist a requested non-default ACL so GetBucketAcl and idempotent
 		// recreation observe it (private is the default and is not stored).
@@ -719,8 +719,10 @@ func hasExplicitBucketACL(r *http.Request) bool {
 	return false
 }
 
-// setBucketOwner creates a function that sets the bucket owner from the request context
-func (s3a *S3ApiServer) setBucketOwner(r *http.Request) func(entry *filer_pb.Entry) {
+// setBucketOwner creates a function that sets the bucket owner from the request
+// context, falling back to an already-verified identity when authentication did
+// not go through the middleware that records it there.
+func (s3a *S3ApiServer) setBucketOwner(r *http.Request, verified *Identity) 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 &&
@@ -733,6 +735,16 @@ func (s3a *S3ApiServer) setBucketOwner(r *http.Request) func(entry *filer_pb.Ent
 	// 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 verified != nil && s3a.iam != nil && s3a.iam.identityAnonymous != nil &&
+		verified.Name == s3a.iam.identityAnonymous.Name {
+		verified = nil
+	}
+	if currentIdentityId == "" && verified != nil {
+		currentIdentityId = verified.Name
+	}
+	if accountId == "" && verified != nil && verified.Account != nil {
+		accountId = verified.Account.Id
+	}
 	if accountId == AccountAnonymous.Id || (s3a.iam != nil && !s3a.iam.isEnabled()) {
 		accountId = ""
 	}
@@ -753,7 +765,7 @@ func (s3a *S3ApiServer) setBucketOwner(r *http.Request) func(entry *filer_pb.Ent
 }
 
 // autoCreateBucket creates a bucket if it doesn't exist, setting the owner from the request context
-// Only users with admin permissions are allowed to auto-create buckets
+// Only identities that may administer the named bucket are allowed to auto-create it
 func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error {
 	// Validate the bucket name before auto-creating
 	if err := s3bucket.VerifyS3BucketName(bucket); err != nil {
@@ -764,16 +776,18 @@ func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error {
 		return fmt.Errorf("auto-create bucket %s: %w", bucket, ErrAutoCreateDisabled)
 	}
 
-	// Check if user has admin permissions
-	if !s3a.isUserAdmin(r) {
+	// Check if user may administer this bucket
+	adminIdentity, allowed := s3a.userCanAdminBucket(r, bucket)
+	if !allowed {
 		return fmt.Errorf("auto-create bucket %s: %w", bucket, ErrAutoCreatePermissionDenied)
 	}
 
-	identityId := s3_constants.GetIdentityNameFromContext(r)
 	var bucketCrtime int64
+	var ownerId string
 	if err := s3a.mkdirExclusive(s3a.option.BucketsPath, bucket, func(entry *filer_pb.Entry) {
 		bucketCrtime = entry.Attributes.Crtime
-		s3a.setBucketOwner(r)(entry)
+		s3a.setBucketOwner(r, adminIdentity)(entry)
+		ownerId = bucketEntryOwner(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.
@@ -799,7 +813,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
-				s3a.setBucketOwner(r)(entry)
+				s3a.setBucketOwner(r, adminIdentity)(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 {
@@ -825,9 +839,9 @@ func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error {
 		s3a.bucketConfigCache.RemoveNegativeCache(bucket)
 	}
 
-	if identityId != "" {
-		if err := s3a.addBucketToOwnerIndex(identityId, bucket, bucketCrtime); err != nil {
-			glog.Warningf("autoCreateBucket: owner index add %s/%s: %v", identityId, bucket, err)
+	if ownerId != "" {
+		if err := s3a.addBucketToOwnerIndex(ownerId, bucket, bucketCrtime); err != nil {
+			glog.Warningf("autoCreateBucket: owner index add %s/%s: %v", ownerId, bucket, err)
 		}
 	}
 
@@ -900,6 +914,43 @@ func (s3a *S3ApiServer) isUserAdmin(r *http.Request) bool {
 	return identity != nil && identity.isAdmin()
 }
 
+// userCanAdminBucket securely checks if the authenticated identity may
+// administer the named bucket: a global Admin, or a bucket-scoped grant such
+// as "Admin:prefix*" matching the bucket name — the same rule explicit
+// CreateBucket already enforces, including attached IAM policy denies.
+// The verified identity is returned so callers can attribute ownership even
+// when no middleware recorded it in the request context.
+func (s3a *S3ApiServer) userCanAdminBucket(r *http.Request, bucket string) (*Identity, bool) {
+	if identityObj := s3_constants.GetIdentityFromContext(r); identityObj != nil {
+		if identity, ok := identityObj.(*Identity); ok {
+			return identity, s3a.verifyBucketAdmin(r, identity, bucket)
+		}
+	}
+	if s3a.iam == nil {
+		return nil, false
+	}
+	identity, errCode := s3a.iam.authRequest(r, s3_constants.ACTION_ADMIN)
+	if errCode != s3err.ErrNone {
+		return nil, false
+	}
+	return identity, s3a.verifyBucketAdmin(r, identity, bucket)
+}
+
+// verifyBucketAdmin evaluates bucket administration as s3:CreateBucket through
+// the same policy-aware path explicit creation uses: a synthesized PUT
+// /bucket request makes attached IAM policies resolve (and deny) against the
+// bucket ARN regardless of the real request's method or query.
+func (s3a *S3ApiServer) verifyBucketAdmin(r *http.Request, identity *Identity, bucket string) bool {
+	if identity == nil {
+		return false
+	}
+	if s3a.iam == nil {
+		return identity.CanDo(s3_constants.ACTION_ADMIN, bucket, "")
+	}
+	createReq := objectKeyPolicyRequest(r, http.MethodPut, bucket, "", "")
+	return s3a.iam.VerifyActionPermission(createReq, identity, s3_constants.ACTION_ADMIN, bucket, "") == s3err.ErrNone
+}
+
 // isBucketPublicRead checks if a bucket allows anonymous read access based on its cached ACL status
 func (s3a *S3ApiServer) isBucketPublicRead(bucket string) bool {
 	// Get bucket configuration which contains cached public-read status

weed/s3api/s3api_bucket_handlers_misc_test.go

@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -278,6 +279,151 @@ func TestHandleAutoCreateBucketNonAdmin(t *testing.T) {
 	}
 }
 
+// autoCreateBucketFiler records every entry a bucket auto-create writes.
+type autoCreateBucketFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	created []*filer_pb.Entry
+}
+
+func (f *autoCreateBucketFiler) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	f.created = append(f.created, req.Entry)
+	return &filer_pb.CreateEntryResponse{}, nil
+}
+
+// A bucket-scoped admin grant ("Admin:prefix*") authorizes auto-create for the
+// buckets it matches — the same rule explicit CreateBucket enforces — and no
+// others.
+func TestAutoCreateBucketScopedAdmin(t *testing.T) {
+	cases := []struct {
+		name    string
+		actions []Action
+		bucket  string
+		allowed bool
+	}{
+		{"global admin", []Action{"Admin"}, "app-uploads", true},
+		{"matching prefix", []Action{"Admin:app-*"}, "app-uploads", true},
+		{"exact bucket", []Action{"Admin:app-uploads"}, "app-uploads", true},
+		{"non-matching prefix", []Action{"Admin:app-*"}, "other-bucket", false},
+		{"write-scoped grant is not admin", []Action{"Write:app-*"}, "app-uploads", false},
+		{"unscoped write", []Action{"Write"}, "app-uploads", false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			filer := &autoCreateBucketFiler{}
+			s3a := newHeadBucketTestServer(t, filer)
+			s3a.option.AutoCreateBucket = true
+			req := newBucketRequest(http.MethodPut, tc.bucket, "", "")
+			req = req.WithContext(recordIdentityInContext(req, &Identity{
+				Name:    "app",
+				Actions: tc.actions,
+			}))
+			rec := httptest.NewRecorder()
+
+			if got := s3a.handleAutoCreateBucket(rec, req, tc.bucket, "PutObjectHandler"); got != tc.allowed {
+				t.Fatalf("handleAutoCreateBucket = %v, want %v (status %d, body %s)", got, tc.allowed, rec.Code, rec.Body.String())
+			}
+			if !tc.allowed {
+				if rec.Code != http.StatusForbidden {
+					t.Fatalf("status = %d, want %d", rec.Code, http.StatusForbidden)
+				}
+				return
+			}
+			require.NotEmpty(t, filer.created)
+			require.Equal(t, tc.bucket, filer.created[0].Name)
+			require.Equal(t, []byte("app"), filer.created[0].Extended[s3_constants.AmzIdentityId])
+		})
+	}
+}
+
+// An attached IAM policy governs auto-create the same way it governs explicit
+// CreateBucket: a Deny on s3:CreateBucket blocks creation even when a scoped
+// Admin grant matches, an Allow for it succeeds, and a policy that says
+// nothing about it leaves only the bare-Admin floor (scoped grants are not
+// consulted once policies attach).
+func TestAutoCreateBucketAttachedPolicy(t *testing.T) {
+	const bucket = "app-uploads"
+	arn := fmt.Sprintf("arn:aws:s3:::%s", bucket)
+	cases := []struct {
+		name    string
+		actions []Action
+		policy  string
+		allowed bool
+	}{
+		{
+			name:    "deny beats scoped admin",
+			actions: []Action{"Admin:app-*"},
+			policy: fmt.Sprintf(`{"Version":"2012-10-17","Statement":[`+
+				`{"Effect":"Allow","Action":"s3:PutObject","Resource":"%s/*"},`+
+				`{"Effect":"Deny","Action":"s3:CreateBucket","Resource":"%s"}]}`, arn, arn),
+			allowed: false,
+		},
+		{
+			name:    "allow without scoped grant",
+			actions: []Action{"Write:app-*"},
+			policy: fmt.Sprintf(`{"Version":"2012-10-17","Statement":[`+
+				`{"Effect":"Allow","Action":"s3:CreateBucket","Resource":"%s"}]}`, arn),
+			allowed: true,
+		},
+		{
+			name:    "scoped admin with unrelated policy is denied",
+			actions: []Action{"Admin:app-*"},
+			policy: fmt.Sprintf(`{"Version":"2012-10-17","Statement":[`+
+				`{"Effect":"Allow","Action":"s3:PutObject","Resource":"%s/*"}]}`, arn),
+			allowed: false,
+		},
+		{
+			name:    "bare admin survives unrelated policy",
+			actions: []Action{"Admin"},
+			policy: fmt.Sprintf(`{"Version":"2012-10-17","Statement":[`+
+				`{"Effect":"Allow","Action":"s3:PutObject","Resource":"%s/*"}]}`, arn),
+			allowed: true,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			filer := &autoCreateBucketFiler{}
+			s3a := newHeadBucketTestServer(t, filer)
+			s3a.option.AutoCreateBucket = true
+			s3a.iam = &IdentityAccessManagement{}
+			require.NoError(t, s3a.iam.PutPolicy("auto-create-test", tc.policy))
+			req := newBucketRequest(http.MethodPut, bucket, "", "")
+			req = req.WithContext(recordIdentityInContext(req, &Identity{
+				Name:        "app",
+				Actions:     tc.actions,
+				PolicyNames: []string{"auto-create-test"},
+			}))
+			rec := httptest.NewRecorder()
+
+			if got := s3a.handleAutoCreateBucket(rec, req, bucket, "PutObjectHandler"); got != tc.allowed {
+				t.Fatalf("handleAutoCreateBucket = %v, want %v (status %d, body %s)", got, tc.allowed, rec.Code, rec.Body.String())
+			}
+			if !tc.allowed {
+				if rec.Code != http.StatusForbidden {
+					t.Fatalf("status = %d, want %d", rec.Code, http.StatusForbidden)
+				}
+				require.Empty(t, filer.created)
+				return
+			}
+			require.NotEmpty(t, filer.created)
+			require.Equal(t, bucket, filer.created[0].Name)
+		})
+	}
+}
+
+// An identity authenticated outside the middleware (IAM integration, policy
+// signature) records no request-context identity and no internal account
+// header; the verified identity still supplies the bucket owner.
+func TestSetBucketOwnerVerifiedIdentityFallback(t *testing.T) {
+	s3a := newHeadBucketTestServer(t, &autoCreateBucketFiler{})
+	req := newBucketRequest(http.MethodPut, "app-uploads", "", "")
+	entry := &filer_pb.Entry{}
+
+	s3a.setBucketOwner(req, &Identity{Name: "app", Account: &Account{Id: "acct-app"}})(entry)
+
+	require.Equal(t, []byte("app"), entry.Extended[s3_constants.AmzIdentityId])
+	require.Equal(t, []byte("acct-app"), entry.Extended[s3_constants.ExtAmzOwnerKey])
+}
+
 func TestUploadMissingBucketAutoCreateDisabled(t *testing.T) {
 	cases := []struct {
 		name    string