From c113180d0bbf1446ba146fbd6d625c31715c211b Mon Sep 17 00:00:00 2001
From: Chris Rice <chris.rice@lytx.com>
Date: Fri, 8 Sep 2023 08:58:23 -0700
Subject: [PATCH] Add support for 'none' authentication for kafka while still
 allowing SSL

---
 internal/endpoint/kafka.go | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/internal/endpoint/kafka.go b/internal/endpoint/kafka.go
index 4379d24d..960d0193 100644
--- a/internal/endpoint/kafka.go
+++ b/internal/endpoint/kafka.go
@@ -140,6 +140,24 @@ func (conn *KafkaConn) Send(msg string) error {
 			}
 
 			cfg.Net.TLS.Config = &tlsConfig
+		case "none":
+			// This path allows to either provide a custom ca certificate
+			// or, because RootCAs is nil, is using the hosts ca set
+			// to verify the server certificate
+			if conn.ep.Kafka.SSL {
+				tlsConfig := tls.Config{}
+
+				if conn.ep.Kafka.CACertFile != "" {
+					caCertPool, err := loadRootTLSCert(conn.ep.Kafka.CACertFile)
+					if err != nil {
+						return err
+					}
+					tlsConfig.RootCAs = &caCertPool
+				}
+
+				cfg.Net.TLS.Enable = true
+				cfg.Net.TLS.Config = &tlsConfig
+			}
 		}
 
 		c, err := sarama.NewSyncProducer([]string{uri}, cfg)