NETOBSERV-974 Implement SASL for Kafka (producer+consumer) (#424)

* Implement SASL for Kafka (producer+consumer)

* fix test

Author: jotak

go.mod

@@ -93,6 +93,8 @@ require (
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/subosito/gotenv v1.2.0 // indirect
+	github.com/xdg/scram v1.0.5 // indirect
+	github.com/xdg/stringprep v1.0.3 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	golang.org/x/crypto v0.5.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5 // indirect

pkg/api/encode_kafka.go

@@ -18,14 +18,15 @@
 package api
 
 type EncodeKafka struct {
-	Address      string     `yaml:"address" json:"address" doc:"address of kafka server"`
-	Topic        string     `yaml:"topic" json:"topic" doc:"kafka topic to write to"`
-	Balancer     string     `yaml:"balancer,omitempty" json:"balancer,omitempty" enum:"KafkaEncodeBalancerEnum" doc:"one of the following:"`
-	WriteTimeout int64      `yaml:"writeTimeout,omitempty" json:"writeTimeout,omitempty" doc:"timeout (in seconds) for write operation performed by the Writer"`
-	ReadTimeout  int64      `yaml:"readTimeout,omitempty" json:"readTimeout,omitempty" doc:"timeout (in seconds) for read operation performed by the Writer"`
-	BatchBytes   int64      `yaml:"batchBytes,omitempty" json:"batchBytes,omitempty" doc:"limit the maximum size of a request in bytes before being sent to a partition"`
-	BatchSize    int        `yaml:"batchSize,omitempty" json:"batchSize,omitempty" doc:"limit on how many messages will be buffered before being sent to a partition"`
-	TLS          *ClientTLS `yaml:"tls" json:"tls" doc:"TLS client configuration (optional)"`
+	Address      string      `yaml:"address" json:"address" doc:"address of kafka server"`
+	Topic        string      `yaml:"topic" json:"topic" doc:"kafka topic to write to"`
+	Balancer     string      `yaml:"balancer,omitempty" json:"balancer,omitempty" enum:"KafkaEncodeBalancerEnum" doc:"one of the following:"`
+	WriteTimeout int64       `yaml:"writeTimeout,omitempty" json:"writeTimeout,omitempty" doc:"timeout (in seconds) for write operation performed by the Writer"`
+	ReadTimeout  int64       `yaml:"readTimeout,omitempty" json:"readTimeout,omitempty" doc:"timeout (in seconds) for read operation performed by the Writer"`
+	BatchBytes   int64       `yaml:"batchBytes,omitempty" json:"batchBytes,omitempty" doc:"limit the maximum size of a request in bytes before being sent to a partition"`
+	BatchSize    int         `yaml:"batchSize,omitempty" json:"batchSize,omitempty" doc:"limit on how many messages will be buffered before being sent to a partition"`
+	TLS          *ClientTLS  `yaml:"tls" json:"tls" doc:"TLS client configuration (optional)"`
+	SASL         *SASLConfig `yaml:"sasl" json:"sasl" doc:"SASL configuration (optional)"`
 }
 
 type KafkaEncodeBalancerEnum struct {

pkg/api/enum.go

@@ -28,6 +28,7 @@ type enums struct {
 	TransformFilterOperationEnum  TransformFilterOperationEnum
 	TransformGenericOperationEnum TransformGenericOperationEnum
 	KafkaEncodeBalancerEnum       KafkaEncodeBalancerEnum
+	SASLTypeEnum                  SASLTypeEnum
 	ConnTrackOperationEnum        ConnTrackOperationEnum
 	ConnTrackOutputRecordTypeEnum ConnTrackOutputRecordTypeEnum
 	DecoderEnum                   DecoderEnum

pkg/api/ingest_kafka.go

@@ -18,16 +18,17 @@
 package api
 
 type IngestKafka struct {
-	Brokers           []string   `yaml:"brokers,omitempty" json:"brokers,omitempty" doc:"list of kafka broker addresses"`
-	Topic             string     `yaml:"topic,omitempty" json:"topic,omitempty" doc:"kafka topic to listen on"`
-	GroupId           string     `yaml:"groupid,omitempty" json:"groupid,omitempty" doc:"separate groupid for each consumer on specified topic"`
-	GroupBalancers    []string   `yaml:"groupBalancers,omitempty" json:"groupBalancers,omitempty" doc:"list of balancing strategies (range, roundRobin, rackAffinity)"`
-	StartOffset       string     `yaml:"startOffset,omitempty" json:"startOffset,omitempty" doc:"FirstOffset (least recent - default) or LastOffset (most recent) offset available for a partition"`
-	BatchReadTimeout  int64      `yaml:"batchReadTimeout,omitempty" json:"batchReadTimeout,omitempty" doc:"how often (in milliseconds) to process input"`
-	Decoder           Decoder    `yaml:"decoder,omitempty" json:"decoder" doc:"decoder to use (E.g. json or protobuf)"`
-	BatchMaxLen       int        `yaml:"batchMaxLen,omitempty" json:"batchMaxLen,omitempty" doc:"the number of accumulated flows before being forwarded for processing"`
-	PullQueueCapacity int        `yaml:"pullQueueCapacity,omitempty" json:"pullQueueCapacity,omitempty" doc:"the capacity of the queue use to store pulled flows"`
-	PullMaxBytes      int        `yaml:"pullMaxBytes,omitempty" json:"pullMaxBytes,omitempty" doc:"the maximum number of bytes being pulled from kafka"`
-	CommitInterval    int64      `yaml:"commitInterval,omitempty" json:"commitInterval,omitempty" doc:"the interval (in milliseconds) at which offsets are committed to the broker.  If 0, commits will be handled synchronously."`
-	TLS               *ClientTLS `yaml:"tls" json:"tls" doc:"TLS client configuration (optional)"`
+	Brokers           []string    `yaml:"brokers,omitempty" json:"brokers,omitempty" doc:"list of kafka broker addresses"`
+	Topic             string      `yaml:"topic,omitempty" json:"topic,omitempty" doc:"kafka topic to listen on"`
+	GroupId           string      `yaml:"groupid,omitempty" json:"groupid,omitempty" doc:"separate groupid for each consumer on specified topic"`
+	GroupBalancers    []string    `yaml:"groupBalancers,omitempty" json:"groupBalancers,omitempty" doc:"list of balancing strategies (range, roundRobin, rackAffinity)"`
+	StartOffset       string      `yaml:"startOffset,omitempty" json:"startOffset,omitempty" doc:"FirstOffset (least recent - default) or LastOffset (most recent) offset available for a partition"`
+	BatchReadTimeout  int64       `yaml:"batchReadTimeout,omitempty" json:"batchReadTimeout,omitempty" doc:"how often (in milliseconds) to process input"`
+	Decoder           Decoder     `yaml:"decoder,omitempty" json:"decoder" doc:"decoder to use (E.g. json or protobuf)"`
+	BatchMaxLen       int         `yaml:"batchMaxLen,omitempty" json:"batchMaxLen,omitempty" doc:"the number of accumulated flows before being forwarded for processing"`
+	PullQueueCapacity int         `yaml:"pullQueueCapacity,omitempty" json:"pullQueueCapacity,omitempty" doc:"the capacity of the queue use to store pulled flows"`
+	PullMaxBytes      int         `yaml:"pullMaxBytes,omitempty" json:"pullMaxBytes,omitempty" doc:"the maximum number of bytes being pulled from kafka"`
+	CommitInterval    int64       `yaml:"commitInterval,omitempty" json:"commitInterval,omitempty" doc:"the interval (in milliseconds) at which offsets are committed to the broker.  If 0, commits will be handled synchronously."`
+	TLS               *ClientTLS  `yaml:"tls" json:"tls" doc:"TLS client configuration (optional)"`
+	SASL              *SASLConfig `yaml:"sasl" json:"sasl" doc:"SASL configuration (optional)"`
 }

pkg/api/sasl.go

@@ -0,0 +1,16 @@
+package api
+
+type SASLConfig struct {
+	Type             string
+	ClientIDPath     string `yaml:"clientIDPath,omitempty" json:"clientIDPath,omitempty" doc:"path to the client ID / SASL username"`
+	ClientSecretPath string `yaml:"clientSecretPath,omitempty" json:"clientSecretPath,omitempty" doc:"path to the client secret / SASL password"`
+}
+
+type SASLTypeEnum struct {
+	Plain       string `yaml:"plain" json:"plain" doc:"Plain SASL"`
+	ScramSHA512 string `yaml:"scramSHA512" json:"scramSHA512" doc:"SCRAM/SHA512 SASL"`
+}
+
+func SASLTypeName(operation string) string {
+	return GetEnumName(SASLTypeEnum{}, operation)
+}

pkg/api/tls.go

@@ -62,5 +62,5 @@ func (c *ClientTLS) Build() (*tls.Config, error) {
 		}
 		return tlsConfig, nil
 	}
-	return nil, nil
+	return tlsConfig, nil
 }

pkg/config/pipeline_builder_test.go

@@ -150,7 +150,7 @@ func TestKafkaPromPipeline(t *testing.T) {
 
 	b, err = json.Marshal(params[0])
 	require.NoError(t, err)
-	require.JSONEq(t, `{"name":"ingest","ingest":{"type":"kafka","kafka":{"brokers":["http://kafka"],"topic":"netflows","groupid":"my-group","decoder":{"type":"json"},"tls":{"insecureSkipVerify":true,"caCertPath":"/ca.crt"}}}}`, string(b))
+	require.JSONEq(t, `{"name":"ingest","ingest":{"type":"kafka","kafka":{"brokers":["http://kafka"],"topic":"netflows","groupid":"my-group","decoder":{"type":"json"},"sasl":null,"tls":{"insecureSkipVerify":true,"caCertPath":"/ca.crt"}}}}`, string(b))
 
 	b, err = json.Marshal(params[1])
 	require.NoError(t, err)

pkg/pipeline/encode/encode_kafka.go

@@ -24,8 +24,8 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/operational"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/segmentio/kafka-go"
 	kafkago "github.com/segmentio/kafka-go"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
@@ -95,7 +95,7 @@ func NewEncodeKafka(opMetrics *operational.Metrics, params config.StageParam) (E
 		writeTimeoutSecs = config.WriteTimeout
 	}
 
-	transport := kafka.Transport{}
+	transport := kafkago.Transport{}
 	if config.TLS != nil {
 		log.Infof("Using TLS configuration: %v", config.TLS)
 		tlsConfig, err := config.TLS.Build()
@@ -105,6 +105,14 @@ func NewEncodeKafka(opMetrics *operational.Metrics, params config.StageParam) (E
 		transport.TLS = tlsConfig
 	}
 
+	if config.SASL != nil {
+		m, err := utils.SetupSASLMechanism(config.SASL)
+		if err != nil {
+			return nil, err
+		}
+		transport.SASL = m
+	}
+
 	// connect to the kafka server
 	kafkaWriter := kafkago.Writer{
 		Addr:         kafkago.TCP(config.Address),

pkg/pipeline/ingest/ingest_kafka.go

@@ -232,6 +232,14 @@ func NewIngestKafka(opMetrics *operational.Metrics, params config.StageParam) (I
 		dialer.TLS = tlsConfig
 	}
 
+	if jsonIngestKafka.SASL != nil {
+		m, err := utils.SetupSASLMechanism(jsonIngestKafka.SASL)
+		if err != nil {
+			return nil, err
+		}
+		dialer.SASLMechanism = m
+	}
+
 	readerConfig := kafkago.ReaderConfig{
 		Brokers:        jsonIngestKafka.Brokers,
 		Topic:          jsonIngestKafka.Topic,

pkg/pipeline/utils/sasl.go

@@ -0,0 +1,40 @@
+package utils
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/netobserv/flowlogs-pipeline/pkg/api"
+	"github.com/segmentio/kafka-go/sasl"
+	"github.com/segmentio/kafka-go/sasl/plain"
+	"github.com/segmentio/kafka-go/sasl/scram"
+)
+
+func SetupSASLMechanism(cfg *api.SASLConfig) (sasl.Mechanism, error) {
+	// Read client ID
+	id, err := os.ReadFile(cfg.ClientIDPath)
+	if err != nil {
+		return nil, err
+	}
+	strId := strings.TrimSpace(string(id))
+	// Read password
+	pwd, err := os.ReadFile(cfg.ClientSecretPath)
+	if err != nil {
+		return nil, err
+	}
+	strPwd := strings.TrimSpace(string(pwd))
+	var mechanism sasl.Mechanism
+	switch cfg.Type {
+	case api.SASLTypeName("Plain"):
+		mechanism = plain.Mechanism{Username: strId, Password: strPwd}
+	case api.SASLTypeName("ScramSHA512"):
+		mechanism, err = scram.Mechanism(scram.SHA512, strId, strPwd)
+	default:
+		return nil, fmt.Errorf("Unknown SASL type: %s", cfg.Type)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return mechanism, nil
+}