diff --git a/go.mod b/go.mod index f1d69689..05bbf59f 100644 --- a/go.mod +++ b/go.mod @@ -74,6 +74,7 @@ require ( github.com/lestrrat-go/strftime v1.0.6 github.com/sercand/kuberesolver/v6 v6.0.1 github.com/shirou/gopsutil/v3 v3.24.5 + github.com/xdg-go/scram v1.1.2 go.etcd.io/etcd/api/v3 v3.5.13 k8s.io/api v0.31.2 k8s.io/apimachinery v0.31.2 @@ -152,7 +153,6 @@ require ( github.com/x448/float16 v0.8.4 // indirect github.com/xanzy/ssh-agent v0.3.3 // indirect github.com/xdg-go/pbkdf2 v1.0.0 // indirect - github.com/xdg-go/scram v1.1.2 // indirect github.com/xdg-go/stringprep v1.0.4 // indirect github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect diff --git a/mq/kafka/config.go b/mq/kafka/config.go index 1857bbef..338a9bd9 100644 --- a/mq/kafka/config.go +++ b/mq/kafka/config.go @@ -34,10 +34,8 @@ func BuildConsumerGroupConfig(conf *Config, initial int64, autoCommitEnable bool if conf.ConsumerFetchMaxBytes > 0 { kfk.Consumer.Fetch.Max = int32(conf.ConsumerFetchMaxBytes) } - if conf.Username != "" || conf.Password != "" { - kfk.Net.SASL.Enable = true - kfk.Net.SASL.User = conf.Username - kfk.Net.SASL.Password = conf.Password + if err := configureSASL(kfk, conf); err != nil { + return nil, err } if conf.TLS.EnableTLS { tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify) @@ -53,7 +51,7 @@ func BuildConsumerGroupConfig(conf *Config, initial int64, autoCommitEnable bool func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error) { cg, err := sarama.NewConsumerGroup(addr, groupID, conf) if err != nil { - return nil, errs.WrapMsg(err, "NewConsumerGroup failed", "addr", addr, "groupID", groupID, "conf", *conf) + return nil, errs.WrapMsg(err, "new consumer group failed", "addr", addr, "groupID", groupID) } return cg, nil } @@ -63,10 +61,8 @@ func BuildProducerConfig(conf Config) (*sarama.Config, error) { kfk.Producer.Return.Successes = true kfk.Producer.Return.Errors = true kfk.Producer.Partitioner = sarama.NewHashPartitioner - if conf.Username != "" || conf.Password != "" { - kfk.Net.SASL.Enable = true - kfk.Net.SASL.User = conf.Username - kfk.Net.SASL.Password = conf.Password + if err := configureSASL(kfk, &conf); err != nil { + return nil, err } switch strings.ToLower(conf.ProducerAck) { case "no_response": @@ -99,10 +95,40 @@ func BuildProducerConfig(conf Config) (*sarama.Config, error) { return kfk, nil } +func configureSASL(kfk *sarama.Config, conf *Config) error { + mechanism := strings.ToUpper(strings.TrimSpace(conf.SASLMechanism)) + if mechanism == "" { + if conf.Username == "" && conf.Password == "" { + return nil + } + mechanism = string(sarama.SASLTypePlaintext) + } + + switch sarama.SASLMechanism(mechanism) { + case sarama.SASLTypePlaintext: + case sarama.SASLTypeSCRAMSHA256, sarama.SASLTypeSCRAMSHA512: + if conf.Username == "" { + return errs.New("kafka SASL username is required").Wrap() + } + if conf.Password == "" { + return errs.New("kafka SASL password is required").Wrap() + } + kfk.Net.SASL.SCRAMClientGeneratorFunc = newSCRAMClientGenerator(sarama.SASLMechanism(mechanism)) + default: + return errs.New("unsupported kafka SASL mechanism", "mechanism", conf.SASLMechanism).Wrap() + } + + kfk.Net.SASL.Enable = true + kfk.Net.SASL.User = conf.Username + kfk.Net.SASL.Password = conf.Password + kfk.Net.SASL.Mechanism = sarama.SASLMechanism(mechanism) + return nil +} + func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error) { producer, err := sarama.NewSyncProducer(addr, conf) if err != nil { - return nil, errs.WrapMsg(err, "NewSyncProducer failed", "addr", addr, "conf", *conf) + return nil, errs.WrapMsg(err, "new sync producer failed", "addr", addr) } return producer, nil } @@ -119,6 +145,7 @@ type TLSConfig struct { type Config struct { Username string `yaml:"username"` Password string `yaml:"password"` + SASLMechanism string `yaml:"saslMechanism"` ProducerAck string `yaml:"producerAck"` CompressType string `yaml:"compressType"` MaxMessageBytes int `yaml:"maxMessageBytes"` diff --git a/mq/kafka/kafka_test.go b/mq/kafka/kafka_test.go index 836d0a14..a6d29407 100644 --- a/mq/kafka/kafka_test.go +++ b/mq/kafka/kafka_test.go @@ -1,7 +1,244 @@ package kafka -import "testing" +import ( + "strings" + "testing" -func TestProducer(t *testing.T) { + "github.com/IBM/sarama" +) +func TestBuildConfigSASL(t *testing.T) { + tests := []struct { + name string + config Config + saslEnabled bool + mechanism sarama.SASLMechanism + hasSCRAMGenerator bool + }{ + { + name: "disabled", + }, + { + name: "legacy plain", + config: Config{ + Username: "user", + Password: "password", + }, + saslEnabled: true, + mechanism: sarama.SASLTypePlaintext, + }, + { + name: "explicit plain", + config: Config{ + Username: "user", + Password: "password", + SASLMechanism: "plain", + }, + saslEnabled: true, + mechanism: sarama.SASLTypePlaintext, + }, + { + name: "scram sha 256", + config: Config{ + Username: "user", + Password: "password", + SASLMechanism: "SCRAM-SHA-256", + }, + saslEnabled: true, + mechanism: sarama.SASLTypeSCRAMSHA256, + hasSCRAMGenerator: true, + }, + { + name: "scram sha 512", + config: Config{ + Username: "user", + Password: "password", + SASLMechanism: "scram-sha-512", + }, + saslEnabled: true, + mechanism: sarama.SASLTypeSCRAMSHA512, + hasSCRAMGenerator: true, + }, + } + + builders := []struct { + name string + build func(config Config) (*sarama.Config, error) + }{ + { + name: "producer", + build: BuildProducerConfig, + }, + { + name: "consumer", + build: func(config Config) (*sarama.Config, error) { + return BuildConsumerGroupConfig(&config, sarama.OffsetNewest, false) + }, + }, + } + + for _, builder := range builders { + t.Run(builder.name, func(t *testing.T) { + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config, err := builder.build(tt.config) + if err != nil { + t.Fatalf("build config failed: %v", err) + } + if config.Net.SASL.Enable != tt.saslEnabled { + t.Fatalf("unexpected SASL enabled state: %v", config.Net.SASL.Enable) + } + if !tt.saslEnabled { + return + } + if config.Net.SASL.Mechanism != tt.mechanism { + t.Fatalf("unexpected SASL mechanism: %s", config.Net.SASL.Mechanism) + } + if config.Net.SASL.User != tt.config.Username { + t.Fatalf("unexpected SASL username: %s", config.Net.SASL.User) + } + if config.Net.SASL.Password != tt.config.Password { + t.Fatal("unexpected SASL password") + } + if (config.Net.SASL.SCRAMClientGeneratorFunc != nil) != tt.hasSCRAMGenerator { + t.Fatal("unexpected SCRAM generator state") + } + if tt.hasSCRAMGenerator { + client := config.Net.SASL.SCRAMClientGeneratorFunc() + if err := client.Begin(tt.config.Username, tt.config.Password, ""); err != nil { + t.Fatalf("initialize SCRAM client failed: %v", err) + } + } + }) + } + }) + } +} + +func TestBuildConfigSASLErrors(t *testing.T) { + tests := []struct { + name string + config Config + errMessage string + }{ + { + name: "unsupported mechanism", + config: Config{ + Username: "user", + Password: "password", + SASLMechanism: "GSSAPI", + }, + errMessage: "unsupported kafka SASL mechanism", + }, + { + name: "scram username required", + config: Config{ + Password: "password", + SASLMechanism: "SCRAM-SHA-512", + }, + errMessage: "kafka SASL username is required", + }, + { + name: "scram password required", + config: Config{ + Username: "user", + SASLMechanism: "SCRAM-SHA-512", + }, + errMessage: "kafka SASL password is required", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := BuildProducerConfig(tt.config) + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), tt.errMessage) { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} + +func TestSCRAMClientBeginWrapsError(t *testing.T) { + client := newSCRAMClientGenerator(sarama.SASLTypeSCRAMSHA512)() + err := client.Begin("\a", "password", "") + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "initialize kafka SCRAM client failed") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestSCRAMClientBeginDoesNotExposePassword(t *testing.T) { + const passwordMarker = "secret-password" + + client := newSCRAMClientGenerator(sarama.SASLTypeSCRAMSHA512)() + err := client.Begin("user", passwordMarker+"\a", "") + if err == nil { + t.Fatal("expected error") + } + if strings.Contains(err.Error(), passwordMarker) { + t.Fatalf("password leaked in error: %v", err) + } +} + +func TestSCRAMClientStepWrapsError(t *testing.T) { + client := newSCRAMClientGenerator(sarama.SASLTypeSCRAMSHA512)() + if err := client.Begin("user", "password", ""); err != nil { + t.Fatalf("initialize SCRAM client failed: %v", err) + } + if _, err := client.Step(""); err != nil { + t.Fatalf("create SCRAM client first message failed: %v", err) + } + _, err := client.Step("invalid-server-first-message") + if err == nil { + t.Fatal("expected error") + } + if !strings.Contains(err.Error(), "advance kafka SCRAM client failed") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestClientCreationErrorsDoNotExposePassword(t *testing.T) { + const password = "secret-password-must-not-leak" + + tests := []struct { + name string + create func() error + }{ + { + name: "producer", + create: func() error { + config := sarama.NewConfig() + config.Net.SASL.Password = password + _, err := NewProducer(config, nil) + return err + }, + }, + { + name: "consumer group", + create: func() error { + config := sarama.NewConfig() + config.Net.SASL.Password = password + config.Consumer.Group.Session.Timeout = 0 + _, err := NewConsumerGroup(config, nil, "group") + return err + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.create() + if err == nil { + t.Fatal("expected error") + } + if strings.Contains(err.Error(), password) { + t.Fatalf("password leaked in error: %v", err) + } + }) + } } diff --git a/mq/kafka/scram_client.go b/mq/kafka/scram_client.go new file mode 100644 index 00000000..abf50f9d --- /dev/null +++ b/mq/kafka/scram_client.go @@ -0,0 +1,57 @@ +// Copyright 2026 OpenIM open source community. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "github.com/IBM/sarama" + "github.com/openimsdk/tools/errs" + "github.com/xdg-go/scram" +) + +type scramClient struct { + hashGenerator scram.HashGeneratorFcn + conversation *scram.ClientConversation +} + +func newSCRAMClientGenerator(mechanism sarama.SASLMechanism) func() sarama.SCRAMClient { + hashGenerator := scram.SHA256 + if mechanism == sarama.SASLTypeSCRAMSHA512 { + hashGenerator = scram.SHA512 + } + return func() sarama.SCRAMClient { + return &scramClient{hashGenerator: hashGenerator} + } +} + +func (s *scramClient) Begin(userName, password, authzID string) error { + client, err := s.hashGenerator.NewClient(userName, password, authzID) + if err != nil { + return errs.New("initialize kafka SCRAM client failed").Wrap() + } + s.conversation = client.NewConversation() + return nil +} + +func (s *scramClient) Step(challenge string) (string, error) { + response, err := s.conversation.Step(challenge) + if err != nil { + return "", errs.WrapMsg(err, "advance kafka SCRAM client failed") + } + return response, nil +} + +func (s *scramClient) Done() bool { + return s.conversation.Done() +} diff --git a/mq/kafka/verify.go b/mq/kafka/verify.go index 0a09eed4..d5c9004f 100644 --- a/mq/kafka/verify.go +++ b/mq/kafka/verify.go @@ -16,7 +16,6 @@ package kafka import ( "context" - "fmt" "github.com/IBM/sarama" "github.com/openimsdk/tools/errs" @@ -29,13 +28,13 @@ func CheckTopics(ctx context.Context, conf *Config, topics []string) error { } cli, err := sarama.NewClient(conf.Addr, kfk) if err != nil { - return errs.WrapMsg(err, "NewClient failed", "config: ", fmt.Sprintf("%+v", conf)) + return errs.WrapMsg(err, "new kafka client failed", "addr", conf.Addr) } defer cli.Close() existingTopics, err := cli.Topics() if err != nil { - return errs.WrapMsg(err, "Failed to list topics") + return errs.WrapMsg(err, "failed to list topics") } existingTopicsMap := make(map[string]bool) @@ -58,7 +57,7 @@ func CheckHealth(ctx context.Context, conf *Config) error { } cli, err := sarama.NewClient(conf.Addr, kfk) if err != nil { - return errs.WrapMsg(err, "NewClient failed", "config: ", fmt.Sprintf("%+v", conf)) + return errs.WrapMsg(err, "new kafka client failed", "addr", conf.Addr) } defer cli.Close()