Harden HTTP Response Body Handling

adapters/audienceNetwork/facebook.go

@@ -433,7 +433,7 @@ func Builder(bidderName openrtb_ext.BidderName, config config.Adapter, server co
 	return bidder, nil
 }
 
-func (a *adapter) MakeTimeoutNotification(req *adapters.RequestData) (*adapters.RequestData, []error) {
+func (a *adapter) MakeTimeoutNotification(req *adapters.RequestData) (*adapters.RequestData, error) {
 	var (
 		rID   string
 		pubID string
@@ -445,15 +445,13 @@ func (a *adapter) MakeTimeoutNotification(req *adapters.RequestData) (*adapters.
 	// corresponding imp's ID
 	rID, err = jsonparser.GetString(req.Body, "id")
 	if err != nil {
-		return &adapters.RequestData{}, []error{err}
+		return &adapters.RequestData{}, err
 	}
 
 	// The publisher ID is expected in the app object
 	pubID, err = jsonparser.GetString(req.Body, "app", "publisher", "id")
 	if err != nil {
-		return &adapters.RequestData{}, []error{
-			errors.New("path app.publisher.id not found in the request"),
-		}
+		return &adapters.RequestData{}, errors.New("path app.publisher.id not found in the request")
 	}
 
 	uri := fmt.Sprintf("https://www.facebook.com/audiencenetwork/nurl/?partner=%s&app=%s&auction=%s&ortb_loss_code=2", a.platformID, pubID, rID)

adapters/bidder.go

@@ -51,7 +51,7 @@ type TimeoutBidder interface {
 	//
 	// Do note that if MakeRequests returns multiple requests, and more than one of these times out, MakeTimeoutNotice will be called
 	// once for each timed out request.
-	MakeTimeoutNotification(req *RequestData) (*RequestData, []error)
+	MakeTimeoutNotification(req *RequestData) (*RequestData, error)
 }
 
 // BidderResponse wraps the server's response with the list of bids and the currency used by the bidder.

analytics/agma/sender.go

@@ -5,6 +5,7 @@ import (
 	"compress/gzip"
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"time"
@@ -74,6 +75,12 @@ func createHttpSender(httpClient *http.Client, endpoint config.AgmaAnalyticsHttp
 			glog.Errorf("[agmaAnalytics] Sending request failed %v", err)
 			return err
 		}
+		defer func() {
+			if _, err := io.Copy(io.Discard, resp.Body); err != nil {
+				glog.Errorf("[agmaAnalytics] Draining response body failed: %v", err)
+			}
+			resp.Body.Close()
+		}()
 
 		if resp.StatusCode != http.StatusOK {
 			glog.Errorf("[agmaAnalytics] Wrong code received %d instead of %d", resp.StatusCode, http.StatusOK)

analytics/pubstack/config.go

@@ -2,24 +2,38 @@ package pubstack
 
 import (
 	"encoding/json"
+	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"time"
 
 	"github.com/docker/go-units"
+	"github.com/golang/glog"
 )
 
 func fetchConfig(client *http.Client, endpoint *url.URL) (*Configuration, error) {
 	res, err := client.Get(endpoint.String())
 	if err != nil {
 		return nil, err
 	}
+	defer func() {
+		// read the entire response body to ensure full connection reuse if there's an
+		// error while decoding the json
+		if _, err := io.Copy(io.Discard, res.Body); err != nil {
+			glog.Errorf("[pubstack] Draining config response body failed: %v", err)
+		}
+		res.Body.Close()
+	}()
+
+	if res.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected status code: %d", res.StatusCode)
+	}
 
-	defer res.Body.Close()
 	c := Configuration{}
 	err = json.NewDecoder(res.Body).Decode(&c)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to decode config: %w", err)
 	}
 	return &c, nil
 }

analytics/pubstack/eventchannel/sender.go

@@ -3,6 +3,7 @@ package eventchannel
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"path"
@@ -27,7 +28,12 @@ func NewHttpSender(client *http.Client, endpoint string) Sender {
 		if err != nil {
 			return err
 		}
-		resp.Body.Close()
+		defer func() {
+			if _, err := io.Copy(io.Discard, resp.Body); err != nil {
+				glog.Errorf("[pubstack] Draining sender response body failed: %v", err)
+			}
+			resp.Body.Close()
+		}()
 
 		if resp.StatusCode != http.StatusOK {
 			glog.Errorf("[pubstack] Wrong code received %d instead of %d", resp.StatusCode, http.StatusOK)

currency/rate_converter.go

@@ -52,14 +52,20 @@ func (rc *RateConverter) fetch() (*Rates, error) {
 	if err != nil {
 		return nil, err
 	}
+	defer func() {
+		// read the entire response body to ensure full connection reuse if there's an
+		// invalid status code
+		if _, err := io.Copy(io.Discard, response.Body); err != nil {
+			glog.Errorf("Error draining conversion rates response body: %v", err)
+		}
+		response.Body.Close()
+	}()
 
 	if response.StatusCode >= 400 {
 		message := fmt.Sprintf("The currency rates request failed with status code %d", response.StatusCode)
 		return nil, &errortypes.BadServerResponse{Message: message}
 	}
 
-	defer response.Body.Close()
-
 	bytesJSON, err := io.ReadAll(response.Body)
 	if err != nil {
 		return nil, err

exchange/bidder.go

@@ -636,6 +636,7 @@ func (bidder *BidderAdapter) doRequestImpl(ctx context.Context, req *adapters.Re
 			err:     err,
 		}
 	}
+	defer httpResp.Body.Close()
 
 	respBody, err := io.ReadAll(httpResp.Body)
 	if err != nil {
@@ -644,7 +645,6 @@ func (bidder *BidderAdapter) doRequestImpl(ctx context.Context, req *adapters.Re
 			err:     err,
 		}
 	}
-	defer httpResp.Body.Close()
 
 	if httpResp.StatusCode < 200 || httpResp.StatusCode >= 400 {
 		if httpResp.StatusCode >= 500 {
@@ -671,14 +671,25 @@ func (bidder *BidderAdapter) doRequestImpl(ctx context.Context, req *adapters.Re
 func (bidder *BidderAdapter) doTimeoutNotification(timeoutBidder adapters.TimeoutBidder, req *adapters.RequestData, logger util.LogMsg) {
 	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
 	defer cancel()
-	toReq, errL := timeoutBidder.MakeTimeoutNotification(req)
-	if toReq != nil && len(errL) == 0 {
+
+	toReq, errMakeTimeoutNotification := timeoutBidder.MakeTimeoutNotification(req)
+	if toReq != nil && errMakeTimeoutNotification == nil {
 		httpReq, err := http.NewRequest(toReq.Method, toReq.Uri, bytes.NewBuffer(toReq.Body))
 		if err == nil {
 			httpReq.Header = req.Headers
 			httpResp, err := ctxhttp.Do(ctx, bidder.Client, httpReq)
+			if err == nil {
+				defer func() {
+					if _, err := io.Copy(io.Discard, httpResp.Body); err != nil {
+						glog.Errorf("TimeoutNotification: Draining response body failed %v", err)
+					}
+					httpResp.Body.Close()
+				}()
+			}
+
 			success := (err == nil && httpResp.StatusCode >= 200 && httpResp.StatusCode < 300)
 			bidder.me.RecordTimeoutNotice(success)
+
 			if bidder.config.Debug.TimeoutNotification.Log && !(bidder.config.Debug.TimeoutNotification.FailOnly && success) {
 				var msg string
 				if err == nil {
@@ -697,16 +708,15 @@ func (bidder *BidderAdapter) doTimeoutNotification(timeoutBidder adapters.Timeou
 			}
 		}
 	} else if bidder.config.Debug.TimeoutNotification.Log {
-		reqJSON, err := jsonutil.Marshal(req)
+		reqJSON, errMarshal := jsonutil.Marshal(req)
 		var msg string
-		if err == nil {
-			msg = fmt.Sprintf("TimeoutNotification: Failed to generate timeout request: error(%s), bidder request(%s)", errL[0].Error(), string(reqJSON))
+		if errMarshal == nil {
+			msg = fmt.Sprintf("TimeoutNotification: Failed to generate timeout request: error(%s), bidder request(%s)", errMarshal.Error(), string(reqJSON))
 		} else {
-			msg = fmt.Sprintf("TimeoutNotification: Failed to generate timeout request: error(%s), bidder request marshal failed(%s)", errL[0].Error(), err.Error())
+			msg = fmt.Sprintf("TimeoutNotification: Failed to generate timeout request: error(%s), bidder request marshal failed(%s)", errMakeTimeoutNotification.Error(), errMarshal.Error())
 		}
 		util.LogRandomSample(msg, logger, bidder.config.Debug.TimeoutNotification.SamplingRate)
 	}
-
 }
 
 type httpCallInfo struct {

exchange/bidder_test.go

@@ -2473,7 +2473,7 @@ func (bidder *notifyingBidder) MakeBids(internalRequest *openrtb2.BidRequest, ex
 	return nil, nil
 }
 
-func (bidder *notifyingBidder) MakeTimeoutNotification(req *adapters.RequestData) (*adapters.RequestData, []error) {
+func (bidder *notifyingBidder) MakeTimeoutNotification(req *adapters.RequestData) (*adapters.RequestData, error) {
 	return &bidder.notifyRequest, nil
 }
 

floors/fetcher.go

@@ -238,13 +238,13 @@ func (f *PriceFloorFetcher) fetchAndValidate(config config.AccountFloorFetch) (*
 	}
 
 	if len(floorResp) > (config.MaxFileSizeKB * 1024) {
-		glog.Errorf("Recieved invalid floor data from URL: %s, reason : floor file size is greater than MaxFileSize", config.URL)
+		glog.Errorf("Received invalid floor data from URL: %s, reason : floor file size is greater than MaxFileSize", config.URL)
 		return nil, 0
 	}
 
 	var priceFloors openrtb_ext.PriceFloorRules
 	if err = json.Unmarshal(floorResp, &priceFloors.Data); err != nil {
-		glog.Errorf("Recieved invalid price floor json from URL: %s", config.URL)
+		glog.Errorf("Received invalid price floor json from URL: %s", config.URL)
 		return nil, 0
 	}
 
@@ -271,6 +271,14 @@ func (f *PriceFloorFetcher) fetchFloorRulesFromURL(config config.AccountFloorFet
 	if err != nil {
 		return nil, 0, errors.New("error while getting response from url : " + err.Error())
 	}
+	defer func() {
+		// read the entire response body to ensure full connection reuse if there's an
+		// invalid status code
+		if _, err := io.Copy(io.Discard, httpResp.Body); err != nil {
+			glog.Errorf("Error while draining fetched floor response body: %v", err)
+		}
+		httpResp.Body.Close()
+	}()
 
 	if httpResp.StatusCode != http.StatusOK {
 		return nil, 0, errors.New("no response from server")
@@ -291,7 +299,6 @@ func (f *PriceFloorFetcher) fetchFloorRulesFromURL(config config.AccountFloorFet
 	if err != nil {
 		return nil, 0, errors.New("unable to read response")
 	}
-	defer httpResp.Body.Close()
 
 	return respBody, maxAge, nil
 }

pbs/usersync.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"strings"
@@ -45,7 +46,15 @@ func (deps *UserSyncDeps) VerifyRecaptcha(response string) error {
 	if err != nil {
 		return err
 	}
-	defer resp.Body.Close()
+	defer func() {
+		// read the entire response body to ensure full connection reuse if there's an
+		// error while decoding the json
+		if _, err := io.Copy(io.Discard, resp.Body); err != nil {
+			glog.Errorf("Captcha verify draining response body failed: %v", err)
+		}
+		resp.Body.Close()
+	}()
+
 	var gr = googleResponse{}
 	if err := json.NewDecoder(resp.Body).Decode(&gr); err != nil {
 		return err

stored_requests/backends/http_fetcher/fetcher.go

@@ -92,6 +92,7 @@ func (fetcher *HttpFetcher) FetchRequests(ctx context.Context, requestIDs []stri
 		return nil, nil, []error{err}
 	}
 	defer httpResp.Body.Close()
+
 	requestData, impData, errs = unpackResponse(httpResp)
 	return
 }
@@ -142,6 +143,7 @@ func (fetcher *HttpFetcher) FetchAccounts(ctx context.Context, accountIDs []stri
 		}
 	}
 	defer httpResp.Body.Close()
+
 	respBytes, err := io.ReadAll(httpResp.Body)
 	if err != nil {
 		return nil, []error{

stored_requests/events/http/http.go

@@ -96,6 +96,7 @@ type HTTPEvents struct {
 func (e *HTTPEvents) fetchAll() {
 	ctx, cancel := e.ctxProducer()
 	defer cancel()
+
 	resp, err := ctxhttp.Get(ctx, e.client, e.Endpoint)
 	if respObj, ok := e.parse(e.Endpoint, resp, err); ok &&
 		(len(respObj.StoredRequests) > 0 || len(respObj.StoredImps) > 0 || len(respObj.StoredResponses) > 0 || len(respObj.Accounts) > 0) {