diff --git a/weed/s3api/s3_sse_s3_integration_test.go b/weed/s3api/s3_sse_s3_integration_test.go index 454e0b5ce..2a2a14a58 100644 --- a/weed/s3api/s3_sse_s3_integration_test.go +++ b/weed/s3api/s3_sse_s3_integration_test.go @@ -489,9 +489,10 @@ func TestBuildMultipartSSES3Reader_InvalidIVLength(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - closed := false + fetchCalled := false fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) { - return &closeTrackingReadCloser{Reader: bytes.NewReader([]byte("whatever")), closed: &closed}, nil + fetchCalled = true + return io.NopCloser(bytes.NewReader([]byte("whatever"))), nil } chunks := []*filer_pb.FileChunk{ @@ -511,24 +512,29 @@ func TestBuildMultipartSSES3Reader_InvalidIVLength(t *testing.T) { if !strings.Contains(err.Error(), "invalid IV length") { t.Errorf("expected 'invalid IV length' in error, got: %v", err) } - if !closed { - t.Error("chunk reader for the bad chunk was not closed on error") + // Validation runs upfront before any chunk fetch, so no volume-server + // HTTP connection should have been opened on the failure path. + if fetchCalled { + t.Error("fetchChunk was called for an invalid-IV chunk; metadata validation should fail before any fetch") } }) } } -// TestBuildMultipartSSES3Reader_ClosesAppendedOnError verifies that when a -// later chunk fails (e.g., malformed metadata), readers already appended for -// earlier valid chunks are closed so volume-server HTTP connections do not leak. -func TestBuildMultipartSSES3Reader_ClosesAppendedOnError(t *testing.T) { +// TestBuildMultipartSSES3Reader_RejectsBadChunkBeforeAnyFetch verifies that +// when any chunk's metadata is malformed, the helper returns an error WITHOUT +// having opened a volume-server HTTP connection for any chunk. Per-chunk +// metadata is validated upfront precisely so a bad chunk in position N does +// not leak open HTTP responses for chunks 0..N-1 (the original eager +// implementation depended on a closeAppendedReaders cleanup path; this test +// pins the stronger contract: nothing is opened in the first place). +func TestBuildMultipartSSES3Reader_RejectsBadChunkBeforeAnyFetch(t *testing.T) { keyManager := initSSES3KeyManagerForTest(t) // First chunk: valid SSE-S3 chunk. cipher1, meta1 := encryptSSES3Part(t, []byte("first chunk plaintext")) - // Second chunk: missing per-chunk metadata, triggers error after first is - // already appended. + // Second chunk: missing per-chunk metadata, triggers error. chunks := []*filer_pb.FileChunk{ { FileId: "1,good", @@ -546,36 +552,20 @@ func TestBuildMultipartSSES3Reader_ClosesAppendedOnError(t *testing.T) { }, } - firstClosed := false - secondClosed := false + fetched := map[string]int{} fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) { - switch c.GetFileIdString() { - case "1,good": - return &closeTrackingReadCloser{Reader: bytes.NewReader(cipher1), closed: &firstClosed}, nil - case "2,bad": - return &closeTrackingReadCloser{Reader: bytes.NewReader([]byte("x")), closed: &secondClosed}, nil - } - return nil, fmt.Errorf("unexpected chunk %s", c.GetFileIdString()) + fetched[c.GetFileIdString()]++ + return io.NopCloser(bytes.NewReader([]byte("x"))), nil } _, err := buildMultipartSSES3Reader(chunks, keyManager, fetch) if err == nil { t.Fatal("expected error from missing chunk metadata, got nil") } - if !firstClosed { - t.Error("previously appended chunk reader was not closed on error") + if !strings.Contains(err.Error(), "missing per-chunk metadata") { + t.Errorf("expected 'missing per-chunk metadata' in error, got: %v", err) } - if !secondClosed { - t.Error("chunk reader for the failing chunk was not closed on error") + if len(fetched) != 0 { + t.Errorf("expected no chunks fetched on validation failure, got %v", fetched) } } - -type closeTrackingReadCloser struct { - io.Reader - closed *bool -} - -func (r *closeTrackingReadCloser) Close() error { - *r.closed = true - return nil -} diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 40ea157fa..511321c28 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -2786,6 +2786,15 @@ func (s3a *S3ApiServer) createMultipartSSES3DecryptedReaderDirect(ctx context.Co // SSE-S3 chunks. Chunks are fetched via fetchChunk and decrypted using their // per-chunk metadata (each multipart part has its own DEK and IV). Exposed as a // standalone helper so tests can inject a mock chunk fetcher. +// +// All per-chunk metadata is validated upfront so a malformed chunk fails fast +// without opening any HTTP connections to volume servers. The actual chunk +// fetch and decryption happens LAZILY as the returned reader is read: at most +// one chunk's HTTP connection is open at a time. Eagerly opening every chunk's +// HTTP response (the previous behavior) caused later chunks' connections to +// sit idle while earlier chunks were still being consumed, which under load +// could trip volume-server idle/keepalive limits and yield truncated reads +// (issue #8908). func buildMultipartSSES3Reader(chunks []*filer_pb.FileChunk, keyManager *SSES3KeyManager, fetchChunk func(*filer_pb.FileChunk) (io.ReadCloser, error)) (io.Reader, error) { // Sort a copy of the slice so callers do not observe their input chunks // reordered (the backing array is shared with entry.Chunks, which other @@ -2796,82 +2805,138 @@ func buildMultipartSSES3Reader(chunks []*filer_pb.FileChunk, keyManager *SSES3Ke return sortedChunks[i].GetOffset() < sortedChunks[j].GetOffset() }) - // Create readers for each chunk, decrypting them independently - readers := make([]io.Reader, 0, len(sortedChunks)) - - // Close any readers already appended to `readers` on error paths, to avoid - // leaking volume-server HTTP connections. - closeAppendedReaders := func() { - for _, r := range readers { - if closer, ok := r.(io.Closer); ok { - closer.Close() - } - } - } - + // Validate every chunk's SSE-S3 metadata before returning a reader. This + // keeps the eager-validation contract that callers and tests rely on + // (malformed metadata fails immediately), without holding open any + // volume-server HTTP connections. + preparedChunks := make([]preparedMultipartChunk, 0, len(sortedChunks)) for _, chunk := range sortedChunks { - // Get this chunk's encrypted data - chunkReader, err := fetchChunk(chunk) + if chunk.GetSseType() != filer_pb.SSEType_SSE_S3 { + preparedChunks = append(preparedChunks, preparedMultipartChunk{chunk: chunk}) + continue + } + if len(chunk.GetSseMetadata()) == 0 { + return nil, fmt.Errorf("SSE-S3 chunk %s missing per-chunk metadata", chunk.GetFileIdString()) + } + meta, err := DeserializeSSES3Metadata(chunk.GetSseMetadata(), keyManager) if err != nil { - closeAppendedReaders() - return nil, fmt.Errorf("failed to create chunk reader: %v", err) + return nil, fmt.Errorf("failed to deserialize SSE-S3 metadata for chunk %s: %v", chunk.GetFileIdString(), err) } - - // Handle based on chunk's encryption type - if chunk.GetSseType() == filer_pb.SSEType_SSE_S3 { - // Check if this chunk has per-chunk SSE-S3 metadata - if len(chunk.GetSseMetadata()) == 0 { - chunkReader.Close() - closeAppendedReaders() - return nil, fmt.Errorf("SSE-S3 chunk %s missing per-chunk metadata", chunk.GetFileIdString()) - } - - // Deserialize the per-chunk SSE-S3 metadata to get the IV - chunkSSES3Metadata, err := DeserializeSSES3Metadata(chunk.GetSseMetadata(), keyManager) - if err != nil { - chunkReader.Close() - closeAppendedReaders() - return nil, fmt.Errorf("failed to deserialize SSE-S3 metadata for chunk %s: %v", chunk.GetFileIdString(), err) - } - - // Use the IV from the chunk metadata. DeserializeSSES3Metadata does - // not require an IV, so validate the length here before it reaches - // cipher.NewCTR, which would otherwise panic on a nil or short IV. - iv := chunkSSES3Metadata.IV - if len(iv) != s3_constants.AESBlockSize { - chunkReader.Close() - closeAppendedReaders() - return nil, fmt.Errorf("SSE-S3 chunk %s has invalid IV length %d (expected %d)", - chunk.GetFileIdString(), len(iv), s3_constants.AESBlockSize) - } - glog.V(4).Infof("Decrypting SSE-S3 chunk %s with KeyID=%s, IV length=%d", - chunk.GetFileIdString(), chunkSSES3Metadata.KeyID, len(iv)) - - // Create decrypted reader for this chunk - decryptedChunkReader, decErr := CreateSSES3DecryptedReader(chunkReader, chunkSSES3Metadata, iv) - if decErr != nil { - chunkReader.Close() - closeAppendedReaders() - return nil, fmt.Errorf("failed to decrypt SSE-S3 chunk: %v", decErr) - } - - // Use the streaming decrypted reader directly - readers = append(readers, struct { - io.Reader - io.Closer - }{ - Reader: decryptedChunkReader, - Closer: chunkReader, - }) - glog.V(4).Infof("Added streaming decrypted reader for SSE-S3 chunk %s", chunk.GetFileIdString()) - } else { - // Non-SSE-S3 chunk, use as-is - readers = append(readers, chunkReader) - glog.V(4).Infof("Added non-encrypted reader for chunk %s", chunk.GetFileIdString()) + // DeserializeSSES3Metadata does not require an IV, so validate the + // length here before it reaches cipher.NewCTR, which would otherwise + // panic on a nil or short IV. + if len(meta.IV) != s3_constants.AESBlockSize { + return nil, fmt.Errorf("SSE-S3 chunk %s has invalid IV length %d (expected %d)", + chunk.GetFileIdString(), len(meta.IV), s3_constants.AESBlockSize) } + // Capture meta and chunk by-value into the wrap closure so each + // prepared entry decrypts with its own per-chunk key + IV. + fileId := chunk.GetFileIdString() + preparedChunks = append(preparedChunks, preparedMultipartChunk{ + chunk: chunk, + wrap: func(raw io.ReadCloser) (io.Reader, error) { + glog.V(4).Infof("Decrypting SSE-S3 chunk %s with KeyID=%s, IV length=%d", + fileId, meta.KeyID, len(meta.IV)) + dec, err := CreateSSES3DecryptedReader(raw, meta, meta.IV) + if err != nil { + return nil, fmt.Errorf("failed to decrypt SSE-S3 chunk: %v", err) + } + return dec, nil + }, + }) } - return NewMultipartSSEReader(readers), nil + return &lazyMultipartChunkReader{ + chunks: preparedChunks, + fetch: fetchChunk, + }, nil +} + +// preparedMultipartChunk pairs a chunk with the per-SSE wrapping logic the +// lazy reader applies to its raw HTTP body. wrap is nil for chunks that +// stream as-is (no SSE on the chunk, even though the object is multipart-SSE); +// otherwise wrap is the SSE-specific decryption setup, which receives the +// already-opened raw chunk body and returns the plaintext reader. +type preparedMultipartChunk struct { + chunk *filer_pb.FileChunk + wrap func(raw io.ReadCloser) (io.Reader, error) +} + +// lazyMultipartChunkReader streams a sequence of multipart chunks one at a +// time. It opens each chunk's underlying HTTP fetch (and applies the +// SSE-specific decryption wrapper) only when the previous chunk has been +// fully consumed, so volume-server connections do not pile up for large +// objects. This is the same shape used by all three SSE multipart read +// paths (SSE-S3, SSE-KMS, SSE-C); only the per-chunk wrap closure differs. +type lazyMultipartChunkReader struct { + chunks []preparedMultipartChunk + fetch func(*filer_pb.FileChunk) (io.ReadCloser, error) + idx int + current io.Reader // current chunk's plaintext reader (or raw reader for non-SSE chunks) + closer io.Closer // current chunk's underlying HTTP body, to close on advance/Close + finished bool +} + +func (l *lazyMultipartChunkReader) Read(p []byte) (int, error) { + for { + if l.finished { + return 0, io.EOF + } + if l.current == nil { + if l.idx >= len(l.chunks) { + l.finished = true + return 0, io.EOF + } + pc := l.chunks[l.idx] + l.idx++ + chunkReader, err := l.fetch(pc.chunk) + if err != nil { + l.finished = true + return 0, fmt.Errorf("failed to create chunk reader: %v", err) + } + if pc.wrap == nil { + // Non-SSE chunk in an otherwise SSE-multipart object: stream + // raw bytes through. + l.current = chunkReader + l.closer = chunkReader + glog.V(4).Infof("Streaming non-encrypted chunk %s", pc.chunk.GetFileIdString()) + } else { + wrapped, wrapErr := pc.wrap(chunkReader) + if wrapErr != nil { + chunkReader.Close() + l.finished = true + return 0, wrapErr + } + l.current = wrapped + l.closer = chunkReader + } + } + n, err := l.current.Read(p) + if err == io.EOF { + closeErr := l.closer.Close() + l.current = nil + l.closer = nil + if n > 0 { + return n, nil + } + if closeErr != nil { + glog.V(2).Infof("Error closing chunk reader: %v", closeErr) + } + continue + } + return n, err + } +} + +func (l *lazyMultipartChunkReader) Close() error { + l.finished = true + if l.closer != nil { + err := l.closer.Close() + l.current = nil + l.closer = nil + return err + } + return nil } // createEncryptedChunkReader creates a reader for a single encrypted chunk diff --git a/weed/s3api/sses3_multipart_repro_test.go b/weed/s3api/sses3_multipart_repro_test.go new file mode 100644 index 000000000..035be7e21 --- /dev/null +++ b/weed/s3api/sses3_multipart_repro_test.go @@ -0,0 +1,280 @@ +package s3api + +import ( + "bytes" + "crypto/rand" + "fmt" + "io" + "sync/atomic" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +// TestMultipartSSES3RealisticEndToEnd reproduces the production multipart SSE-S3 +// flow where ALL parts share the same DEK and baseIV (the upload-init key/IV), +// and each part is encrypted with partOffset=0. Each part is then chunked at 8MB +// boundaries the way UploadReaderInChunks does. After completion, the chunks +// have global offsets but per-chunk stored IVs derived from part-local offsets. +// +// buildMultipartSSES3Reader is then run on the assembled chunks; the +// concatenated decrypted output must equal the concatenation of the part +// plaintexts. This is the round trip that fails in #8908 if anything in the +// encrypt or decrypt path is inconsistent. +func TestMultipartSSES3RealisticEndToEnd(t *testing.T) { + keyManager := initSSES3KeyManagerForTest(t) + + // One DEK and one baseIV, shared by all parts (the upload-init values). + key, err := GenerateSSES3Key() + if err != nil { + t.Fatalf("GenerateSSES3Key: %v", err) + } + baseIV := make([]byte, s3_constants.AESBlockSize) + if _, err := rand.Read(baseIV); err != nil { + t.Fatalf("rand.Read baseIV: %v", err) + } + + const chunkSize = int64(8 * 1024 * 1024) + + // Realistic mix of part sizes: small (one chunk), exact 8MB, >8MB (two + // chunks), much larger (multiple chunks). + partSizes := []int{ + 5 * 1024 * 1024, // 5MB (single chunk) + 8 * 1024 * 1024, // 8MB exactly (single chunk, full) + 8*1024*1024 + 123, // crosses chunk boundary (two chunks) + 17 * 1024 * 1024, // three chunks + 1234, // tiny + } + + parts := make([][]byte, len(partSizes)) + for i, n := range partSizes { + parts[i] = makeRandomPlaintext(t, n) + } + + // Build the chunks list the way completion would produce it: encrypt each + // part with partOffset=0, slice the ciphertext at chunkSize boundaries, + // store per-chunk metadata IV = calculateIVWithOffset(baseIV, partLocalOff), + // then assign GLOBAL offsets to the FileChunk. + type chunkBlob struct { + fid string + ciphertext []byte + } + var chunks []*filer_pb.FileChunk + chunkData := map[string][]byte{} + var globalOffset int64 + for partIdx, partPlaintext := range parts { + encReader, _, err := CreateSSES3EncryptedReaderWithBaseIV(bytes.NewReader(partPlaintext), key, baseIV, 0) + if err != nil { + t.Fatalf("CreateSSES3EncryptedReaderWithBaseIV(part %d): %v", partIdx, err) + } + ciphertext, err := io.ReadAll(encReader) + if err != nil { + t.Fatalf("read encrypted part %d: %v", partIdx, err) + } + + for partLocalOff := int64(0); partLocalOff < int64(len(ciphertext)); partLocalOff += chunkSize { + end := partLocalOff + chunkSize + if end > int64(len(ciphertext)) { + end = int64(len(ciphertext)) + } + cipherSlice := ciphertext[partLocalOff:end] + + chunkIV, _ := calculateIVWithOffset(baseIV, partLocalOff) + chunkKey := &SSES3Key{ + Key: key.Key, + KeyID: key.KeyID, + Algorithm: key.Algorithm, + IV: chunkIV, + } + meta, err := SerializeSSES3Metadata(chunkKey) + if err != nil { + t.Fatalf("SerializeSSES3Metadata(part %d off %d): %v", partIdx, partLocalOff, err) + } + + fid := fmt.Sprintf("%d,%d", partIdx+1, partLocalOff) + chunks = append(chunks, &filer_pb.FileChunk{ + FileId: fid, + Offset: globalOffset, // global offset assigned at completion + Size: uint64(end - partLocalOff), + SseType: filer_pb.SSEType_SSE_S3, + SseMetadata: meta, + }) + chunkData[fid] = cipherSlice + globalOffset += end - partLocalOff + } + } + + fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) { + data, ok := chunkData[c.GetFileIdString()] + if !ok { + return nil, fmt.Errorf("unexpected chunk %s", c.GetFileIdString()) + } + return io.NopCloser(bytes.NewReader(data)), nil + } + + reader, err := buildMultipartSSES3Reader(chunks, keyManager, fetch) + if err != nil { + t.Fatalf("buildMultipartSSES3Reader: %v", err) + } + got, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("ReadAll decrypted output: %v", err) + } + + want := bytes.Join(parts, nil) + if !bytes.Equal(got, want) { + idx := firstMismatch(got, want) + end := idx + 32 + if end > len(got) { + end = len(got) + } + if end > len(want) { + end = len(want) + } + t.Fatalf("decrypted output mismatch at byte %d (total len got=%d want=%d)\n got: %x\nwant: %x", + idx, len(got), len(want), got[idx:end], want[idx:end]) + } +} + +// TestBuildMultipartSSES3Reader_LazyChunkFetch pins the lazy behavior of +// buildMultipartSSES3Reader: chunk N's HTTP fetch only happens after chunk +// N-1 has been fully consumed. The original eager loop opened every chunk's +// HTTP response upfront and held them open while io.MultiReader walked +// through readers[0]; for objects with many chunks (e.g. a 200MB Docker image +// blob), this could trip volume-server idle/keepalive limits and produce +// truncated reads at the client (issue #8908). +// +// The test installs a fetch hook that tracks how many chunks have been +// opened and when each one is closed, and verifies: +// - At any point during streaming, at most one chunk's reader is open. +// - The number of opened chunks grows as bytes are read out, not upfront. +// - All chunks are closed when the outer reader is fully drained. +func TestBuildMultipartSSES3Reader_LazyChunkFetch(t *testing.T) { + keyManager := initSSES3KeyManagerForTest(t) + + key, err := GenerateSSES3Key() + if err != nil { + t.Fatalf("GenerateSSES3Key: %v", err) + } + baseIV := make([]byte, s3_constants.AESBlockSize) + if _, err := rand.Read(baseIV); err != nil { + t.Fatalf("rand.Read baseIV: %v", err) + } + + // Many small chunks (mirrors many-part Docker Registry uploads). + const numChunks = 8 + const chunkPayload = 1024 + plaintexts := make([][]byte, numChunks) + chunkData := map[string][]byte{} + chunks := make([]*filer_pb.FileChunk, 0, numChunks) + for i := 0; i < numChunks; i++ { + plaintexts[i] = makeRandomPlaintext(t, chunkPayload) + + // Encrypt as a fresh "part" with partOffset=0 (matching putToFiler). + encReader, _, err := CreateSSES3EncryptedReaderWithBaseIV(bytes.NewReader(plaintexts[i]), key, baseIV, 0) + if err != nil { + t.Fatalf("encrypt chunk %d: %v", i, err) + } + ciphertext, err := io.ReadAll(encReader) + if err != nil { + t.Fatalf("read ciphertext %d: %v", i, err) + } + + chunkIV, _ := calculateIVWithOffset(baseIV, 0) + chunkKey := &SSES3Key{ + Key: key.Key, + KeyID: key.KeyID, + Algorithm: key.Algorithm, + IV: chunkIV, + } + meta, err := SerializeSSES3Metadata(chunkKey) + if err != nil { + t.Fatalf("serialize meta %d: %v", i, err) + } + + fid := fmt.Sprintf("vol,c%d", i) + chunks = append(chunks, &filer_pb.FileChunk{ + FileId: fid, + Offset: int64(i) * chunkPayload, + Size: uint64(chunkPayload), + SseType: filer_pb.SSEType_SSE_S3, + SseMetadata: meta, + }) + chunkData[fid] = ciphertext + } + + var openCount int64 // total opens + var liveCount int64 // currently open + var maxLive int64 + + fetch := func(c *filer_pb.FileChunk) (io.ReadCloser, error) { + atomic.AddInt64(&openCount, 1) + if live := atomic.AddInt64(&liveCount, 1); live > atomic.LoadInt64(&maxLive) { + atomic.StoreInt64(&maxLive, live) + } + data, ok := chunkData[c.GetFileIdString()] + if !ok { + return nil, fmt.Errorf("unexpected chunk %s", c.GetFileIdString()) + } + return &liveTrackingReadCloser{Reader: bytes.NewReader(data), live: &liveCount}, nil + } + + reader, err := buildMultipartSSES3Reader(chunks, keyManager, fetch) + if err != nil { + t.Fatalf("buildMultipartSSES3Reader: %v", err) + } + + // Construction alone must not have opened any chunk reader. + if got := atomic.LoadInt64(&openCount); got != 0 { + t.Fatalf("expected no chunks opened before any Read, got %d", got) + } + + // Read first byte: should open chunk 0 only. + one := make([]byte, 1) + if n, err := reader.Read(one); n != 1 || err != nil { + t.Fatalf("first Read: n=%d err=%v", n, err) + } + if got := atomic.LoadInt64(&openCount); got != 1 { + t.Errorf("after first byte, expected 1 chunk opened, got %d", got) + } + if got := atomic.LoadInt64(&liveCount); got != 1 { + t.Errorf("after first byte, expected 1 chunk live, got %d", got) + } + + // Drain the rest. + rest, err := io.ReadAll(reader) + if err != nil { + t.Fatalf("drain: %v", err) + } + got := append(one, rest...) + want := bytes.Join(plaintexts, nil) + if !bytes.Equal(got, want) { + idx := firstMismatch(got, want) + t.Fatalf("decrypted output mismatch at byte %d (got len %d, want len %d)", idx, len(got), len(want)) + } + if got := atomic.LoadInt64(&openCount); got != int64(numChunks) { + t.Errorf("expected exactly %d chunk opens after drain, got %d", numChunks, got) + } + if got := atomic.LoadInt64(&maxLive); got > 1 { + t.Errorf("expected at most 1 chunk reader live at a time (lazy), saw peak of %d", got) + } + if got := atomic.LoadInt64(&liveCount); got != 0 { + t.Errorf("expected all chunks closed after drain, %d still live", got) + } +} + +type liveTrackingReadCloser struct { + io.Reader + live *int64 + once bool +} + +func (r *liveTrackingReadCloser) Close() error { + if r.once { + return nil + } + r.once = true + atomic.AddInt64(r.live, -1) + return nil +}