Implement TLS authentication for kafka mqt (#1300)

* use secrets to store keys and certificates

pkg/mqtrigger/messageQueue/kafka.go

@@ -17,10 +17,13 @@ limitations under the License.
 package messageQueue
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 
 	sarama "github.com/Shopify/sarama"
@@ -39,6 +42,8 @@ type (
 		routerUrl string
 		brokers   []string
 		version   sarama.KafkaVersion
+		authKeys  map[string][]byte
+		tls       bool
 	}
 )
 
@@ -64,9 +69,23 @@ func makeKafkaMessageQueue(logger *zap.Logger, routerUrl string, mqCfg MessageQu
 		version:   kafkaVersion,
 	}
 
+	if tls, _ := strconv.ParseBool(os.Getenv("TLS_ENABLED")); tls == true {
+		kafka.tls = true
+
+		authKeys := make(map[string][]byte)
+
+		if mqCfg.Secrets == nil {
+			return nil, errors.New("no secrets were loaded")
+		}
+
+		authKeys["caCert"] = mqCfg.Secrets["caCert"]
+		authKeys["userCert"] = mqCfg.Secrets["userCert"]
+		authKeys["userKey"] = mqCfg.Secrets["userKey"]
+		kafka.authKeys = authKeys
+	}
+
 	logger.Info("created kafka queue", zap.Any("kafka brokers", kafka.brokers),
 		zap.Any("kafka version", kafka.version))
-
 	return kafka, nil
 }
 
@@ -83,6 +102,28 @@ func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubs
 	consumerConfig.Consumer.Return.Errors = true
 	consumerConfig.Group.Return.Notifications = true
 	consumerConfig.Config.Version = kafka.version
+
+	// Create new producer
+	producerConfig := sarama.NewConfig()
+	producerConfig.Producer.RequiredAcks = sarama.WaitForAll
+	producerConfig.Producer.Retry.Max = 10
+	producerConfig.Producer.Return.Successes = true
+	producerConfig.Version = kafka.version
+
+	// Setup TLS for both producer and consumer
+	if kafka.tls {
+		consumerConfig.Net.TLS.Enable = true
+		producerConfig.Net.TLS.Enable = true
+		tlsConfig, err := kafka.getTLSConfig()
+
+		if err != nil {
+			return nil, err
+		}
+
+		producerConfig.Net.TLS.Config = tlsConfig
+		consumerConfig.Net.TLS.Config = tlsConfig
+	}
+
 	consumer, err := cluster.NewConsumer(kafka.brokers, string(trigger.Metadata.UID), []string{trigger.Spec.Topic}, consumerConfig)
 	kafka.logger.Info("created a new consumer", zap.Strings("brokers", kafka.brokers),
 		zap.String("input topic", trigger.Spec.Topic),
@@ -91,17 +132,10 @@ func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubs
 		zap.String("trigger name", trigger.Metadata.Name),
 		zap.String("function namespace", trigger.Metadata.Namespace),
 		zap.String("function name", trigger.Spec.FunctionReference.Name))
-
 	if err != nil {
-		panic(err)
+		return nil, err
 	}
 
-	// Create new producer
-	producerConfig := sarama.NewConfig()
-	producerConfig.Producer.RequiredAcks = sarama.WaitForAll
-	producerConfig.Producer.Retry.Max = 10
-	producerConfig.Producer.Return.Successes = true
-	producerConfig.Version = kafka.version
 	producer, err := sarama.NewSyncProducer(kafka.brokers, producerConfig)
 	kafka.logger.Info("created a new producer", zap.Strings("brokers", kafka.brokers),
 		zap.String("input topic", trigger.Spec.Topic),
@@ -112,7 +146,7 @@ func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubs
 		zap.String("function name", trigger.Spec.FunctionReference.Name))
 
 	if err != nil {
-		panic(err)
+		return nil, err
 	}
 
 	// consume errors
@@ -142,6 +176,27 @@ func (kafka Kafka) subscribe(trigger *fv1.MessageQueueTrigger) (messageQueueSubs
 	return consumer, nil
 }
 
+func (kafka Kafka) getTLSConfig() (*tls.Config, error) {
+	tlsConfig := tls.Config{}
+	cert, err := tls.X509KeyPair(kafka.authKeys["userCert"], kafka.authKeys["userKey"])
+	if err != nil {
+		return nil, err
+	}
+
+	tlsConfig.Certificates = []tls.Certificate{cert}
+
+	if err != nil {
+		return nil, err
+	}
+
+	caCertPool := x509.NewCertPool()
+	caCertPool.AppendCertsFromPEM(kafka.authKeys["caCert"])
+	tlsConfig.RootCAs = caCertPool
+	tlsConfig.BuildNameToCertificate()
+
+	return &tlsConfig, nil
+}
+
 func (kafka Kafka) unsubscribe(subscription messageQueueSubscription) error {
 	return subscription.(*cluster.Consumer).Close()
 }

pkg/mqtrigger/messageQueue/messageQueue.go

@@ -42,8 +42,9 @@ type (
 	requestType int
 
 	MessageQueueConfig struct {
-		MQType string
-		Url    string
+		MQType  string
+		Url     string
+		Secrets map[string][]byte
 	}
 
 	MessageQueue interface {