Repository: knative-sandbox/eventing-kafka
Pull request #205 (2020-11-19 14:23:36)

Move AuthConfig code to one place and add tests
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

## Proposed Changes

- If no `saslType` is provided, we should default to "PLAIN" (the source only supports plain)
- moving code to one central place
- adding tests

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/205#" title="Author self-approved">matzew</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/master/OWNERS)~~ [matzew]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205?src=pr&el=h1) Report
> Merging [#205](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205?src=pr&el=desc) (3df2b40) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/a6322c779b10496874edbaa0f2c4c25bf7655e2b?el=desc) (a6322c7) will **increase** coverage by `0.62%`.
> The diff coverage is `87.50%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #205      +/-   ##
==========================================
+ Coverage   75.42%   76.04%   +0.62%     
==========================================
  Files         112      112              
  Lines        4280     4266      -14     
==========================================
+ Hits         3228     3244      +16     
+ Misses        860      829      -31     
- Partials      192      193       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pkg/channel/consolidated/utils/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3V0aWxzL3V0aWwuZ28=) | `91.83% <ø> (ø)` | |
| [pkg/source/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205/diff?src=pr&el=tree#diff-cGtnL3NvdXJjZS9jbGllbnQuZ28=) | `65.00% <85.71%> (+19.54%)` | :arrow_up: |
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `50.00% <100.00%> (+5.30%)` | :arrow_up: |
| [...tributed/common/kafka/admin/eventhubcache/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL2FkbWluL2V2ZW50aHViY2FjaGUvY2FjaGUuZ28=) | `65.06% <0.00%> (-2.41%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205?src=pr&el=footer). Last update [a6322c7...274140a](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/205?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329430142304391168/artifacts/line-cov.html#file0) | 56.3% | 63.6% | 7.2
[pkg/source/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329430142304391168/artifacts/line-cov.html#file2) | 50.0% | 71.2% | 21.2

lberk: /assign
knative-prow-robot: @matzew: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | a2fd59f4e18fc496f82d56dfe1414bb0739d826c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-unit-tests/1329458037374586880) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/205). [Your PR dashboard](https://gubernator.knative.dev/pr/matzew).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329458037403947008/artifacts/line-cov.html#file0) | 56.3% | 63.6% | 7.2

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329460066624999424/artifacts/line-cov.html#file0) | 56.3% | 63.6% | 7.2
[pkg/source/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329460066624999424/artifacts/line-cov.html#file2) | 50.0% | 70.4% | 20.4

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329462293733314560/artifacts/line-cov.html#file0) | 56.3% | 63.6% | 7.2
[pkg/source/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329462293733314560/artifacts/line-cov.html#file2) | 50.0% | 71.2% | 21.2

lberk: /lgtm
thanks!
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329466133887062016/artifacts/line-cov.html#file0) | 56.3% | 63.4% | 7.1
[pkg/source/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329466133887062016/artifacts/line-cov.html#file2) | 50.0% | 71.1% | 21.1

knative-prow-robot: New changes are detected. LGTM label has been removed.
lberk: /lgtm
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329467534054789120/artifacts/line-cov.html#file0) | 56.3% | 63.4% | 7.1
[pkg/source/client.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/205/pull-knative-sandbox-eventing-kafka-go-coverage/1329467534054789120/artifacts/line-cov.html#file2) | 50.0% | 71.1% | 21.1


Patch:
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher.go b/pkg/channel/consolidated/dispatcher/dispatcher.go
index eb555f64f..9cb850125 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher.go
@@ -95,35 +95,10 @@ func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs) (*KafkaDispat
 	conf.Consumer.Return.Errors = true    // Returns the errors in ConsumerGroup#Errors() https://godoc.org/github.com/Shopify/sarama#ConsumerGroup
 	conf.Producer.Return.Successes = true // Must be enabled for sync producer
 
-	// Get the auth info
-	if args.KafkaAuthConfig != nil {
-		// tls
-		if args.KafkaAuthConfig.TLS != nil {
-			conf.Net.TLS.Enable = true
-			tlsConfig, err := source.NewTLSConfig(args.KafkaAuthConfig.TLS.Usercert, args.KafkaAuthConfig.TLS.Userkey, args.KafkaAuthConfig.TLS.Cacert)
-			if err != nil {
-				return nil, err
-			}
-			conf.Net.TLS.Config = tlsConfig
-		}
-		// SASL
-		if args.KafkaAuthConfig.SASL != nil {
-			conf.Net.SASL.Enable = true
-			conf.Net.SASL.Handshake = true
-
-			// if SASLTypeSCRAMSHA256 is provided we use that. Defaulting to SASLTypeSCRAMSHA512
-			if args.KafkaAuthConfig.SASL.SaslType == utils.SASLTypeSCRAMSHA256 {
-				conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &source.XDGSCRAMClient{HashGeneratorFcn: source.SHA256} }
-				conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256
-
-			} else {
-				conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &source.XDGSCRAMClient{HashGeneratorFcn: source.SHA512} }
-				conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
-			}
-
-			conf.Net.SASL.User = args.KafkaAuthConfig.SASL.User
-			conf.Net.SASL.Password = args.KafkaAuthConfig.SASL.Password
-		}
+	// append the auth info:
+	err := source.UpdateSaramaConfigWithKafkaAuthConfig(conf, args.KafkaAuthConfig)
+	if err != nil {
+		return nil, fmt.Errorf("Error updating the Sarama Auth config: %w", err)
 	}
 
 	producer, err := sarama.NewSyncProducer(args.Brokers, conf)
diff --git a/pkg/channel/consolidated/utils/util.go b/pkg/channel/consolidated/utils/util.go
index 58a29c325..8a3627db0 100644
--- a/pkg/channel/consolidated/utils/util.go
+++ b/pkg/channel/consolidated/utils/util.go
@@ -46,9 +46,6 @@ const (
 	SaslPassword = "password"
 	SaslType     = "saslType"
 
-	SASLTypeSCRAMSHA256 = "scram_sha256"
-	SASLTypeSCRAMSHA512 = "scram_sha512"
-
 	KafkaChannelSeparator = "."
 
 	knativeKafkaTopicPrefix = "knative-messaging-kafka"
diff --git a/pkg/source/client.go b/pkg/source/client.go
index 78b099dca..c4b275fc3 100644
--- a/pkg/source/client.go
+++ b/pkg/source/client.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"fmt"
 	"time"
 
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
@@ -132,13 +133,22 @@ func MakeAdminClient(clientID string, kafkaAuthCfg *utils.KafkaAuthConfig, boots
 	saramaConf.Version = sarama.V2_0_0_0
 	saramaConf.ClientID = clientID
 
+	err := UpdateSaramaConfigWithKafkaAuthConfig(saramaConf, kafkaAuthCfg)
+	if err != nil {
+		return nil, fmt.Errorf("Error creating the Admin Client: %w", err)
+	}
+
+	return sarama.NewClusterAdmin(bootstrapServers, saramaConf)
+}
+
+func UpdateSaramaConfigWithKafkaAuthConfig(saramaConf *sarama.Config, kafkaAuthCfg *utils.KafkaAuthConfig) error {
 	if kafkaAuthCfg != nil {
 		// tls
 		if kafkaAuthCfg.TLS != nil {
 			saramaConf.Net.TLS.Enable = true
 			tlsConfig, err := NewTLSConfig(kafkaAuthCfg.TLS.Usercert, kafkaAuthCfg.TLS.Userkey, kafkaAuthCfg.TLS.Cacert)
 			if err != nil {
-				return nil, err
+				return fmt.Errorf("Error creating TLS config: %w", err)
 			}
 			saramaConf.Net.TLS.Config = tlsConfig
 		}
@@ -147,20 +157,25 @@ func MakeAdminClient(clientID string, kafkaAuthCfg *utils.KafkaAuthConfig, boots
 			saramaConf.Net.SASL.Enable = true
 			saramaConf.Net.SASL.Handshake = true
 
-			// if SASLTypeSCRAMSHA256 is provided we use that. Defaulting to SASLTypeSCRAMSHA512
-			if kafkaAuthCfg.SASL.SaslType == utils.SASLTypeSCRAMSHA256 {
+			// if SaslType is not provided we are defaulting to PLAIN
+			if kafkaAuthCfg.SASL.SaslType == "" {
+				kafkaAuthCfg.SASL.SaslType = sarama.SASLTypePlaintext
+			}
+
+			if kafkaAuthCfg.SASL.SaslType == sarama.SASLTypeSCRAMSHA256 {
 				saramaConf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA256} }
 				saramaConf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256
-			} else {
+			}
+
+			if kafkaAuthCfg.SASL.SaslType == sarama.SASLTypeSCRAMSHA512 {
 				saramaConf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} }
 				saramaConf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
 			}
-
 			saramaConf.Net.SASL.User = kafkaAuthCfg.SASL.User
 			saramaConf.Net.SASL.Password = kafkaAuthCfg.SASL.Password
 		}
 	}
-	return sarama.NewClusterAdmin(bootstrapServers, saramaConf)
+	return nil
 }
 
 // verifyCertSkipHostname verifies certificates in the same way that the


Test patch:
diff --git a/pkg/source/client_test.go b/pkg/source/client_test.go
index 1d1a811fa..4ca06ef07 100644
--- a/pkg/source/client_test.go
+++ b/pkg/source/client_test.go
@@ -29,11 +29,122 @@ import (
 	"testing"
 	"time"
 
+	"knative.dev/eventing-kafka/pkg/channel/consolidated/utils"
+
 	"github.com/Shopify/sarama"
 
 	"github.com/stretchr/testify/require"
 )
 
+func TestUpdateSaramaConfigWithKafkaAuthConfig(t *testing.T) {
+
+	cert, key := generateCert(t)
+
+	testCases := map[string]struct {
+		kafkaAuthCfg  *utils.KafkaAuthConfig
+		enabledTLS    bool
+		enabledSASL   bool
+		salsMechanism string
+	}{
+		"No Auth": {
+			enabledTLS:  false,
+			enabledSASL: false,
+		},
+		"Only SASL-PLAIN Auth": {
+			kafkaAuthCfg: &utils.KafkaAuthConfig{
+				SASL: &utils.KafkaSaslConfig{
+					User:     "my-user",
+					Password: "super-secret",
+				},
+			},
+			enabledTLS:    false,
+			enabledSASL:   true,
+			salsMechanism: "",
+		},
+		"Only SASL-SCRAM-SHA-256 Auth": {
+			kafkaAuthCfg: &utils.KafkaAuthConfig{
+				SASL: &utils.KafkaSaslConfig{
+					User:     "my-user",
+					Password: "super-secret",
+					SaslType: "SCRAM-SHA-256",
+				},
+			},
+			enabledTLS:    false,
+			enabledSASL:   true,
+			salsMechanism: "SCRAM-SHA-256",
+		},
+		"Only SASL-SCRAM-SHA-512 Auth": {
+			kafkaAuthCfg: &utils.KafkaAuthConfig{
+				SASL: &utils.KafkaSaslConfig{
+					User:     "my-user",
+					Password: "super-secret",
+					SaslType: "SCRAM-SHA-512",
+				},
+			},
+			enabledTLS:    false,
+			enabledSASL:   true,
+			salsMechanism: "SCRAM-SHA-512",
+		},
+		"Only TLS Auth": {
+			kafkaAuthCfg: &utils.KafkaAuthConfig{
+				TLS: &utils.KafkaTlsConfig{
+					Cacert:   cert,
+					Usercert: cert,
+					Userkey:  key,
+				},
+			},
+			enabledTLS:  true,
+			enabledSASL: false,
+		},
+		"SASL and TLS Auth": {
+			kafkaAuthCfg: &utils.KafkaAuthConfig{
+				SASL: &utils.KafkaSaslConfig{
+					User:     "my-user",
+					Password: "super-secret",
+					SaslType: "SCRAM-SHA-512",
+				},
+				TLS: &utils.KafkaTlsConfig{
+					Cacert:   cert,
+					Usercert: cert,
+					Userkey:  key,
+				},
+			},
+			enabledTLS:    true,
+			enabledSASL:   true,
+			salsMechanism: "SCRAM-SHA-512",
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+
+			// Perform The Test
+			config := sarama.NewConfig()
+			UpdateSaramaConfigWithKafkaAuthConfig(config, tc.kafkaAuthCfg)
+
+			saslEnabled := config.Net.SASL.Enable
+			if saslEnabled != tc.enabledSASL {
+				t.Errorf("SASL config is wrong")
+			}
+			if saslEnabled {
+				if tc.salsMechanism != string(config.Net.SASL.Mechanism) {
+					t.Errorf("SASL Mechanism is wrong")
+				}
+			}
+
+			tlsEnabled := config.Net.TLS.Enable
+			if tlsEnabled != tc.enabledTLS {
+				t.Errorf("TLS config is wrong")
+			}
+			if tlsEnabled {
+				if config.Net.TLS.Config == nil {
+					t.Errorf("TLS config is wrong")
+				}
+			}
+		})
+	}
+}
+
 func TestNewTLSConfig(t *testing.T) {
 	cert, key := generateCert(t)
 
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 432a4899b..ba2a54491 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -364,6 +364,7 @@ function create_auth_secrets() {
   kubectl create secret --namespace knative-eventing generic strimzi-sasl-secret \
     --from-literal=ca.crt="$STRIMZI_CRT" \
     --from-literal=password="$SASL_PASSWD" \
+    --from-literal=saslType="SCRAM-SHA-512" \
     --from-literal=user="my-sasl-user"
 }