Adds tests, role binding. Dispatcher sets up TLS config

Author: gab-satchi

config/channels/in-memory-channel/200-imc-dispatcher-serviceaccount.yaml

@@ -36,3 +36,17 @@ roleRef:
   kind: ClusterRole
   name: imc-dispatcher
   apiGroup: rbac.authorization.k8s.io
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: tls-role-binding
+  namespace: knative-eventing
+subjects:
+- kind: ServiceAccount
+  name: imc-dispatcher
+  apiGroup: ""
+roleRef:
+  kind: Role
+  name: tls-role
+  apiGroup: rbac.authorization.k8s.io

config/tls/role.yaml

@@ -0,0 +1,31 @@
+# Copyright 2023 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.
+
+# This is a role to read Secrets within the system namespace for eventing.
+# Role should be used by any component setting up a TLS server
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: tls-role
+  namespace: knative-eventing
+rules:
+- apiGroups: 
+    - ""
+  resources:
+    - secrets
+  verbs:
+    - get
+    - list
+    - watch

pkg/channel/fanout/fanout_message_handler_test.go

@@ -325,6 +325,12 @@ func testFanoutMessageHandler(t *testing.T, async bool, receiverFunc channel.Unb
 		t.Fatal(err)
 	}
 
+	calledChan := make(chan bool, 1)
+	recvOptionFunc := func(*channel.MessageReceiver) error {
+		calledChan <- true
+		return nil
+	}
+
 	h, err := NewFanoutMessageHandler(
 		logger,
 		channel.NewMessageDispatcher(logger),
@@ -333,7 +339,9 @@ func testFanoutMessageHandler(t *testing.T, async bool, receiverFunc channel.Unb
 			AsyncHandler:  async,
 		},
 		reporter,
+		recvOptionFunc,
 	)
+	<-calledChan
 	if err != nil {
 		t.Fatal("NewHandler failed =", err)
 	}

pkg/channel/multichannelfanout/multi_channel_fanout_message_handler.go

@@ -45,12 +45,6 @@ type MultiChannelMessageHandler interface {
 	CountChannelHandlers() int
 }
 
-// makeChannelKeyFromConfig creates the channel key for a given channelConfig. It is a helper around
-// MakeChannelKey.
-func makeChannelKeyFromConfig(config ChannelConfig) string {
-	return config.HostName
-}
-
 // Handler is an http.Handler that introspects the incoming request to determine what Channel it is
 // on, and then delegates handling of that request to the single fanout.FanoutMessageHandler corresponding to
 // that Channel.
@@ -70,21 +64,26 @@ func NewMessageHandler(_ context.Context, logger *zap.Logger, messageDispatcher
 
 // NewMessageHandlerWithConfig creates a new Handler with the specified configuration. This is really meant for tests
 // where you want to apply a fully specified configuration for tests. Reconciler operates on single channel at a time.
-func NewMessageHandlerWithConfig(_ context.Context, logger *zap.Logger, messageDispatcher channel.MessageDispatcher, conf Config, reporter channel.StatsReporter) (*MessageHandler, error) {
+func NewMessageHandlerWithConfig(_ context.Context, logger *zap.Logger, messageDispatcher channel.MessageDispatcher, conf Config, reporter channel.StatsReporter, recvOptions ...channel.MessageReceiverOptions) (*MessageHandler, error) {
 	handlers := make(map[string]fanout.MessageHandler, len(conf.ChannelConfigs))
 
 	for _, cc := range conf.ChannelConfigs {
-		key := makeChannelKeyFromConfig(cc)
-		handler, err := fanout.NewFanoutMessageHandler(logger, messageDispatcher, cc.FanoutConfig, reporter)
-		if err != nil {
-			logger.Error("Failed creating new fanout handler.", zap.Error(err))
-			return nil, err
-		}
-		if _, present := handlers[key]; present {
-			logger.Error("Duplicate channel key", zap.String("channelKey", key))
-			return nil, fmt.Errorf("duplicate channel key: %v", key)
+		keys := []string{cc.HostName, cc.Path}
+		for _, key := range keys {
+			if key == "" {
+				continue
+			}
+			handler, err := fanout.NewFanoutMessageHandler(logger, messageDispatcher, cc.FanoutConfig, reporter, recvOptions...)
+			if err != nil {
+				logger.Error("Failed creating new fanout handler.", zap.Error(err))
+				return nil, err
+			}
+			if _, present := handlers[key]; present {
+				logger.Error("Duplicate channel key", zap.String("channelKey", key))
+				return nil, fmt.Errorf("duplicate channel key: %v", key)
+			}
+			handlers[key] = handler
 		}
-		handlers[key] = handler
 	}
 	return &MessageHandler{
 		logger:   logger,

pkg/channel/multichannelfanout/multi_channel_fanout_message_handler_test.go

@@ -122,19 +122,63 @@ func TestServeHTTPMessageHandler(t *testing.T) {
 		config             Config
 		eventID            *string
 		respStatusCode     int
-		key                string
+		hostKey            string
+		pathKey            string
+		recvOptions        []channel.MessageReceiverOptions
 		expectedStatusCode int
 	}{
-		"non-existent channel": {
+		"non-existent channel host based": {
 			config:             Config{},
-			key:                "default.does-not-exist",
+			hostKey:            "default.does-not-exist",
+			expectedStatusCode: http.StatusInternalServerError,
+		},
+		"non-existent channel path based": {
+			hostKey: "first-channel.default",
+			config: Config{
+				ChannelConfigs: []ChannelConfig{
+					{
+						Namespace: "ns",
+						Name:      "name",
+						HostName:  "first-channel.default",
+						FanoutConfig: fanout.Config{
+							Subscriptions: []fanout.Subscription{
+								{
+									Reply: replaceDomain,
+								},
+							},
+						},
+					},
+				},
+			},
+			pathKey:            "some-namespace/wrong-channel",
 			expectedStatusCode: http.StatusInternalServerError,
 		},
 		"bad host": {
 			config:             Config{},
-			key:                "no-dot",
+			hostKey:            "no-dot",
 			expectedStatusCode: http.StatusInternalServerError,
 		},
+		"malformed path": {
+			hostKey: "first-channel.default",
+			config: Config{
+				ChannelConfigs: []ChannelConfig{
+					{
+						Namespace: "ns",
+						Name:      "name",
+						HostName:  "first-channel.default",
+						FanoutConfig: fanout.Config{
+							Subscriptions: []fanout.Subscription{
+								{
+									Reply: replaceDomain,
+								},
+							},
+						},
+					},
+				},
+			},
+			pathKey:            "missing-forward-slash",
+			expectedStatusCode: http.StatusBadRequest,
+		},
 		"pass through failure": {
 			config: Config{
 				ChannelConfigs: []ChannelConfig{
@@ -153,7 +197,7 @@ func TestServeHTTPMessageHandler(t *testing.T) {
 				},
 			},
 			respStatusCode:     http.StatusInternalServerError,
-			key:                "first-channel.default",
+			hostKey:            "first-channel.default",
 			expectedStatusCode: http.StatusInternalServerError,
 		},
 		"invalid event": {
@@ -188,7 +232,7 @@ func TestServeHTTPMessageHandler(t *testing.T) {
 			},
 			eventID:            ptr.String(""), // invalid id
 			respStatusCode:     http.StatusOK,
-			key:                "second-channel.default",
+			hostKey:            "second-channel.default",
 			expectedStatusCode: http.StatusBadRequest,
 		},
 		"choose channel": {
@@ -222,8 +266,33 @@ func TestServeHTTPMessageHandler(t *testing.T) {
 				},
 			},
 			respStatusCode:     http.StatusOK,
-			key:                "second-channel.default",
+			hostKey:            "second-channel.default",
+			expectedStatusCode: http.StatusAccepted,
+		},
+		"path based": {
+			config: Config{
+				ChannelConfigs: []ChannelConfig{
+					{
+
+						Namespace: "ns",
+						Name:      "name",
+						HostName:  "first-channel.default",
+						Path:      "default/first-channel",
+						FanoutConfig: fanout.Config{
+							Subscriptions: []fanout.Subscription{
+								{
+									Subscriber: replaceDomain,
+								},
+							},
+						},
+					},
+				},
+			},
+			respStatusCode:     http.StatusOK,
+			hostKey:            "host.should.be.ignored",
+			pathKey:            "default/first-channel",
 			expectedStatusCode: http.StatusAccepted,
+			recvOptions:        []channel.MessageReceiverOptions{channel.ResolveMessageChannelFromPath(channel.ParseChannelFromPath)},
 		},
 	}
 	for n, tc := range testCases {
@@ -236,7 +305,7 @@ func TestServeHTTPMessageHandler(t *testing.T) {
 
 			logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
 			reporter := channel.NewStatsReporter("testcontainer", "testpod")
-			h, err := NewMessageHandlerWithConfig(context.TODO(), logger, channel.NewMessageDispatcher(logger), tc.config, reporter)
+			h, err := NewMessageHandlerWithConfig(context.TODO(), logger, channel.NewMessageDispatcher(logger), tc.config, reporter, tc.recvOptions...)
 			if err != nil {
 				t.Fatalf("Unexpected NewHandler error: '%v'", err)
 			}
@@ -255,7 +324,7 @@ func TestServeHTTPMessageHandler(t *testing.T) {
 			event.SetSource("testsource")
 			event.SetData(cloudevents.ApplicationJSON, "{}")
 
-			req := httptest.NewRequest(http.MethodPost, "http://"+tc.key+"/", nil)
+			req := httptest.NewRequest(http.MethodPost, "http://"+tc.hostKey+"/"+tc.pathKey, nil)
 			err = bindingshttp.WriteRequest(ctx, binding.ToMessage(&event), req)
 			if err != nil {
 				t.Fatal(err)

pkg/eventingtls/eventingtlstesting/eventingtlstesting.go

@@ -49,17 +49,7 @@ func StartServer(ctx context.Context, t *testing.T, port int, requests chan<- *n
 		Name:      "tls-secret",
 	}
 
-	_ = secretinformer.Get(ctx).Informer().GetStore().Add(&corev1.Secret{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace: secret.Namespace,
-			Name:      secret.Name,
-		},
-		Data: map[string][]byte{
-			"tls.key": Key,
-			"tls.crt": Crt,
-		},
-		Type: corev1.SecretTypeTLS,
-	})
+	_ = secretinformer.Get(ctx).Informer().GetStore().Add(GetTLSSecret(secret.Namespace, secret.Name))
 
 	serverTLSConfig := eventingtls.NewDefaultServerConfig()
 	serverTLSConfig.GetCertificate = eventingtls.GetCertificateFromSecret(ctx, secretinformer.Get(ctx), kubeclient.Get(ctx), secret)
@@ -187,3 +177,17 @@ O2dgzikq8iSy1BlRsVw=
 -----END CERTIFICATE-----
 `)
 }
+
+func GetTLSSecret(namespace, name string) *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+		Data: map[string][]byte{
+			"tls.key": Key,
+			"tls.crt": Crt,
+		},
+		Type: corev1.SecretTypeTLS,
+	}
+}

pkg/reconciler/inmemorychannel/dispatcher/controller.go

@@ -21,7 +21,9 @@ import (
 	"time"
 
 	"knative.dev/pkg/injection"
+	"knative.dev/pkg/system"
 
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 
 	"github.com/google/uuid"
@@ -50,6 +52,8 @@ import (
 	inmemorychannelinformer "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/inmemorychannel"
 	inmemorychannelreconciler "knative.dev/eventing/pkg/client/injection/reconciler/messaging/v1/inmemorychannel"
 	"knative.dev/eventing/pkg/inmemorychannel"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	secretinformer "knative.dev/pkg/injection/clients/namespacedkube/informers/core/v1/secret"
 )
 
 const (
@@ -58,6 +62,7 @@ const (
 	httpPort      = 8080
 	httpsPort     = 8443
 	finalizerName = "imc-dispatcher"
+	tlsSecretName = "imc-dispatcher-server-tls"
 )
 
 type envConfig struct {
@@ -151,15 +156,24 @@ func NewController(
 	httpDispatcher := inmemorychannel.NewMessageDispatcher(httpArgs)
 	httpReceiver := httpDispatcher.GetReceiver()
 
+	secret := types.NamespacedName{
+		Namespace: system.Namespace(),
+		Name:      tlsSecretName,
+	}
+	serverTLSConfig := eventingtls.NewDefaultServerConfig()
+	serverTLSConfig.GetCertificate = eventingtls.GetCertificateFromSecret(ctx, secretinformer.Get(ctx), kubeclient.Get(ctx), secret)
+	tlsConfig, err := eventingtls.GetTLSServerConfig(serverTLSConfig)
+	if err != nil {
+		logger.Panicf("unable to get tls config: %s", err)
+	}
 	httpsArgs := &inmemorychannel.InMemoryMessageDispatcherArgs{
 		Port:         httpsPort,
 		ReadTimeout:  readTimeout,
 		WriteTimeout: writeTimeout,
 		Handler:      sh,
 		Logger:       logger.Desugar(),
 
-		// TODO: add tls config
-		HTTPMessageReceiverOptions: []kncloudevents.HTTPMessageReceiverOption{},
+		HTTPMessageReceiverOptions: []kncloudevents.HTTPMessageReceiverOption{kncloudevents.WithTLSConfig(tlsConfig)},
 	}
 	httpsDispatcher := inmemorychannel.NewMessageDispatcher(httpsArgs)
 	httpsReceiver := httpsDispatcher.GetReceiver()