Support for scram sasl mechanisms

README.md

@@ -98,6 +98,7 @@ This image is configurable using different flags
 | sasl.handshake               | true       | Only set this to false if using a non-Kafka SASL proxy                                              |
 | sasl.username                |            | SASL user name                                                                                      |
 | sasl.password                |            | SASL user password                                                                                  |
+| sasl.mechanism               |            | SASL mechanism can be plain, sha512, sha256
 | tls.enabled                  | false      | Connect using TLS                                                                                   |
 | tls.ca-file                  |            | The optional certificate authority file for TLS client authentication                               |
 | tls.cert-file                |            | The optional certificate file for client authentication                                             |

kafka_exporter.go

@@ -66,6 +66,7 @@ type kafkaOpts struct {
 	useSASLHandshake         bool
 	saslUsername             string
 	saslPassword             string
+	saslMechanism            string
 	useTLS                   bool
 	tlsCAFile                string
 	tlsCertFile              string
@@ -124,6 +125,20 @@ func NewExporter(opts kafkaOpts, topicFilter string, groupFilter string) (*Expor
 	config.Version = kafkaVersion
 
 	if opts.useSASL {
+		// Convert to lowercase so that SHA512 and SHA256 is still valid
+		opts.saslMechanism = strings.ToLower(opts.saslMechanism)
+		switch opts.saslMechanism {
+		case "sha512":
+			config.Net.SASL.SCRAMClient = &XDGSCRAMClient{HashGeneratorFcn: SHA512}
+			config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)
+		case "sha256":
+			config.Net.SASL.SCRAMClient = &XDGSCRAMClient{HashGeneratorFcn: SHA256}
+			config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA256)
+		case "plain":
+		default:
+			plog.Fatalf("invalid sasl mechanism \"%s\": can only be \"sha256\", \"sha512\" or \"plain\"", opts.saslMechanism)
+		}
+
 		config.Net.SASL.Enable = true
 		config.Net.SASL.Handshake = opts.useSASLHandshake
 
@@ -481,6 +496,7 @@ func main() {
 	kingpin.Flag("sasl.handshake", "Only set this to false if using a non-Kafka SASL proxy.").Default("true").BoolVar(&opts.useSASLHandshake)
 	kingpin.Flag("sasl.username", "SASL user name.").Default("").StringVar(&opts.saslUsername)
 	kingpin.Flag("sasl.password", "SASL user password.").Default("").StringVar(&opts.saslPassword)
+	kingpin.Flag("sasl.mechanism", "The SASL SCRAM SHA algorithm sha256 or sha512 as mechanism").Default("").StringVar(&opts.saslMechanism)
 	kingpin.Flag("tls.enabled", "Connect using TLS.").Default("false").BoolVar(&opts.useTLS)
 	kingpin.Flag("tls.ca-file", "The optional certificate authority file for TLS client authentication.").Default("").StringVar(&opts.tlsCAFile)
 	kingpin.Flag("tls.cert-file", "The optional certificate file for client authentication.").Default("").StringVar(&opts.tlsCertFile)

scram_client.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"crypto/sha256"
+	"crypto/sha512"
+	"hash"
+
+	"github.com/xdg/scram"
+)
+
+var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }
+var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
+
+type XDGSCRAMClient struct {
+	*scram.Client
+	*scram.ClientConversation
+	scram.HashGeneratorFcn
+}
+
+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
+}
+
+func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
+	response, err = x.ClientConversation.Step(challenge)
+	return
+}
+
+func (x *XDGSCRAMClient) Done() bool {
+	return x.ClientConversation.Done()
+}