diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 549c36aae..860b7f654 100644 --- a/weed/operation/upload_content.go +++ b/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 diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index e1f3558ba..8923d33e5 100644 --- a/weed/operation/upload_content_test.go +++ b/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)) + } +} diff --git a/weed/s3api/s3api_encrypted_volume_copy_test.go b/weed/s3api/s3api_encrypted_volume_copy_test.go index 169a0f616..ad53a261c 100644 --- a/weed/s3api/s3api_encrypted_volume_copy_test.go +++ b/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) + } + }) + } +} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index a24758b82..0338af5ad 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/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 ) diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index f165c04ac..775852c35 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/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 } diff --git a/weed/s3api/s3api_object_handlers_copy_etag_test.go b/weed/s3api/s3api_object_handlers_copy_etag_test.go index 11036cd3e..e49559e86 100644 --- a/weed/s3api/s3api_object_handlers_copy_etag_test.go +++ b/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) + } + }) + } +} diff --git a/weed/s3api/s3api_object_handlers_copy_part_sse.go b/weed/s3api/s3api_object_handlers_copy_part_sse.go index a7887d850..1bcc05a25 100644 --- a/weed/s3api/s3api_object_handlers_copy_part_sse.go +++ b/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) }