cloud/amazon: add option to skip TLS verification for backup/restore

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

Author: sravotto

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.Config{
+				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
@@ -202,7 +204,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 +239,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 +345,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 +363,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,7 +586,13 @@ 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.Config{
+			Bucket:             s.opts.bucket,
+			Client:             s.storageOptions.ClientName,
+			Cloud:              "aws",
+			InsecureSkipVerify: s.opts.skipTLSVerify,
+		})
 	if err != nil {
 		return s3Client{}, "", err
 	}

pkg/cloud/amazon/s3_storage_test.go

@@ -442,6 +442,53 @@ 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")
+		// Set AWSSkipTLSVerify to use with HTTPS Server with self signed certificates.
+		q.Add(AWSSkipTLSVerify, "true")
+
+		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("")
+
+		storage, err := cloud.MakeExternalStorage(ctx, conf, ioConf, testSettings, clientFactory,
+			nil, nil, cloud.NilMetrics)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer storage.Close()
+		require.True(t, storage.Conf().S3Config.SkipTLSVerify)
+		_, _, err = storage.ReadFile(ctx, "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.Config{
+			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"
 )
 
+// Config defines the settings for building the underlying transport.
+type Config 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 Config,
 ) (*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 Config,
 ) (*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.Config{
+		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.Config{
+			Cloud:  "http",
+			Bucket: "base",
+			Client: clientName,
+		})
 	if err != nil {
 		return nil, err
 	}

pkg/cmd/roachtest/tests/s3_clone_backup_restore.go

@@ -24,48 +24,76 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
+type s3CloneSecureOption int
+
+const (
+	s3ClonePlain      s3CloneSecureOption = iota // Use HTTP
+	s3CloneRequireTLS                            // Use HTTPS, but skip certification verification.
+	s3CloneVerifyTLS                             // Use HTTPS and verify certificates.
+)
+
+var s3CloneSecureOptions = []s3CloneSecureOption{s3ClonePlain, s3CloneRequireTLS, s3CloneVerifyTLS}
+
+func (o s3CloneSecureOption) String() string {
+	switch o {
+	case s3ClonePlain:
+		return "plain"
+	case s3CloneRequireTLS:
+		return "require"
+	case s3CloneVerifyTLS:
+		return "verify"
+	default:
+		panic("invalid option")
+	}
+}
+
 // registerBackupS3Clones validates backup/restore compatibility with S3 clones.
 func registerBackupS3Clones(r registry.Registry) {
 	// Running against a microceph cluster deployed on a GCE instance.
 	for _, cephVersion := range []string{"reef", "squid"} {
-		r.Add(registry.TestSpec{
-			Name:                      fmt.Sprintf("backup/ceph/%s", cephVersion),
-			Owner:                     registry.OwnerFieldEng,
-			Cluster:                   r.MakeClusterSpec(4, spec.WorkloadNodeCount(1)),
-			EncryptionSupport:         registry.EncryptionMetamorphic,
-			Leases:                    registry.MetamorphicLeases,
-			CompatibleClouds:          registry.Clouds(spec.GCE),
-			Suites:                    registry.Suites(registry.Nightly),
-			TestSelectionOptOutSuites: registry.Suites(registry.Nightly),
-			Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
-				v := s3BackupRestoreValidator{
-					t:            t,
-					c:            c,
-					crdbNodes:    c.CRDBNodes(),
-					csvPort:      8081,
-					importNode:   c.Node(1),
-					rows:         1000,
-					workloadNode: c.WorkloadNode(),
-				}
-				v.startCluster(ctx)
-				ceph := cephManager{
-					t:      t,
-					c:      c,
-					bucket: backupTestingBucket,
-					// For now, we use the workload node as the cephNode
-					cephNodes: c.Node(c.Spec().NodeCount),
-					key:       randomString(32),
-					secret:    randomString(64),
-					// reef `microceph enable rgw` does not support `--ssl-certificate`
-					// so we'll test a non-secure version.
-					secure:  cephVersion != "reef",
-					version: cephVersion,
-				}
-				ceph.install(ctx)
-				defer ceph.cleanup(ctx)
-				v.validateBackupRestore(ctx, ceph)
-			},
-		})
+		for _, secureOption := range s3CloneSecureOptions {
+			if cephVersion == "reef" && secureOption != s3ClonePlain {
+				// reef `microceph enable rgw` does not support `--ssl-certificate`
+				// so we'll test only a non-secure version.
+				continue
+			}
+			r.Add(registry.TestSpec{
+				Name:                      fmt.Sprintf("backup/ceph/%s/%s", cephVersion, secureOption),
+				Owner:                     registry.OwnerFieldEng,
+				Cluster:                   r.MakeClusterSpec(4, spec.WorkloadNodeCount(1)),
+				EncryptionSupport:         registry.EncryptionMetamorphic,
+				Leases:                    registry.MetamorphicLeases,
+				CompatibleClouds:          registry.Clouds(spec.GCE),
+				Suites:                    registry.Suites(registry.Nightly),
+				TestSelectionOptOutSuites: registry.Suites(registry.Nightly),
+				Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
+					v := s3BackupRestoreValidator{
+						t:            t,
+						c:            c,
+						crdbNodes:    c.CRDBNodes(),
+						csvPort:      8081,
+						importNode:   c.Node(1),
+						rows:         1000,
+						workloadNode: c.WorkloadNode(),
+					}
+					v.startCluster(ctx)
+					ceph := cephManager{
+						t:      t,
+						c:      c,
+						bucket: backupTestingBucket,
+						// For now, we use the workload node as the cephNode
+						cephNodes: c.Node(c.Spec().NodeCount),
+						key:       randomString(32),
+						secret:    randomString(64),
+						secure:    secureOption,
+						version:   cephVersion,
+					}
+					ceph.install(ctx)
+					defer ceph.cleanup(ctx)
+					v.validateBackupRestore(ctx, ceph)
+				},
+			})
+		}
 	}
 
 	r.Add(registry.TestSpec{