add basic sasl_ssl support:

common.go

@@ -164,6 +164,24 @@ func randomString(length int) string {
 	return fmt.Sprintf("%x", buf)[:length]
 }
 
+func setupSaslMechanism(auth authConfig, saramaCfg *sarama.Config) (error) {
+	switch strings.ToLower(auth.SASLMechanism) {
+	case "plain", "":
+		saramaCfg.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypePlaintext)
+		return nil
+	case "scram-sha512":
+		saramaCfg.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} }
+		saramaCfg.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)
+		return nil
+	case "scram-sha256":
+		saramaCfg.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA256} }
+		saramaCfg.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA256)
+		return nil
+	default:
+		return fmt.Errorf("Unsupported auth sasl mechanism: %#v", auth.SASLMechanism)
+	}
+}
+
 // setupCerts takes the paths to a tls certificate, CA, and certificate key in
 // a PEM format and returns a constructed tls.Config object.
 func setupCerts(certPath, caPath, keyPath string) (*tls.Config, error) {
@@ -207,6 +225,8 @@ type authConfig struct {
 	ClientCertKey     string `json:"client-certificate-key"`
 	SASLPlainUser     string `json:"sasl_plain_user"`
 	SASLPlainPassword string `json:"sasl_plain_password"`
+	SASLMechanism     string `json:"sasl_mechanism"`
+	SecurityProtocol  string `json:"security_protocol"`
 }
 
 func setupAuth(auth authConfig, saramaCfg *sarama.Config) error {
@@ -218,7 +238,7 @@ func setupAuth(auth authConfig, saramaCfg *sarama.Config) error {
 	case "TLS":
 		return setupAuthTLS(auth, saramaCfg)
 	case "TLS-1way":
-		return setupAuthTLS1Way(auth, saramaCfg)
+		return setupAuthTLS(auth, saramaCfg)
 	case "SASL":
 		return setupSASL(auth, saramaCfg)
 	default:
@@ -230,38 +250,24 @@ func setupSASL(auth authConfig, saramaCfg *sarama.Config) error {
 	saramaCfg.Net.SASL.Enable = true
 	saramaCfg.Net.SASL.User = auth.SASLPlainUser
 	saramaCfg.Net.SASL.Password = auth.SASLPlainPassword
-	return nil
-}
+	err := setupSaslMechanism(auth, saramaCfg)
 
-func setupAuthTLS1Way(auth authConfig, saramaCfg *sarama.Config) error {
-	saramaCfg.Net.TLS.Enable = true
-	saramaCfg.Net.TLS.Config = &tls.Config{}
-
-	if auth.CACert == "" {
-		return nil
-	}
-
-	caString, err := os.ReadFile(auth.CACert)
 	if err != nil {
-		return fmt.Errorf("failed to read ca-certificate err=%v", err)
+		return err;
 	}
 
-	caPool := x509.NewCertPool()
-	ok := caPool.AppendCertsFromPEM(caString)
-	if !ok {
-		failf("unable to add ca-certificate at %s to certificate pool", auth.CACert)
+	if (strings.EqualFold(auth.SecurityProtocol, "SASL_SSL")) {
+		return setupAuthTLS(auth, saramaCfg)
 	}
-
-	tlsCfg := &tls.Config{RootCAs: caPool}
-	tlsCfg.BuildNameToCertificate()
-
-	saramaCfg.Net.TLS.Config = tlsCfg
 	return nil
 }
 
 func setupAuthTLS(auth authConfig, saramaCfg *sarama.Config) error {
-	if auth.CACert == "" || auth.ClientCert == "" || auth.ClientCertKey == "" {
-		return fmt.Errorf("client-certificate, client-certificate-key and ca-certificate are required - got auth=%#v", auth)
+	saramaCfg.Net.TLS.Enable = true
+	saramaCfg.Net.TLS.Config = &tls.Config{}
+
+	if auth.CACert == "" {
+		return nil
 	}
 
 	caString, err := os.ReadFile(auth.CACert)
@@ -275,17 +281,20 @@ func setupAuthTLS(auth authConfig, saramaCfg *sarama.Config) error {
 		failf("unable to add ca-certificate at %s to certificate pool", auth.CACert)
 	}
 
-	clientCert, err := tls.LoadX509KeyPair(auth.ClientCert, auth.ClientCertKey)
-	if err != nil {
-		return err
-	}
+	var tlsCfg *tls.Config
+	if auth.ClientCert != "" && auth.ClientCertKey != "" {
+		clientCert, err := tls.LoadX509KeyPair(auth.ClientCert, auth.ClientCertKey)
+		if err != nil {
+			return err
+		}
+		tlsCfg = &tls.Config{RootCAs: caPool, Certificates: []tls.Certificate{clientCert}}
+	} else {
+		tlsCfg = &tls.Config{RootCAs: caPool}
 
-	tlsCfg := &tls.Config{RootCAs: caPool, Certificates: []tls.Certificate{clientCert}}
+	}
 	tlsCfg.BuildNameToCertificate()
 
-	saramaCfg.Net.TLS.Enable = true
 	saramaCfg.Net.TLS.Config = tlsCfg
-
 	return nil
 }
 

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/markusmobius/go-dateparser v1.2.1
 	github.com/stretchr/testify v1.8.4
+	github.com/xdg-go/scram v1.1.2
 	golang.org/x/crypto v0.13.0
 )
 
@@ -36,6 +37,8 @@ require (
 	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	github.com/tetratelabs/wazero v1.5.0 // indirect
 	github.com/wasilibs/go-re2 v1.4.0 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
 	golang.org/x/net v0.15.0 // indirect
 	golang.org/x/sys v0.12.0 // indirect

go.sum

@@ -77,6 +77,12 @@ github.com/wasilibs/go-re2 v1.4.0 h1:Jp6BM8G/zajgY1BCQUm3i7oGMdR1gA5EBv87wGd2ysc
 github.com/wasilibs/go-re2 v1.4.0/go.mod h1:hLzlKjEgON+17hWjikLx8hJBkikyjQH/lsqCy9t6tIY=
 github.com/wasilibs/nottinygc v0.4.0 h1:h1TJMihMC4neN6Zq+WKpLxgd9xCFMw7O9ETLwY2exJQ=
 github.com/wasilibs/nottinygc v0.4.0/go.mod h1:oDcIotskuYNMpqMF23l7Z8uzD4TC0WXHK8jetlB3HIo=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
@@ -114,6 +120,7 @@ golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=

scram_client.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"crypto/sha256"
+	"crypto/sha512"
+	"hash"
+
+	"github.com/xdg-go/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()
+}