feat: support kafka scram auth

server/kafka_runtime.go

@@ -10,6 +10,8 @@ import (
 	"time"
 
 	"github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl"
+	"github.com/segmentio/kafka-go/sasl/scram"
 	"github.com/tinode/chat/server/logs"
 )
 
@@ -40,6 +42,10 @@ type kafkaConfig struct {
 	RequiredAcks     int               `json:"required_acks"`
 	AllowAutoTopic   bool              `json:"allow_auto_topic_creation"`
 	TLSEnabled       bool              `json:"tls_enabled"`
+	SASLEnabled      bool              `json:"sasl_enabled"`
+	SASLMechanism    string            `json:"sasl_mechanism"`
+	SASLUsername     string            `json:"sasl_username"`
+	SASLPassword     string            `json:"sasl_password"`
 	MaxRetryCount    int               `json:"max_retry_count"`
 	RetryBackoffMS   []int             `json:"retry_backoff_ms"`
 	DeadRetentionSec int               `json:"dead_letter_retention_seconds"`
@@ -99,6 +105,10 @@ func initKafkaRuntime(raw json.RawMessage) error {
 	if len(cfg.Brokers) == 0 {
 		return errors.New("kafka: enabled but brokers are not configured")
 	}
+	saslMechanism, err := kafkaSASLMechanism(cfg)
+	if err != nil {
+		return err
+	}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	rt := &kafkaRuntime{
@@ -113,6 +123,7 @@ func initKafkaRuntime(raw json.RawMessage) error {
 	if cfg.TLSEnabled {
 		transport.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
 	}
+	transport.SASL = saslMechanism
 	rt.writer = &kafka.Writer{
 		Addr:                   kafka.TCP(cfg.Brokers...),
 		Balancer:               &kafka.Hash{},
@@ -161,6 +172,26 @@ func normalizeKafkaConfig(cfg *kafkaConfig) {
 	if cfg.DeadRetentionSec <= 0 {
 		cfg.DeadRetentionSec = 90 * 24 * 3600
 	}
+	if cfg.SASLMechanism == "" {
+		cfg.SASLMechanism = "SCRAM-SHA-512"
+	}
+}
+
+func kafkaSASLMechanism(cfg kafkaConfig) (sasl.Mechanism, error) {
+	if !cfg.SASLEnabled && cfg.SASLUsername == "" {
+		return nil, nil
+	}
+	if cfg.SASLUsername == "" || cfg.SASLPassword == "" {
+		return nil, errors.New("kafka: sasl enabled but username or password is empty")
+	}
+	switch strings.ToUpper(strings.ReplaceAll(cfg.SASLMechanism, "_", "-")) {
+	case "", "SCRAM-SHA-512":
+		return scram.Mechanism(scram.SHA512, cfg.SASLUsername, cfg.SASLPassword)
+	case "SCRAM-SHA-256":
+		return scram.Mechanism(scram.SHA256, cfg.SASLUsername, cfg.SASLPassword)
+	default:
+		return nil, errors.New("kafka: unsupported sasl mechanism " + cfg.SASLMechanism)
+	}
 }
 
 func kafkaTopics(overrides map[string]string) map[string]string {
@@ -291,6 +322,11 @@ func (rt *kafkaRuntime) dialer() *kafka.Dialer {
 	if rt.config.TLSEnabled {
 		dialer.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
 	}
+	if mechanism, err := kafkaSASLMechanism(rt.config); err != nil {
+		logs.Err.Println("Kafka SASL config failed:", err)
+	} else {
+		dialer.SASLMechanism = mechanism
+	}
 	return dialer
 }