e2e test for broker topic and trigger consumer id templates being used (

#3769)

Co-authored-by: Marek Schmidt <[email protected]>

test/e2e_new/broker_sasl_ssl_test.go

@@ -99,6 +99,20 @@ func TestBrokerAuthSslSaslScram512(t *testing.T) {
 	env.Test(ctx, t, features.SetupBrokerAuthSslSaslScram512(ctx))
 }
 
+func TestRestrictedBrokerAuthSslSaslScram512(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, features.SetupBrokerAuthRestrictedSslSaslScram512(ctx))
+}
+
 func TestBrokerNotReadyWithoutAuthSecret(t *testing.T) {
 	t.Parallel()
 

test/kafka/kafka_setup.sh

@@ -33,9 +33,12 @@ kubectl -n kafka apply -f $(dirname $0)/kafka-ephemeral.yaml || kubectl -n kafka
 echo "Create TLS user"
 kubectl apply -n kafka -f $(dirname $0)/user-tls.yaml
 
-echo "Create SASL SCRUM 512 user"
+echo "Create SASL SCRAM 512 user"
 kubectl apply -n kafka -f $(dirname $0)/user-sasl-scram-512.yaml
 
+echo "Create SASL SCRAM 512 user with limited privileges"
+kubectl apply -n kafka -f $(dirname $0)/user-sasl-scram-512-restricted.yaml
+
 # Waits until the given object exists.
 # Parameters: $1 - the kind of the object.
 #             $2 - object's name.

test/kafka/user-sasl-scram-512-restricted.yaml

@@ -0,0 +1,46 @@
+# Copyright 2020 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaUser
+metadata:
+  name: my-restricted-sasl-user
+  labels:
+    strimzi.io/cluster: my-cluster
+spec:
+  authentication:
+    type: scram-sha-512
+  authorization:
+    type: simple
+    acls:
+      # Example ACL rules for Broker with names following knative default brokers.topic.template
+      - resource:
+          type: topic
+          name: knative-broker-
+          patternType: prefix
+        operations:
+          - Create
+          - Describe
+          - Read
+          - Write
+          - Delete
+        host: "*"
+      # Example ACL rules for Consumer Group ID following knative default triggers.consumergroup.template
+      - resource:
+          type: group
+          name: knative-trigger-
+          patternType: prefix
+        operations:
+          - Read
+        host: "*"

test/pkg/run.go

@@ -35,10 +35,11 @@ const (
 	NumPartitions     = 10
 	ReplicationFactor = 3
 
-	KafkaClusterNamespace = "kafka"
-	TlsUserSecretName     = "my-tls-user"
-	SaslUserSecretName    = "my-sasl-user"
-	CaSecretName          = "my-cluster-cluster-ca-cert"
+	KafkaClusterNamespace        = "kafka"
+	TlsUserSecretName            = "my-tls-user"
+	SaslUserSecretName           = "my-sasl-user"
+	SaslRestrictedUserSecretName = "my-restricted-sasl-user"
+	CaSecretName                 = "my-cluster-cluster-ca-cert"
 )
 
 var (

test/rekt/features/broker_auth.go

@@ -64,6 +64,11 @@ func SetupBrokerAuthSslSaslScram512(ctx context.Context) *feature.Feature {
 		kafkaauthsecret.WithSslSaslScram512Data(ctx))
 }
 
+func SetupBrokerAuthRestrictedSslSaslScram512(ctx context.Context) *feature.Feature {
+	return SetupBrokerAuth(testpkg.BootstrapServersSslSaslScram,
+		kafkaauthsecret.WithRestrictedSslSaslScram512Data(ctx))
+}
+
 func SetupBrokerAuth(bootstrapServer string, authSecretOptions ...manifest.CfgFn) *feature.Feature {
 	f := feature.NewFeatureNamed("Broker with Kafka Auth Secret")
 

test/rekt/resources/kafkaauthsecret/kafkaauthsecret.go

@@ -103,3 +103,23 @@ func WithSslSaslScram512Data(ctx context.Context) manifest.CfgFn {
 		"password":       saslUserSecret.Data["password"],
 	})
 }
+
+func WithRestrictedSslSaslScram512Data(ctx context.Context) manifest.CfgFn {
+	caSecret, err := client.Get(ctx).CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(ctx, pkg.CaSecretName, metav1.GetOptions{})
+	if err != nil {
+		panic(err)
+	}
+
+	saslRestrictedUserSecret, err := client.Get(ctx).CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(ctx, pkg.SaslRestrictedUserSecretName, metav1.GetOptions{})
+	if err != nil {
+		panic(err)
+	}
+
+	return secret.WithData(map[string][]byte{
+		"protocol":       []byte("SASL_SSL"),
+		"sasl.mechanism": []byte("SCRAM-SHA-512"),
+		"ca.crt":         caSecret.Data["ca.crt"],
+		"user":           []byte(pkg.SaslRestrictedUserSecretName),
+		"password":       saslRestrictedUserSecret.Data["password"],
+	})
+}