Harden HTTP Response Body Handling (#4489)

Author: SyntaxNode

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)

config/config.go

@@ -472,15 +472,19 @@ type Analytics struct {
 }
 
 type CurrencyConverter struct {
-	FetchURL             string `mapstructure:"fetch_url"`
-	FetchIntervalSeconds int    `mapstructure:"fetch_interval_seconds"`
-	StaleRatesSeconds    int    `mapstructure:"stale_rates_seconds"`
+	FetchURL                 string `mapstructure:"fetch_url"`
+	FetchTimeoutMilliseconds int    `mapstructure:"fetch_timeout_ms"`
+	FetchIntervalSeconds     int    `mapstructure:"fetch_interval_seconds"`
+	StaleRatesSeconds        int    `mapstructure:"stale_rates_seconds"`
 }
 
 func (cfg *CurrencyConverter) validate(errs []error) []error {
 	if cfg.FetchIntervalSeconds < 0 {
 		errs = append(errs, fmt.Errorf("currency_converter.fetch_interval_seconds must be in the range [0, %d]. Got %d", 0xffff, cfg.FetchIntervalSeconds))
 	}
+	if cfg.FetchTimeoutMilliseconds < 0 {
+		errs = append(errs, fmt.Errorf("currency_converter.fetch_timeout_ms must be 0 or greater. Got %d", cfg.FetchTimeoutMilliseconds))
+	}
 	return errs
 }
 
@@ -1173,6 +1177,7 @@ func SetupViper(v *viper.Viper, filename string, bidderInfos BidderInfos) {
 	v.SetDefault("ccpa.enforce", false)
 	v.SetDefault("lmt.enforce", true)
 	v.SetDefault("currency_converter.fetch_url", "https://cdn.jsdelivr.net/gh/prebid/currency-file@1/latest.json")
+	v.SetDefault("currency_converter.fetch_timeout_ms", 60000)      // 60 seconds
 	v.SetDefault("currency_converter.fetch_interval_seconds", 1800) // fetch currency rates every 30 minutes
 	v.SetDefault("currency_converter.stale_rates_seconds", 0)
 	v.SetDefault("default_request.type", "")

currency/currency_test.go

@@ -53,6 +53,7 @@ func TestGetAuctionCurrencyRates(t *testing.T) {
 
 		return NewRateConverter(
 			mockCurrencyClient,
+			60*time.Second,
 			"currency.fake.com",
 			24*time.Hour,
 		)

currency/rate_converter.go

@@ -1,6 +1,7 @@
 package currency
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -16,6 +17,7 @@ import (
 // RateConverter holds the currencies conversion rates dictionary
 type RateConverter struct {
 	httpClient          httpClient
+	httpTimeout         time.Duration
 	staleRatesThreshold time.Duration
 	syncSourceURL       string
 	rates               atomic.Value // Should only hold Rates struct
@@ -27,11 +29,13 @@ type RateConverter struct {
 // NewRateConverter returns a new RateConverter
 func NewRateConverter(
 	httpClient httpClient,
+	httpTimeout time.Duration,
 	syncSourceURL string,
 	staleRatesThreshold time.Duration,
 ) *RateConverter {
 	return &RateConverter{
 		httpClient:          httpClient,
+		httpTimeout:         httpTimeout,
 		staleRatesThreshold: staleRatesThreshold,
 		syncSourceURL:       syncSourceURL,
 		rates:               atomic.Value{},
@@ -43,7 +47,10 @@ func NewRateConverter(
 
 // fetch allows to retrieve the currencies rates from the syncSourceURL provided
 func (rc *RateConverter) fetch() (*Rates, error) {
-	request, err := http.NewRequest("GET", rc.syncSourceURL, nil)
+	ctx, cancel := context.WithTimeout(context.Background(), rc.httpTimeout)
+	defer cancel()
+
+	request, err := http.NewRequestWithContext(ctx, "GET", rc.syncSourceURL, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -52,23 +59,29 @@ 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)
+		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
+		return nil, fmt.Errorf("the currency rates request failed: %v", err)
 	}
 
 	updatedRates := &Rates{}
 	err = jsonutil.UnmarshalValid(bytesJSON, updatedRates)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("the currency rates request failed to parse json: %v", err)
 	}
 
 	return updatedRates, err

currency/rate_converter_test.go

@@ -123,6 +123,7 @@ func TestReadWriteRates(t *testing.T) {
 		}
 		currencyConverter := NewRateConverter(
 			&http.Client{},
+			60*time.Second,
 			url,
 			24*time.Hour,
 		)
@@ -181,6 +182,7 @@ func TestRateStaleness(t *testing.T) {
 	// Execute:
 	currencyConverter := NewRateConverter(
 		&http.Client{},
+		60*time.Second,
 		mockedHttpServer.URL,
 		30*time.Second, // stale rates threshold
 	)
@@ -268,6 +270,7 @@ func TestRatesAreNeverConsideredStale(t *testing.T) {
 	// Execute:
 	currencyConverter := NewRateConverter(
 		&http.Client{},
+		60*time.Second,
 		mockedHttpServer.URL,
 		0*time.Millisecond, // stale rates threshold
 	)
@@ -321,6 +324,7 @@ func TestRace(t *testing.T) {
 	interval := 1 * time.Millisecond
 	currencyConverter := NewRateConverter(
 		mockedHttpClient,
+		60*time.Second,
 		"currency.fake.com",
 		24*time.Hour,
 	)

endpoints/openrtb2/auction_benchmark_test.go

@@ -94,7 +94,7 @@ func BenchmarkOpenrtbEndpoint(b *testing.B) {
 		nilMetrics,
 		infos,
 		gdprPermsBuilder,
-		currency.NewRateConverter(&http.Client{}, "", time.Duration(0)),
+		currency.NewRateConverter(&http.Client{}, 60*time.Second, "", time.Duration(0)),
 		empty_fetcher.EmptyFetcher{},
 		&adscert.NilSigner{},
 		macros.NewStringIndexBasedReplacer(),