home/telegraf
2
0
Fork 0
mirror of https://github.com/wahyd4/telegraf.git synced 2026-08-09 04:36:05 +10:00
Code Issues Packages Projects Releases Wiki Activity

Add SCRAM SASL support to kafka plugins

Browse Source
This commit is contained in:
Daniel Nelson
2019-07-11 13:51:01 -07:00
parent c9107015b0
commit 4f49189acf
5 changed files with 218 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Gopkg.lock
Generated
+113 -3
View File
@@ -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",
internal/kafka/sasl.go
+53
View File
@@ -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
}
internal/kafka/scram.go
+37
View File
@@ -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()
}
plugins/inputs/kafka_consumer/kafka_consumer.go
+8 -9
View File
@@ -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) {
plugins/outputs/kafka/kafka.go
+7 -10
View File
@@ -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)