feat(cdc): Add superflag to enable TLS without CA or certs. (#7946) (#8097)

This will attempt to connect to Kafka over TLS using the system certs.

* Add helper function x.TLSBaseConfig.

  Sets the min TLS version to v1.2 along with the minimum cipher suites.

Author: danielmai

worker/server_state.go

@@ -42,7 +42,7 @@ const (
 	BadgerDefaults = `compression=snappy; numgoroutines=8;`
 	CacheDefaults  = `size-mb=1024; percentage=50,30,20;`
 	CDCDefaults    = `file=; kafka=; sasl_user=; sasl_password=; ca_cert=; client_cert=; ` +
-		`client_key=; sasl-mechanism=PLAIN;`
+		`client_key=; sasl-mechanism=PLAIN; tls=false;`
 	GraphQLDefaults = `introspection=true; debug=false; extensions=true; poll-interval=1s; `
 	LambdaDefaults  = `url=; num=1; port=20000; restart-after=30s; `
 	LimitDefaults   = `mutations=allow; query-edge=1000000; normalize-node=10000; ` +

worker/sink_handler.go

@@ -86,8 +86,18 @@ func newKafkaSink(config *z.SuperFlag) (Sink, error) {
 	saramaConf.Producer.Return.Successes = true
 	saramaConf.Producer.Return.Errors = true
 
-	if config.GetPath("ca-cert") != "" {
-		tlsCfg := &tls.Config{}
+	if config.GetBool("tls") && config.GetPath("ca-cert") == "" {
+		tlsCfg := x.TLSBaseConfig()
+		var pool *x509.CertPool
+		var err error
+		if pool, err = x509.SystemCertPool(); err != nil {
+			return nil, err
+		}
+		tlsCfg.RootCAs = pool
+		saramaConf.Net.TLS.Enable = true
+		saramaConf.Net.TLS.Config = tlsCfg
+	} else if config.GetPath("ca-cert") != "" {
+		tlsCfg := x.TLSBaseConfig()
 		var pool *x509.CertPool
 		var err error
 		if pool, err = x509.SystemCertPool(); err != nil {

x/tls_helper.go

@@ -263,11 +263,34 @@ func setupClientAuth(authType string) (tls.ClientAuthType, error) {
 	return tls.NoClientCert, nil
 }
 
+// TLSBaseConfig returns a *tls.Config with the base set of security
+// requirements (minimum TLS v1.2 and set of cipher suites)
+func TLSBaseConfig() *tls.Config {
+	tlsCfg := new(tls.Config)
+	tlsCfg.MinVersion = tls.VersionTLS12
+	tlsCfg.CipherSuites = []uint16{
+		tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305,
+		tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
+		tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
+		tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
+		tls.TLS_RSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_RSA_WITH_AES_256_CBC_SHA,
+	}
+	return tlsCfg
+}
+
 // GenerateServerTLSConfig creates and returns a new *tls.Config with the
 // configuration provided.
 func GenerateServerTLSConfig(config *TLSHelperConfig) (tlsCfg *tls.Config, err error) {
 	if config.CertRequired {
-		tlsCfg = new(tls.Config)
+		tlsCfg = TLSBaseConfig()
 		cert, err := tls.LoadX509KeyPair(config.Cert, config.Key)
 		if err != nil {
 			return nil, err
@@ -286,23 +309,6 @@ func GenerateServerTLSConfig(config *TLSHelperConfig) (tlsCfg *tls.Config, err e
 		}
 		tlsCfg.ClientAuth = auth
 
-		tlsCfg.MinVersion = tls.VersionTLS12
-		tlsCfg.CipherSuites = []uint16{
-			tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305,
-			tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
-			tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
-			tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
-			tls.TLS_RSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_RSA_WITH_AES_256_CBC_SHA,
-		}
-
 		return tlsCfg, nil
 	}
 	return nil, nil