From 456ce4cc92b6a24cc6087bebacd22f3d810d5de7 Mon Sep 17 00:00:00 2001
From: Praveen raj Mani <praveen@minio.io>
Date: Wed, 18 Sep 2019 23:43:04 +0530
Subject: [PATCH] Add rootCAs support to Kafka & MQTT (#8236)

Fixes #8211
---
 cmd/config-current.go     | 7 +++++++
 pkg/event/target/kafka.go | 6 +++++-
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git a/cmd/config-current.go b/cmd/config-current.go
index 11bcf52e2..da86a59b6 100644
--- a/cmd/config-current.go
+++ b/cmd/config-current.go
@@ -341,6 +341,9 @@ func (s *serverConfig) TestNotificationTargets() error {
 		if !v.Enable {
 			continue
 		}
+		if v.TLS.Enable {
+			v.TLS.RootCAs = globalRootCAs
+		}
 		t, err := target.NewKafkaTarget(k, v, GlobalServiceDoneCh)
 		if err != nil {
 			return fmt.Errorf("kafka(%s): %s", k, err.Error())
@@ -352,6 +355,7 @@ func (s *serverConfig) TestNotificationTargets() error {
 		if !v.Enable {
 			continue
 		}
+		v.RootCAs = globalRootCAs
 		t, err := target.NewMQTTTarget(k, v, GlobalServiceDoneCh)
 		if err != nil {
 			return fmt.Errorf("mqtt(%s): %s", k, err.Error())
@@ -690,6 +694,9 @@ func getNotificationTargets(config *serverConfig) *event.TargetList {
 
 	for id, args := range config.Notify.Kafka {
 		if args.Enable {
+			if args.TLS.Enable {
+				args.TLS.RootCAs = globalRootCAs
+			}
 			newTarget, err := target.NewKafkaTarget(id, args, GlobalServiceDoneCh)
 			if err != nil {
 				logger.LogIf(context.Background(), err)
diff --git a/pkg/event/target/kafka.go b/pkg/event/target/kafka.go
index 3dda0ae90..23e00846b 100644
--- a/pkg/event/target/kafka.go
+++ b/pkg/event/target/kafka.go
@@ -18,6 +18,7 @@ package target
 
 import (
 	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"errors"
 	"net"
@@ -40,6 +41,7 @@ type KafkaArgs struct {
 	QueueLimit uint64      `json:"queueLimit"`
 	TLS        struct {
 		Enable     bool               `json:"enable"`
+		RootCAs    *x509.CertPool     `json:"-"`
 		SkipVerify bool               `json:"skipVerify"`
 		ClientAuth tls.ClientAuthType `json:"clientAuth"`
 	} `json:"tls"`
@@ -198,7 +200,9 @@ func NewKafkaTarget(id string, args KafkaArgs, doneCh <-chan struct{}) (*KafkaTa
 
 	config.Net.TLS.Enable = args.TLS.Enable
 	tlsConfig := &tls.Config{
-		ClientAuth: args.TLS.ClientAuth,
+		ClientAuth:         args.TLS.ClientAuth,
+		InsecureSkipVerify: args.TLS.SkipVerify,
+		RootCAs:            args.TLS.RootCAs,
 	}
 	config.Net.TLS.Config = tlsConfig