[e2e] SASL/TLS with implicit CA Kafka Broker test

test/e2e-common.sh

@@ -531,3 +531,63 @@ function setup_kafka_channel_auth() {
 function install_eventing_core_test_tls_resources() {
     ko apply -Rf "${repo_root_dir}/vendor/knative.dev/eventing/test/config/tls" || return $?
 }
+
+function mount_knative_eventing_bundle() {
+  echo "Mounting knative-eventing-bundle ConfigMap to data-plane components"
+
+  for receiver in kafka-broker-receiver kafka-channel-receiver kafka-sink-receiver
+  do
+    kubectl get deployment -n knative-eventing "$receiver" -o json | \
+      jq '.spec.template.spec.volumes |= . + [{"name":"knative-eventing-bundle","configMap":{"defaultMode":420,"name":"knative-eventing-bundle"}}]' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$receiver"'") | .volumeMounts)  |= . + [{"mountPath":"/etc/knative-eventing-bundle/knative-eventing-bundle.jks","name":"knative-eventing-bundle","readOnly":true,"subPath":"knative-eventing-bundle.jks"}]' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$receiver"'") | .env[] | select(.name=="JAVA_TOOL_OPTIONS") | .value) |= . + " -Djavax.net.ssl.trustStore=/etc/knative-eventing-bundle/knative-eventing-bundle.jks"' | \
+      kubectl apply -f -
+  done
+
+  for dispatcher in kafka-broker-dispatcher kafka-channel-dispatcher kafka-source-dispatcher
+  do
+    kubectl get statefulset -n knative-eventing "$dispatcher" -o json | \
+      jq '.spec.template.spec.volumes |= . + [{"name":"knative-eventing-bundle","configMap":{"defaultMode":420,"name":"knative-eventing-bundle"}}]' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$dispatcher"'") | .volumeMounts)  |= . + [{"mountPath":"/etc/knative-eventing-bundle/knative-eventing-bundle.jks","name":"knative-eventing-bundle","readOnly":true,"subPath":"knative-eventing-bundle.jks"}]' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$dispatcher"'") | .env[] | select(.name=="JAVA_TOOL_OPTIONS") | .value) |= . + " -Djavax.net.ssl.trustStore=/etc/knative-eventing-bundle/knative-eventing-bundle.jks"' | \
+      kubectl apply -f -
+  done
+
+  echo "Mounting knative-eventing-bundle ConfigMap to kafka-controller"
+
+  kubectl get deployment -n knative-eventing kafka-controller -o json | \
+    jq '.spec.template.spec.volumes |= . + [{"name":"knative-eventing-bundle","configMap":{"defaultMode":420,"name":"knative-eventing-bundle"}}]' | \
+    jq '(.spec.template.spec.containers[] | select(.name=="controller") | .volumeMounts) |= . + [{"mountPath":"/etc/knative-eventing-bundle/knative-eventing-bundle.pem","name":"knative-eventing-bundle","readOnly":true,"subPath":"knative-eventing-bundle.pem"}]' | \
+    jq '(.spec.template.spec.containers[] | select(.name=="controller") | .env) |= . + [{"name":"SSL_CERT_DIR","value":"/etc/knative-eventing-bundle:/etc/ssl/certs"}]' | \
+    kubectl apply -f -
+}
+
+function unmount_knative_eventing_bundle() {
+  echo "Unmounting knative-eventing-bundle ConfigMap from data-plane components"
+
+  for receiver in kafka-broker-receiver kafka-channel-receiver kafka-sink-receiver
+  do
+    kubectl get deployment -n knative-eventing "$receiver" -o json | \
+      jq '(.spec.template.spec.volumes[] | select(.name=="knative-eventing-bundle")) |= empty' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$receiver"'") | .volumeMounts[] | select(.name=="knative-eventing-bundle")) |= empty' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$receiver"'") | .env[] | select(.name=="JAVA_TOOL_OPTIONS") | .value) |= (. | sub(" -Djavax.net.ssl.trustStore=/etc/knative-eventing-bundle/knative-eventing-bundle.jks";""))' | \
+      kubectl apply -f -
+  done
+
+  for dispatcher in kafka-broker-dispatcher kafka-channel-dispatcher kafka-source-dispatcher
+  do
+    kubectl get statefulset -n knative-eventing "$dispatcher" -o json | \
+      jq '(.spec.template.spec.volumes[] | select(.name=="knative-eventing-bundle")) |= empty' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$dispatcher"'") | .volumeMounts[] | select(.name=="knative-eventing-bundle")) |= empty' | \
+      jq '(.spec.template.spec.containers[] | select(.name=="'"$dispatcher"'") | .env[] | select(.name=="JAVA_TOOL_OPTIONS") | .value) |= (. | sub(" -Djavax.net.ssl.trustStore=/etc/knative-eventing-bundle/knative-eventing-bundle.jks";""))' | \
+      kubectl apply -f -
+  done
+
+  echo "Unmounting knative-eventing-bundle ConfigMap from kafka-controller"
+
+  kubectl get deployment -n knative-eventing kafka-controller -o json | \
+    jq '(.spec.template.spec.volumes[] | select(.name=="knative-eventing-bundle")) |= empty' | \
+    jq '(.spec.template.spec.containers[] | select(.name=="controller") | .volumeMounts[] | select(.name=="knative-eventing-bundle")) |= empty' | \
+    jq '(.spec.template.spec.containers[] | select(.name=="controller") | .env[] | select(.name=="SSL_CERT_DIR")) |= empty' | \
+    kubectl apply -f -
+}

test/e2e_new/broker_sasl_ssl_implicit_ca_test.go

@@ -0,0 +1,44 @@
+//go:build e2e && implicitca
+// +build e2e,implicitca
+
+/*
+ * Copyright 2024 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.
+ */
+
+package e2e_new
+
+import (
+	"testing"
+
+	"knative.dev/eventing-kafka-broker/test/rekt/features"
+	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/k8s"
+	"knative.dev/reconciler-test/pkg/knative"
+)
+
+func TestBrokerAuthImplicitCASslSaslScram512(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.SetupBrokerAuthImplicitCASslSaslScram512(ctx))
+}

test/kafka/kafka-certificate.yaml

@@ -0,0 +1,30 @@
+# Copyright 2024 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: cert-manager.io/v1
+kind: Certificate
+metadata:
+  name: my-cluster-kafka
+spec:
+  secretName: my-cluster-kafka-certmanager-cert
+  dnsNames:
+    - my-cluster-kafka-bootstrap.kafka
+    - my-cluster-kafka-bootstrap.kafka.svc
+    - my-cluster-kafka-0.my-cluster-kafka-brokers.kafka.svc
+    - my-cluster-kafka-1.my-cluster-kafka-brokers.kafka.svc
+    - my-cluster-kafka-2.my-cluster-kafka-brokers.kafka.svc
+  issuerRef:
+    name: knative-eventing-ca-issuer
+    kind: ClusterIssuer
+    group: cert-manager.io

test/kafka/kafka-ephemeral.yaml

@@ -41,6 +41,20 @@ spec:
         tls: true
         authentication:
           type: scram-sha-512
+      # protocol=SASL_SSL
+      # sasl.mechanism=SCRAM-SHA-512
+      # TLS Certificate signed by knative-eventing-ca-issuer ClusterIssuer
+      - name: cmsasltls
+        port: 9194
+        type: internal
+        tls: true
+        authentication:
+          type: scram-sha-512
+        configuration:
+          brokerCertChainAndKey:
+            certificate: tls.crt
+            key: tls.key
+            secretName: my-cluster-kafka-certmanager-cert
       # protocol=SASL_PLAINTEXT
       # sasl.mechanism=SCRAM-SHA-512
       - name: saslplain

test/kafka/kafka_setup.sh

@@ -26,6 +26,9 @@ kubectl create namespace kafka --dry-run=client -o yaml | kubectl apply -f -
 header "Applying Strimzi Cluster Operator file"
 cat $(dirname $0)/strimzi-cluster-operator.yaml | sed 's/namespace: .*/namespace: kafka/' | sed "s/cluster.local/${CLUSTER_SUFFIX}/g" | kubectl apply -n kafka -f -
 
+echo "Create Kafka Certificate"
+kubectl -n kafka apply -f $(dirname $0)/kafka-certificate.yaml
+
 sleep 10
 
 kubectl -n kafka apply -f $(dirname $0)/kafka-ephemeral.yaml || kubectl -n kafka apply -f $(dirname $0)/kafka-ephemeral.yaml

test/pkg/run.go

@@ -26,11 +26,12 @@ const (
 	rerunTimes = 5
 
 	// Kafka bootstrap server.
-	BootstrapServersPlaintext     = "my-cluster-kafka-bootstrap.kafka:9092"
-	BootstrapServersSsl           = "my-cluster-kafka-bootstrap.kafka:9093"
-	BootstrapServersTlsNoAuth     = "my-cluster-kafka-bootstrap.kafka:9096"
-	BootstrapServersSaslPlaintext = "my-cluster-kafka-bootstrap.kafka:9095"
-	BootstrapServersSslSaslScram  = "my-cluster-kafka-bootstrap.kafka:9094"
+	BootstrapServersPlaintext              = "my-cluster-kafka-bootstrap.kafka:9092"
+	BootstrapServersSsl                    = "my-cluster-kafka-bootstrap.kafka:9093"
+	BootstrapServersTlsNoAuth              = "my-cluster-kafka-bootstrap.kafka:9096"
+	BootstrapServersSaslPlaintext          = "my-cluster-kafka-bootstrap.kafka:9095"
+	BootstrapServersSslSaslScram           = "my-cluster-kafka-bootstrap.kafka:9094"
+	BootstrapServersImplicitCASslSaslScram = "my-cluster-kafka-bootstrap.kafka:9194"
 
 	NumPartitions     = 10
 	ReplicationFactor = 3

test/reconciler-tests.sh

@@ -50,6 +50,14 @@ go_test_e2e -tags=e2e,cloudevents -timeout=1h ./test/e2e_new_channel/... || fail
 
 go_test_e2e -tags=deletecm ./test/e2e_new/... || fail_test "E2E (new deletecm) suite failed"
 
+echo "Running 'Implicit CA' E2E Reconciler tests with knative-eventing-bundle mounted"
+
+mount_knative_eventing_bundle
+
+go_test_e2e -tags=e2e,implicitca -timeout=5m ./test/e2e_new -run ImplicitCA || fail_test "E2E (new - implicitca) suite failed"
+
+unmount_knative_eventing_bundle
+
 echo "Running E2E Reconciler tests with consumergroup id template changed"
 
 kubectl apply -f "$(dirname "$0")/config-kafka-features/new-cg-id.yaml"

test/rekt/features/broker_auth.go

@@ -75,6 +75,11 @@ func SetupBrokerAuthSslSaslScram512(ctx context.Context) *feature.Feature {
 		kafkaauthsecret.WithSslSaslScram512Data(ctx))
 }
 
+func SetupBrokerAuthImplicitCASslSaslScram512(ctx context.Context) *feature.Feature {
+	return SetupBrokerAuth(testpkg.BootstrapServersImplicitCASslSaslScram,
+		kafkaauthsecret.WithImplicitCASslSaslScram512Data(ctx))
+}
+
 func SetupBrokerAuthRestrictedSslSaslScram512(ctx context.Context) *feature.Feature {
 	return SetupBrokerAuth(testpkg.BootstrapServersSslSaslScram,
 		kafkaauthsecret.WithRestrictedSslSaslScram512Data(ctx))

test/rekt/resources/kafkaauthsecret/kafkaauthsecret.go

@@ -104,6 +104,21 @@ func WithSslSaslScram512Data(ctx context.Context) manifest.CfgFn {
 	})
 }
 
+// WithImplicitCASslSaslScram512Data same as WithSslSaslScram512Data but using an implicit CA configured in the control-plane and data-plane containers
+func WithImplicitCASslSaslScram512Data(ctx context.Context) manifest.CfgFn {
+	saslUserSecret, err := client.Get(ctx).CoreV1().Secrets(pkg.KafkaClusterNamespace).Get(ctx, pkg.SaslUserSecretName, metav1.GetOptions{})
+	if err != nil {
+		panic(err)
+	}
+
+	return secret.WithData(map[string][]byte{
+		"protocol":       []byte("SASL_SSL"),
+		"sasl.mechanism": []byte("SCRAM-SHA-512"),
+		"user":           []byte(pkg.SaslUserSecretName),
+		"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 {