Add mTLS, SASL/PLAIN, SASL/SCRAM authentication support

clients/pkg/promtail/scrapeconfig/scrapeconfig.go

@@ -5,6 +5,9 @@ import (
 	"reflect"
 	"time"
 
+	"github.com/Shopify/sarama"
+	"github.com/grafana/dskit/flagext"
+
 	promconfig "github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/discovery"
@@ -244,6 +247,52 @@ type KafkaTargetConfig struct {
 
 	// Rebalancing strategy to use. (e.g sticky, roundrobin or range)
 	Assignor string `yaml:"assignor"`
+
+	// Authentication strategy with Kafka brokers
+	Authentication KafkaAuthentication `yaml:"authentication"`
+}
+
+// KafkaAuthenticationType specifies method to authenticate with Kafka brokers
+type KafkaAuthenticationType string
+
+const (
+	// KafkaAuthenticationTypeNone represents using no authentication
+	KafkaAuthenticationTypeNone = "none"
+	// KafkaAuthenticationTypeSSL represents using SSL/TLS to authenticate
+	KafkaAuthenticationTypeSSL = "ssl"
+	// KafkaAuthenticationTypeSASL represents using SASL to authenticate
+	KafkaAuthenticationTypeSASL = "sasl"
+)
+
+// KafkaAuthentication describe the configuration for authentication with Kafka brokers
+type KafkaAuthentication struct {
+	// Type is authentication type
+	// Possible values: none, sasl and ssl (defaults to none).
+	Type KafkaAuthenticationType `yaml:"type"`
+
+	// TLSConfig is used for TLS encryption and authentication with Kafka brokers
+	TLSConfig promconfig.TLSConfig `yaml:"tls_config,omitempty"`
+
+	// SASLConfig is used for SASL authentication with Kafka brokers
+	SASLConfig KafkaSASLConfig `yaml:"sasl_config,omitempty"`
+}
+
+// KafkaSASLConfig describe the SASL configuration for authentication with Kafka brokers
+type KafkaSASLConfig struct {
+	// SASL mechanism. Supports PLAIN, SCRAM-SHA-256 and SCRAM-SHA-512
+	Mechanism sarama.SASLMechanism `yaml:"mechanism"`
+
+	// SASL Username
+	User string `yaml:"user"`
+
+	// SASL Password for the User
+	Password flagext.Secret `yaml:"password"`
+
+	// UseTLS sets whether TLS is used with SASL
+	UseTLS bool `yaml:"use_tls"`
+
+	// TLSConfig is used for SASL over TLS. It is used only when UseTLS is true
+	TLSConfig promconfig.TLSConfig `yaml:",inline"`
 }
 
 // GcplogTargetConfig describes a scrape config to pull logs from any pubsub topic.

clients/pkg/promtail/targets/kafka/authentication.go

@@ -0,0 +1,69 @@
+package kafka
+
+import (
+	"crypto/sha256"
+	"crypto/sha512"
+	"crypto/tls"
+	"crypto/x509"
+	"os"
+
+	promconfig "github.com/prometheus/common/config"
+	"github.com/xdg-go/scram"
+)
+
+func createTLSConfig(cfg promconfig.TLSConfig) (*tls.Config, error) {
+	tc := &tls.Config{
+		InsecureSkipVerify: cfg.InsecureSkipVerify,
+		ServerName:         cfg.ServerName,
+	}
+	// load ca cert
+	if len(cfg.CAFile) > 0 {
+		caCert, err := os.ReadFile(cfg.CAFile)
+		if err != nil {
+			return nil, err
+		}
+		caCertPool := x509.NewCertPool()
+		caCertPool.AppendCertsFromPEM(caCert)
+		tc.RootCAs = caCertPool
+	}
+	// load client cert
+	if len(cfg.CertFile) > 0 && len(cfg.KeyFile) > 0 {
+		cert, err := tls.LoadX509KeyPair(cfg.CertFile, cfg.KeyFile)
+		if err != nil {
+			return nil, err
+		}
+		tc.Certificates = []tls.Certificate{cert}
+	}
+	return tc, nil
+}
+
+// copied from https://github.com/Shopify/sarama/blob/44627b731c60bb90efe25573e7ef2b3f8df3fa23/examples/sasl_scram_client/scram_client.go
+var (
+	SHA256 scram.HashGeneratorFcn = sha256.New
+	SHA512 scram.HashGeneratorFcn = sha512.New
+)
+
+// XDGSCRAMClient implements sarama.SCRAMClient
+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()
+}

clients/pkg/promtail/targets/kafka/target_syncer.go

@@ -73,6 +73,10 @@ func NewSyncer(
 	default:
 		return nil, fmt.Errorf("unrecognized consumer group partition assignor: %s", cfg.KafkaConfig.Assignor)
 	}
+	config, err = withAuthentication(*config, cfg.KafkaConfig.Authentication)
+	if err != nil {
+		return nil, fmt.Errorf("error setting up kafka authentication: %w", err)
+	}
 	client, err := sarama.NewClient(cfg.KafkaConfig.Brokers, config)
 	if err != nil {
 		return nil, fmt.Errorf("error creating kafka client: %w", err)
@@ -113,6 +117,74 @@ func NewSyncer(
 	return t, nil
 }
 
+func withAuthentication(cfg sarama.Config, authCfg scrapeconfig.KafkaAuthentication) (*sarama.Config, error) {
+	if len(authCfg.Type) == 0 || authCfg.Type == scrapeconfig.KafkaAuthenticationTypeNone {
+		return &cfg, nil
+	}
+
+	switch authCfg.Type {
+	case scrapeconfig.KafkaAuthenticationTypeSSL:
+		return withSSLAuthentication(cfg, authCfg)
+	case scrapeconfig.KafkaAuthenticationTypeSASL:
+		return withSASLAuthentication(cfg, authCfg)
+	default:
+		return nil, fmt.Errorf("unsupported authentication type %s", authCfg.Type)
+	}
+}
+
+func withSSLAuthentication(cfg sarama.Config, authCfg scrapeconfig.KafkaAuthentication) (*sarama.Config, error) {
+	cfg.Net.TLS.Enable = true
+	tc, err := createTLSConfig(authCfg.TLSConfig)
+	if err != nil {
+		return nil, err
+	}
+	cfg.Net.TLS.Config = tc
+	return &cfg, nil
+}
+
+func withSASLAuthentication(cfg sarama.Config, authCfg scrapeconfig.KafkaAuthentication) (*sarama.Config, error) {
+	cfg.Net.SASL.Enable = true
+	cfg.Net.SASL.User = authCfg.SASLConfig.User
+	cfg.Net.SASL.Password = authCfg.SASLConfig.Password.Value
+	cfg.Net.SASL.Mechanism = authCfg.SASLConfig.Mechanism
+	if cfg.Net.SASL.Mechanism == "" {
+		cfg.Net.SASL.Mechanism = sarama.SASLTypePlaintext
+	}
+
+	supportedMechanism := []string{
+		sarama.SASLTypeSCRAMSHA512,
+		sarama.SASLTypeSCRAMSHA256,
+		sarama.SASLTypePlaintext,
+	}
+	if !util.StringSliceContains(supportedMechanism, string(authCfg.SASLConfig.Mechanism)) {
+		return nil, fmt.Errorf("error unsupported sasl mechanism: %s", authCfg.SASLConfig.Mechanism)
+	}
+
+	if cfg.Net.SASL.Mechanism == sarama.SASLTypeSCRAMSHA512 {
+		cfg.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+			return &XDGSCRAMClient{
+				HashGeneratorFcn: SHA512,
+			}
+		}
+	}
+	if cfg.Net.SASL.Mechanism == sarama.SASLTypeSCRAMSHA256 {
+		cfg.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+			return &XDGSCRAMClient{
+				HashGeneratorFcn: SHA256,
+			}
+		}
+	}
+	if authCfg.SASLConfig.UseTLS {
+		tc, err := createTLSConfig(authCfg.SASLConfig.TLSConfig)
+		if err != nil {
+			return nil, err
+		}
+		cfg.Net.TLS.Config = tc
+		cfg.Net.TLS.Enable = true
+	}
+	return &cfg, nil
+}
+
 func (ts *TargetSyncer) loop() {
 	topicChanged := make(chan []string)
 	ts.wg.Add(2)

clients/pkg/promtail/targets/kafka/target_syncer_test.go

@@ -6,6 +6,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/grafana/dskit/flagext"
+	"github.com/prometheus/common/config"
+
 	"github.com/Shopify/sarama"
 	"github.com/go-kit/log"
 	"github.com/grafana/loki/clients/pkg/promtail/client/fake"
@@ -208,3 +211,144 @@ func Test_validateConfig(t *testing.T) {
 		})
 	}
 }
+
+func Test_withAuthentication(t *testing.T) {
+	var (
+		tlsConf = config.TLSConfig{
+			CAFile:             "testdata/example.com.ca.pem",
+			CertFile:           "testdata/example.com.pem",
+			KeyFile:            "testdata/example.com-key.pem",
+			ServerName:         "example.com",
+			InsecureSkipVerify: true,
+		}
+		expectedTLSConf, _ = createTLSConfig(config.TLSConfig{
+			CAFile:             "testdata/example.com.ca.pem",
+			CertFile:           "testdata/example.com.pem",
+			KeyFile:            "testdata/example.com-key.pem",
+			ServerName:         "example.com",
+			InsecureSkipVerify: true,
+		})
+		cfg = sarama.NewConfig()
+	)
+
+	// no authentication
+	noAuthCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeNone,
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, noAuthCfg.Net.TLS.Enable)
+	assert.Equal(t, false, noAuthCfg.Net.SASL.Enable)
+	assert.NoError(t, noAuthCfg.Validate())
+
+	// specify unsupported auth type
+	illegalAuthTypeCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: "illegal",
+	})
+	assert.NotNil(t, err)
+	assert.Nil(t, illegalAuthTypeCfg)
+
+	// mTLS authentication
+	mTLSCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type:      scrapeconfig.KafkaAuthenticationTypeSSL,
+		TLSConfig: tlsConf,
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, true, mTLSCfg.Net.TLS.Enable)
+	assert.NotNil(t, mTLSCfg.Net.TLS.Config)
+	assert.Equal(t, "example.com", mTLSCfg.Net.TLS.Config.ServerName)
+	assert.Equal(t, true, mTLSCfg.Net.TLS.Config.InsecureSkipVerify)
+	assert.Equal(t, expectedTLSConf.Certificates, mTLSCfg.Net.TLS.Config.Certificates)
+	assert.NotNil(t, mTLSCfg.Net.TLS.Config.RootCAs)
+	assert.NoError(t, mTLSCfg.Validate())
+
+	// mTLS authentication expect ignore sasl
+	mTLSCfg, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type:      scrapeconfig.KafkaAuthenticationTypeSSL,
+		TLSConfig: tlsConf,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeSCRAMSHA256,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+			UseTLS: false,
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, mTLSCfg.Net.SASL.Enable)
+
+	// SASL/PLAIN
+	saslCfg, err := withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypePlaintext,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, saslCfg.Net.TLS.Enable)
+	assert.Equal(t, true, saslCfg.Net.SASL.Enable)
+	assert.Equal(t, "user", saslCfg.Net.SASL.User)
+	assert.Equal(t, "pass", saslCfg.Net.SASL.Password)
+	assert.Equal(t, sarama.SASLTypePlaintext, string(saslCfg.Net.SASL.Mechanism))
+	assert.NoError(t, saslCfg.Validate())
+
+	// SASL/SCRAM
+	saslCfg, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeSCRAMSHA512,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, false, saslCfg.Net.TLS.Enable)
+	assert.Equal(t, true, saslCfg.Net.SASL.Enable)
+	assert.Equal(t, "user", saslCfg.Net.SASL.User)
+	assert.Equal(t, "pass", saslCfg.Net.SASL.Password)
+	assert.Equal(t, sarama.SASLTypeSCRAMSHA512, string(saslCfg.Net.SASL.Mechanism))
+	assert.NoError(t, saslCfg.Validate())
+
+	// SASL unsupported mechanism
+	_, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeGSSAPI,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+		},
+	})
+	assert.Error(t, err)
+	assert.Equal(t, err.Error(), "error unsupported sasl mechanism: GSSAPI")
+
+	// SASL over TLS
+	saslCfg, err = withAuthentication(*cfg, scrapeconfig.KafkaAuthentication{
+		Type: scrapeconfig.KafkaAuthenticationTypeSASL,
+		SASLConfig: scrapeconfig.KafkaSASLConfig{
+			Mechanism: sarama.SASLTypeSCRAMSHA512,
+			User:      "user",
+			Password: flagext.Secret{
+				Value: "pass",
+			},
+			UseTLS:    true,
+			TLSConfig: tlsConf,
+		},
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, true, saslCfg.Net.TLS.Enable)
+	assert.Equal(t, true, saslCfg.Net.SASL.Enable)
+	assert.NotNil(t, saslCfg.Net.TLS.Config)
+	assert.Equal(t, "example.com", saslCfg.Net.TLS.Config.ServerName)
+	assert.Equal(t, true, saslCfg.Net.TLS.Config.InsecureSkipVerify)
+	assert.Equal(t, expectedTLSConf.Certificates, saslCfg.Net.TLS.Config.Certificates)
+	assert.NotNil(t, saslCfg.Net.TLS.Config.RootCAs)
+	assert.NoError(t, saslCfg.Validate())
+}