[v0.14] Reuse transports with same settings for Fleet CLI (#4410)

Author: aruiz14

internal/bundlereader/auth.go

@@ -19,6 +19,13 @@ type Auth struct {
 	BasicHTTP          bool   `json:"basicHTTP,omitempty"`
 }
 
+func toByte(v bool) byte {
+	if v {
+		return byte(1)
+	}
+	return byte(0)
+}
+
 func ReadHelmAuthFromSecret(ctx context.Context, c client.Reader, req types.NamespacedName) (Auth, error) {
 	if req.Name == "" {
 		return Auth{}, nil

internal/bundlereader/charturl.go

@@ -2,14 +2,19 @@ package bundlereader
 
 import (
 	"context"
+	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/binary"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/Masterminds/semver/v3"
 	fleet "github.com/rancher/fleet/pkg/apis/fleet.cattle.io/v1alpha1"
@@ -22,6 +27,16 @@ import (
 	"oras.land/oras-go/v2/registry/remote/errcode"
 )
 
+const (
+	// safety timeout to prevent unbounded requests
+	httpClientTimeout = 5 * time.Minute
+)
+
+var (
+	transportsCache      = map[string]http.RoundTripper{}
+	transportsCacheMutex sync.RWMutex
+)
+
 // ChartVersion returns the version of the helm chart from a helm repo server, by
 // inspecting the repo's index.yaml
 func ChartVersion(location fleet.HelmOptions, a Auth) (string, error) {
@@ -234,25 +249,69 @@ func GetOCITag(r *remote.Repository, v string) (string, error) {
 }
 
 func getHTTPClient(auth Auth) *http.Client {
-	client := &http.Client{}
+	return &http.Client{
+		Transport: transportForAuth(auth.InsecureSkipVerify, auth.CABundle),
+		Timeout:   httpClientTimeout,
+	}
+}
+
+func transportHash(insecureSkipVerify bool, caBundle []byte) string {
+	hash := sha256.New()
+
+	// Write a length prefix for every field to avoid collisions
+	lenBuf := make([]byte, 8)
+	writeField := func(data []byte) {
+		binary.LittleEndian.PutUint64(lenBuf, uint64(len(data)))
+		hash.Write(lenBuf)
+		hash.Write(data)
+	}
+
+	for _, v := range [][]byte{ // values to hash
+		caBundle,
+		{toByte(insecureSkipVerify)},
+	} {
+		writeField(v)
+	}
 
+	return hex.EncodeToString(hash.Sum(nil))
+}
+
+func transportForAuth(insecureSkipVerify bool, caBundle []byte) http.RoundTripper {
+	// We don't need the full hash
+	hash := transportHash(insecureSkipVerify, caBundle)
+
+	// Fast path: valid transport already exists
+	transportsCacheMutex.RLock()
+	rt, ok := transportsCache[hash]
+	transportsCacheMutex.RUnlock()
+	if ok {
+		return rt
+	}
+
+	transportsCacheMutex.Lock()
+	defer transportsCacheMutex.Unlock()
+
+	// Check again using write lock
+	if rt, ok := transportsCache[hash]; ok {
+		return rt
+	}
+
+	// Create new transport
 	transport := http.DefaultTransport.(*http.Transport).Clone()
 	transport.TLSClientConfig = &tls.Config{
-		InsecureSkipVerify: auth.InsecureSkipVerify, // nolint:gosec
+		InsecureSkipVerify: insecureSkipVerify, //nolint:gosec
 	}
-
-	if auth.CABundle != nil {
+	if caBundle != nil {
 		pool, err := x509.SystemCertPool()
 		if err != nil {
 			pool = x509.NewCertPool()
 		}
-		pool.AppendCertsFromPEM(auth.CABundle)
+		pool.AppendCertsFromPEM(caBundle)
 
 		transport.TLSClientConfig.RootCAs = pool
 		transport.TLSClientConfig.MinVersion = tls.VersionTLS12
 	}
 
-	client.Transport = transport
-
-	return client
+	transportsCache[hash] = transport
+	return transport
 }

internal/bundlereader/loaddirectory.go

@@ -3,8 +3,6 @@ package bundlereader
 import (
 	"bufio"
 	"context"
-	"crypto/tls"
-	"crypto/x509"
 	"encoding/base64"
 	"fmt"
 	"io/fs"
@@ -179,7 +177,7 @@ func loadDirectory(ctx context.Context, opts loadOpts, dir directory) ([]fleet.B
 		if opts.compress || !utf8.Valid(data) {
 			content, err := content.Base64GZ(data)
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("decoding compressed base64 data: %w", err)
 			}
 			r.Content = content
 			r.Encoding = "base64+gz"
@@ -211,7 +209,7 @@ func GetContent(ctx context.Context, base, source, version string, auth Auth, di
 	if hasOCIURL.MatchString(source) {
 		source, err = downloadOCIChart(source, version, temp, auth)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("downloading OCI chart from %q: %w", orgSource, err)
 		}
 	}
 
@@ -275,7 +273,7 @@ func GetContent(ctx context.Context, base, source, version string, auth Auth, di
 	}
 
 	if _, err := client.Get(ctx, req); err != nil {
-		return nil, err
+		return nil, fmt.Errorf("retrieving file from %q: %w", source, err)
 	}
 
 	files := map[string][]byte{}
@@ -312,7 +310,7 @@ func GetContent(ctx context.Context, base, source, version string, auth Auth, di
 			// try to update possible dependencies.
 			if !disableDepsUpdate && helmupdater.ChartYAMLExists(path) {
 				if err = helmupdater.UpdateHelmDependencies(path); err != nil {
-					return err
+					return fmt.Errorf("updating helm dependencies: %w", err)
 				}
 			}
 			// Skip .fleetignore'd and hidden directories
@@ -432,7 +430,7 @@ func downloadOCIChart(name, version, path string, auth Auth) (string, error) {
 
 func newHttpGetter(auth Auth) *getter.HttpGetter {
 	httpGetter := &getter.HttpGetter{
-		Client: &http.Client{},
+		Client: getHTTPClient(auth),
 	}
 
 	if auth.Username != "" && auth.Password != "" {
@@ -441,25 +439,6 @@ func newHttpGetter(auth Auth) *getter.HttpGetter {
 		httpGetter.Header = header
 	}
 
-	transport := http.DefaultTransport.(*http.Transport).Clone()
-	if auth.CABundle != nil {
-		pool, err := x509.SystemCertPool()
-		if err != nil {
-			pool = x509.NewCertPool()
-		}
-		pool.AppendCertsFromPEM(auth.CABundle)
-		transport.TLSClientConfig = &tls.Config{
-			RootCAs:            pool,
-			MinVersion:         tls.VersionTLS12,
-			InsecureSkipVerify: auth.InsecureSkipVerify, // nolint:gosec
-		}
-	} else if auth.InsecureSkipVerify {
-		transport.TLSClientConfig = &tls.Config{
-			InsecureSkipVerify: auth.InsecureSkipVerify, // nolint:gosec
-		}
-	}
-	httpGetter.Client.Transport = transport
-
 	return httpGetter
 }