feat (kafa-es): Implements SCRAM functionality and enables SCRAM-SHA5…

…12/256 SASL * add InsecureSkipVerify property Signed-off-by: Usama Kaleem <[email protected]>
argoproj · Feb 10, 2022 · ea5b5a5 · ea5b5a5
1 parent 097e87d
commit ea5b5a5
Show file tree

Hide file tree

Showing 16 changed files with 258 additions and 128 deletions.
diff --git a/USERS.md b/USERS.md
@@ -21,6 +21,7 @@ Organizations below are **officially** using Argo Events. Please send a PR with
 1. [Helio](https://helio.exchange)
 1. [InsideBoard](https://www.insideboard.com)
 1. [Intuit](https://www.intuit.com/)
+1. [Mobimeo GmbH](https://mobimeo.com/en/home/)
 1. [OneCause](https://www.onecause.com/)
 1. [Produvar](https://www.produvar.com/)
 1. [ProPoint Solutions](https://supersalon.com)

diff --git a/api/jsonschema/schema.json b/api/jsonschema/schema.json
@@ -238,6 +238,10 @@
         "clientKeySecret": {
           "$ref": "#/definitions/io.k8s.api.core.v1.SecretKeySelector",
           "description": "ClientKeySecret refers to the secret that contains the client key"
+        },
+        "insecureSkipVerify": {
+          "description": "If true, skips creation of TLSConfig with certs and creates an empty TLSConfig. (Defaults to false)",
+          "type": "boolean"
         }
       },
       "type": "object"

diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json
diff --git a/common/util.go b/common/util.go
@@ -210,12 +210,20 @@ func GenerateEnvFromConfigMapSpec(selector *v1.ConfigMapKeySelector) v1.EnvFromS
 	}
 }
 
-// GetTLSConfig returns a tls configuration for given cert and key.
+// GetTLSConfig returns a tls configuration for given cert and key or skips the certs if InsecureSkipVerify is true.
 func GetTLSConfig(config *apicommon.TLSConfig) (*tls.Config, error) {
 	if config == nil {
 		return nil, errors.New("TLSConfig is nil")
 	}
 
+	if config.InsecureSkipVerify {
+		tlsConfig := &tls.Config{
+			InsecureSkipVerify: true,
+			ClientAuth:         0,
+		}
+		return tlsConfig, nil
+	}
+
 	var caCertPath, clientCertPath, clientKeyPath string
 	var err error
 	if config.CACertSecret != nil {

diff --git a/common/util_test.go b/common/util_test.go
@@ -174,27 +174,33 @@ func TestVolumesFromSecretsOrConfigMaps(t *testing.T) {
 	})
 }
 
-func fakeTLSConfig(t *testing.T) *apicommon.TLSConfig {
+func fakeTLSConfig(t *testing.T, insecureSkipVerify bool) *apicommon.TLSConfig {
 	t.Helper()
-	return &apicommon.TLSConfig{
-		CACertSecret: &corev1.SecretKeySelector{
-			Key: "fake-key1",
-			LocalObjectReference: corev1.LocalObjectReference{
-				Name: "fake-name1",
+	if insecureSkipVerify == true {
+		return &apicommon.TLSConfig{
+			InsecureSkipVerify: true,
+		}
+	} else {
+		return &apicommon.TLSConfig{
+			CACertSecret: &corev1.SecretKeySelector{
+				Key: "fake-key1",
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: "fake-name1",
+				},
 			},
-		},
-		ClientCertSecret: &corev1.SecretKeySelector{
-			Key: "fake-key2",
-			LocalObjectReference: corev1.LocalObjectReference{
-				Name: "fake-name2",
+			ClientCertSecret: &corev1.SecretKeySelector{
+				Key: "fake-key2",
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: "fake-name2",
+				},
 			},
-		},
-		ClientKeySecret: &corev1.SecretKeySelector{
-			Key: "fake-key3",
-			LocalObjectReference: corev1.LocalObjectReference{
-				Name: "fake-name3",
+			ClientKeySecret: &corev1.SecretKeySelector{
+				Key: "fake-key3",
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: "fake-name3",
+				},
 			},
-		},
+		}
 	}
 }
 
@@ -207,7 +213,7 @@ func TestGetTLSConfig(t *testing.T) {
 	})
 
 	t.Run("test clientKeySecret is set, clientCertSecret is empty", func(t *testing.T) {
-		c := fakeTLSConfig(t)
+		c := fakeTLSConfig(t, false)
 		c.CACertSecret = nil
 		c.ClientCertSecret = nil
 		_, err := GetTLSConfig(c)
@@ -216,7 +222,7 @@ func TestGetTLSConfig(t *testing.T) {
 	})
 
 	t.Run("test only caCertSecret is set", func(t *testing.T) {
-		c := fakeTLSConfig(t)
+		c := fakeTLSConfig(t, false)
 		c.ClientCertSecret = nil
 		c.ClientKeySecret = nil
 		_, err := GetTLSConfig(c)
@@ -225,15 +231,15 @@ func TestGetTLSConfig(t *testing.T) {
 	})
 
 	t.Run("test clientCertSecret and clientKeySecret are set", func(t *testing.T) {
-		c := fakeTLSConfig(t)
+		c := fakeTLSConfig(t, false)
 		c.CACertSecret = nil
 		_, err := GetTLSConfig(c)
 		assert.NotNil(t, err)
 		assert.True(t, strings.Contains(err.Error(), "failed to load client cert key pair"))
 	})
 
 	t.Run("test all of 3 are set", func(t *testing.T) {
-		c := fakeTLSConfig(t)
+		c := fakeTLSConfig(t, false)
 		_, err := GetTLSConfig(c)
 		assert.NotNil(t, err)
 		assert.True(t, strings.Contains(err.Error(), "failed to read ca cert file"))

diff --git a/eventsources/sources/kafka/scram_client.go b/eventsources/sources/kafka/scram_client.go
@@ -0,0 +1,37 @@
+package kafka
+
+import (
+	"crypto/sha256"
+	"crypto/sha512"
+
+	"github.com/xdg-go/scram"
+)
+
+var (
+	SHA256 scram.HashGeneratorFcn = sha256.New
+	SHA512 scram.HashGeneratorFcn = 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/eventsources/sources/kafka/start.go b/eventsources/sources/kafka/start.go
@@ -279,6 +279,11 @@ func getSaramaConfig(kafkaEventSource *v1alpha1.KafkaEventSource, log *zap.Sugar
 		config.Net.SASL.Enable = true
 
 		config.Net.SASL.Mechanism = sarama.SASLMechanism(kafkaEventSource.SASL.GetMechanism())
+		if config.Net.SASL.Mechanism == "SCRAM-SHA-512" {
+			config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} }
+		} else if config.Net.SASL.Mechanism == "SCRAM-SHA-256" {
+			config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA256} }
+		}
 
 		user, err := common.GetSecretFromVolume(kafkaEventSource.SASL.UserSecret)
 		if err != nil {

diff --git a/go.mod b/go.mod
@@ -91,6 +91,8 @@ require (
 	github.com/felixge/httpsnoop v1.0.2 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/stringprep v1.0.2 // indirect
 )
 
 require (
@@ -237,6 +239,7 @@ require (
 	github.com/valyala/fasthttp v1.9.0 // indirect
 	github.com/valyala/gozstd v1.7.0 // indirect
 	github.com/xanzy/ssh-agent v0.3.0 // indirect
+	github.com/xdg-go/scram v1.1.0
 	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
 	github.com/xeipuuv/gojsonschema v1.1.0 // indirect

diff --git a/go.sum b/go.sum
@@ -1183,9 +1183,12 @@ github.com/xanzy/go-gitlab v0.54.4 h1:3CFEdQ9O+bFx3BsyuOK0gqgLPwnT2rwnPOjudV07wT
 github.com/xanzy/go-gitlab v0.54.4/go.mod h1:F0QEXwmqiBUxCgJm8fE9S+1veX4XC9Z4cfaAbqwk4YM=
 github.com/xanzy/ssh-agent v0.3.0 h1:wUMzuKtKilRgBAD1sUb8gOwwRr2FGoBVumcjoOACClI=
 github.com/xanzy/ssh-agent v0.3.0/go.mod h1:3s9xbODqPuuhK9JV1R321M/FlMZSBvE5aY6eAcqrDh0=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
 github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
 github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=
+github.com/xdg-go/scram v1.1.0 h1:d70R37I0HrDLsafRrMBXyrD4lmQbCHE873t00Vr0gm0=
 github.com/xdg-go/scram v1.1.0/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=
+github.com/xdg-go/stringprep v1.0.2 h1:6iq84/ryjjeRmMJwxutI51F2GIPlP5BfTvXHeYjyhBc=
 github.com/xdg-go/stringprep v1.0.2/go.mod h1:8F9zXuvzgwmyT5DUm4GUfZGDdT3W+LCvS6+da4O5kxM=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v0.0.0-20180714160509-73f8eece6fdc/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=

diff --git a/pkg/apis/common/common.go b/pkg/apis/common/common.go
@@ -132,6 +132,9 @@ type TLSConfig struct {
 	ClientCertSecret *corev1.SecretKeySelector `json:"clientCertSecret,omitempty" protobuf:"bytes,2,opt,name=clientCertSecret"`
 	// ClientKeySecret refers to the secret that contains the client key
 	ClientKeySecret *corev1.SecretKeySelector `json:"clientKeySecret,omitempty" protobuf:"bytes,3,opt,name=clientKeySecret"`
+	// If true, skips creation of TLSConfig with certs and creates an empty TLSConfig. (Defaults to false)
+	// +optional
+	InsecureSkipVerify bool `json:"insecureSkipVerify,omitempty" protobuf:"varint,4,opt,name=insecureSkipVerify"`
 }
 
 // SASLConfig refers to SASL configuration for a client