s3: keep multipart part encryption on the initiate-time decision (#12134)

* s3: keep multipart part encryption on the initiate-time decision

putToFiler applied the *current* bucket default encryption when writing a
staged part, so a PutBucketEncryption landing between two UploadPart calls
produced an object whose parts disagree: some chunks ciphertext, some
plaintext. Completion then marks the object by its majority and GetObject
fails decrypting the rest. Parts staged under .uploads now skip the
bucket-default lookup; the decision snapshotted into the upload entry at
CreateMultipartUpload is replayed through request headers instead, which
matches AWS (default encryption applies at initiation).

* s3: mark multipart part writes via request context, not the staged path

Inferring "is a part" from the .uploads file path let an ordinary PutObject
named .uploads/x.part skip bucket default encryption entirely. The part
decision now travels on the request context from the two callers that
stage parts (UploadPart, UploadPartCopy); a client-chosen key cannot reach
it.

weed/s3api/s3api_bucket_default_encryption_test.go

@@ -114,3 +114,107 @@ func TestPutObjectAppliesBucketDefaultEncryptionWithVolumeCipher(t *testing.T) {
 		})
 	}
 }
+
+// A part staged under .uploads must not pick up a bucket default encryption
+// change: the upload's encryption is fixed at CreateMultipartUpload, so a part
+// written after PutBucketEncryption would otherwise mix plaintext and
+// ciphertext parts and the completed object would fail to decrypt.
+func TestPutObjectPartSkipsBucketDefaultEncryption(t *testing.T) {
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{
+		volume:  volume,
+		entries: map[string]*filer_pb.Entry{},
+		apply:   true,
+	}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set("b", &BucketConfig{
+		Name: "b",
+		Encryption: &s3_pb.EncryptionConfiguration{
+			SseAlgorithm: "AES256",
+		},
+	})
+
+	filePath := "/buckets/b/.uploads/up1/0002_deadbeef.part"
+	r := withMultipartPart(httptest.NewRequest(http.MethodPut, "/b/o?partNumber=2&uploadId=up1", nil))
+	etag, code, sseMeta := s3a.putToFiler(r, filePath, strings.NewReader("part data"), "b", "", 2, 0, nil, false, "")
+	if code != s3err.ErrNone {
+		t.Fatalf("putToFiler = %v, want %v", code, s3err.ErrNone)
+	}
+	if etag == "" {
+		t.Fatal("expected non-empty etag")
+	}
+	if sseMeta.SSEType != "" {
+		t.Fatalf("part unexpectedly encrypted, SSEType=%s", sseMeta.SSEType)
+	}
+
+	entry, ok := filerImpl.entries[filePath]
+	if !ok {
+		t.Fatalf("entry not found on filer at %s", filePath)
+	}
+	if _, has := entry.Extended[s3_constants.AmzServerSideEncryption]; has {
+		t.Fatal("part entry carries SSE attributes from bucket default encryption")
+	}
+	for i, chunk := range entry.Chunks {
+		if chunk.SseType != filer_pb.SSEType_NONE {
+			t.Errorf("chunk %d: expected SseType NONE, got %v", i, chunk.SseType)
+		}
+	}
+}
+
+// An ordinary PutObject whose key happens to look like a staged part path
+// (.uploads/...part) is still a top-level object write and must get the
+// bucket's default encryption; only calls explicitly marked as part writes
+// skip it.
+func TestPutObjectUnderUploadsFolderKeepsBucketDefaultEncryption(t *testing.T) {
+	km := GetSSES3KeyManager()
+	oldSuperKey := km.superKey
+	km.superKey = make([]byte, 32)
+	for i := range km.superKey {
+		km.superKey[i] = byte(i + 1)
+	}
+	t.Cleanup(func() {
+		km.superKey = oldSuperKey
+	})
+
+	volume := startFakeVolumeServer(t)
+	filerImpl := &ambiguousPutFiler{
+		volume:  volume,
+		entries: map[string]*filer_pb.Entry{},
+		apply:   true,
+	}
+	s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
+	s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
+	s3a.bucketConfigCache.Set("b", &BucketConfig{
+		Name: "b",
+		Encryption: &s3_pb.EncryptionConfiguration{
+			SseAlgorithm: "AES256",
+		},
+	})
+
+	filePath := "/buckets/b/.uploads/notes.part"
+	r := httptest.NewRequest(http.MethodPut, "/b/.uploads/notes.part", nil)
+	etag, code, sseMeta := s3a.putToFiler(r, filePath, strings.NewReader("ordinary object body"), "b", ".uploads/notes.part", 1, 0, nil, false, "")
+	if code != s3err.ErrNone {
+		t.Fatalf("putToFiler = %v, want %v", code, s3err.ErrNone)
+	}
+	if etag == "" {
+		t.Fatal("expected non-empty etag")
+	}
+	if sseMeta.SSEType != s3_constants.SSETypeS3 {
+		t.Fatalf("expected SSE response metadata type %s, got %s", s3_constants.SSETypeS3, sseMeta.SSEType)
+	}
+
+	entry, ok := filerImpl.entries[filePath]
+	if !ok {
+		t.Fatalf("entry not found on filer at %s", filePath)
+	}
+	if !bytes.Equal(entry.Extended[s3_constants.AmzServerSideEncryption], []byte("AES256")) {
+		t.Fatal("expected object entry to keep bucket default AES256 encryption")
+	}
+	for i, chunk := range entry.Chunks {
+		if chunk.SseType != filer_pb.SSEType_SSE_S3 {
+			t.Errorf("chunk %d: expected SseType SSE_S3, got %v", i, chunk.SseType)
+		}
+	}
+}

weed/s3api/s3api_object_handlers_copy_part_sse.go

@@ -404,7 +404,7 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption(
 	applyDestChecksumHeaderToCopyRequest(cloned, uploadEntry)
 
 	filePath := s3a.genPartUploadPath(dstBucket, uploadID, partID)
-	tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil, false, s3a.toFilerPath(dstBucket, dstObject))
+	tag, code, putSSE := s3a.putToFiler(withMultipartPart(cloned), filePath, srcReader, dstBucket, "", partID, 0, nil, false, s3a.toFilerPath(dstBucket, dstObject))
 	if code != s3err.ErrNone {
 		return "", SSEResponseMetadata{}, code
 	}

weed/s3api/s3api_object_handlers_multipart.go

@@ -492,7 +492,7 @@ func (s3a *S3ApiServer) PutObjectPartHandler(w http.ResponseWriter, r *http.Requ
 	// start the TTL clock before CompleteMultipartUpload ever assembled
 	// the object. filer.conf storage rules are the opposite case: they
 	// place the bytes the part becomes, so they resolve against the object.
-	etag, errCode, sseMetadata := s3a.putToFiler(r, filePath, dataReader, bucket, "", partID, 0, nil, false, s3a.toFilerPath(bucket, object))
+	etag, errCode, sseMetadata := s3a.putToFiler(withMultipartPart(r), filePath, dataReader, bucket, "", partID, 0, nil, false, s3a.toFilerPath(bucket, object))
 	if errCode != s3err.ErrNone {
 		glog.Errorf("PutObjectPart: putToFiler failed with error code %v for bucket=%s, object=%s, partNumber=%d",
 			errCode, bucket, object, partID)

weed/s3api/s3api_object_handlers_put.go

@@ -511,6 +511,21 @@ func (s3a *S3ApiServer) uploadChunkSize() int32 {
 	return sizeMB * 1024 * 1024
 }
 
+// multipartPartContextKey marks a putToFiler request as writing a staged
+// multipart part. Encryption for the upload is decided at
+// CreateMultipartUpload and replayed through request headers, so bucket
+// default encryption must not be re-evaluated for part writes.
+type multipartPartContextKey struct{}
+
+func withMultipartPart(r *http.Request) *http.Request {
+	return r.WithContext(context.WithValue(r.Context(), multipartPartContextKey{}, true))
+}
+
+func isMultipartPartRequest(r *http.Request) bool {
+	marked, _ := r.Context().Value(multipartPartContextKey{}).(bool)
+	return marked
+}
+
 // putToFiler writes one chunk of object bytes (a full PutObject body, a
 // single MPU part, a copy-part destination). lifecycleTTLSec is non-zero
 // only for top-level PutObject paths where the lifecycle XML's
@@ -590,7 +605,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
 
 	// Apply bucket default encryption if no explicit encryption was provided
 	// This implements AWS S3 behavior where bucket default encryption automatically applies
-	if !hasExplicitEncryption(customerKey, sseKMSKey, sseS3Key) {
+	if !isMultipartPartRequest(r) && !hasExplicitEncryption(customerKey, sseKMSKey, sseS3Key) {
 		glog.V(4).Infof("putToFiler: no explicit encryption detected, checking for bucket default encryption")
 
 		// Apply bucket default encryption and get the result