S3: commit SSE GET status only after the first read succeeds (#10935)

The SSE streaming path kept writing 200/206 from filer metadata before fetching or decrypting anything, so a missing needle or failed decrypt setup surfaced as a broken 200 body. Same deferral as the plain path: the status commits on the first body write, and every failure before that returns to the handler for a clean S3 error response.
2026-09-01 21:59:44 +00:00 · 2026-08-24 18:36:46 -07:00
parent 115756dd41
commit d2c470af1b
2 changed files with 156 additions and 44 deletions
@@ -67,11 +67,17 @@ type commitOnFirstWrite struct {
 }

 func (c *commitOnFirstWrite) Write(p []byte) (int, error) {
+	c.Commit()
+	return c.w.Write(p)
+}
+
+// Commit commits the response if the first body write has not already done so;
+// a zero-length body never writes, so a successful stream must end with this.
+func (c *commitOnFirstWrite) Commit() {
 	if !c.committed {
 		c.committed = true
 		c.commit()
 	}
-	return c.w.Write(p)
 }

 // adjustRangeForPart adjusts a client's Range header to absolute offsets within a part.
@@ -1227,11 +1233,7 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		// Streaming error after WriteHeader was called - response already partially written
 		return newStreamErrorWithResponse(err)
 	}
-	if !body.committed {
-		// a zero-length body never writes, so commit the empty response here
-		body.committed = true
-		body.commit()
-	}
+	body.Commit()
 	glog.V(4).Infof("streamFromVolumeServers: streamFn completed successfully, wrote %d bytes", written)
 	return nil
 }
@@ -1466,28 +1468,26 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 	}
 	keyValidateTime = time.Since(tKeyValidate)

-	// Set response headers
-	// IMPORTANT: Set ALL headers BEFORE calling WriteHeader (headers are ignored after WriteHeader)
-	tHeaderSet := time.Now()
-	s3a.setResponseHeaders(w, r, entry, totalSize)
-	s3a.addSSEResponseHeadersFromEntry(w, r, entry, sseType)
-
-	// Override/add range-specific headers if this is a range request
-	if isRangeRequest {
-		w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
-		w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
-	}
-	headerSetTime = time.Since(tHeaderSet)
-
-	// Now write status code (headers are all set)
-	if isRangeRequest {
-		w.WriteHeader(http.StatusPartialContent)
-	} else {
-		w.WriteHeader(http.StatusOK)
-	}
-
-	// Track time to first byte metric
-	TimeToFirstByte(r.Method, t0, r)
+	// Headers and status are committed on the first body write, once the first
+	// fetch and decryption step has succeeded -- the same deferral as
+	// streamFromVolumeServers, so a missing needle or broken decrypt setup
+	// surfaces as a clean 5xx instead of a 200 with a broken body.
+	body := &commitOnFirstWrite{w: w, commit: func() {
+		tHeaderSet := time.Now()
+		s3a.setResponseHeaders(w, r, entry, totalSize)
+		s3a.addSSEResponseHeadersFromEntry(w, r, entry, sseType)
+		if isRangeRequest {
+			w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
+			w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
+		}
+		headerSetTime = time.Since(tHeaderSet)
+		if isRangeRequest {
+			w.WriteHeader(http.StatusPartialContent)
+		} else {
+			w.WriteHeader(http.StatusOK)
+		}
+		TimeToFirstByte(r.Method, t0, r)
+	}}

 	// Full Range Optimization: Use ViewFromChunks to only fetch/decrypt needed chunks
 	tDecryptSetup := time.Now()
@@ -1496,7 +1496,7 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 	if isRangeRequest {
 		glog.V(2).Infof("Using range-aware SSE decryption for offset=%d size=%d", offset, size)
 		streamFetchTime = 0 // No full stream fetch in range-aware path
-		written, err := s3a.streamDecryptedRangeFromChunks(r.Context(), w, entry, offset, size, sseType, decryptionKey)
+		written, err := s3a.streamDecryptedRangeFromChunks(r.Context(), body, entry, offset, size, sseType, decryptionKey)
 		decryptSetupTime = time.Since(tDecryptSetup)
 		copyTime = decryptSetupTime // Streaming is included in decrypt setup for range-aware path
 		// Track traffic even on partial writes for accurate egress accounting
@@ -1504,9 +1504,13 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 			BucketTrafficSent(written, r)
 		}
 		if err != nil {
-			// Error after WriteHeader - response already written
+			if !body.committed {
+				// nothing written yet -- the caller writes the S3 error response
+				return err
+			}
 			return newStreamErrorWithResponse(err)
 		}
+		body.Commit()
 		return nil
 	}

@@ -1541,15 +1545,13 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry)
 			streamFetchTime = time.Since(tStreamFetch)
 			if streamErr != nil {
-				// Error after WriteHeader - response already written
-				return newStreamErrorWithResponse(streamErr)
+				return streamErr
 			}
 			defer encryptedReader.Close()

 			iv := entry.Extended[s3_constants.SeaweedFSSSEIV]
 			if len(iv) == 0 {
-				// Error after WriteHeader - response already written
-				return newStreamErrorWithResponse(fmt.Errorf("SSE-C IV not found in entry metadata"))
+				return fmt.Errorf("SSE-C IV not found in entry metadata")
 			}
 			glog.V(2).Infof("SSE-C decryption: IV length=%d, KeyMD5=%s", len(iv), customerKey.KeyMD5)
 			decryptedReader, err = CreateSSECDecryptedReader(encryptedReader, customerKey, iv)
@@ -1579,8 +1581,7 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry)
 			streamFetchTime = time.Since(tStreamFetch)
 			if streamErr != nil {
-				// Error after WriteHeader - response already written
-				return newStreamErrorWithResponse(streamErr)
+				return streamErr
 			}
 			defer encryptedReader.Close()

@@ -1612,16 +1613,14 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 			encryptedReader, streamErr := s3a.getEncryptedStreamFromVolumes(r.Context(), entry)
 			streamFetchTime = time.Since(tStreamFetch)
 			if streamErr != nil {
-				// Error after WriteHeader - response already written
-				return newStreamErrorWithResponse(streamErr)
+				return streamErr
 			}
 			defer encryptedReader.Close()

 			keyManager := GetSSES3KeyManager()
 			iv, ivErr := GetSSES3IV(entry, sseS3Key, keyManager)
 			if ivErr != nil {
-				// Error after WriteHeader - response already written
-				return newStreamErrorWithResponse(fmt.Errorf("failed to get SSE-S3 IV: %w", ivErr))
+				return fmt.Errorf("failed to get SSE-S3 IV: %w", ivErr)
 			}
 			glog.V(2).Infof("SSE-S3 decryption: KeyID=%s, IV length=%d", sseS3Key.KeyID, len(iv))
 			decryptedReader, err = CreateSSES3DecryptedReader(encryptedReader, sseS3Key, iv)
@@ -1631,8 +1630,7 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r

 	if err != nil {
 		glog.Errorf("SSE decryption error (%s): %v", sseType, err)
-		// Error after WriteHeader - response already written
-		return newStreamErrorWithResponse(fmt.Errorf("failed to create decrypted reader: %w", err))
+		return fmt.Errorf("failed to create decrypted reader: %w", err)
 	}

 	// Close the decrypted reader to avoid leaking HTTP bodies
@@ -1647,7 +1645,7 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 	// Stream full decrypted object to client
 	tCopy := time.Now()
 	buf := make([]byte, 128*1024)
-	copied, copyErr := io.CopyBuffer(w, decryptedReader, buf)
+	copied, copyErr := io.CopyBuffer(body, decryptedReader, buf)
 	copyTime = time.Since(tCopy)
 	// Track traffic even on partial writes for accurate egress accounting
 	if copied > 0 {
@@ -1655,9 +1653,13 @@ func (s3a *S3ApiServer) streamFromVolumeServersWithSSE(w http.ResponseWriter, r
 	}
 	if copyErr != nil {
 		glog.Errorf("Failed to copy full object: copied %d bytes: %v", copied, copyErr)
-		// Error after WriteHeader - response already written
+		if !body.committed {
+			// nothing written yet -- the caller writes the S3 error response
+			return copyErr
+		}
 		return newStreamErrorWithResponse(copyErr)
 	}
+	body.Commit()
 	glog.V(3).Infof("Full object request: copied %d bytes", copied)
 	return nil
 }
@@ -0,0 +1,110 @@
+package s3api
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// the SSE mirror of the plain-path deferred commit: the 200 must not be
+// committed until the first fetched-and-decrypted byte is written
+func TestS3SSEStreamCommitsStatusOnFirstWrite(t *testing.T) {
+	plaintext := []byte("0123456789")
+	keyPair := GenerateTestSSECKey(1)
+
+	newSSECRequest := func(rangeHeader string) *http.Request {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		t.Cleanup(cancel)
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil).WithContext(ctx)
+		SetupTestSSECHeaders(r, keyPair)
+		if rangeHeader != "" {
+			r.Header.Set("Range", rangeHeader)
+		}
+		return r
+	}
+
+	encrypt := func(t *testing.T, r *http.Request) (ciphertext, iv []byte) {
+		customerKey, err := ParseSSECHeaders(r)
+		require.NoError(t, err)
+		encReader, iv, err := CreateSSECEncryptedReader(bytes.NewReader(plaintext), customerKey)
+		require.NoError(t, err)
+		ciphertext, err = io.ReadAll(encReader)
+		require.NoError(t, err)
+		return ciphertext, iv
+	}
+
+	t.Run("readable SSE-C object still streams a 200 with the plaintext", func(t *testing.T) {
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-sseok", nil))
+		r := newSSECRequest("")
+		ciphertext, iv := encrypt(t, r)
+		entry := &filer_pb.Entry{
+			Name:       "obj.bin",
+			Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(plaintext))},
+			Content:    ciphertext,
+			Extended: map[string][]byte{
+				s3_constants.AmzServerSideEncryptionCustomerAlgorithm: []byte("AES256"),
+				s3_constants.AmzServerSideEncryptionCustomerKeyMD5:    []byte(keyPair.KeyMD5),
+				s3_constants.SeaweedFSSSEIV:                           iv,
+			},
+		}
+		w := httptest.NewRecorder()
+
+		err := s3a.streamFromVolumeServersWithSSE(w, r, entry, s3_constants.SSETypeC, "mybucket", "dir/obj.bin", "")
+
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusOK, w.Code)
+		assert.Equal(t, plaintext, w.Body.Bytes())
+		assert.Equal(t, "AES256", w.Header().Get(s3_constants.AmzServerSideEncryptionCustomerAlgorithm))
+	})
+
+	unreadableEntry := func(t *testing.T, r *http.Request) *filer_pb.Entry {
+		_, iv := encrypt(t, r)
+		return &filer_pb.Entry{
+			Name:       "obj.bin",
+			Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(plaintext))},
+			Chunks:     []*filer_pb.FileChunk{{FileId: "1,0123456789ab", Size: uint64(len(plaintext))}},
+			Extended: map[string][]byte{
+				s3_constants.AmzServerSideEncryptionCustomerKeyMD5: []byte(keyPair.KeyMD5),
+				s3_constants.SeaweedFSSSEIV:                        iv,
+			},
+		}
+	}
+
+	assertNothingCommitted := func(t *testing.T, w *httptest.ResponseRecorder, err error) {
+		require.Error(t, err)
+		var streamErr *StreamError
+		assert.False(t, errors.As(err, &streamErr) && streamErr.ResponseWritten,
+			"the caller must still own the error response")
+		assert.Zero(t, w.Body.Len(), "no body bytes may precede the failure")
+	}
+
+	t.Run("unreadable SSE-C object leaves the response uncommitted", func(t *testing.T) {
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-ssefail", nil))
+		r := newSSECRequest("")
+		w := httptest.NewRecorder()
+
+		err := s3a.streamFromVolumeServersWithSSE(w, r, unreadableEntry(t, r), s3_constants.SSETypeC, "mybucket", "dir/obj.bin", "")
+
+		assertNothingCommitted(t, w, err)
+	})
+
+	t.Run("unreadable SSE-C range leaves the response uncommitted", func(t *testing.T) {
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-sserange", nil))
+		r := newSSECRequest("bytes=2-5")
+		w := httptest.NewRecorder()
+
+		err := s3a.streamFromVolumeServersWithSSE(w, r, unreadableEntry(t, r), s3_constants.SSETypeC, "mybucket", "dir/obj.bin", "")
+
+		assertNothingCommitted(t, w, err)
+	})
+}