added support for SASL/SCRAM on Kafka bucket notifications. (#9168)

fixes #9167

pkg/event/target/kafka.go

@@ -46,6 +46,7 @@ const (
 	KafkaSASL          = "sasl"
 	KafkaSASLUsername  = "sasl_username"
 	KafkaSASLPassword  = "sasl_password"
+	KafkaSASLMechanism = "sasl_mechanism"
 	KafkaClientTLSCert = "client_tls_cert"
 	KafkaClientTLSKey  = "client_tls_key"
 	KafkaVersion       = "version"
@@ -61,6 +62,7 @@ const (
 	EnvKafkaSASLEnable    = "MINIO_NOTIFY_KAFKA_SASL"
 	EnvKafkaSASLUsername  = "MINIO_NOTIFY_KAFKA_SASL_USERNAME"
 	EnvKafkaSASLPassword  = "MINIO_NOTIFY_KAFKA_SASL_PASSWORD"
+	EnvKafkaSASLMechanism = "MINIO_NOTIFY_KAFKA_SASL_MECHANISM"
 	EnvKafkaClientTLSCert = "MINIO_NOTIFY_KAFKA_CLIENT_TLS_CERT"
 	EnvKafkaClientTLSKey  = "MINIO_NOTIFY_KAFKA_CLIENT_TLS_KEY"
 	EnvKafkaVersion       = "MINIO_NOTIFY_KAFKA_VERSION"
@@ -83,9 +85,10 @@ type KafkaArgs struct {
 		ClientTLSKey  string             `json:"clientTLSKey"`
 	} `json:"tls"`
 	SASL struct {
-		Enable   bool   `json:"enable"`
-		User     string `json:"username"`
-		Password string `json:"password"`
+		Enable    bool   `json:"enable"`
+		User      string `json:"username"`
+		Password  string `json:"password"`
+		Mechanism string `json:"mechanism"`
 	} `json:"sasl"`
 }
 
@@ -255,6 +258,16 @@ func NewKafkaTarget(id string, args KafkaArgs, doneCh <-chan struct{}, loggerOnc
 
 	config.Net.SASL.User = args.SASL.User
 	config.Net.SASL.Password = args.SASL.Password
+	if args.SASL.Mechanism == "sha512" {
+		config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: KafkaSHA512} }
+		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)
+	} else if args.SASL.Mechanism == "sha256" {
+		config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: KafkaSHA256} }
+		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA256)
+	} else {
+		// default to PLAIN
+		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypePlaintext)
+	}
 	config.Net.SASL.Enable = args.SASL.Enable
 
 	tlsConfig, err := saramatls.NewConfig(args.TLS.ClientTLSCert, args.TLS.ClientTLSKey)

pkg/event/target/kafka_scram_client.go

@@ -0,0 +1,70 @@
+/*
+ * MinIO Cloud Storage, (C) 2020 MinIO, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package target
+
+import (
+	"crypto/sha512"
+	"hash"
+
+	"github.com/minio/sha256-simd"
+	"github.com/xdg/scram"
+)
+
+// KafkaSHA256 is a function that returns a crypto/sha256 hasher and should be used
+// to create Client objects configured for SHA-256 hashing.
+var KafkaSHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }
+
+// KafkaSHA512 is a function that returns a crypto/sha512 hasher and should be used
+// to create Client objects configured for SHA-512 hashing.
+var KafkaSHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
+
+// XDGSCRAMClient implements the client-side of an authentication
+// conversation with a server.  A new conversation must be created for
+// each authentication attempt.
+type XDGSCRAMClient struct {
+	*scram.Client
+	*scram.ClientConversation
+	scram.HashGeneratorFcn
+}
+
+// Begin constructs a SCRAM client component based on a given hash.Hash
+// factory receiver.  This constructor will normalize the username, password
+// and authzID via the SASLprep algorithm, as recommended by RFC-5802.  If
+// SASLprep fails, the method returns an error.
+func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) {
+	x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID)
+	if err != nil {
+		return err
+	}
+	x.ClientConversation = x.Client.NewConversation()
+	return nil
+}
+
+// Step takes a string provided from a server (or just an empty string for the
+// very first conversation step) and attempts to move the authentication
+// conversation forward.  It returns a string to be sent to the server or an
+// error if the server message is invalid.  Calling Step after a conversation
+// completes is also an error.
+func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
+	response, err = x.ClientConversation.Step(challenge)
+	return
+}
+
+// Done returns true if the conversation is completed or has errored.
+func (x *XDGSCRAMClient) Done() bool {
+	return x.ClientConversation.Done()
+}