s3: drain request body before error response (#11334)

* s3: drain request body before error response

* s3: keep oversized request bodies drainable

weed/s3api/request_body.go

@@ -0,0 +1,23 @@
+package s3api
+
+import (
+	"errors"
+	"io"
+	"net/http"
+)
+
+var errRequestBodyTooLarge = errors.New("request body too large")
+
+func readRequestBody(r *http.Request, limit int64) ([]byte, error) {
+	if r.Body == nil {
+		return nil, nil
+	}
+	body, err := io.ReadAll(io.LimitReader(r.Body, limit+1))
+	if err != nil {
+		return nil, err
+	}
+	if int64(len(body)) > limit {
+		return nil, errRequestBodyTooLarge
+	}
+	return body, nil
+}

weed/s3api/request_body_test.go

@@ -0,0 +1,24 @@
+package s3api
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestReadRequestBodyLeavesOversizedBodyDrainable(t *testing.T) {
+	req := httptest.NewRequest(http.MethodPost, "/", nil)
+	req.Body = io.NopCloser(strings.NewReader("abcdef"))
+
+	body, err := readRequestBody(req, 3)
+	remaining, readErr := io.ReadAll(req.Body)
+
+	assert.ErrorIs(t, err, errRequestBodyTooLarge)
+	assert.Nil(t, body)
+	assert.NoError(t, readErr)
+	assert.Equal(t, "ef", string(remaining))
+}

weed/s3api/s3api_bucket_handlers.go

@@ -8,7 +8,6 @@ import (
 	"encoding/xml"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"net/url"
 	"sort"
@@ -1130,12 +1129,10 @@ func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWr
 		return
 	}
 
-	r.Body = http.MaxBytesReader(w, r.Body, maxBucketLifecycleConfigurationSize)
-	lifecycleXML, err := io.ReadAll(r.Body)
+	lifecycleXML, err := readRequestBody(r, maxBucketLifecycleConfigurationSize)
 	if err != nil {
 		glog.Warningf("PutBucketLifecycleConfigurationHandler read body: %s", err)
-		var maxBytesErr *http.MaxBytesError
-		if errors.As(err, &maxBytesErr) {
+		if errors.Is(err, errRequestBodyTooLarge) {
 			s3err.WriteErrorResponse(w, r, s3err.ErrEntityTooLarge)
 			return
 		}

weed/s3api/s3api_bucket_handlers_misc.go

@@ -3,7 +3,6 @@ package s3api
 import (
 	"encoding/xml"
 	"errors"
-	"io"
 	"net/http"
 	"strings"
 
@@ -92,9 +91,8 @@ func (s3a *S3ApiServer) PutBucketRequestPaymentHandler(w http.ResponseWriter, r
 		return
 	}
 
-	r.Body = http.MaxBytesReader(w, r.Body, putBucketRequestPaymentMaxBodyBytes)
 	defer r.Body.Close()
-	body, err := io.ReadAll(r.Body)
+	body, err := readRequestBody(r, putBucketRequestPaymentMaxBodyBytes)
 	if err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
 		return

weed/s3api/s3api_bucket_quota_handlers.go

@@ -1,6 +1,7 @@
 package s3api
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -57,11 +58,16 @@ func (s3a *S3ApiServer) PutBucketQuotaHandler(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	r.Body = http.MaxBytesReader(w, r.Body, putBucketQuotaMaxBodyBytes)
 	defer r.Body.Close()
 
+	body, err := readRequestBody(r, putBucketQuotaMaxBodyBytes)
+	if err != nil {
+		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
+		return
+	}
+
 	var req bucketQuotaRequest
-	dec := json.NewDecoder(r.Body)
+	dec := json.NewDecoder(bytes.NewReader(body))
 	if err := dec.Decode(&req); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrMalformedXML)
 		return

weed/s3api/s3api_server.go

@@ -709,10 +709,8 @@ func (s3a *S3ApiServer) UnifiedPostHandler(w http.ResponseWriter, r *http.Reques
 	// Save the body first so we can restore it for STS handler signature verification
 	var bodyBytes []byte
 	if r.Body != nil {
-		// Limit body size to prevent DoS attacks
-		r.Body = http.MaxBytesReader(w, r.Body, iamRequestBodyLimit)
 		var err error
-		bodyBytes, err = io.ReadAll(r.Body)
+		bodyBytes, err = readRequestBody(r, iamRequestBodyLimit)
 		if err != nil {
 			glog.Errorf("failed to read request body: %v", err)
 			s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)

weed/s3api/s3err/error_handler.go

@@ -3,9 +3,11 @@ package s3err
 import (
 	"bytes"
 	"encoding/xml"
+	"io"
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/aws/aws-sdk-go/private/protocol/xml/xmlutil"
 	"github.com/gorilla/mux"
@@ -18,6 +20,8 @@ type mimeType string
 const (
 	mimeNone mimeType = ""
 	MimeXML  mimeType = "application/xml"
+
+	errorResponseBodyDrainTimeout = 30 * time.Second
 )
 
 func WriteAwsXMLResponse(w http.ResponseWriter, r *http.Request, statusCode int, result interface{}) {
@@ -62,10 +66,22 @@ func WriteErrorResponseWithMessage(w http.ResponseWriter, r *http.Request, error
 	if message != "" {
 		errorResponse.Message = message
 	}
+	drainRequestBody(w, r)
 	WriteXMLResponse(w, r, apiError.HTTPStatusCode, errorResponse)
 	PostLog(r, apiError.HTTPStatusCode, errorCode)
 }
 
+func drainRequestBody(w http.ResponseWriter, r *http.Request) {
+	if r == nil || r.Body == nil || r.Body == http.NoBody {
+		return
+	}
+	rc := http.NewResponseController(w)
+	if err := rc.SetReadDeadline(time.Now().Add(errorResponseBodyDrainTimeout)); err == nil {
+		defer rc.SetReadDeadline(time.Time{})
+	}
+	_, _ = io.Copy(io.Discard, r.Body)
+}
+
 func getRESTErrorResponse(err APIError, resource string, bucket, object, requestID string) RESTErrorResponse {
 	return RESTErrorResponse{
 		Code:       err.Code,

weed/s3api/s3err/error_handler_test.go

@@ -1,10 +1,13 @@
 package s3err
 
 import (
+	"bytes"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"regexp"
 	"testing"
+	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/seaweedfs/seaweedfs/weed/util/request_id"
@@ -26,6 +29,31 @@ func TestWriteErrorResponseReusesRequestID(t *testing.T) {
 	assert.Equal(t, "req-123", extractRequestIDFromBody(rr.Body.String()))
 }
 
+func TestWriteErrorResponseDrainsRequestBodyBeforeWriting(t *testing.T) {
+	body := &trackingReadCloser{data: bytes.Repeat([]byte("a"), 1024)}
+	req := httptest.NewRequest(http.MethodPut, "/bucket/object", nil)
+	req.Body = body
+	req.ContentLength = int64(body.remaining())
+	req = mux.SetURLVars(req, map[string]string{
+		"bucket": "bucket",
+		"object": "object",
+	})
+
+	rr := &drainCheckingResponseWriter{
+		header: make(http.Header),
+		body:   body,
+	}
+
+	WriteErrorResponse(rr, req, ErrInternalError)
+
+	assert.Empty(t, rr.writeHeaderErr)
+	assert.Equal(t, 0, body.remaining())
+	assert.Equal(t, http.StatusInternalServerError, rr.status)
+	assert.Len(t, rr.readDeadlines, 2)
+	assert.False(t, rr.readDeadlines[0].IsZero())
+	assert.True(t, rr.readDeadlines[1].IsZero())
+}
+
 func extractRequestIDFromBody(body string) string {
 	re := regexp.MustCompile(`<RequestId>([^<]+)</RequestId>`)
 	matches := re.FindStringSubmatch(body)
@@ -34,3 +62,57 @@ func extractRequestIDFromBody(body string) string {
 	}
 	return matches[1]
 }
+
+type trackingReadCloser struct {
+	data []byte
+}
+
+func (t *trackingReadCloser) Read(p []byte) (int, error) {
+	if len(t.data) == 0 {
+		return 0, io.EOF
+	}
+	n := copy(p, t.data)
+	t.data = t.data[n:]
+	return n, nil
+}
+
+func (t *trackingReadCloser) Close() error {
+	return nil
+}
+
+func (t *trackingReadCloser) remaining() int {
+	return len(t.data)
+}
+
+type drainCheckingResponseWriter struct {
+	header         http.Header
+	body           *trackingReadCloser
+	status         int
+	writeHeaderErr string
+	readDeadlines  []time.Time
+}
+
+func (d *drainCheckingResponseWriter) Header() http.Header {
+	return d.header
+}
+
+func (d *drainCheckingResponseWriter) Write(p []byte) (int, error) {
+	if d.status == 0 {
+		d.WriteHeader(http.StatusOK)
+	}
+	return len(p), nil
+}
+
+func (d *drainCheckingResponseWriter) WriteHeader(status int) {
+	d.status = status
+	if d.body.remaining() != 0 {
+		d.writeHeaderErr = "request body was not drained before WriteHeader"
+	}
+}
+
+func (d *drainCheckingResponseWriter) Flush() {}
+
+func (d *drainCheckingResponseWriter) SetReadDeadline(t time.Time) error {
+	d.readDeadlines = append(d.readDeadlines, t)
+	return nil
+}