Repository: apache/pulsar-client-go
Pull request #855 (2022-09-30 08:34:07)

Add TLS transport config
### Motivation

Add a TLS transport config for the client.

In the old version, we can only use the TLS authentication to config the TLS transport, so we cannot use the other authentication, which is confusing, so we need to add a TLS transport config for the client, which would resolve the issue I mentioned.

When using this config with the TLS authentication, the TLS authentication config is preferred.



Patch:
diff --git a/pulsar/client.go b/pulsar/client.go
index 22b12ef..29a582f 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -105,6 +105,12 @@ type ClientOptions struct {
 	// Example: `Authentication: NewAuthenticationTLS("my-cert.pem", "my-key.pem")`
 	Authentication
 
+	// Set the path to the TLS key file
+	TLSKeyFilePath string
+
+	// Set the path to the TLS certificate file
+	TLSCertificateFile string
+
 	// Set the path to the trusted TLS certificate file
 	TLSTrustCertsFilePath string
 
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index e7fa642..712e197 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -71,6 +71,8 @@ func newClient(options ClientOptions) (Client, error) {
 	case "pulsar+ssl", "https":
 		tlsConfig = &internal.TLSOptions{
 			AllowInsecureConnection: options.TLSAllowInsecureConnection,
+			KeyFile:                 options.TLSKeyFilePath,
+			CertFile:                options.TLSCertificateFile,
 			TrustCertsFilePath:      options.TLSTrustCertsFilePath,
 			ValidateHostname:        options.TLSValidateHostname,
 		}
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 4196396..6a410a2 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -47,6 +47,8 @@ const (
 )
 
 type TLSOptions struct {
+	KeyFile                 string
+	CertFile                string
 	TrustCertsFilePath      string
 	AllowInsecureConnection bool
 	ValidateHostname        bool
@@ -982,6 +984,14 @@ func (c *connection) getTLSConfig() (*tls.Config, error) {
 		tlsConfig.ServerName = c.physicalAddr.Hostname()
 	}
 
+	if c.tlsOptions.CertFile != "" || c.tlsOptions.KeyFile != "" {
+		cert, err := tls.LoadX509KeyPair(c.tlsOptions.CertFile, c.tlsOptions.KeyFile)
+		if err != nil {
+			return nil, errors.New(err.Error())
+		}
+		tlsConfig.Certificates = []tls.Certificate{cert}
+	}
+
 	cert, err := c.auth.GetTLSCertificate()
 	if err != nil {
 		return nil, err
diff --git a/pulsar/internal/http_client.go b/pulsar/internal/http_client.go
index 8c494dc..3c68dac 100644
--- a/pulsar/internal/http_client.go
+++ b/pulsar/internal/http_client.go
@@ -345,6 +345,14 @@ func getDefaultTransport(tlsConfig *TLSOptions) (http.RoundTripper, error) {
 			cfg.RootCAs = x509.NewCertPool()
 			cfg.RootCAs.AppendCertsFromPEM(rootCA)
 		}
+
+		if tlsConfig.CertFile != "" || tlsConfig.KeyFile != "" {
+			cert, err := tls.LoadX509KeyPair(tlsConfig.CertFile, tlsConfig.KeyFile)
+			if err != nil {
+				return nil, errors.New(err.Error())
+			}
+			cfg.Certificates = []tls.Certificate{cert}
+		}
 		transport.TLSClientConfig = cfg
 	}
 	transport.MaxIdleConnsPerHost = 10


Test patch:
diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index ba8f6eb..c98a542 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -1066,3 +1066,38 @@ func TestHTTPSBasicAuth(t *testing.T) {
 
 	client.Close()
 }
+
+func testTLSTransportWithBasicAuth(t *testing.T, url string) {
+	t.Helper()
+
+	basicAuth, err := NewAuthenticationBasic("admin", "123456")
+	require.NoError(t, err)
+	require.NotNil(t, basicAuth)
+
+	client, err := NewClient(ClientOptions{
+		URL:                   url,
+		TLSCertificateFile:    tlsClientCertPath,
+		TLSKeyFilePath:        tlsClientKeyPath,
+		TLSTrustCertsFilePath: caCertsPath,
+		Authentication:        basicAuth,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newTopicName(),
+	})
+
+	require.NoError(t, err)
+	require.NotNil(t, producer)
+
+	client.Close()
+}
+
+func TestServiceUrlTLSWithTLSTransportWithBasicAuth(t *testing.T) {
+	testTLSTransportWithBasicAuth(t, serviceURLTLS)
+}
+
+func TestWebServiceUrlTLSWithTLSTransportWithBasicAuth(t *testing.T) {
+	testTLSTransportWithBasicAuth(t, webServiceURLTLS)
+}