aws/request: Fix HTTP Request Body race condition.

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