Skip to content

feat(kafka): support SCRAM SASL mechanisms (#141) #257

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jackliu145 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(kafka): support SCRAM SASL mechanisms (#141) #257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
47 changes: 37 additions & 10 deletions mq/kafka/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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
}
Expand All @@ -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":
Expand Down Expand Up @@ -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
}
Expand All @@ -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"`
Expand Down
241 changes: 239 additions & 2 deletions mq/kafka/kafka_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
}
})
}
}
Loading