From ca734a48ab37eb71f584e68fd5414432be992133 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 28 Aug 2026 00:31:08 +0800 Subject: [PATCH] feat: support kafka scram auth --- docker/tinode/Dockerfile | 4 ++++ docker/tinode/Dockerfile.build | 4 ++++ docker/tinode/config.template | 4 ++++ server/kafka_runtime.go | 36 ++++++++++++++++++++++++++++++++++ 4 files changed, 48 insertions(+) diff --git a/docker/tinode/Dockerfile b/docker/tinode/Dockerfile index 9fdd69a..7de3560 100644 --- a/docker/tinode/Dockerfile +++ b/docker/tinode/Dockerfile @@ -78,6 +78,10 @@ ENV KAFKA_READ_TIMEOUT_MS=3000 ENV KAFKA_REQUIRED_ACKS=1 ENV KAFKA_ALLOW_AUTO_TOPIC_CREATION=false ENV KAFKA_TLS_ENABLED=false +ENV KAFKA_SASL_ENABLED=false +ENV KAFKA_SASL_MECHANISM=SCRAM-SHA-512 +ENV KAFKA_SASL_USERNAME= +ENV KAFKA_SASL_PASSWORD= ENV KAFKA_MAX_RETRY_COUNT=5 ENV KAFKA_RETRY_BACKOFF_MS='[10000,30000,60000,300000,900000]' ENV KAFKA_DEAD_LETTER_RETENTION_SECONDS=7776000 diff --git a/docker/tinode/Dockerfile.build b/docker/tinode/Dockerfile.build index 8c70b94..3c7a66a 100644 --- a/docker/tinode/Dockerfile.build +++ b/docker/tinode/Dockerfile.build @@ -58,6 +58,10 @@ ENV KAFKA_READ_TIMEOUT_MS=3000 ENV KAFKA_REQUIRED_ACKS=1 ENV KAFKA_ALLOW_AUTO_TOPIC_CREATION=false ENV KAFKA_TLS_ENABLED=false +ENV KAFKA_SASL_ENABLED=false +ENV KAFKA_SASL_MECHANISM=SCRAM-SHA-512 +ENV KAFKA_SASL_USERNAME= +ENV KAFKA_SASL_PASSWORD= ENV KAFKA_MAX_RETRY_COUNT=5 ENV KAFKA_RETRY_BACKOFF_MS='[10000,30000,60000,300000,900000]' ENV KAFKA_DEAD_LETTER_RETENTION_SECONDS=7776000 diff --git a/docker/tinode/config.template b/docker/tinode/config.template index fb45d5a..8f1fef5 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -128,6 +128,10 @@ "required_acks": $KAFKA_REQUIRED_ACKS, "allow_auto_topic_creation": $KAFKA_ALLOW_AUTO_TOPIC_CREATION, "tls_enabled": $KAFKA_TLS_ENABLED, + "sasl_enabled": $KAFKA_SASL_ENABLED, + "sasl_mechanism": "$KAFKA_SASL_MECHANISM", + "sasl_username": "$KAFKA_SASL_USERNAME", + "sasl_password": "$KAFKA_SASL_PASSWORD", "max_retry_count": $KAFKA_MAX_RETRY_COUNT, "retry_backoff_ms": $KAFKA_RETRY_BACKOFF_MS, "dead_letter_retention_seconds": $KAFKA_DEAD_LETTER_RETENTION_SECONDS, diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go index b90db29..c306b21 100644 --- a/server/kafka_runtime.go +++ b/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 }