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

@@ -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)

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) {

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
 // ---------------------------------------------------------------------------

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},

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)
+	metadataType := "None"
 	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
+	if hasSSEC {
+		metadataType = s3_constants.SSETypeC
 	} else {
-				// No content, both headers - default to SSE-C
-				return s3_constants.SSETypeC
+		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)