From 8a68337256b4160e40f8f04689a5bcf8a4532c06 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 5 Sep 2026 10:16:17 -0700 Subject: [PATCH] filer: pack SSE chunks into manifests (#11175) * filer: pack SSE chunks into manifests * s3: resolve encrypted manifests before reads * s3: scope encrypted manifest resolution to ranges --- weed/filer/filechunk_manifest.go | 7 - .../filer/filechunk_manifest_rollback_test.go | 14 -- weed/filer/filechunk_manifest_test.go | 41 ++++ weed/s3api/s3api_chunk_manifest_test.go | 206 +++++++++++++++++- weed/s3api/s3api_object_handlers.go | 62 ++---- 5 files changed, 266 insertions(+), 64 deletions(-) diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index 73dc79751..35fdc5063 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -232,13 +232,6 @@ func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrin // it returns references none of them. deleteChunks may be nil where the caller // has no deleter to offer; then the blobs are only named in the log. func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, deleteChunks func([]*filer_pb.FileChunk), inputChunks []*filer_pb.FileChunk) (chunks []*filer_pb.FileChunk, err error) { - // Don't manifestize SSE-encrypted chunks to preserve per-chunk metadata - for _, chunk := range inputChunks { - if chunk.GetSseType() != 0 { // Any SSE type (SSE-C or SSE-KMS) - return inputChunks, nil - } - } - var saved []*filer_pb.FileChunk record := func(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) { chunk, saveErr := saveFunc(reader, name, offset, tsNs, expectedDataSize) diff --git a/weed/filer/filechunk_manifest_rollback_test.go b/weed/filer/filechunk_manifest_rollback_test.go index 61c899480..7dcf07217 100644 --- a/weed/filer/filechunk_manifest_rollback_test.go +++ b/weed/filer/filechunk_manifest_rollback_test.go @@ -75,20 +75,6 @@ func TestMaybeManifestizeBelowThreshold(t *testing.T) { } } -func TestMaybeManifestizeSkipsSse(t *testing.T) { - store := &fakeManifestStore{} - chunks := flatTestChunks(ManifestBatch + 50) - chunks[0].SseType = filer_pb.SSEType_SSE_S3 - - result, err := MaybeManifestize(store.save, store.delete, chunks) - if err != nil { - t.Fatalf("MaybeManifestize: %v", err) - } - if len(result) != len(chunks) || store.saves != 0 { - t.Fatalf("SSE chunks must not be folded, got %d chunks, %d saves", len(result), store.saves) - } -} - // A fold that fails midway must hand back the caller's own list and delete the // manifest blobs its earlier batches already uploaded. func TestMaybeManifestizeRollsBackPartialFold(t *testing.T) { diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index 9e51722f4..2036a8d11 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -243,6 +243,47 @@ func TestManifestRoundTripPreservesChunks(t *testing.T) { } } +func TestMaybeManifestizePreservesSSEMetadata(t *testing.T) { + for _, sseType := range []filer_pb.SSEType{ + filer_pb.SSEType_SSE_C, + filer_pb.SSEType_SSE_KMS, + filer_pb.SSEType_SSE_S3, + } { + t.Run(sseType.String(), func(t *testing.T) { + store := newTestManifestStore() + chunks := make([]*filer_pb.FileChunk, ManifestBatch+1) + for i := range chunks { + chunks[i] = testChunk(1, uint64(i+1), uint32(i+1), int64(i*8), 8, int64(i+1)) + chunks[i].SseType = sseType + chunks[i].SseMetadata = []byte(fmt.Sprintf("metadata-%d", i)) + } + + packed, err := MaybeManifestize(store.saveFunc(), nil, chunks) + if err != nil { + t.Fatal(err) + } + manifests, remainder := SeparateManifestChunks(packed) + if len(manifests) != 1 || len(remainder) != 1 { + t.Fatalf("got %d manifests and %d remaining chunks", len(manifests), len(remainder)) + } + + resolved, err := store.resolve(manifests[0]) + if err != nil { + t.Fatal(err) + } + for i, chunk := range resolved { + if chunk.GetSseType() != sseType { + t.Fatalf("chunk %d SSE type = %s, want %s", i, chunk.GetSseType(), sseType) + } + wantMetadata := fmt.Sprintf("metadata-%d", i) + if string(chunk.GetSseMetadata()) != wantMetadata { + t.Fatalf("chunk %d SSE metadata = %q, want %q", i, chunk.GetSseMetadata(), wantMetadata) + } + } + }) + } +} + // --------------------------------------------------------------------------- // Compact resolved overlapping manifests: older sub-chunks become garbage // --------------------------------------------------------------------------- diff --git a/weed/s3api/s3api_chunk_manifest_test.go b/weed/s3api/s3api_chunk_manifest_test.go index 5035b5278..6ce26f8f3 100644 --- a/weed/s3api/s3api_chunk_manifest_test.go +++ b/weed/s3api/s3api_chunk_manifest_test.go @@ -1,11 +1,214 @@ package s3api import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync/atomic" "testing" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/proto" + + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/wdclient" + "github.com/stretchr/testify/require" ) +type manifestVolumeFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + volumeServer string +} + +func (f *manifestVolumeFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { + locations := make(map[string]*filer_pb.Locations, len(req.VolumeIds)) + for _, volumeID := range req.VolumeIds { + locations[volumeID] = &filer_pb.Locations{Locations: []*filer_pb.Location{{Url: f.volumeServer}}} + } + return &filer_pb.LookupVolumeResponse{LocationsMap: locations}, nil +} + +func TestDetectPrimarySSETypeFromManifestedEntry(t *testing.T) { + s3a := &S3ApiServer{} + manifest := &filer_pb.FileChunk{IsChunkManifest: true} + for _, test := range []struct { + name string + extended map[string][]byte + want string + }{ + { + name: "SSE-C", + extended: map[string][]byte{ + s3_constants.AmzServerSideEncryptionCustomerAlgorithm: []byte(s3_constants.SSEAlgorithmAES256), + }, + want: s3_constants.SSETypeC, + }, + { + name: "SSE-KMS", + extended: map[string][]byte{ + s3_constants.AmzServerSideEncryption: []byte(s3_constants.SSEAlgorithmKMS), + }, + want: s3_constants.SSETypeKMS, + }, + { + name: "SSE-S3", + extended: map[string][]byte{ + s3_constants.AmzServerSideEncryption: []byte(s3_constants.SSEAlgorithmAES256), + }, + want: s3_constants.SSETypeS3, + }, + } { + t.Run(test.name, func(t *testing.T) { + entry := &filer_pb.Entry{Chunks: []*filer_pb.FileChunk{manifest}, Extended: test.extended} + require.Equal(t, test.want, s3a.detectPrimarySSEType(entry)) + }) + } +} + +func TestSSECReadsResolveChunkManifests(t *testing.T) { + keyPair := GenerateTestSSECKey(9) + customerKey := &SSECustomerKey{Algorithm: s3_constants.SSEAlgorithmAES256, Key: keyPair.Key, KeyMD5: keyPair.KeyMD5} + parts := [][]byte{[]byte("first encrypted part"), []byte("second encrypted part")} + objects := make(map[string][]byte) + chunks := make([]*filer_pb.FileChunk, 0, len(parts)) + var plaintext []byte + var firstIV []byte + var offset int64 + for i, part := range parts { + encrypted, iv, err := CreateSSECEncryptedReader(bytes.NewReader(part), customerKey) + require.NoError(t, err) + ciphertext, err := io.ReadAll(encrypted) + require.NoError(t, err) + metadata, err := SerializeSSECMetadata(iv, keyPair.KeyMD5, 0) + require.NoError(t, err) + + chunk := &filer_pb.FileChunk{ + Fid: &filer_pb.FileId{VolumeId: uint32(8 + i), FileKey: 1, Cookie: 1}, + Offset: offset, + Size: uint64(len(part)), + SseType: filer_pb.SSEType_SSE_C, + SseMetadata: metadata, + } + objects[chunk.GetFileIdString()] = ciphertext + chunks = append(chunks, chunk) + plaintext = append(plaintext, part...) + offset += int64(len(part)) + if i == 0 { + firstIV = iv + } + } + + manifests := make([]*filer_pb.FileChunk, 0, len(chunks)) + for i, chunk := range chunks { + serializedChunks := []*filer_pb.FileChunk{proto.Clone(chunk).(*filer_pb.FileChunk)} + filer_pb.BeforeEntrySerialization(serializedChunks) + manifestData, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: serializedChunks}) + require.NoError(t, err) + manifest := &filer_pb.FileChunk{ + Fid: &filer_pb.FileId{VolumeId: uint32(17 + i), FileKey: 1, Cookie: 1}, + Offset: chunk.Offset, + Size: chunk.Size, + IsChunkManifest: true, + } + objects[manifest.GetFileIdString()] = manifestData + manifests = append(manifests, manifest) + } + + var failFirstManifest atomic.Bool + volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fileID := strings.TrimPrefix(r.URL.Path, "/") + data, found := objects[fileID] + if fileID == manifests[0].GetFileIdString() && failFirstManifest.Load() { + found = false + } + if !found { + http.NotFound(w, r) + return + } + _, _ = w.Write(data) + })) + t.Cleanup(volumeServer.Close) + filerAddress := startFakeFiler(t, &manifestVolumeFiler{volumeServer: strings.TrimPrefix(volumeServer.URL, "http://")}) + filerClient := wdclient.NewFilerClient( + []pb.ServerAddress{filerAddress}, + grpc.WithTransportCredentials(insecure.NewCredentials()), + "", + ) + t.Cleanup(filerClient.Close) + s3a := &S3ApiServer{option: &S3ApiServerOption{}, filerClient: filerClient} + + newEntry := func() *filer_pb.Entry { + entryManifests := make([]*filer_pb.FileChunk, len(manifests)) + for i, manifest := range manifests { + entryManifests[i] = proto.Clone(manifest).(*filer_pb.FileChunk) + } + return &filer_pb.Entry{ + Name: "object", + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(plaintext))}, + Chunks: entryManifests, + Extended: map[string][]byte{ + s3_constants.AmzServerSideEncryptionCustomerAlgorithm: []byte(s3_constants.SSEAlgorithmAES256), + s3_constants.AmzServerSideEncryptionCustomerKeyMD5: []byte(keyPair.KeyMD5), + s3_constants.SeaweedFSSSEIV: firstIV, + }, + } + } + newRequest := func() *http.Request { + r := httptest.NewRequest(http.MethodGet, "/bucket/object", nil) + SetupTestSSECHeaders(r, keyPair) + return r + } + + t.Run("full object", func(t *testing.T) { + entry := newEntry() + sseType := s3a.detectPrimarySSEType(entry) + w := httptest.NewRecorder() + err := s3a.streamFromVolumeServersWithSSE(w, newRequest(), entry, sseType, "bucket", "object", "") + require.NoError(t, err) + require.Equal(t, plaintext, w.Body.Bytes()) + }) + + t.Run("range", func(t *testing.T) { + failFirstManifest.Store(true) + entry := newEntry() + sseType := s3a.detectPrimarySSEType(entry) + r := newRequest() + start, end := len(parts[0])+2, len(parts[0])+8 + r.Header.Set("Range", "bytes="+strconv.Itoa(start)+"-"+strconv.Itoa(end)) + w := httptest.NewRecorder() + err := s3a.streamFromVolumeServersWithSSE(w, r, entry, sseType, "bucket", "object", "") + require.NoError(t, err) + require.Equal(t, plaintext[start:end+1], w.Body.Bytes()) + }) + + t.Run("invalid range", func(t *testing.T) { + entry := newEntry() + r := newRequest() + r.Header.Set("Range", "bytes="+strconv.Itoa(len(plaintext))+"-") + w := httptest.NewRecorder() + err := s3a.streamFromVolumeServersWithSSE(w, r, entry, s3a.detectPrimarySSEType(entry), "bucket", "object", "") + require.Error(t, err) + require.Equal(t, http.StatusRequestedRangeNotSatisfiable, w.Code) + }) + + t.Run("wrong key", func(t *testing.T) { + entry := newEntry() + r := httptest.NewRequest(http.MethodGet, "/bucket/object", nil) + SetupTestSSECHeaders(r, GenerateTestSSECKey(10)) + w := httptest.NewRecorder() + err := s3a.streamFromVolumeServersWithSSE(w, r, entry, s3a.detectPrimarySSEType(entry), "bucket", "object", "") + require.Error(t, err) + require.Equal(t, http.StatusForbidden, w.Code) + }) +} + func TestPartRange(t *testing.T) { chunks := []*filer_pb.FileChunk{ {FileId: "1,a", Offset: 0, Size: 8}, @@ -13,19 +216,16 @@ func TestPartRange(t *testing.T) { {FileId: "1,c", Offset: 16, Size: 24}, } - // Byte offsets win even when the chunk indexes no longer match the list. start, end, ok := partRange(&PartBoundaryInfo{StartChunk: 40, EndChunk: 80, StartOffset: 16, EndOffset: 40}, chunks) if !ok || start != 16 || end != 39 { t.Errorf("offset boundary: got [%d,%d] ok=%v, want [16,39]", start, end, ok) } - // Legacy record: range from chunk indexes. start, end, ok = partRange(&PartBoundaryInfo{StartChunk: 1, EndChunk: 3}, chunks) if !ok || start != 8 || end != 39 { t.Errorf("legacy boundary: got [%d,%d] ok=%v, want [8,39]", start, end, ok) } - // Legacy record with indexes off the list must report, not panic. for _, b := range []*PartBoundaryInfo{ {StartChunk: 2, EndChunk: 9}, {StartChunk: -1, EndChunk: 2}, diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 58d0eb3e4..142665c66 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -1458,6 +1458,9 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r body.Commit() return nil } + if _, err := s3a.flattenManifestChunks(r.Context(), entry); err != nil { + return fmt.Errorf("resolve encrypted chunk manifests: %w", err) + } // Full object path: Optimize multipart vs single-part var decryptedReader io.Reader @@ -1613,9 +1616,12 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r // This implements the filer's ViewFromChunks approach for optimal range performance // Returns the number of bytes written and any error func (s3a *S3ApiServer) streamDecryptedRangeFromChunks(ctx context.Context, w io.Writer, entry *filer_pb.Entry, offset int64, size int64, sseType string, decryptionKey interface{}) (int64, error) { - // Use filer's ViewFromChunks to resolve only needed chunks for the range lookupFileIdFn := s3a.createLookupFileIdFunction() - chunkViews := filer.ViewFromChunks(ctx, lookupFileIdFn, entry.GetChunks(), offset, size) + resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, entry.GetChunks(), offset, offset+size, s3a.filerClient) + if err != nil { + return 0, err + } + chunkViews := filer.ViewFromChunks(ctx, nil, resolvedChunks, offset, size) totalWritten := int64(0) targetOffset := offset @@ -1637,7 +1643,7 @@ func (s3a *S3ApiServer) streamDecryptedRangeFromChunks(ctx context.Context, w io // Find the corresponding FileChunk for this chunkView var fileChunk *filer_pb.FileChunk - for _, chunk := range entry.GetChunks() { + for _, chunk := range resolvedChunks { if chunk.GetFileIdString() == chunkView.FileId { fileChunk = chunk break @@ -2653,48 +2659,24 @@ func (s3a *S3ApiServer) addObjectLockHeadersToResponse(w http.ResponseWriter, en // detectPrimarySSEType determines the primary SSE type by examining chunk metadata func (s3a *S3ApiServer) detectPrimarySSEType(entry *filer_pb.Entry) string { - // Safety check: handle nil entry if entry == nil { return "None" } - if len(entry.GetChunks()) == 0 { - // No chunks - check object-level metadata only (single objects or smallContent) - hasSSEC := entry.Extended[s3_constants.AmzServerSideEncryptionCustomerAlgorithm] != nil - hasSSEKMS := entry.Extended[s3_constants.AmzServerSideEncryption] != nil - - // Check for SSE-S3: algorithm is AES256 but no customer key - if hasSSEKMS && !hasSSEC { - // Distinguish SSE-S3 from SSE-KMS: check the algorithm value and the presence of a KMS key ID - sseAlgo := string(entry.Extended[s3_constants.AmzServerSideEncryption]) - switch sseAlgo { - case s3_constants.SSEAlgorithmAES256: - // Could be SSE-S3 or SSE-KMS, check for KMS key ID - if _, hasKMSKey := entry.Extended[s3_constants.AmzServerSideEncryptionAwsKmsKeyId]; hasKMSKey { - return s3_constants.SSETypeKMS - } - // No KMS key, this is SSE-S3 - return s3_constants.SSETypeS3 - case s3_constants.SSEAlgorithmKMS: - return s3_constants.SSETypeKMS - default: - // Unknown or unsupported algorithm - return "None" - } - } else if hasSSEC && !hasSSEKMS { - return s3_constants.SSETypeC - } else if hasSSEC && hasSSEKMS { - // Both present - this should only happen during cross-encryption copies - // Use content to determine actual encryption state - if len(entry.Content) > 0 { - // smallContent - check if it's encrypted (heuristic: random-looking data) - return s3_constants.SSETypeC // Default to SSE-C for mixed case - } else { - // No content, both headers - default to SSE-C - return s3_constants.SSETypeC + metadataType := "None" + hasSSEC := entry.Extended[s3_constants.AmzServerSideEncryptionCustomerAlgorithm] != nil + if hasSSEC { + metadataType = s3_constants.SSETypeC + } else { + switch string(entry.Extended[s3_constants.AmzServerSideEncryption]) { + case s3_constants.SSEAlgorithmAES256: + metadataType = s3_constants.SSETypeS3 + if _, hasKMSKey := entry.Extended[s3_constants.AmzServerSideEncryptionAwsKmsKeyId]; hasKMSKey { + metadataType = s3_constants.SSETypeKMS } + case s3_constants.SSEAlgorithmKMS: + metadataType = s3_constants.SSETypeKMS } - return "None" } // Count chunk types to determine primary (multipart objects) @@ -2735,7 +2717,7 @@ func (s3a *S3ApiServer) detectPrimarySSEType(entry *filer_pb.Entry) string { return s3_constants.SSETypeS3 } - return "None" + return metadataType } // createMultipartSSECDecryptedReaderDirect creates a reader that decrypts each chunk independently for multipart SSE-C objects (direct volume path)