s3tables: only the admin repairs a table bucket that has no owner metadata (#11873)

* s3tables: only the admin repairs a table bucket that has no owner metadata

CreateNamespace wrote bucket metadata naming the caller as owner before any
permission check ran, so an ordinary caller that found the marker without
metadata was written in as the bucket's owner and passed every check after.
A bucket in that state is answered with the operation's own denial; the
admin or a trusted internal handler may still repair it.

* s3tables: repair the marker-only bucket under the entry lock

A creator finishing its own create between the missing-metadata check and
the repair write would have its owner overwritten. The mutation now adopts
whatever metadata is there and writes only while it is still absent.

* s3tables: gate the repair on the Admin action, not an operation grant

hasAdminAction also answers s3tables:*, so a non-admin identity holding
the wildcard operation grant could pass the repair check and write itself
in as the bucket's owner.

weed/s3api/s3tables/handler_namespace.go

@@ -11,8 +11,29 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 )
 
+// errOwnerlessBucket marks a table bucket that has the marker but no
+// metadata: damaged, and answered with the operation's own denial rather than
+// letting the caller name itself the bucket's owner before any check runs.
+var errOwnerlessBucket = fmt.Errorf("table bucket has no owner metadata")
+
+// mayRepairOwnerlessBucket reports whether the caller may repair a bucket in
+// that state: the admin or a trusted internal handler. hasAdminAction also
+// answers s3tables:*, an operation grant, not administrator status.
+func (h *S3TablesHandler) mayRepairOwnerlessBucket(r *http.Request) bool {
+	if h.trusted || h.getAccountID(r) == s3_constants.AccountAdminId {
+		return true
+	}
+	for _, action := range getIdentityActions(r) {
+		if action == s3_constants.ACTION_ADMIN {
+			return true
+		}
+	}
+	return false
+}
+
 // handleCreateNamespace creates a new namespace in a table bucket
 func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error {
 	glog.Errorf("S3Tables: handleCreateNamespace called")
@@ -66,16 +87,29 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 				if entryResp.Entry == nil || !IsTableBucketEntry(entryResp.Entry) {
 					return filer_pb.ErrNotFound
 				}
-				bucketMetadata = tableBucketMetadata{
-					Name:           bucketName,
-					CreatedAt:      time.Now(),
-					OwnerAccountID: ownerAccountID,
+				if !h.mayRepairOwnerlessBucket(r) {
+					return errOwnerlessBucket
 				}
-				metadataBytes, err := json.Marshal(&bucketMetadata)
-				if err != nil {
-					return fmt.Errorf("failed to marshal bucket metadata: %w", err)
-				}
-				if err := h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata, metadataBytes); err != nil {
+				// Repair the marker-only entry under the entry lock so a
+				// creator finishing its own create in the meantime keeps its
+				// owner: adopt whatever metadata is there, write only when it
+				// is still absent.
+				if err := h.mutateEntryExtended(r.Context(), client, bucketPath, func(extended map[string][]byte) error {
+					if data, ok := extended[ExtendedKeyMetadata]; ok {
+						return json.Unmarshal(data, &bucketMetadata)
+					}
+					bucketMetadata = tableBucketMetadata{
+						Name:           bucketName,
+						CreatedAt:      time.Now(),
+						OwnerAccountID: ownerAccountID,
+					}
+					metadataBytes, err := json.Marshal(&bucketMetadata)
+					if err != nil {
+						return fmt.Errorf("failed to marshal bucket metadata: %w", err)
+					}
+					extended[ExtendedKeyMetadata] = metadataBytes
+					return nil
+				}); err != nil {
 					return err
 				}
 			} else {
@@ -101,6 +135,10 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 	})
 
 	if err != nil {
+		if errors.Is(err, errOwnerlessBucket) {
+			h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create namespace in this bucket")
+			return ErrAccessDenied
+		}
 		if errors.Is(err, filer_pb.ErrNotFound) {
 			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName))
 		} else {

weed/s3api/s3tables/handler_ownerless_test.go

@@ -0,0 +1,88 @@
+package s3tables
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A table bucket with the marker but no metadata is damaged, not ownerless for
+// the taking: an ordinary caller is answered with the operation's own denial,
+// and the admin or a trusted handler may repair it.
+func TestCreateNamespaceOnAnOwnerlessBucket(t *testing.T) {
+	const bucket = "broken"
+	plant := func(fs *s3tablestest.MemFiler) {
+		fs.Put(TablesPath, bucket, map[string][]byte{ExtendedKeyTableBucket: []byte("true")})
+	}
+	arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, bucket)
+	require.NoError(t, err)
+	request := &CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{"ns"}}
+
+	t.Run("an outsider gets the denial and writes nothing", func(t *testing.T) {
+		fs := s3tablestest.Start(t)
+		plant(fs)
+		m := NewManager()
+		body, err := json.Marshal(request)
+		require.NoError(t, err)
+		identity := &testIdentity{Name: "outsider", Account: &testIdentityAccount{Id: "outsider"}, Actions: []string{"Read"}}
+		req, err := newManagerRequest(s3_constants.SetIdentityInContext(context.Background(), identity), "CreateNamespace", body, "outsider")
+		require.NoError(t, err)
+
+		recorder := httptest.NewRecorder()
+		m.handler.HandleRequest(recorder, req, NewManagerClient(fs.Client))
+		assert.Equal(t, http.StatusForbidden, recorder.Code, recorder.Body.String())
+		entry := fs.Get(TablesPath, bucket)
+		require.NotNil(t, entry)
+		assert.Empty(t, entry.Extended[ExtendedKeyMetadata], "the denied caller wrote itself in as the bucket owner")
+		assert.Nil(t, fs.Get(GetTableBucketPath(bucket), "ns"))
+	})
+
+	t.Run("a trusted handler repairs it and owns it", func(t *testing.T) {
+		fs := s3tablestest.Start(t)
+		plant(fs)
+		m := NewManager()
+		m.SetTrusted(true)
+		body, err := json.Marshal(request)
+		require.NoError(t, err)
+		identity := &testIdentity{Name: "svc", Account: &testIdentityAccount{Id: "svc"}, Actions: []string{"Read"}}
+		req, err := newManagerRequest(s3_constants.SetIdentityInContext(context.Background(), identity), "CreateNamespace", body, "svc")
+		require.NoError(t, err)
+
+		recorder := httptest.NewRecorder()
+		m.handler.HandleRequest(recorder, req, NewManagerClient(fs.Client))
+		assert.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
+		entry := fs.Get(TablesPath, bucket)
+		require.NotNil(t, entry)
+		var metadata tableBucketMetadata
+		require.NoError(t, json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata))
+		assert.Equal(t, "svc", metadata.OwnerAccountID)
+		require.NotNil(t, fs.Get(GetTableBucketPath(bucket), "ns"))
+	})
+
+	t.Run("an untrusted admin repairs it too", func(t *testing.T) {
+		fs := s3tablestest.Start(t)
+		plant(fs)
+		m := NewManager()
+		body, err := json.Marshal(request)
+		require.NoError(t, err)
+		identity := &testIdentity{Name: "admin", Account: &testIdentityAccount{Id: s3_constants.AccountAdminId}, Actions: []string{"Admin"}}
+		req, err := newManagerRequest(s3_constants.SetIdentityInContext(context.Background(), identity), "CreateNamespace", body, "admin")
+		require.NoError(t, err)
+
+		recorder := httptest.NewRecorder()
+		m.handler.HandleRequest(recorder, req, NewManagerClient(fs.Client))
+		assert.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String())
+		entry := fs.Get(TablesPath, bucket)
+		require.NotNil(t, entry)
+		var metadata tableBucketMetadata
+		require.NoError(t, json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata))
+		assert.Equal(t, s3_constants.AccountAdminId, metadata.OwnerAccountID)
+	})
+}