Eventing TLS: Test ApiServerSource with eventshub TLS receiver as sink

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

DEVELOPMENT.md

@@ -161,6 +161,17 @@ ko apply -f config/brokers/mt-channel-broker/
 Depending on your needs you might want to install other
 [Broker implementations](https://github.com/knative/eventing/tree/main/docs/broker).
 
+## Install Cert-Manager
+
+Install the Cert-manager operator to run e2e tests for TLS
+
+```shell
+kubectl apply -f third_party/cert-manager
+```
+
+Depending on your needs you might want to install other
+[Broker implementations](https://github.com/knative/eventing/tree/main/docs/broker).
+
 ## Enable Sugar controller
 
 If you are running e2e tests that leverage the Sugar Controller, you will need

config/core/resources/apiserversource.yaml

@@ -136,6 +136,9 @@ spec:
                   uri:
                     description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                     type: string
+                  CACerts:
+                    description: CACerts is the Certification Authority (CA) certificates in PEM format that the source trusts when sending events to the sink.
+                    type: string
               namespaceSelector:
                 description: NamespaceSelector is a label selector to capture the namespaces that should be watched by the source.
                 type: object

test/rekt/apiserversource_test.go

@@ -24,13 +24,15 @@ import (
 	"time"
 
 	"knative.dev/pkg/system"
+	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
-	apiserversourcefeatures "knative.dev/eventing/test/rekt/features/apiserversource"
 	_ "knative.dev/pkg/system/testing"
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
+
+	apiserversourcefeatures "knative.dev/eventing/test/rekt/features/apiserversource"
 )
 
 // TestApiServerSourceValidationWebhookConfigurationOnCreate tests if the webhook
@@ -79,6 +81,21 @@ func TestApiServerSourceDataPlane_SinkTypes(t *testing.T) {
 	env.TestSet(ctx, t, apiserversourcefeatures.DataPlane_SinkTypes())
 }
 
+func TestApiServerSourceDataPlaneTLS(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		//environment.Managed(t),
+		eventshub.WithTLS(t),
+	)
+
+	env.Test(ctx, t, apiserversourcefeatures.SendsEventsWithTLS())
+}
+
 func TestApiServerSourceDataPlane_EventModes(t *testing.T) {
 	t.Parallel()
 

test/rekt/features/apiserversource/data_plane.go

@@ -21,17 +21,21 @@ import (
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/v2/test"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/reconciler-test/pkg/eventshub"
-	eventasssert "knative.dev/reconciler-test/pkg/eventshub/assert"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/pkg/resources/service"
 
+	eventasssert "knative.dev/reconciler-test/pkg/eventshub/assert"
+
+	"knative.dev/reconciler-test/pkg/resources/pod"
+
 	"knative.dev/eventing/pkg/apis/sources"
 	v1 "knative.dev/eventing/pkg/apis/sources/v1"
 	"knative.dev/eventing/test/rekt/resources/account_role"
@@ -41,7 +45,6 @@ import (
 	"knative.dev/eventing/test/rekt/resources/namespace"
 	"knative.dev/eventing/test/rekt/resources/pingsource"
 	"knative.dev/eventing/test/rekt/resources/trigger"
-	"knative.dev/reconciler-test/pkg/resources/pod"
 )
 
 const (
@@ -110,7 +113,7 @@ func SendsEventsWithSinkRef() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode(v1.ResourceMode),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion: "v1",
 			Kind:       "Event",
@@ -147,7 +150,7 @@ func SendsEventsWithSinkUri() *feature.Feature {
 		cfg := []manifest.CfgFn{
 			apiserversource.WithServiceAccountName(sacmName),
 			apiserversource.WithEventMode(v1.ResourceMode),
-			apiserversource.WithSink(nil, sinkuri.String()),
+			apiserversource.WithSink(&duckv1.Destination{URI: sinkuri}),
 			apiserversource.WithResources(v1.APIVersionKindSelector{
 				APIVersion: "v1",
 				Kind:       "Event",
@@ -165,6 +168,47 @@ func SendsEventsWithSinkUri() *feature.Feature {
 	return f
 }
 
+func SendsEventsWithTLS() *feature.Feature {
+	source := feature.MakeRandomK8sName("apiserversource")
+	sink := feature.MakeRandomK8sName("sink")
+
+	f := feature.NewFeatureNamed("Send events to TLS sink")
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiverTLS))
+
+	sacmName := feature.MakeRandomK8sName("apiserversource")
+	f.Requirement("Create Service Account for ApiServerSource with RBAC for v1.Event resources",
+		setupAccountAndRoleForPods(sacmName))
+
+	cfg := []manifest.CfgFn{
+		apiserversource.WithServiceAccountName(sacmName),
+		apiserversource.WithEventMode(v1.ResourceMode),
+		apiserversource.WithResources(v1.APIVersionKindSelector{
+			APIVersion: "v1",
+			Kind:       "Event",
+		}),
+	}
+
+	f.Requirement("install ApiServerSource", func(ctx context.Context, t feature.T) {
+		d := service.AsDestinationRef(sink)
+		d.CACerts = eventshub.GetCaCerts(ctx)
+
+		cfg = append(cfg, apiserversource.WithSink(d))
+		apiserversource.Install(source, cfg...)(ctx, t)
+	})
+	f.Requirement("ApiServerSource goes ready", apiserversource.IsReady(source))
+
+	f.Stable("ApiServerSource as event source").
+		Must("delivers events on sink with ref",
+			eventasssert.OnStore(sink).
+				Match(eventasssert.MatchKind(eventshub.EventReceived)).
+				MatchEvent(test.HasType("dev.knative.apiserver.resource.update")).
+				AtLeast(1),
+		)
+
+	return f
+}
+
 // SendsEventsWithEventTypes tests apiserversource to a ready broker.
 func SendsEventsWithEventTypes() *feature.Feature {
 	source := feature.MakeRandomK8sName("source")
@@ -194,7 +238,7 @@ func SendsEventsWithEventTypes() *feature.Feature {
 		cfg := []manifest.CfgFn{
 			apiserversource.WithServiceAccountName(sacmName),
 			apiserversource.WithEventMode(v1.ResourceMode),
-			apiserversource.WithSink(nil, brokeruri.String()),
+			apiserversource.WithSink(&duckv1.Destination{URI: brokeruri}),
 			apiserversource.WithResources(v1.APIVersionKindSelector{
 				APIVersion: "v1",
 				Kind:       "Event",
@@ -229,7 +273,7 @@ func SendsEventsWithObjectReferencePayload() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode(v1.ReferenceMode),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion: "v1",
 			Kind:       "Pod",
@@ -272,7 +316,7 @@ func SendsEventsWithResourceEventPayload() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode(v1.ResourceMode),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion: "v1",
 			Kind:       "Pod",
@@ -315,7 +359,7 @@ func SendsEventsForAllResources() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode("Reference"),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion: "v1",
 			Kind:       "Pod",
@@ -368,7 +412,7 @@ func SendsEventsForAllResourcesWithNamespaceSelector() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode("Reference"),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion: "v1",
 			Kind:       "Pod",
@@ -441,7 +485,7 @@ func SendsEventsForAllResourcesWithEmptyNamespaceSelector() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode("Reference"),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion: "sources.knative.dev/v1",
 			Kind:       "PingSource",
@@ -499,7 +543,7 @@ func SendsEventsForLabelMatchingResources() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode("Reference"),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion:    "v1",
 			Kind:          "Pod",
@@ -593,7 +637,7 @@ func SendEventsForLabelExpressionMatchingResources() *feature.Feature {
 	cfg := []manifest.CfgFn{
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode("Reference"),
-		apiserversource.WithSink(service.AsKReference(sink), ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion:    "v1",
 			Kind:          "Pod",
@@ -682,7 +726,7 @@ func SendsEventsWithRetries() *feature.Feature {
 		cfg := []manifest.CfgFn{
 			apiserversource.WithServiceAccountName(sacmName),
 			apiserversource.WithEventMode(v1.ReferenceMode),
-			apiserversource.WithSink(nil, sinkuri.String()),
+			apiserversource.WithSink(&duckv1.Destination{URI: sinkuri}),
 			apiserversource.WithResources(v1.APIVersionKindSelector{
 				APIVersion:    "v1",
 				Kind:          "Pod",

test/rekt/features/apiserversource/readiness.go

@@ -18,13 +18,13 @@ package apiserversource
 
 import (
 	rbacv1 "k8s.io/api/rbac/v1"
-	v1 "knative.dev/eventing/pkg/apis/sources/v1"
-	"knative.dev/eventing/test/rekt/resources/account_role"
-	"knative.dev/eventing/test/rekt/resources/apiserversource"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
 	"knative.dev/reconciler-test/pkg/resources/service"
+
+	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+	"knative.dev/eventing/test/rekt/resources/account_role"
+	"knative.dev/eventing/test/rekt/resources/apiserversource"
 )
 
 // GoesReady returns a feature testing if an ApiServerSource becomes ready.
@@ -60,11 +60,7 @@ func Install(name string, cfg ...manifest.CfgFn) *feature.Feature {
 	cfg = append(cfg,
 		apiserversource.WithServiceAccountName(sacmName),
 		apiserversource.WithEventMode(v1.ResourceMode),
-		apiserversource.WithSink(&duckv1.KReference{
-			Kind:       "Service",
-			Name:       sink,
-			APIVersion: "v1",
-		}, ""),
+		apiserversource.WithSink(service.AsDestinationRef(sink)),
 		apiserversource.WithResources(v1.APIVersionKindSelector{
 			APIVersion: "v1",
 			Kind:       "Event",

test/rekt/features/apiserversource/webhook_validation_smoke.go

@@ -20,12 +20,13 @@ import (
 	"context"
 
 	"github.com/stretchr/testify/assert"
+	"knative.dev/reconciler-test/pkg/resources/service"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/reconciler-test/pkg/feature"
+
 	v1 "knative.dev/eventing/pkg/apis/sources/v1"
 	"knative.dev/eventing/test/rekt/resources/apiserversource"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/reconciler-test/pkg/feature"
 )
 
 func CreateWithInvalidSpec() *feature.Feature {
@@ -53,11 +54,7 @@ func UpdateWithInvalidSpec(name string) *feature.Feature {
 func createApiServerSourceWithInvalidSpec(name string) func(ctx context.Context, t feature.T) {
 	return func(ctx context.Context, t feature.T) {
 		_, err := apiserversource.InstallLocalYaml(ctx, name,
-			apiserversource.WithSink(&duckv1.KReference{
-				Kind:       "Service",
-				Name:       "foo-svc",
-				APIVersion: "v1",
-			}, ""),
+			apiserversource.WithSink(service.AsDestinationRef("foo-svc")),
 			apiserversource.WithResources(v1.APIVersionKindSelector{
 				APIVersion: "v1",
 				Kind:       "Event",

test/rekt/resources/apiserversource/apiserversource.go

@@ -19,15 +19,17 @@ package apiserversource
 import (
 	"context"
 	"embed"
+	"strings"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	v1 "knative.dev/eventing/pkg/apis/sources/v1"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 
+	v1 "knative.dev/eventing/pkg/apis/sources/v1"
+
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/manifest"
 )
@@ -79,15 +81,24 @@ func WithEventMode(eventMode string) manifest.CfgFn {
 }
 
 // WithSink adds the sink related config to a ApiServerSource spec.
-func WithSink(ref *duckv1.KReference, uri string) manifest.CfgFn {
+func WithSink(d *duckv1.Destination) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, set := cfg["sink"]; !set {
 			cfg["sink"] = map[string]interface{}{}
 		}
 		sink := cfg["sink"].(map[string]interface{})
 
-		if uri != "" {
-			sink["uri"] = uri
+		ref := d.Ref
+		uri := d.URI
+
+		if d.CACerts != nil {
+			// This is a multi-line string and should be indented accordingly.
+			// Replace "new line" with "new line + spaces".
+			sink["CACerts"] = strings.ReplaceAll(*d.CACerts, "\n", "\n      ")
+		}
+
+		if uri != nil {
+			sink["uri"] = uri.String()
 		}
 		if ref != nil {
 			if _, set := sink["ref"]; !set {

test/rekt/resources/apiserversource/apiserversource.yaml

@@ -76,6 +76,10 @@ spec:
       name: {{ .sink.ref.name }}
       apiVersion: {{ .sink.ref.apiVersion }}
     {{ end }}
+    {{ if .sink.CACerts }}
+    CACerts: |-
+      {{ .sink.CACerts }}
+    {{ end }}
     {{ if .sink.uri }}
     uri: {{ .sink.uri }}
     {{ end }}