Merge #147914

147914: cloud/amazon: add option to skip TLS verification for backup/restore r=sravotto a=sravotto

Previously, backup/restore jobs running against secure, on-premises S3 object stores that use self signed certificate would fail, unless cloudstorage.http.custom_ca is set.
However, customers want the ability to bypass the certificate validation in test environments.

To address this, we added a new URL parameter (AWS_SKIP_TLS_VERIFY) that can be set in the backup URL to bypass certificate validation. Note that a CA certificate is still required; this parameter means that the client will not verify the certificate.

Warning: Use this query parameter with caution, as it creates MITM vulnerabilities.

This change also adds a test case in roachtest for verification against a microCeph cluster.

Epic: none

Fixes: 147910

Release note: None

Co-authored-by: Silvano Ravotto <[email protected]>

Author: craig[bot]

pkg/cloud/amazon/aws_kms.go

@@ -144,7 +144,11 @@ func MakeAWSKMS(ctx context.Context, uri string, env cloud.KMSEnv) (cloud.KMS, e
 			return nil, errors.New(
 				"custom endpoints disallowed for aws kms due to --aws-kms-disable-http flag")
 		}
-		client, err := cloud.MakeHTTPClient(env.ClusterSettings(), cloud.NilMetrics, "aws", "KMS", "")
+		client, err := cloud.MakeHTTPClient(env.ClusterSettings(), cloud.NilMetrics,
+			cloud.HTTPClientConfig{
+				Bucket: "KMS",
+				Cloud:  "aws",
+			})
 		if err != nil {
 			return nil, err
 		}

pkg/cloud/amazon/s3_storage.go

@@ -59,6 +59,8 @@ const (
 	AWSUsePathStyle = "AWS_USE_PATH_STYLE"
 	// AWSSkipChecksumParam is the query parameter for SkipChecksum in S3 options.
 	AWSSkipChecksumParam = "AWS_SKIP_CHECKSUM"
+	// AWSSkipTLSVerify is the query parameter for skipping certificate verification.
+	AWSSkipTLSVerify = "AWS_SKIP_TLS_VERIFY"
 
 	// AWSServerSideEncryptionMode is the query parameter in an AWS URI, for the
 	// mode to be used for server side encryption. It can either be AES256 or
@@ -87,6 +89,9 @@ const (
 	scheme = "s3"
 
 	checksumAlgorithm = types.ChecksumAlgorithmSha256
+
+	// TODO(yevgeniy): Revisit retry logic.  Retrying 10 times seems arbitrary.
+	defaultRetryMaxAttempts = 10
 )
 
 // NightlyEnvVarS3Params maps param keys that get added to an S3
@@ -120,6 +125,16 @@ type s3Storage struct {
 	cached *s3Client
 }
 
+// retryMaxAttempts defines how many times we will retry a
+// S3 request on a retriable error.
+var retryMaxAttempts = defaultRetryMaxAttempts
+
+// InjectTestingRetryMaxAttempts is used to change the
+// default retries for tests that need quick fail.
+func InjectTestingRetryMaxAttempts(maxAttempts int) {
+	retryMaxAttempts = maxAttempts
+}
+
 // customRetryer implements the `request.Retryer` interface and allows for
 // customization of the retry behaviour of an AWS client.
 type customRetryer struct{}
@@ -202,7 +217,8 @@ type s3ClientConfig struct {
 	assumeRoleProvider                                           roleProvider
 	delegateRoleProviders                                        []roleProvider
 
-	skipChecksum bool
+	skipChecksum  bool
+	skipTLSVerify bool
 	// log.V(2) decides session init params so include it in key.
 	verbose bool
 }
@@ -236,6 +252,7 @@ func clientConfig(conf *cloudpb.ExternalStorage_S3) s3ClientConfig {
 		endpoint:              conf.Endpoint,
 		usePathStyle:          conf.UsePathStyle,
 		skipChecksum:          conf.SkipChecksum,
+		skipTLSVerify:         conf.SkipTLSVerify,
 		region:                conf.Region,
 		bucket:                conf.Bucket,
 		accessKey:             conf.AccessKey,
@@ -341,7 +358,15 @@ func parseS3URL(uri *url.URL) (cloudpb.ExternalStorage, error) {
 			return cloudpb.ExternalStorage{}, errors.Wrapf(err, "cannot parse %s as bool", AWSSkipChecksumParam)
 		}
 	}
-
+	skipTLSVerifyStr := s3URL.ConsumeParam(AWSSkipTLSVerify)
+	skipTLSVerifyBool := false
+	if skipTLSVerifyStr != "" {
+		var err error
+		skipTLSVerifyBool, err = strconv.ParseBool(skipTLSVerifyStr)
+		if err != nil {
+			return cloudpb.ExternalStorage{}, errors.Wrapf(err, "cannot parse %s as bool", AWSSkipTLSVerify)
+		}
+	}
 	conf.S3Config = &cloudpb.ExternalStorage_S3{
 		Bucket:                s3URL.Host,
 		Prefix:                s3URL.Path,
@@ -351,6 +376,7 @@ func parseS3URL(uri *url.URL) (cloudpb.ExternalStorage, error) {
 		Endpoint:              s3URL.ConsumeParam(AWSEndpointParam),
 		UsePathStyle:          pathStyleBool,
 		SkipChecksum:          skipChecksumBool,
+		SkipTLSVerify:         skipTLSVerifyBool,
 		Region:                s3URL.ConsumeParam(S3RegionParam),
 		Auth:                  s3URL.ConsumeParam(cloud.AuthParam),
 		ServerEncMode:         s3URL.ConsumeParam(AWSServerSideEncryptionMode),
@@ -573,21 +599,24 @@ func (s *s3Storage) newClient(ctx context.Context) (s3Client, string, error) {
 		loadOptions = append(loadOptions, option)
 	}
 
-	client, err := cloud.MakeHTTPClient(s.settings, s.metrics, "aws", s.opts.bucket, s.storageOptions.ClientName)
+	client, err := cloud.MakeHTTPClient(s.settings, s.metrics,
+		cloud.HTTPClientConfig{
+			Bucket:             s.opts.bucket,
+			Client:             s.storageOptions.ClientName,
+			Cloud:              "aws",
+			InsecureSkipVerify: s.opts.skipTLSVerify,
+		})
 	if err != nil {
 		return s3Client{}, "", err
 	}
 	addLoadOption(config.WithHTTPClient(client))
 
-	// TODO(yevgeniy): Revisit retry logic.  Retrying 10 times seems arbitrary.
-	retryMaxAttempts := 10
 	addLoadOption(config.WithRetryMaxAttempts(retryMaxAttempts))
 
 	addLoadOption(config.WithLogger(newLogAdapter(ctx)))
 	if s.opts.verbose {
 		addLoadOption(config.WithClientLogMode(awsVerboseLogging))
 	}
-
 	config.WithRetryer(func() aws.Retryer {
 		return retry.NewStandard(func(opts *retry.StandardOptions) {
 			opts.MaxAttempts = retryMaxAttempts

pkg/cloud/amazon/s3_storage_test.go

@@ -442,6 +442,70 @@ func TestPutS3Endpoint(t *testing.T) {
 			t.Fatal(err)
 		}
 	})
+
+	t.Run("skip-tls-verify", func(t *testing.T) {
+		ctx := context.Background()
+		srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+		defer srv.Close()
+		q := make(url.Values)
+		// Convert IP address to `localhost` to exercise the DNS-defining path in the S3 client.
+		localhostURL := strings.Replace(srv.URL, "127.0.0.1", "localhost", -1)
+		q.Add(AWSEndpointParam, localhostURL)
+		q.Add(AWSAccessKeyParam, "key")
+		q.Add(AWSSecretParam, "secret")
+		q.Add(S3RegionParam, "region")
+		q.Add(AWSUsePathStyle, "true")
+
+		// Verify that it fails without the flag.
+		u := url.URL{
+			Scheme:   "s3",
+			Host:     "bucket",
+			Path:     "subdir1/subdir2",
+			RawQuery: q.Encode(),
+		}
+
+		user := username.RootUserName()
+		ioConf := base.ExternalIODirConfig{}
+
+		conf, err := cloud.ExternalStorageConfFromURI(u.String(), user)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Setup a sink for the given args.
+		testSettings := cluster.MakeTestingClusterSettings()
+		clientFactory := blobs.TestBlobServiceClient("")
+		// Force to fail quickly.
+		InjectTestingRetryMaxAttempts(1)
+		storage, err := cloud.MakeExternalStorage(ctx, conf, ioConf, testSettings, clientFactory,
+			nil, nil, cloud.NilMetrics)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer storage.Close()
+		_, _, err = storage.ReadFile(ctx, "test file", cloud.ReadOptions{NoFileSize: true})
+		require.Error(t, err)
+
+		// Set AWSSkipTLSVerify to use with HTTPS Server with self signed certificates.
+		q.Add(AWSSkipTLSVerify, "true")
+		u.RawQuery = q.Encode()
+
+		confWithSkipVerify, err := cloud.ExternalStorageConfFromURI(u.String(), user)
+		if err != nil {
+			t.Fatal(err)
+		}
+		storageWithSkipVerify, err := cloud.MakeExternalStorage(ctx, confWithSkipVerify, ioConf, testSettings, clientFactory,
+			nil, nil, cloud.NilMetrics)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer storageWithSkipVerify.Close()
+		require.True(t, storageWithSkipVerify.Conf().S3Config.SkipTLSVerify)
+		_, _, err = storageWithSkipVerify.ReadFile(ctx, "another test file", cloud.ReadOptions{NoFileSize: true})
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
 }
 
 func TestS3DisallowCustomEndpoints(t *testing.T) {

pkg/cloud/azure/azure_storage.go

@@ -221,7 +221,12 @@ func makeAzureStorage(
 	}
 
 	options := args.ExternalStorageOptions()
-	t, err := cloud.MakeHTTPClient(args.Settings, args.MetricsRecorder, "azure", dest.AzureConfig.Container, options.ClientName)
+	t, err := cloud.MakeHTTPClient(args.Settings, args.MetricsRecorder,
+		cloud.HTTPClientConfig{
+			Bucket: dest.AzureConfig.Container,
+			Client: options.ClientName,
+			Cloud:  "azure",
+		})
 	if err != nil {
 		return nil, errors.Wrap(err, "azure: unable to create transport")
 	}

pkg/cloud/cloud_io.go

@@ -28,6 +28,21 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
+// HTTPClientConfig defines the settings for building the underlying transport.
+type HTTPClientConfig struct {
+	// Bucket is name of the bucket. Used to label metrics.
+	Bucket string
+	// Client type (e.g. CDC vs backup). Used to label metrics.
+	Client string
+	// Cloud is the storage provider. Used to label metrics.
+	Cloud string
+	// InsecureSkipVerify controls whether a client verifies the server's
+	// certificate chain and host name. If InsecureSkipVerify is true, crypto/tls
+	// accepts any certificate presented by the server and any host name in that
+	// certificate. In this mode, TLS is susceptible to machine-in-the-middle attacks.
+	InsecureSkipVerify bool
+}
+
 // Timeout is a cluster setting used for cloud storage interactions.
 var Timeout = settings.RegisterDurationSetting(
 	settings.ApplicationLevel,
@@ -77,9 +92,9 @@ var httpMetrics = settings.RegisterBoolSetting(
 // MakeHTTPClient makes an http client configured with the common settings used
 // for interacting with cloud storage (timeouts, retries, CA certs, etc).
 func MakeHTTPClient(
-	settings *cluster.Settings, metrics *Metrics, cloud, bucket, client string,
+	settings *cluster.Settings, metrics *Metrics, config HTTPClientConfig,
 ) (*http.Client, error) {
-	t, err := MakeTransport(settings, metrics, cloud, bucket, client)
+	t, err := MakeTransport(settings, metrics, config)
 	if err != nil {
 		return nil, err
 	}
@@ -99,10 +114,12 @@ func MakeHTTPClientForTransport(t http.RoundTripper) (*http.Client, error) {
 // used for interacting with cloud storage (timeouts, retries, CA certs, etc).
 // Prefer MakeHTTPClient where possible.
 func MakeTransport(
-	settings *cluster.Settings, metrics *Metrics, cloud, bucket, client string,
+	settings *cluster.Settings, metrics *Metrics, config HTTPClientConfig,
 ) (*http.Transport, error) {
 	var tlsConf *tls.Config
-	if pem := httpCustomCA.Get(&settings.SV); pem != "" {
+	if config.InsecureSkipVerify {
+		tlsConf = &tls.Config{InsecureSkipVerify: true}
+	} else if pem := httpCustomCA.Get(&settings.SV); pem != "" {
 		roots, err := x509.SystemCertPool()
 		if err != nil {
 			return nil, errors.Wrap(err, "could not load system root CA pool")
@@ -121,7 +138,7 @@ func MakeTransport(
 	// most bulk jobs.
 	t.MaxIdleConnsPerHost = 64
 	if metrics != nil {
-		t.DialContext = metrics.NetMetrics.Wrap(t.DialContext, cloud, bucket, client)
+		t.DialContext = metrics.NetMetrics.Wrap(t.DialContext, config.Cloud, config.Bucket, config.Client)
 	}
 	return t, nil
 }

pkg/cloud/cloudpb/external_storage.proto

@@ -63,7 +63,8 @@ message ExternalStorage {
     string temp_token = 5;
     string endpoint = 6;
     bool use_path_style=16;
-    bool skip_checksum=17;
+    bool skip_checksum=17; 
+    bool skip_tls_verify=18 [(gogoproto.customname) = "SkipTLSVerify"];
     string region = 7;
     string auth = 8;
     string server_enc_mode  = 9;
@@ -93,7 +94,7 @@ message ExternalStorage {
     // list so that the role specified in AssumeRoleProvider can be assumed.
     repeated AssumeRoleProvider delegate_role_providers = 15 [(gogoproto.nullable) = false];
 
-    // Next ID: 18;
+    // Next ID: 19;
   }
 
   message GCS {

pkg/cloud/gcp/gcs_storage.go

@@ -186,7 +186,11 @@ func makeGCSStorage(
 	}
 
 	clientName := args.ExternalStorageOptions().ClientName
-	baseTransport, err := cloud.MakeTransport(args.Settings, args.MetricsRecorder, "gcs", conf.Bucket, clientName)
+	baseTransport, err := cloud.MakeTransport(args.Settings, args.MetricsRecorder, cloud.HTTPClientConfig{
+		Bucket: conf.Bucket,
+		Client: clientName,
+		Cloud:  "gcs",
+	})
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create http transport")
 	}

pkg/cloud/httpsink/http_storage.go

@@ -67,7 +67,12 @@ func MakeHTTPStorage(
 	}
 
 	clientName := args.ExternalStorageOptions().ClientName
-	client, err := cloud.MakeHTTPClient(args.Settings, args.MetricsRecorder, "http", base, clientName)
+	client, err := cloud.MakeHTTPClient(args.Settings, args.MetricsRecorder,
+		cloud.HTTPClientConfig{
+			Cloud:  "http",
+			Bucket: "base",
+			Client: clientName,
+		})
 	if err != nil {
 		return nil, err
 	}