feat: kafka sasl auth

eventsources/sources/kafka/start.go

@@ -270,6 +270,26 @@ func getSaramaConfig(kafkaEventSource *v1alpha1.KafkaEventSource, log *zap.Sugar
 		config.Version = version
 	}
 
+	if kafkaEventSource.SASL != nil {
+		config.Net.SASL.Enable = true
+
+		config.Net.SASL.Mechanism = kafkaEventSource.SASL.GetMechanism()
+
+		user, err := common.GetSecretFromVolume(kafkaEventSource.SASL.User)
+		if err != nil {
+			log.Errorf("Error getting user value from secret: %v", err)
+			return nil, err
+		}
+		config.Net.SASL.User = user
+
+		password, err := common.GetSecretFromVolume(kafkaEventSource.SASL.Password)
+		if err != nil {
+			log.Errorf("Error getting password value from secret: %v", err)
+			return nil, err
+		}
+		config.Net.SASL.Password = password
+	}
+
 	if kafkaEventSource.TLS != nil {
 		tlsConfig, err := common.GetTLSConfig(kafkaEventSource.TLS)
 		if err != nil {

eventsources/sources/kafka/validate.go

@@ -46,5 +46,8 @@ func validate(eventSource *v1alpha1.KafkaEventSource) error {
 	if eventSource.TLS != nil {
 		return apicommon.ValidateTLSConfig(eventSource.TLS)
 	}
+	if eventSource.SASL != nil {
+		return apicommon.ValidateSASLConfig(eventSource.SASL)
+	}
 	return nil
 }

pkg/apis/common/common.go

@@ -17,6 +17,7 @@ limitations under the License.
 package common
 
 import (
+	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 )
 
@@ -128,6 +129,19 @@ type TLSConfig struct {
 	DeprecatedClientKeyPath string `json:"clientKeyPath,omitempty" protobuf:"bytes,6,opt,name=clientKeyPath"`
 }
 
+// SASLConfig refers to SASL configuration for a client
+type SASLConfig struct {
+	// SASLMechanism is the name of the enabled SASL mechanism.
+	// Possible values: OAUTHBEARER, PLAIN (defaults to PLAIN).
+	// +optional
+	Mechanism string `json:"mechanism,omitempty" protobuf:"bytes,1,opt,name=mechanism"`
+	// User is the authentication identity (authcid) to present for
+	// SASL/PLAIN or SASL/SCRAM authentication
+	User *corev1.SecretKeySelector `json:"user,omitempty" protobuf:"bytes,2,opt,name=user"`
+	// Password for SASL/PLAIN authentication
+	Password *corev1.SecretKeySelector `json:"password,omitempty" protobuf:"bytes,3,opt,name=password"`
+}
+
 // Backoff for an operation
 type Backoff struct {
 	// The initial duration in nanoseconds or strings like "1s", "3m"
@@ -153,3 +167,13 @@ type Metadata struct {
 	Annotations map[string]string `json:"annotations,omitempty" protobuf:"bytes,1,rep,name=annotations"`
 	Labels      map[string]string `json:"labels,omitempty" protobuf:"bytes,2,rep,name=labels"`
 }
+
+func (s SASLConfig) GetMechanism() sarama.SASLMechanism {
+	switch s.Mechanism {
+	case "PLAIN", "OAUTHBEARER", "SCRAM-SHA-256", "SCRAM-SHA-512", "GSSAPI":
+		return sarama.SASLMechanism(s.Mechanism)
+	default:
+		// default to PLAINTEXT mechanism
+		return sarama.SASLMechanism(sarama.SASLTypePlaintext)
+	}
+}