s3: fix UploadPartCopy with volume-data encryption (#10971)

* operation: give an encrypted chunk the plaintext ETag

With -encryptVolumeData the volume server stores ciphertext, so it cannot
echo a Content-MD5 back and the chunk lands with an empty ETag. Every ETag
derived from those chunks then comes out empty for a single chunk, or
d41d8cd98f00b204e9800998ecf8427e-N for several.

The caller already hashes the plaintext to send as Content-MD5, so keep that
digest as the chunk ETag instead of dropping it, and compute it for a
WantMd5 caller under cipher too.

* s3: re-encrypt a part copy from a volume-encrypted source

UploadPartCopy raw-copies source chunks when neither side uses SSE, which
also caught -encryptVolumeData sources. Those chunks are ciphertext a
whole-chunk cipher key decrypts, so copying a byte range out of one and
keeping the key leaves a destination that fails authentication on GET, and
the copied chunks carry no ETag for the part result to report.

Route them through the re-encrypting path already used for SSE: it reads the
source as plaintext, hashes the part, and writes the destination under the
gateway's own encryption.

* s3: fetch only the range a part copy asked for

The re-encrypting UploadPartCopy path opened the source at offset 0 and threw
the prefix away, so assembling an object part by part read the source once per
part. Now that volume-encrypted sources take this path too, that is the common
case rather than an SSE corner.

The chunk stream already seeks, so hand it the range.

* s3: reject an unsatisfiable copy-source-range

A part copy has no way to report a short part, so a range reaching past the
source cannot be clamped the way a GET clamps one. The fast path silently
produced a part shorter than asked for, or an empty one; the re-encrypting
path pads with zeros, so a 2 MiB source copied as bytes=1048576-9999999 came
back as 1 MiB of data followed by 7.5 MiB of nothing.

Answer InvalidRange instead, which is what s3-tests'
test_multipart_copy_invalid_range expects.

weed/operation/upload_content.go

@@ -250,8 +250,9 @@ func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assi
 
 	// Hash the buffer we already hold so the server echoes Content-MD5 back as
 	// the chunk ETag (std-base64 of the raw digest, the form ParseUpload
-	// verifies). Never under cipher: the server sees only ciphertext.
-	if uploadOption.WantMd5 && uploadOption.Md5 == "" && !uploadOption.Cipher {
+	// verifies). Under cipher the server never sees the header, but the digest
+	// still becomes the chunk ETag.
+	if uploadOption.WantMd5 && uploadOption.Md5 == "" {
 		digest := md5.Sum(data)
 		uploadOption.Md5 = base64.StdEncoding.EncodeToString(digest[:])
 	}
@@ -412,6 +413,9 @@ func (uploader *Uploader) doUploadData(ctx context.Context, data []byte, option
 		uploadResult.Name = option.Filename
 		uploadResult.Mime = option.MimeType
 		uploadResult.CipherKey = cipherKey
+		// The volume server only ever hashes the ciphertext it stored, so the
+		// chunk ETag has to come from the caller's plaintext digest.
+		uploadResult.ContentMd5 = option.Md5
 		uploadResult.Size = uint32(clearDataLen)
 		if contentIsGzipped {
 			uploadResult.Gzip = 1

weed/operation/upload_content_test.go

@@ -3,6 +3,8 @@ package operation
 import (
 	"bytes"
 	"context"
+	"crypto/md5"
+	"encoding/base64"
 	"errors"
 	"fmt"
 	"io"
@@ -16,6 +18,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"google.golang.org/grpc"
 )
 
@@ -455,3 +458,60 @@ func TestUploadRetryStopsOnContextCancel(t *testing.T) {
 		t.Fatalf("dial attempts = %d, want at most 1 after cancellation", got)
 	}
 }
+
+// A ciphered upload hands the volume server ciphertext, so it cannot echo a
+// Content-MD5 back; without the caller's plaintext digest the chunk lands with
+// no ETag and every ETag computed from those chunks comes out empty
+// (issue #10968).
+func TestCipherUploadKeepsPlaintextETag(t *testing.T) {
+	payload := bytes.Repeat([]byte("encrypt me\n"), 4096)
+	digest := md5.Sum(payload)
+	wantMd5 := base64.StdEncoding.EncodeToString(digest[:])
+
+	var storedNeedle *needle.Needle
+	var sentContentMd5 string
+	var parseErr error
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		sentContentMd5 = r.Header.Get("Content-MD5")
+		storedNeedle, _, _, parseErr = needle.CreateNeedleFromRequest(r, false, 1024*1024, &bytes.Buffer{})
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusCreated)
+		_, _ = io.WriteString(w, `{"name":"payload","size":45056}`)
+	}))
+	defer server.Close()
+
+	uploader := newUploader(server.Client())
+	result, err := uploader.UploadData(context.Background(), payload, &UploadOption{
+		UploadUrl:   server.URL + "/3,01637037d6",
+		Cipher:      true,
+		Md5:         wantMd5,
+		MaxAttempts: 1,
+	})
+	if err != nil {
+		t.Fatalf("cipher upload failed: %v", err)
+	}
+	if parseErr != nil {
+		t.Fatalf("parse ciphered upload: %v", parseErr)
+	}
+	if sentContentMd5 != "" {
+		t.Fatalf("Content-MD5 %q sent to the volume server, which only sees ciphertext", sentContentMd5)
+	}
+	if result.ContentMd5 != wantMd5 {
+		t.Fatalf("chunk ETag = %q, want the plaintext digest %q", result.ContentMd5, wantMd5)
+	}
+	if bytes.Equal(storedNeedle.Data, payload) {
+		t.Fatal("volume server stored plaintext for a ciphered upload")
+	}
+	decrypted, err := util.Decrypt(storedNeedle.Data, util.CipherKey(result.CipherKey))
+	if err != nil {
+		t.Fatalf("decrypt stored needle: %v", err)
+	}
+	if result.Gzip > 0 {
+		if decrypted, err = util.DecompressData(decrypted); err != nil {
+			t.Fatalf("decompress stored needle: %v", err)
+		}
+	}
+	if !bytes.Equal(decrypted, payload) {
+		t.Fatalf("decrypted %d bytes, want %d", len(decrypted), len(payload))
+	}
+}

weed/s3api/s3api_encrypted_volume_copy_test.go

@@ -2,6 +2,8 @@ package s3api
 
 import (
 	"bytes"
+	"context"
+	"io"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -173,3 +175,96 @@ func TestEncryptedVolumeCopyScenario(t *testing.T) {
 		t.Log("✓ All chunk metadata properly preserved for encrypted volume copy scenario")
 	})
 }
+
+// A volume-encrypted source must take the re-encrypting UploadPartCopy path:
+// the raw chunk copy slices ciphertext the destination's whole-chunk cipher key
+// can no longer decrypt, and reports the part's ETag from chunks that carry
+// none (issue #10968).
+func TestSourceEntryIsEncryptedForVolumeCipher(t *testing.T) {
+	testCases := []struct {
+		name  string
+		entry *filer_pb.Entry
+		want  bool
+	}{
+		{
+			name:  "nil entry",
+			entry: nil,
+		},
+		{
+			name: "plaintext chunks",
+			entry: &filer_pb.Entry{Chunks: []*filer_pb.FileChunk{
+				{FileId: "1,abc123", Size: 1024, ETag: "etag1"},
+			}},
+		},
+		{
+			name: "volume-encrypted chunk",
+			entry: &filer_pb.Entry{Chunks: []*filer_pb.FileChunk{
+				{FileId: "1,abc123", Size: 1024, CipherKey: util.GenCipherKey()},
+			}},
+			want: true,
+		},
+		{
+			name: "volume-encrypted second chunk",
+			entry: &filer_pb.Entry{Chunks: []*filer_pb.FileChunk{
+				{FileId: "1,abc123", Size: 1024, ETag: "etag1"},
+				{FileId: "2,def456", Offset: 1024, Size: 1024, CipherKey: util.GenCipherKey()},
+			}},
+			want: true,
+		},
+		{
+			name: "SSE-S3 chunk",
+			entry: &filer_pb.Entry{Chunks: []*filer_pb.FileChunk{
+				{FileId: "1,abc123", Size: 1024, SseType: filer_pb.SSEType_SSE_S3},
+			}},
+			want: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := sourceEntryIsEncrypted(tc.entry); got != tc.want {
+				t.Errorf("sourceEntryIsEncrypted = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// A ranged part copy asks the chunk stream for its slice; reading the whole
+// object and discarding the prefix would make an N-part copy read the source
+// N/2 times over.
+func TestGetEncryptedStreamFromVolumesRangesInlineContent(t *testing.T) {
+	s3a := &S3ApiServer{}
+	entry := &filer_pb.Entry{Content: []byte("0123456789")}
+
+	testCases := []struct {
+		name   string
+		offset int64
+		size   int64
+		want   string
+	}{
+		{name: "whole content", size: 10, want: "0123456789"},
+		{name: "leading slice", size: 4, want: "0123"},
+		{name: "middle slice", offset: 3, size: 4, want: "3456"},
+		{name: "trailing slice", offset: 6, size: 4, want: "6789"},
+		{name: "size past the end", offset: 8, size: 10, want: "89"},
+		{name: "offset past the end", offset: 10, size: 4},
+		{name: "empty range", size: 0},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			reader, err := s3a.getEncryptedStreamFromVolumes(context.Background(), entry, tc.offset, tc.size)
+			if err != nil {
+				t.Fatalf("getEncryptedStreamFromVolumes: %v", err)
+			}
+			defer reader.Close()
+			got, err := io.ReadAll(reader)
+			if err != nil {
+				t.Fatalf("read: %v", err)
+			}
+			if string(got) != tc.want {
+				t.Errorf("got %q, want %q", got, tc.want)
+			}
+		})
+	}
+}

weed/s3api/s3api_object_handlers.go

@@ -1542,7 +1542,7 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 		} else {
 			// For single-part, get encrypted stream and decrypt
 			tStreamFetch := time.Now()
-			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry)
+			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry, 0, int64(filer.FileSize(entry)))
 			streamFetchTime = time.Since(tStreamFetch)
 			if streamErr != nil {
 				return streamErr
@@ -1578,7 +1578,7 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 		} else {
 			// For single-part, get encrypted stream and decrypt
 			tStreamFetch := time.Now()
-			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry)
+			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry, 0, int64(filer.FileSize(entry)))
 			streamFetchTime = time.Since(tStreamFetch)
 			if streamErr != nil {
 				return streamErr
@@ -1610,7 +1610,7 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 		} else {
 			// For single-part, get encrypted stream and decrypt
 			tStreamFetch := time.Now()
-			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry)
+			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry, 0, int64(filer.FileSize(entry)))
 			streamFetchTime = time.Since(tStreamFetch)
 			if streamErr != nil {
 				return streamErr
@@ -2107,25 +2107,35 @@ func (s3a *S3ApiServer) fetchChunkViewData(ctx context.Context, chunkView *filer
 	return resp.Body, nil
 }
 
-// getEncryptedStreamFromVolumes gets raw encrypted data stream from volume servers
-func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry *filer_pb.Entry) (io.ReadCloser, error) {
+// getEncryptedStreamFromVolumes gets a raw encrypted data stream from volume
+// servers for [offset, offset+size) of the entry.
+func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry *filer_pb.Entry, offset, size int64) (io.ReadCloser, error) {
+	if size <= 0 {
+		return io.NopCloser(bytes.NewReader(nil)), nil
+	}
+
 	// Handle inline content
 	if len(entry.Content) > 0 {
-		return io.NopCloser(bytes.NewReader(entry.Content)), nil
+		content := entry.Content
+		if offset >= int64(len(content)) {
+			content = nil
+		} else {
+			content = content[offset:min(offset+size, int64(len(content)))]
+		}
+		return io.NopCloser(bytes.NewReader(content)), nil
 	}
 
 	// Handle empty files
 	chunks := entry.GetChunks()
 	if len(chunks) == 0 {
-		return io.NopCloser(bytes.NewReader([]byte{})), nil
+		return io.NopCloser(bytes.NewReader(nil)), nil
 	}
 
 	// Reuse shared lookup function to keep volume lookup logic in one place
 	lookupFileIdFn := s3a.createLookupFileIdFunction()
 
 	// Resolve chunks
-	totalSize := int64(filer.FileSize(entry))
-	resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, chunks, 0, totalSize)
+	resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, chunks, offset, offset+size)
 	if err != nil {
 		return nil, err
 	}
@@ -2136,8 +2146,8 @@ func (s3a *S3ApiServer) getEncryptedStreamFromVolumes(ctx context.Context, entry
 		s3a.filerClient,
 		filer.JwtForVolumeServer, // Use filer's JWT function (loads config once, generates JWT locally)
 		resolvedChunks,
-		0,
-		totalSize,
+		offset,
+		size,
 		0,
 		4, // prefetch 4 chunks ahead for overlapped fetching
 	)

weed/s3api/s3api_object_handlers_copy.go

@@ -974,7 +974,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 	rangeHeader := r.Header.Get("x-amz-copy-source-range")
 	var startOffset, endOffset int64
 	if rangeHeader != "" {
-		startOffset, endOffset, err = parseRangeHeader(rangeHeader)
+		startOffset, endOffset, err = parseRangeHeader(rangeHeader, int64(filer.FileSize(entry)))
 		if err != nil {
 			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRange)
 			return
@@ -1022,7 +1022,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	if uploadEntryHasSSE(uploadEntry) || sourceEntryHasSSE(entry) || uploadEntryHasChecksum(uploadEntry) {
+	if uploadEntryHasSSE(uploadEntry) || sourceEntryIsEncrypted(entry) || uploadEntryHasChecksum(uploadEntry) {
 		etag, sseMetadata, errCode := s3a.copyObjectPartViaReencryption(r, entry, startOffset, endOffset, dstBucket, dstObject, uploadID, partID, uploadEntry)
 		if errCode != s3err.ErrNone {
 			s3err.WriteErrorResponse(w, r, errCode)
@@ -1444,8 +1444,10 @@ func (s3a *S3ApiServer) assignNewVolume(dstPath string, expectedDataSize uint64)
 	return assignResult, nil
 }
 
-// parseRangeHeader parses the x-amz-copy-source-range header
-func parseRangeHeader(rangeHeader string) (startOffset, endOffset int64, err error) {
+// parseRangeHeader parses the x-amz-copy-source-range header against a source of
+// fileSize bytes. Unlike a GET, a part copy does not clamp: a range reaching past
+// the source is unsatisfiable, and copying it would pad the part with zeros.
+func parseRangeHeader(rangeHeader string, fileSize int64) (startOffset, endOffset int64, err error) {
 	// Remove "bytes=" prefix if present
 	rangeStr := strings.TrimPrefix(rangeHeader, "bytes=")
 	parts := strings.Split(rangeStr, "-")
@@ -1463,6 +1465,10 @@ func parseRangeHeader(rangeHeader string) (startOffset, endOffset int64, err err
 		return 0, 0, fmt.Errorf("invalid end offset: %w", err)
 	}
 
+	if startOffset < 0 || endOffset < startOffset || endOffset >= fileSize {
+		return 0, 0, fmt.Errorf("range %s is not satisfiable for a %d byte source", rangeHeader, fileSize)
+	}
+
 	return startOffset, endOffset, nil
 }
 

weed/s3api/s3api_object_handlers_copy_etag_test.go

@@ -63,3 +63,48 @@ func newCopyETagTestEntry(t *testing.T, storedETag, computedETag string) *filer_
 	}
 	return entry
 }
+
+// A part copy has no way to report a short part, so an unsatisfiable
+// x-amz-copy-source-range has to be rejected rather than clamped like a GET —
+// the re-encrypting path would otherwise pad the part out with zeros.
+func TestParseRangeHeaderRejectsUnsatisfiableRange(t *testing.T) {
+	const fileSize = 2048
+
+	testCases := []struct {
+		name       string
+		header     string
+		wantStart  int64
+		wantEnd    int64
+		wantReject bool
+	}{
+		{name: "whole source", header: "bytes=0-2047", wantEnd: 2047},
+		{name: "leading slice", header: "bytes=0-1023", wantEnd: 1023},
+		{name: "trailing slice", header: "bytes=1024-2047", wantStart: 1024, wantEnd: 2047},
+		{name: "single byte", header: "bytes=7-7", wantStart: 7, wantEnd: 7},
+		{name: "no bytes prefix", header: "0-1023", wantEnd: 1023},
+		{name: "end past the source", header: "bytes=0-2048", wantReject: true},
+		{name: "start past the source", header: "bytes=4096-8191", wantReject: true},
+		{name: "start at the source size", header: "bytes=2048-2048", wantReject: true},
+		{name: "reversed", header: "bytes=1023-0", wantReject: true},
+		{name: "negative start", header: "bytes=-1-1023", wantReject: true},
+		{name: "malformed", header: "bytes=abc", wantReject: true},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			start, end, err := parseRangeHeader(tc.header, fileSize)
+			if tc.wantReject {
+				if err == nil {
+					t.Fatalf("parseRangeHeader(%q) = %d, %d, want an error", tc.header, start, end)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("parseRangeHeader(%q): %v", tc.header, err)
+			}
+			if start != tc.wantStart || end != tc.wantEnd {
+				t.Errorf("parseRangeHeader(%q) = %d, %d, want %d, %d", tc.header, start, end, tc.wantStart, tc.wantEnd)
+			}
+		})
+	}
+}

weed/s3api/s3api_object_handlers_copy_part_sse.go

@@ -14,6 +14,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
@@ -73,15 +74,17 @@ func uploadEntryHasChecksum(uploadEntry *filer_pb.Entry) bool {
 	return checksumAlgorithmFromHeaderName(headerName) != ChecksumAlgorithmNone
 }
 
-// sourceEntryHasSSE reports whether the source object's chunks are SSE
+// sourceEntryIsEncrypted reports whether the source object's chunks are
 // ciphertext on disk and therefore cannot be raw-copied — they must be
-// decrypted on read.
-func sourceEntryHasSSE(srcEntry *filer_pb.Entry) bool {
+// decrypted on read. That is SSE, and also -encryptVolumeData, whose per-chunk
+// key only ever decrypts a whole chunk and whose chunks carry no ETag for the
+// copied part to report.
+func sourceEntryIsEncrypted(srcEntry *filer_pb.Entry) bool {
 	if srcEntry == nil {
 		return false
 	}
 	for _, c := range srcEntry.GetChunks() {
-		if c.GetSseType() != filer_pb.SSEType_NONE {
+		if c.GetSseType() != filer_pb.SSEType_NONE || len(c.GetCipherKey()) > 0 {
 			return true
 		}
 	}
@@ -150,12 +153,13 @@ func (s3a *S3ApiServer) openSourcePlaintextReader(
 	case s3_constants.SSETypeC:
 		return nil, fmt.Errorf("%w: UploadPartCopy from SSE-C source", errCopySourceSSEUnsupported)
 	default:
-		// Unencrypted source: stream raw bytes and apply range.
-		raw, err := s3a.getEncryptedStreamFromVolumes(ctx, srcEntry)
+		// Plaintext or volume-encrypted source: the chunk stream seeks, so ask
+		// it for the range rather than reading and discarding the prefix.
+		raw, err := s3a.getEncryptedStreamFromVolumes(ctx, srcEntry, startOffset, sliceLen)
 		if err != nil {
-			return nil, fmt.Errorf("open unencrypted source: %w", err)
+			return nil, fmt.Errorf("open source: %w", err)
 		}
-		return applyRange(raw, startOffset, sliceLen)
+		return raw, nil
 	}
 }
 
@@ -221,7 +225,7 @@ func (s3a *S3ApiServer) openSSES3SourcePlaintextReader(
 	if err != nil {
 		return nil, fmt.Errorf("get SSE-S3 IV: %w", err)
 	}
-	encStream, err := s3a.getEncryptedStreamFromVolumes(ctx, srcEntry)
+	encStream, err := s3a.getEncryptedStreamFromVolumes(ctx, srcEntry, 0, int64(filer.FileSize(srcEntry)))
 	if err != nil {
 		return nil, fmt.Errorf("open ciphertext source: %w", err)
 	}