diff --git a/Gopkg.lock b/Gopkg.lock index bcdf6cd0..5007c4b1 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -61,6 +61,14 @@ revision = "1f7cd6cfe0adea687ad44a512dfe76140f804318" version = "v10.12.0" +[[projects]] + digest = "1:e001bd0a052223a9dc871d38e303774759efae171abc702dedd312866400552d" + name = "github.com/DataDog/zstd" + packages = ["."] + pruneopts = "" + revision = "809b919c325d7887bff7bd876162af73db53e878" + version = "v1.4.0" + [[projects]] branch = "master" digest = "1:298712a3ee36b59c3ca91f4183bd75d174d5eaa8b4aed5072831f126e2e752f6" @@ -81,12 +89,12 @@ version = "v0.4.9" [[projects]] - digest = "1:213b41361ad1cb4768add9d26c2e27794c65264eefdb24ed6ea34cdfeeff3f3c" + digest = "1:d305ce4a17657b04b748f841dac052ea29e3f566fdb468491d760a9b9df85648" name = "github.com/Shopify/sarama" packages = ["."] pruneopts = "" - revision = "a6144ae922fd99dd0ea5046c8137acfb7fab0914" - version = "v1.18.0" + revision = "dde3ddda8b4b3a594690086725799ab1573bb895" + version = "v1.23.0" [[projects]] digest = "1:f82b8ac36058904227087141017bb82f4b0fc58272990a4cdae3e2d6d222644e" @@ -631,6 +639,14 @@ pruneopts = "" revision = "6bb64b370b90e7ef1fa532be9e591a81c3493e00" +[[projects]] + digest = "1:0038a7f43b51c8b2a8cd03b5372e73f8eadfe156484c2ae8185ae836f8ebc2cd" + name = "github.com/hashicorp/go-uuid" + packages = ["."] + pruneopts = "" + revision = "4f571afc59f3043a65f8fe6bf46d887b10a01d43" + version = "v1.0.1" + [[projects]] digest = "1:f72168ea995f398bab88e84bd1ff58a983466ba162fb8d50d47420666cd57fad" name = "github.com/hashicorp/serf" @@ -701,6 +717,17 @@ revision = "8faa4453fc7051d1076053f8854077753ab912f2" version = "v3.4.0" +[[projects]] + digest = "1:d45477e90c25c8c6d7d4237281167aa56079382fc042db4b44a8328071649bfa" + name = "github.com/jcmturner/gofork" + packages = [ + "encoding/asn1", + "x/crypto/pbkdf2", + ] + pruneopts = "" + revision = "dc7c13fece037a4a36e2b3c69db4991498d30692" + version = "v1.0.0" + [[projects]] digest = "1:6f49eae0c1e5dab1dafafee34b207aeb7a42303105960944828c2079b92fc88e" name = "github.com/jmespath/go-jmespath" @@ -1197,6 +1224,22 @@ pruneopts = "" revision = "f72d8611297a7cf105da904c04198ad701a60101" +[[projects]] + branch = "master" + digest = "1:ad74f33a69bd6ab0bd7287003b7c1069b94cfb5213eb5597005fe2963d7dfca9" + name = "github.com/xdg/scram" + packages = ["."] + pruneopts = "" + revision = "7eeb5667e42c09cb51bf7b7c28aea8c56767da90" + +[[projects]] + digest = "1:d110a69f3fba1632d3a543ebfbf9a62ea59bd2cf0c34608975d7e2c761c10edc" + name = "github.com/xdg/stringprep" + packages = ["."] + pruneopts = "" + revision = "bd625b8dc1e3b0f57412280ccbcc317f0c69d8db" + version = "v1.0.0" + [[projects]] branch = "master" digest = "1:c5918689b7e187382cc1066bf0260de54ba9d1b323105f46ed2551d2fb4a17c7" @@ -1505,6 +1548,72 @@ revision = "d2d2541c53f18d2a059457998ce2876cc8e67cbf" version = "v0.9.1" +[[projects]] + digest = "1:4777ba481cc12866b89aafb0a67529e7ac48b9aea06a25f3737b2cf5a3ffda12" + name = "gopkg.in/jcmturner/aescts.v1" + packages = ["."] + pruneopts = "" + revision = "f6abebb3171c4c1b1fea279cb7c7325020a26290" + version = "v1.0.1" + +[[projects]] + digest = "1:84c5b1392ef65ad1bb64da4b4d0beb2f204eefc769d6d96082347bb7057cb7b1" + name = "gopkg.in/jcmturner/dnsutils.v1" + packages = ["."] + pruneopts = "" + revision = "13eeb8d49ffb74d7a75784c35e4d900607a3943c" + version = "v1.0.1" + +[[projects]] + digest = "1:502ab576ba8c47c4de77fe3f2b2386adc1a1447bb5afae2ac7bf0edd2b6f7c52" + name = "gopkg.in/jcmturner/gokrb5.v7" + packages = [ + "asn1tools", + "client", + "config", + "credentials", + "crypto", + "crypto/common", + "crypto/etype", + "crypto/rfc3961", + "crypto/rfc3962", + "crypto/rfc4757", + "crypto/rfc8009", + "gssapi", + "iana", + "iana/addrtype", + "iana/adtype", + "iana/asnAppTag", + "iana/chksumtype", + "iana/errorcode", + "iana/etypeID", + "iana/flags", + "iana/keyusage", + "iana/msgtype", + "iana/nametype", + "iana/patype", + "kadmin", + "keytab", + "krberror", + "messages", + "pac", + "types", + ] + pruneopts = "" + revision = "363118e62befa8a14ff01031c025026077fe5d6d" + version = "v7.3.0" + +[[projects]] + digest = "1:f9956ccc103c6208cd50c71ee5191b6fdcc635972c12624ef949c9b20b2bb9d1" + name = "gopkg.in/jcmturner/rpc.v1" + packages = [ + "mstypes", + "ndr", + ] + pruneopts = "" + revision = "99a8ce2fbf8b8087b6ed12a37c61b10f04070043" + version = "v1.1.0" + [[projects]] digest = "1:367baf06b7dbd0ef0bbdd785f6a79f929c96b0c18e9d3b29c0eed1ac3f5db133" name = "gopkg.in/ldap.v2" @@ -1676,6 +1785,7 @@ "github.com/vmware/govmomi/vim25/types", "github.com/wavefronthq/wavefront-sdk-go/senders", "github.com/wvanbergen/kafka/consumergroup", + "github.com/xdg/scram", "golang.org/x/net/context", "golang.org/x/net/html/charset", "golang.org/x/oauth2", diff --git a/internal/kafka/sasl.go b/internal/kafka/sasl.go new file mode 100644 index 00000000..a97a063d --- /dev/null +++ b/internal/kafka/sasl.go @@ -0,0 +1,53 @@ +package kafka + +import ( + "errors" + + "github.com/Shopify/sarama" +) + +var ( + ErrUnknownMechanism = errors.New("unknown mechanism") +) + +type SASLConfig struct { + SASLMechanism string `toml:"sasl_mechanism"` + SASLUsername string `toml:"sasl_username"` + SASLPassword string `toml:"sasl_password"` +} + +type SaramaSASL struct { + Enable bool + Mechanism sarama.SASLMechanism + User string + Password string + SCRAMClientGeneratorFunc func() sarama.SCRAMClient +} + +func (c *SASLConfig) SetSaramaSASLConfig(config *sarama.Config) error { + sasl := config.Net.SASL + switch c.SASLMechanism { + case "PLAIN": + sasl.Mechanism = sarama.SASLTypePlaintext + case "SCRAM-SHA-256": + sasl.Mechanism = sarama.SASLTypeSCRAMSHA256 + sasl.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { + return &XDGSCRAMClient{HashGeneratorFcn: SHA512} + } + case "SCRAM-SHA-512": + sasl.Mechanism = sarama.SASLTypeSCRAMSHA512 + sasl.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { + return &XDGSCRAMClient{HashGeneratorFcn: SHA256} + } + default: + return ErrUnknownMechanism + } + + if c.SASLUsername != "" || c.SASLPassword != "" { + sasl.User = c.SASLUsername + sasl.Password = c.SASLPassword + sasl.Enable = true + } + + return nil +} diff --git a/internal/kafka/scram.go b/internal/kafka/scram.go new file mode 100644 index 00000000..ad8c8bbf --- /dev/null +++ b/internal/kafka/scram.go @@ -0,0 +1,37 @@ +// https://github.com/Shopify/sarama/blob/master/examples/sasl_scram_client/scram_client.go +package kafka + +import ( + "crypto/sha256" + "crypto/sha512" + "hash" + + "github.com/xdg/scram" +) + +var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() } +var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() } + +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() +} diff --git a/plugins/inputs/kafka_consumer/kafka_consumer.go b/plugins/inputs/kafka_consumer/kafka_consumer.go index 545e37f5..8841bbe3 100644 --- a/plugins/inputs/kafka_consumer/kafka_consumer.go +++ b/plugins/inputs/kafka_consumer/kafka_consumer.go @@ -10,6 +10,7 @@ import ( "github.com/Shopify/sarama" cluster "github.com/bsm/sarama-cluster" "github.com/influxdata/telegraf" + "github.com/influxdata/telegraf/internal/kafka" "github.com/influxdata/telegraf/internal/tls" "github.com/influxdata/telegraf/plugins/inputs" "github.com/influxdata/telegraf/plugins/parsers" @@ -38,11 +39,10 @@ type Kafka struct { Version string `toml:"version"` MaxUndeliveredMessages int `toml:"max_undelivered_messages"` Offset string `toml:"offset"` - SASLUsername string `toml:"sasl_username"` - SASLPassword string `toml:"sasl_password"` TopicTag string `toml:"topic_tag"` tls.ClientConfig + kafka.SASLConfig cluster Consumer parser parsers.Parser @@ -81,7 +81,8 @@ var sampleConfig = ` ## Use TLS but skip chain & host verification # insecure_skip_verify = false - ## Optional SASL Config + ## Can be one of "PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512" + # sasl_mechanism = "PLAIN" # sasl_username = "kafka" # sasl_password = "secret" @@ -153,12 +154,10 @@ func (k *Kafka) Start(acc telegraf.Accumulator) error { config.Net.TLS.Config = tlsConfig config.Net.TLS.Enable = true } - if k.SASLUsername != "" && k.SASLPassword != "" { - log.Printf("D! Using SASL auth with username '%s',", - k.SASLUsername) - config.Net.SASL.User = k.SASLUsername - config.Net.SASL.Password = k.SASLPassword - config.Net.SASL.Enable = true + + err = k.SASLConfig.SetSaramaSASLConfig(&config.Config) + if err != nil { + return err } switch strings.ToLower(k.Offset) { diff --git a/plugins/outputs/kafka/kafka.go b/plugins/outputs/kafka/kafka.go index 7ba457c5..8f9fb9d4 100644 --- a/plugins/outputs/kafka/kafka.go +++ b/plugins/outputs/kafka/kafka.go @@ -8,6 +8,7 @@ import ( "github.com/Shopify/sarama" "github.com/influxdata/telegraf" + "github.com/influxdata/telegraf/internal/kafka" tlsint "github.com/influxdata/telegraf/internal/tls" "github.com/influxdata/telegraf/plugins/outputs" "github.com/influxdata/telegraf/plugins/serializers" @@ -44,11 +45,7 @@ type ( CA string tlsint.ClientConfig - - // SASL Username - SASLUsername string `toml:"sasl_username"` - // SASL Password - SASLPassword string `toml:"sasl_password"` + kafka.SASLConfig tlsConfig tls.Config producer sarama.SyncProducer @@ -174,7 +171,8 @@ var sampleConfig = ` ## Use TLS but skip chain & host verification # insecure_skip_verify = false - ## Optional SASL Config + ## Can be one of "PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512" + # sasl_mechanism = "PLAIN" # sasl_username = "kafka" # sasl_password = "secret" @@ -266,10 +264,9 @@ func (k *Kafka) Connect() error { config.Net.TLS.Enable = true } - if k.SASLUsername != "" && k.SASLPassword != "" { - config.Net.SASL.User = k.SASLUsername - config.Net.SASL.Password = k.SASLPassword - config.Net.SASL.Enable = true + err = k.SASLConfig.SetSaramaSASLConfig(config) + if err != nil { + return err } producer, err := sarama.NewSyncProducer(k.Brokers, config)