Eventing TLS: Test ContainerSource with eventshub TLS receiver as sink (

#6957)

Fixes #6937 

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

<!-- Please categorize your changes:
- 🎁 Add new feature
- 🐛 Fix bug
- 🧹 Update or clean up current behavior
- 🗑️ Remove feature or internal logic
-->

- Test ContainerSource with eventshub TLS receiver as sink


### Pre-review Checklist

<!-- If these boxes are not checked, you will be asked to complete these
requirements or explain why they do not apply to your PR. -->

- [ ] **At least 80% unit test coverage**
- [ ] **E2E tests** for any new behavior
- [ ] **Docs PR** for any user-facing impact
- [ ] **Spec PR** for any new API feature
- [ ] **Conformance test** for any change to the spec

**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
ContainerSource supports sending events to TLS endpoints, minimum TLS version is v1.2
```


**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->

---------

Signed-off-by: Vishal Choudhary <sendtovishalchoudhary@gmail.com>
Signed-off-by: Christoph Stäbler <cstabler@redhat.com>
Signed-off-by: Knative Automation <automation@knative.team>
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>
Co-authored-by: Christoph Stäbler <cstabler@redhat.com>
Co-authored-by: knative-automation <automation@knative.team>
Co-authored-by: Matthias Wessendorf <mwessend@redhat.com>

config/core/resources/containersource.yaml

@@ -69,6 +69,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
                 # WARNING: the schema tool can not parse PodTemplateSpec, stub here and redirect to Deployment documentation.
                 template:
                   type: object
@@ -127,6 +130,9 @@ spec:
                 sinkUri:
                   description: SinkURI is the current active sink URI that has been configured for the Source.
                   type: string
+                sinkCACerts:
+                  description: CACerts is the Certification Authority (CA) certificates in PEM format that the source trusts when sending events to the sink.
+                  type: string
       additionalPrinterColumns:
         - name: Sink
           type: string

config/core/resources/sinkbindings.yaml

@@ -75,6 +75,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
                 subject:
                   description: Subject references the resource(s) whose "runtime contract" should be augmented by Binding implementations.
                   type: object
@@ -169,6 +172,9 @@ spec:
                 sinkUri:
                   description: SinkURI is the current active sink URI that has been configured for the Source.
                   type: string
+                sinkCACerts:
+                  description: CACerts is the Certification Authority (CA) certificates in PEM format that the source trusts when sending events to the sink.
+                  type: string
       additionalPrinterColumns:
         - name: Sink
           type: string

test/rekt/container_source_test.go

@@ -25,6 +25,7 @@ import (
 	"knative.dev/eventing/test/rekt/features/containersource"
 	"knative.dev/pkg/system"
 	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 )
@@ -85,3 +86,18 @@ func TestContainerSourceWithArgs(t *testing.T) {
 
 	env.Test(ctx, t, containersource.SendsEventsWithArgs())
 }
+
+func TestContainerSourceWithTLS(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, containersource.SendEventsWithTLSRecieverAsSink())
+}

test/rekt/features/channel/features.go

@@ -39,7 +39,6 @@ import (
 	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/eventing/test/rekt/resources/containersource"
 	"knative.dev/eventing/test/rekt/resources/delivery"
-	"knative.dev/eventing/test/rekt/resources/source"
 	"knative.dev/eventing/test/rekt/resources/subscription"
 )
 
@@ -78,7 +77,7 @@ func ChannelChain(length int, createSubscriberFn func(ref *duckv1.KReference, ur
 	}
 
 	// attach the first channel to the source
-	f.Requirement("install containersource", containersource.Install(cs, containersource.WithSink(channel_impl.AsRef(channels[0]), "")))
+	f.Requirement("install containersource", containersource.Install(cs, containersource.WithSink(channel_impl.AsDestinationRef(channels[0]))))
 	f.Requirement("containersource goes ready", containersource.IsReady(cs))
 
 	f.Assert("chained channels relay events", assert.OnStore(sink).MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).AtLeast(1))
@@ -104,7 +103,7 @@ func DeadLetterSink(createSubscriberFn func(ref *duckv1.KReference, uri string)
 	f.Setup("channel is ready", channel_impl.IsReady(name))
 	f.Setup("subscription is ready", subscription.IsReady(sub))
 
-	f.Requirement("install containersource", containersource.Install(cs, source.WithSink(channel_impl.AsRef(name), "")))
+	f.Requirement("install containersource", containersource.Install(cs, containersource.WithSink(channel_impl.AsDestinationRef(name))))
 	f.Requirement("containersource is ready", containersource.IsReady(cs))
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(name, channel_impl.GVR()))
 
@@ -137,7 +136,7 @@ func DeadLetterSinkGenericChannel(createSubscriberFn func(ref *duckv1.KReference
 	f.Setup("channel is ready", channel.IsReady(name))
 	f.Setup("subscription is ready", subscription.IsReady(sub))
 
-	f.Requirement("install containersource", containersource.Install(cs, source.WithSink(channel.AsRef(name), "")))
+	f.Requirement("install containersource", containersource.Install(cs, containersource.WithSink(channel_impl.AsDestinationRef(name))))
 	f.Requirement("containersource is ready", containersource.IsReady(cs))
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(name, channel.GVR()))
 
@@ -160,7 +159,7 @@ func AsDeadLetterSink(createSubscriberFn func(ref *duckv1.KReference, uri string
 	failer := feature.MakeRandomK8sName("failer")
 	sink := feature.MakeRandomK8sName("sink")
 
-	f.Setup("install containersource", containersource.Install(cs, source.WithSink(channel.AsRef(name), "")))
+	f.Setup("install containersource", containersource.Install(cs, containersource.WithSink(channel_impl.AsDestinationRef(name))))
 
 	f.Setup("install channel", channel.Install(name,
 		channel.WithTemplate(),

test/rekt/features/containersource/features.go

@@ -17,6 +17,7 @@ limitations under the License.
 package containersource
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/cloudevents/sdk-go/v2/test"
@@ -29,7 +30,6 @@ import (
 	"knative.dev/reconciler-test/pkg/resources/service"
 
 	"knative.dev/eventing/test/rekt/resources/containersource"
-	"knative.dev/eventing/test/rekt/resources/pingsource"
 )
 
 func SendsEventsWithSinkRef() *feature.Feature {
@@ -39,7 +39,7 @@ func SendsEventsWithSinkRef() *feature.Feature {
 
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 
-	f.Requirement("install containersource", containersource.Install(source, pingsource.WithSink(service.AsKReference(sink), "")))
+	f.Requirement("install containersource", containersource.Install(source, containersource.WithSink(service.AsDestinationRef(sink))))
 	f.Requirement("containersource goes ready", containersource.IsReady(source))
 
 	f.Stable("containersource as event source").
@@ -56,7 +56,7 @@ func SendsEventsWithSinkURI() *feature.Feature {
 
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 
-	f.Requirement("install containersource", containersource.Install(source, containersource.WithSink(service.AsKReference(sink), "")))
+	f.Requirement("install containersource", containersource.Install(source, containersource.WithSink(service.AsDestinationRef(sink))))
 	f.Requirement("containersource goes ready", containersource.IsReady(source))
 
 	f.Stable("containersource as event source").
@@ -77,7 +77,7 @@ func SendsEventsWithCloudEventOverrides() *feature.Feature {
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 
 	f.Requirement("install containersource", containersource.Install(source,
-		pingsource.WithSink(service.AsKReference(sink), ""),
+		containersource.WithSink(service.AsDestinationRef(sink)),
 		containersource.WithExtensions(extensions),
 		manifest.WithPodAnnotations(map[string]interface{}{
 			"foo": true,
@@ -105,7 +105,7 @@ func SendsEventsWithArgs() *feature.Feature {
 	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiver))
 
 	f.Requirement("install containersource", containersource.Install(source,
-		containersource.WithSink(service.AsKReference(sink), ""),
+		containersource.WithSink(service.AsDestinationRef(sink)),
 		containersource.WithArgs(args),
 	))
 	f.Requirement("containersource goes ready", containersource.IsReady(source))
@@ -118,3 +118,29 @@ func SendsEventsWithArgs() *feature.Feature {
 
 	return f
 }
+
+func SendEventsWithTLSRecieverAsSink() *feature.Feature {
+	source := feature.MakeRandomK8sName("containersource")
+	sink := feature.MakeRandomK8sName("sink")
+	f := feature.NewFeature()
+
+	f.Setup("install sink", eventshub.Install(sink, eventshub.StartReceiverTLS))
+
+	f.Requirement("install ContainerSource", func(ctx context.Context, t feature.T) {
+		d := service.AsDestinationRef(sink)
+		d.CACerts = eventshub.GetCaCerts(ctx)
+
+		containersource.Install(source, containersource.WithSink(d))(ctx, t)
+	})
+	f.Requirement("containersource goes ready", containersource.IsReady(source))
+
+	f.Stable("containersource as event source").
+		Must("delivers events",
+			assert.OnStore(sink).
+				Match(assert.MatchKind(eventshub.EventReceived)).
+				MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).
+				AtLeast(1),
+		)
+
+	return f
+}

test/rekt/features/containersource/readyness.go

@@ -18,8 +18,6 @@ package containersource
 
 import (
 	"knative.dev/eventing/test/rekt/resources/containersource"
-	"knative.dev/eventing/test/rekt/resources/pingsource"
-	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"
@@ -33,11 +31,7 @@ func GoesReady(name string, cfg ...manifest.CfgFn) *feature.Feature {
 	f.Setup("install a service", service.Install(sink,
 		service.WithSelectors(map[string]string{"app": "rekt"})))
 
-	cfg = append(cfg, pingsource.WithSink(&duckv1.KReference{
-		Kind:       "Service",
-		Name:       sink,
-		APIVersion: "v1",
-	}, ""))
+	cfg = append(cfg, containersource.WithSink(service.AsDestinationRef(sink)))
 
 	f.Setup("install a ContainerSource", containersource.Install(name, cfg...))
 

test/rekt/resources/channel_impl/channel_impl.go

@@ -159,5 +159,17 @@ func AsRef(name string) *duckv1.KReference {
 	}
 }
 
+// AsRef returns a KRef for a Channel without namespace.
+func AsDestinationRef(name string) *duckv1.Destination {
+	apiVersion, kind := GVK().ToAPIVersionAndKind()
+	return &duckv1.Destination{
+		Ref: &duckv1.KReference{
+			Kind:       kind,
+			APIVersion: apiVersion,
+			Name:       name,
+		},
+	}
+}
+
 // WithDeadLetterSink adds the dead letter sink related config to a Subscription spec.
 var WithDeadLetterSink = delivery.WithDeadLetterSink

test/rekt/resources/containersource/containersource.go

@@ -19,12 +19,12 @@ package containersource
 import (
 	"context"
 	"embed"
+	"strings"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
-	"knative.dev/eventing/test/rekt/resources/source"
-
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
@@ -67,7 +67,37 @@ func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 }
 
 // WithSink adds the sink related config to a ContainerSource spec.
-var WithSink = source.WithSink
+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{})
+
+		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 {
+				sink["ref"] = map[string]interface{}{}
+			}
+			sref := sink["ref"].(map[string]interface{})
+			sref["apiVersion"] = ref.APIVersion
+			sref["kind"] = ref.Kind
+			sref["namespace"] = ref.Namespace
+			sref["name"] = ref.Name
+		}
+	}
+}
 
 // WithExtensions adds the ceOverrides related config to a ContainerSource spec.
 func WithExtensions(extensions map[string]interface{}) manifest.CfgFn {

test/rekt/resources/containersource/containersource.yaml

@@ -31,13 +31,17 @@ spec:
     {{ if .sink.ref }}
     ref:
       kind: {{ .sink.ref.kind }}
-      namespace: {{ .namespace }}
+      namespace: {{ .sink.ref.namespace }}
       name: {{ .sink.ref.name }}
       apiVersion: {{ .sink.ref.apiVersion }}
     {{ end }}
     {{ if .sink.uri }}
     uri: {{ .sink.uri }}
     {{ end }}
+    {{ if .sink.CACerts }}
+    CACerts: |-
+      {{ .sink.CACerts }}
+    {{ end }}
   template:
     {{ if .podannotations }}
     metadata: