From 5ce7074fcb66bebc87141415673cc625fe3322a6 Mon Sep 17 00:00:00 2001
From: Mostafa Moradian <mostafamoradian0@gmail.com>
Date: Wed, 19 Oct 2022 20:38:04 +0200
Subject: [PATCH] Fix bug in TLS config handling (#170)

* Decouple client and server certificate processing to prevent mTLS enforcement on everything
* Update tests to reflect changes
---
 auth.go      | 45 +++++++++++++++++++++++----------------------
 auth_test.go | 34 ++++++++++++++++++++++++++--------
 2 files changed, 49 insertions(+), 30 deletions(-)

diff --git a/auth.go b/auth.go
index 52d283f..d268186 100644
--- a/auth.go
+++ b/auth.go
@@ -119,9 +119,7 @@ func GetTLSConfig(tlsConfig TLSConfig) (*tls.Config, *Xk6KafkaError) {
 	tlsObject := newTLSObject(tlsConfig)
 
 	if tlsConfig.EnableTLS {
-		if tlsConfig.ClientCertPem == "" &&
-			tlsConfig.ClientKeyPem == "" &&
-			tlsConfig.ServerCaPem == "" {
+		if tlsConfig.ServerCaPem == "" {
 			return tlsObject, nil
 		}
 	} else {
@@ -130,28 +128,31 @@ func GetTLSConfig(tlsConfig TLSConfig) (*tls.Config, *Xk6KafkaError) {
 			noTLSConfig, "No TLS config provided. Continuing with TLS disabled.", nil)
 	}
 
-	// Load the client cert and key if they are provided
-	if err := fileExists(tlsConfig.ClientCertPem); err != nil {
-		return nil, err
-	}
+	if tlsConfig.ClientCertPem != "" && tlsConfig.ClientKeyPem != "" {
+		// Load the client certificate and key if provided
+		if err := fileExists(tlsConfig.ClientCertPem); err != nil {
+			return nil, err
+		}
 
-	if err := fileExists(tlsConfig.ClientKeyPem); err != nil {
-		return nil, err
-	}
+		if err := fileExists(tlsConfig.ClientKeyPem); err != nil {
+			return nil, err
+		}
 
-	var cert tls.Certificate
-	cert, err := tls.LoadX509KeyPair(tlsConfig.ClientCertPem, tlsConfig.ClientKeyPem)
-	if err != nil {
-		return nil, NewXk6KafkaError(
-			failedLoadX509KeyPair,
-			fmt.Sprintf(
-				"Error creating x509 key pair from \"%s\" and \"%s\".",
-				tlsConfig.ClientCertPem,
-				tlsConfig.ClientKeyPem),
-			err)
+		cert, err := tls.LoadX509KeyPair(tlsConfig.ClientCertPem, tlsConfig.ClientKeyPem)
+		if err != nil {
+			return nil, NewXk6KafkaError(
+				failedLoadX509KeyPair,
+				fmt.Sprintf(
+					"Error creating x509 key pair from \"%s\" and \"%s\".",
+					tlsConfig.ClientCertPem,
+					tlsConfig.ClientKeyPem),
+				err)
+		}
+
+		tlsObject.Certificates = []tls.Certificate{cert}
 	}
 
-	// Load the CA cert if it is provided
+	// Load the CA certificate if provided
 	if err := fileExists(tlsConfig.ServerCaPem); err != nil {
 		return nil, err
 	}
@@ -166,6 +167,7 @@ func GetTLSConfig(tlsConfig TLSConfig) (*tls.Config, *Xk6KafkaError) {
 				tlsConfig.ServerCaPem),
 			err)
 	}
+
 	caCertPool := x509.NewCertPool()
 	if ok := caCertPool.AppendCertsFromPEM(caCert); !ok {
 		return nil, NewXk6KafkaError(
@@ -176,7 +178,6 @@ func GetTLSConfig(tlsConfig TLSConfig) (*tls.Config, *Xk6KafkaError) {
 			nil)
 	}
 
-	tlsObject.Certificates = []tls.Certificate{cert}
 	tlsObject.RootCAs = caCertPool
 	return tlsObject, nil
 }
diff --git a/auth_test.go b/auth_test.go
index c0caf3e..f666edc 100644
--- a/auth_test.go
+++ b/auth_test.go
@@ -1,7 +1,6 @@
 package kafka
 
 import (
-	"os"
 	"testing"
 	"time"
 
@@ -146,41 +145,60 @@ func TestTlsConfigFails(t *testing.T) {
 	saslConfig := []*SimpleTLSConfig{
 		{
 			saslConfig: SASLConfig{},
-			tlsConfig:  TLSConfig{EnableTLS: true, ClientCertPem: "test.cer"},
+			tlsConfig:  TLSConfig{},
 			err: &Xk6KafkaError{
-				Code:    fileNotFound,
-				Message: "File not found: test.cer",
+				Code:          noTLSConfig,
+				Message:       "No TLS config provided. Continuing with TLS disabled.",
+				OriginalError: nil,
 			},
 		},
 		{
 			saslConfig: SASLConfig{},
 			tlsConfig: TLSConfig{
 				EnableTLS:     true,
+				ServerCaPem:   "server.cer",
+				ClientCertPem: "client.cer",
+				ClientKeyPem:  "client.pem",
+			},
+			err: &Xk6KafkaError{
+				Code:          fileNotFound,
+				Message:       "File not found: client.cer",
+				OriginalError: nil,
+			},
+		},
+		{
+			saslConfig: SASLConfig{},
+			tlsConfig: TLSConfig{
+				EnableTLS:     true,
+				ServerCaPem:   "server.cer",
 				ClientCertPem: "fixtures/client.cer",
+				ClientKeyPem:  "test.pem",
 			},
 			err: &Xk6KafkaError{
 				Code:          fileNotFound,
-				Message:       "File not found: ",
-				OriginalError: &os.PathError{},
+				Message:       "File not found: test.pem",
+				OriginalError: nil,
 			},
 		},
 		{
 			saslConfig: SASLConfig{},
 			tlsConfig: TLSConfig{
 				EnableTLS:     true,
+				ServerCaPem:   "server.cer",
 				ClientCertPem: "fixtures/client.cer",
 				ClientKeyPem:  "fixtures/client.pem",
 			},
 			err: &Xk6KafkaError{
 				Code:          fileNotFound,
-				Message:       "File not found: ",
-				OriginalError: &os.PathError{},
+				Message:       "File not found: server.cer",
+				OriginalError: nil,
 			},
 		},
 		{
 			saslConfig: SASLConfig{},
 			tlsConfig: TLSConfig{
 				EnableTLS:     true,
+				ServerCaPem:   "fixtures/caroot.cer",
 				ClientCertPem: "fixtures/invalid-client.cer",
 				ClientKeyPem:  "fixtures/invalid-client.pem",
 			},