s3: accept a self-copy that changes the storage class (#12031)

* s3: accept a self-copy that changes the storage class or encryption

AWS changes an object's storage class or encryption by copying the
object onto itself with the new x-amz-storage-class or SSE headers, and
only rejects a self-copy that changes none of metadata, storage class,
website redirect location or encryption. CopyObject rejected every
self-copy without REPLACE metadata or tagging directives as
InvalidRequest, so these requests failed.

* s3: reject an invalid x-amz-storage-class on CopyObject

A self-copy with a bad storage class used to fail as InvalidRequest;
once such copies are admitted the value must be validated like PutObject
does, or it would be stored as opaque metadata.

* test: cover the invalid storage class rejection on self-copy

* s3: only a storage class change admits a self-copy

Admitting encryption changes on a self-copy is unsafe: the metadata-only
path compares only KMS key IDs, so a new encryption context or bucket-key
setting is advertised over the old data key, and SSE-KMS key rotation
onto the same entry keeps the old data key.

Count only a requested storage class that differs from the object's
(STANDARD when none is stored) as a change, so a self-copy to the
current class or with only encryption headers is rejected as before.
Such a self-copy updates the object in place and has to keep its
encryption; one that would re-encrypt it or uses SSE-KMS is rejected
too.

---------

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

weed/s3api/s3api_object_handlers_copy.go

@@ -70,6 +70,14 @@ func isValidDirective(value string) bool {
 	return value == "" || value == DirectiveCopy || value == DirectiveReplace
 }
 
+// copyChangesStorageClass reports whether a copy request asks for a storage
+// class other than the source object's. That makes copying an object onto
+// itself a change AWS accepts, even with the COPY directive.
+func (s3a *S3ApiServer) copyChangesStorageClass(h http.Header, src *filer_pb.Entry) bool {
+	sc := h.Get(s3_constants.AmzStorageClass)
+	return sc != "" && sc != s3a.getStorageClassFromExtended(src.Extended)
+}
+
 // hasPrefixFold reports whether s starts with prefix, ignoring case.
 func hasPrefixFold(s, prefix string) bool {
 	return len(s) >= len(prefix) && strings.EqualFold(s[:len(prefix)], prefix)
@@ -196,6 +204,11 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 		return
 	}
 
+	if sc := r.Header.Get(s3_constants.AmzStorageClass); sc != "" && !validateStorageClass(sc) {
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidStorageClass)
+		return
+	}
+
 	replaceMeta, replaceTagging := replaceDirective(r.Header)
 
 	// If source object is empty or bucket is empty, reply back invalid copy source.
@@ -266,9 +279,10 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 	}
 
 	sameDestination := srcBucket == dstBucket && srcObject == dstObject
+	changesStorageClass := s3a.copyChangesStorageClass(r.Header, entry)
 	// A self-copy into a versioned bucket writes a new version instead of overwriting in
 	// place, so it is not the no-op AWS rejects. It is how an earlier version is restored.
-	if sameDestination && !(replaceMeta || replaceTagging) && srcVersioningState != s3_constants.VersioningEnabled {
+	if sameDestination && !(replaceMeta || replaceTagging || changesStorageClass) && srcVersioningState != s3_constants.VersioningEnabled {
 		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopyDest)
 		return
 	}
@@ -300,7 +314,16 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
 
 	replacesSource := copyReplacesSourceEntry(sameDestination, dstVersioningState, srcVersionId)
 
-	if replacesSource && (replaceMeta || replaceTagging) && s3a.canUseMetadataOnlySelfCopy(entry, r, dstBucket, dstObject) {
+	// A self-copy that only changes the storage class updates the object in
+	// place and has to keep its encryption as stored. The direct strategy
+	// compares only KMS key IDs, so SSE-KMS requests stay rejected too.
+	if replacesSource && changesStorageClass && !(replaceMeta || replaceTagging) &&
+		(IsSSEKMSRequest(r) || !s3a.canUseMetadataOnlySelfCopy(entry, r, dstBucket, dstObject)) {
+		s3err.WriteErrorResponse(w, r, s3err.ErrInvalidCopyDest)
+		return
+	}
+
+	if replacesSource && (replaceMeta || replaceTagging || changesStorageClass) && s3a.canUseMetadataOnlySelfCopy(entry, r, dstBucket, dstObject) {
 		var dstVersionId string
 		var etag string
 		// Bind the entire self-copy update to the snapshot read on each attempt.

weed/s3api/s3api_object_handlers_copy_storage_class_test.go

@@ -0,0 +1,111 @@
+package s3api
+
+import (
+	"encoding/base64"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"google.golang.org/protobuf/proto"
+)
+
+// AWS changes an object's storage class by copying the object onto itself
+// with the COPY directive and a new x-amz-storage-class. Only a self-copy that
+// changes nothing is rejected, and a bucket's default encryption is no change.
+// A self-copy that would also change the encryption stays rejected.
+func TestCopyObjectToItselfChangingStorageClass(t *testing.T) {
+	const bucket, object = "self-copy", "object"
+	kmsObject := map[string]string{
+		s3_constants.AmzServerSideEncryption:            s3_constants.SSEAlgorithmKMS,
+		s3_constants.AmzServerSideEncryptionAwsKmsKeyId: "key-a",
+		s3_constants.AmzServerSideEncryptionContext:     base64.StdEncoding.EncodeToString([]byte(`{"env":"old"}`)),
+	}
+	for _, tt := range []struct {
+		name              string
+		stored            map[string]string // extra object metadata
+		storageClass      string
+		headers           map[string]string // extra request headers
+		defaultEncryption bool
+		wantCode          int
+		wantErr           string
+	}{
+		{name: "no change", wantCode: http.StatusBadRequest, wantErr: "InvalidRequest"},
+		{name: "no change under default encryption", defaultEncryption: true, wantCode: http.StatusBadRequest, wantErr: "InvalidRequest"},
+		{name: "invalid storage class", storageClass: "NOT_A_CLASS", wantCode: http.StatusBadRequest, wantErr: "InvalidStorageClass"},
+		{name: "new storage class", storageClass: "STANDARD_IA", wantCode: http.StatusOK},
+		{name: "current storage class", stored: map[string]string{s3_constants.AmzStorageClass: "STANDARD_IA"},
+			storageClass: "STANDARD_IA", wantCode: http.StatusBadRequest, wantErr: "InvalidRequest"},
+		{name: "STANDARD without a stored class", storageClass: "STANDARD", wantCode: http.StatusBadRequest, wantErr: "InvalidRequest"},
+		{name: "encryption only", headers: map[string]string{s3_constants.AmzServerSideEncryption: "AES256"},
+			wantCode: http.StatusBadRequest, wantErr: "InvalidRequest"},
+		{name: "new storage class and SSE-KMS context", stored: kmsObject, storageClass: "STANDARD_IA", headers: map[string]string{
+			s3_constants.AmzServerSideEncryption:            s3_constants.SSEAlgorithmKMS,
+			s3_constants.AmzServerSideEncryptionAwsKmsKeyId: "key-a",
+			s3_constants.AmzServerSideEncryptionContext:     base64.StdEncoding.EncodeToString([]byte(`{"env":"new"}`)),
+		}, wantCode: http.StatusBadRequest, wantErr: "InvalidRequest"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			filer := &copyACLFiler{ambiguousPutFiler: &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{}}}
+			s3a := newPutTestServer(t, startFakeFiler(t, filer))
+			s3a.iam = &IdentityAccessManagement{}
+			s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+			config := &BucketConfig{Name: bucket, Ownership: s3_constants.OwnershipObjectWriter}
+			if tt.defaultEncryption {
+				config.Encryption = &s3_pb.EncryptionConfiguration{SseAlgorithm: s3_constants.SSEAlgorithmAES256}
+			}
+			s3a.bucketConfigCache.Set(bucket, config)
+			objectPath := "/buckets/" + bucket + "/" + object
+			original := &filer_pb.Entry{Name: object, Content: []byte("data"),
+				Attributes: &filer_pb.FuseAttributes{FileMode: defaultFileMode, FileSize: 4, Mime: "text/plain"},
+				Extended:   map[string][]byte{"X-Amz-Meta-Owner": []byte("kept")}}
+			for k, v := range tt.stored {
+				original.Extended[k] = []byte(v)
+			}
+			filer.entries[objectPath] = proto.Clone(original).(*filer_pb.Entry)
+
+			req := httptest.NewRequest(http.MethodPut, "/"+bucket+"/"+object, nil)
+			req = mux.SetURLVars(req, map[string]string{"bucket": bucket, "object": object})
+			req.Header.Set("X-Amz-Copy-Source", "/"+bucket+"/"+object)
+			req.Header.Set(s3_constants.AmzUserMetaDirective, DirectiveCopy)
+			if tt.storageClass != "" {
+				req.Header.Set(s3_constants.AmzStorageClass, tt.storageClass)
+			}
+			for k, v := range tt.headers {
+				req.Header.Set(k, v)
+			}
+			rr := httptest.NewRecorder()
+			s3a.CopyObjectHandler(rr, req)
+
+			if rr.Code != tt.wantCode {
+				t.Fatalf("status = %d, want %d; body: %s", rr.Code, tt.wantCode, rr.Body.String())
+			}
+			filer.mu.Lock()
+			defer filer.mu.Unlock()
+			stored := filer.entries[objectPath]
+			if tt.wantCode != http.StatusOK {
+				if !strings.Contains(rr.Body.String(), "<Code>"+tt.wantErr+"</Code>") {
+					t.Fatalf("body = %s, want %s", rr.Body.String(), tt.wantErr)
+				}
+				if !proto.Equal(stored, original) {
+					t.Fatalf("rejected copy changed the object: %v", stored)
+				}
+				return
+			}
+			if got := string(stored.Extended[s3_constants.AmzStorageClass]); got != tt.storageClass {
+				t.Fatalf("storage class = %q, want %q", got, tt.storageClass)
+			}
+			if got := string(stored.Extended["X-Amz-Meta-Owner"]); got != "kept" {
+				t.Fatalf("user metadata = %q, want it kept under the COPY directive", got)
+			}
+			if got := string(stored.Content); got != "data" {
+				t.Fatalf("content = %q, want %q", got, "data")
+			}
+		})
+	}
+}