aws/request: Fix HTTP Request Body race condition. (#874)

* aws/request: Fix HTTP Request Body race condition.

Fixes the possible race conditions where the SDK's request body could be
mutated by the HTTP request directly. This could occur because the SDK did
not consistently protect the SDK's request#Request.Body parameter from the
http#Request.Body. This inconsistency prevented the SDK from protecting
against race conditions when HTTP Client's Transport would still be reading
on the http#Request.body and the SDK would be attempting to retry the
API operation.

During a API operation retry the request#Request.Body needs to be rewound
for then next retry of the HTTP request. This is protected now by creating
a strong separation between the request#Request.Body and http#Request.Body.
A per request wrapper is created for each HTTP request attempt that is safe
to use concurrently.

Fix #868, #869, #871

aws/client/client.go

@@ -2,7 +2,6 @@ package client
 
 import (
 	"fmt"
-	"io/ioutil"
 	"net/http/httputil"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -104,8 +103,7 @@ func logRequest(r *request.Request) {
 		// Reset the request body because dumpRequest will re-wrap the r.HTTPRequest's
 		// Body as a NoOpCloser and will not be reset after read by the HTTP
 		// client reader.
-		r.Body.Seek(r.BodyStart, 0)
-		r.HTTPRequest.Body = ioutil.NopCloser(r.Body)
+		r.ResetBody()
 	}
 
 	r.Config.Logger.Log(fmt.Sprintf(logReqMsg, r.ClientInfo.ServiceName, r.Operation.Name, string(dumpedBody)))

aws/request/offset_reader.go

@@ -9,7 +9,7 @@ import (
 // with retrying requests
 type offsetReader struct {
 	buf    io.ReadSeeker
-	lock   sync.RWMutex
+	lock   sync.Mutex
 	closed bool
 }
 
@@ -21,18 +21,19 @@ func newOffsetReader(buf io.ReadSeeker, offset int64) *offsetReader {
 	return reader
 }
 
-// Close is a thread-safe close. Uses the write lock.
+// Close will close the instance of the offset reader's access to
+// the underlying io.ReadSeeker.
 func (o *offsetReader) Close() error {
 	o.lock.Lock()
 	defer o.lock.Unlock()
 	o.closed = true
 	return nil
 }
 
-// Read is a thread-safe read using a read lock.
+// Read is a thread-safe read of the underlying io.ReadSeeker
 func (o *offsetReader) Read(p []byte) (int, error) {
-	o.lock.RLock()
-	defer o.lock.RUnlock()
+	o.lock.Lock()
+	defer o.lock.Unlock()
 
 	if o.closed {
 		return 0, io.EOF
@@ -41,6 +42,14 @@ func (o *offsetReader) Read(p []byte) (int, error) {
 	return o.buf.Read(p)
 }
 
+// Seek is a thread-safe seeking operation.
+func (o *offsetReader) Seek(offset int64, whence int) (int64, error) {
+	o.lock.Lock()
+	defer o.lock.Unlock()
+
+	return o.buf.Seek(offset, whence)
+}
+
 // CloseAndCopy will return a new offsetReader with a copy of the old buffer
 // and close the old buffer.
 func (o *offsetReader) CloseAndCopy(offset int64) *offsetReader {

aws/request/offset_reader_test.go

@@ -24,6 +24,23 @@ func TestOffsetReaderRead(t *testing.T) {
 	assert.Equal(t, buf, tempBuf)
 }
 
+func TestOffsetReaderSeek(t *testing.T) {
+	buf := []byte("testData")
+	reader := newOffsetReader(bytes.NewReader(buf), 0)
+
+	orig, err := reader.Seek(0, 1)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), orig)
+
+	n, err := reader.Seek(0, 2)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(len(buf)), n)
+
+	n, err = reader.Seek(orig, 0)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), n)
+}
+
 func TestOffsetReaderClose(t *testing.T) {
 	buf := []byte("testData")
 	reader := &offsetReader{buf: bytes.NewReader(buf)}

aws/request/request.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"net/url"
 	"reflect"
@@ -42,6 +41,12 @@ type Request struct {
 	LastSignedAt     time.Time
 
 	built bool
+
+	// Need to persist an intermideant body betweend the input Body and HTTP
+	// request body because the HTTP Client's transport can maintain a reference
+	// to the HTTP request's body after the client has returned. This value is
+	// safe to use concurrently and rewraps the input Body for each HTTP request.
+	safeBody *offsetReader
 }
 
 // An Operation is the service API operation to be made.
@@ -135,8 +140,8 @@ func (r *Request) SetStringBody(s string) {
 
 // SetReaderBody will set the request's body reader.
 func (r *Request) SetReaderBody(reader io.ReadSeeker) {
-	r.HTTPRequest.Body = newOffsetReader(reader, 0)
 	r.Body = reader
+	r.ResetBody()
 }
 
 // Presign returns the request's signed URL. Error will be returned
@@ -220,6 +225,24 @@ func (r *Request) Sign() error {
 	return r.Error
 }
 
+// ResetBody rewinds the request body backto its starting position, and
+// set's the HTTP Request body reference. When the body is read prior
+// to being sent in the HTTP request it will need to be rewound.
+func (r *Request) ResetBody() {
+	if r.safeBody != nil {
+		r.safeBody.Close()
+	}
+
+	r.safeBody = newOffsetReader(r.Body, r.BodyStart)
+	r.HTTPRequest.Body = r.safeBody
+}
+
+// GetBody will return an io.ReadSeeker of the Request's underlying
+// input body with a concurrency safe wrapper.
+func (r *Request) GetBody() io.ReadSeeker {
+	return r.safeBody
+}
+
 // Send will send the request returning error if errors are encountered.
 //
 // Send will sign the request prior to sending. All Send Handlers will
@@ -231,6 +254,8 @@ func (r *Request) Sign() error {
 //
 // readLoop() and getConn(req *Request, cm connectMethod)
 // https://github.com/golang/go/blob/master/src/net/http/transport.go
+//
+// Send will not close the request.Request's body.
 func (r *Request) Send() error {
 	for {
 		if aws.BoolValue(r.Retryable) {
@@ -239,21 +264,15 @@ func (r *Request) Send() error {
 					r.ClientInfo.ServiceName, r.Operation.Name, r.RetryCount))
 			}
 
-			var body io.ReadCloser
-			if reader, ok := r.HTTPRequest.Body.(*offsetReader); ok {
-				body = reader.CloseAndCopy(r.BodyStart)
-			} else {
-				if r.Config.Logger != nil {
-					r.Config.Logger.Log("Request body type has been overwritten. May cause race conditions")
-				}
-				r.Body.Seek(r.BodyStart, 0)
-				body = ioutil.NopCloser(r.Body)
-			}
+			// The previous http.Request will have a reference to the r.Body
+			// and the HTTP Client's Transport may still be reading from
+			// the request's body even though the Client's Do returned.
+			r.HTTPRequest = copyHTTPRequest(r.HTTPRequest, nil)
+			r.ResetBody()
 
-			r.HTTPRequest = copyHTTPRequest(r.HTTPRequest, body)
+			// Closing response body to ensure that no response body is leaked
+			// between retry attempts.
 			if r.HTTPResponse != nil && r.HTTPResponse.Body != nil {
-				// Closing response body. Since we are setting a new request to send off, this
-				// response will get squashed and leaked.
 				r.HTTPResponse.Body.Close()
 			}
 		}
@@ -281,7 +300,6 @@ func (r *Request) Send() error {
 			debugLogReqError(r, "Send Request", true, err)
 			continue
 		}
-
 		r.Handlers.UnmarshalMeta.Run(r)
 		r.Handlers.ValidateResponse.Run(r)
 		if r.Error != nil {

aws/signer/v4/v4.go

@@ -366,7 +366,9 @@ func signSDKRequestWithCurrTime(req *request.Request, curTimeFn func() time.Time
 		signingTime = req.LastSignedAt
 	}
 
-	signedHeaders, err := v4.signWithBody(req.HTTPRequest, req.Body, name, region, req.ExpireTime, signingTime)
+	signedHeaders, err := v4.signWithBody(req.HTTPRequest, req.GetBody(),
+		name, region, req.ExpireTime, signingTime,
+	)
 	if err != nil {
 		req.Error = err
 		req.SignedHeaderVals = nil