Authenticate requests from mt-broker-ingress to underlying channel

config/brokers/mt-channel-broker/200-ingress-serviceaccount-oidc.yaml

@@ -0,0 +1 @@
+roles/ingress-serviceaccount-oidc.yaml

config/brokers/mt-channel-broker/roles/ingress-role.yaml

@@ -25,3 +25,11 @@ rules:
       - get
       - list
       - watch
+  - apiGroups:
+      - ""
+    resources:
+      - "serviceaccounts/token"
+    resourceNames:
+      - "mt-broker-ingress-oidc"
+    verbs:
+      - create

config/brokers/mt-channel-broker/roles/ingress-serviceaccount-oidc.yaml

@@ -0,0 +1,21 @@
+# 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.
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: mt-broker-ingress-oidc
+  namespace: knative-eventing
+  labels:
+    app.kubernetes.io/version: devel
+    app.kubernetes.io/name: knative-eventing

pkg/apis/eventing/register.go

@@ -66,6 +66,10 @@ const (
 	// https://www.rfc-editor.org/rfc/rfc7468
 	BrokerChannelCACertsStatusAnnotationKey = "knative.dev/channelCACerts"
 
+	// BrokerChannelAudienceStatusAnnotationKey is the broker status annotation
+	// key used to specify the channels OIDC audience.
+	BrokerChannelAudienceStatusAnnotationKey = "knative.dev/channelAudience"
+
 	// BrokerChannelAPIVersionStatusAnnotationKey is the broker status
 	// annotation key used to specify the APIVersion of the channel for
 	// the triggers to subscribe to.

pkg/broker/ingress/ingress_handler.go

@@ -32,10 +32,10 @@ import (
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
-	"k8s.io/utils/pointer"
 
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/system"
 
 	"knative.dev/eventing/pkg/apis/eventing"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -148,15 +148,18 @@ func (h *Handler) getChannelAddress(broker *eventingv1.Broker) (*duckv1.Addressa
 		return nil, fmt.Errorf("failed to parse channel address url")
 	}
 
-	var caCerts *string
-	certs, present := broker.Status.Annotations[eventing.BrokerChannelCACertsStatusAnnotationKey]
-	if present && certs != "" {
-		caCerts = pointer.String(certs)
-	}
 	addr := &duckv1.Addressable{
-		URL:     url,
-		CACerts: caCerts,
+		URL: url,
+	}
+
+	if certs, ok := broker.Status.Annotations[eventing.BrokerChannelCACertsStatusAnnotationKey]; ok && certs != "" {
+		addr.CACerts = &certs
+	}
+
+	if audience, ok := broker.Status.Annotations[eventing.BrokerChannelAudienceStatusAnnotationKey]; ok && audience != "" {
+		addr.Audience = &audience
 	}
+
 	return addr, nil
 }
 
@@ -318,7 +321,15 @@ func (h *Handler) receive(ctx context.Context, headers http.Header, event *cloud
 		return http.StatusBadRequest, kncloudevents.NoDuration
 	}
 
-	dispatchInfo, err := h.eventDispatcher.SendEvent(ctx, *event, *channelAddress, kncloudevents.WithHeader(headers))
+	opts := []kncloudevents.SendOption{
+		kncloudevents.WithHeader(headers),
+		kncloudevents.WithOIDCAuthentication(&types.NamespacedName{
+			Name:      "mt-broker-ingress-oidc",
+			Namespace: system.Namespace(),
+		}),
+	}
+
+	dispatchInfo, err := h.eventDispatcher.SendEvent(ctx, *event, *channelAddress, opts...)
 	if err != nil {
 		h.Logger.Error("failed to dispatch event", zap.Error(err))
 		return http.StatusInternalServerError, kncloudevents.NoDuration

pkg/reconciler/broker/broker.go

@@ -158,6 +158,10 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, b *eventingv1.Broker) pk
 		b.Status.Annotations[eventing.BrokerChannelCACertsStatusAnnotationKey] = *caCerts
 	}
 
+	if audience := triggerChan.Status.Address.Audience; audience != nil && *audience != "" {
+		b.Status.Annotations[eventing.BrokerChannelAudienceStatusAnnotationKey] = *audience
+	}
+
 	channelStatus := &duckv1.ChannelableStatus{
 		AddressStatus:  triggerChan.Status.AddressStatus,
 		DeliveryStatus: triggerChan.Status.DeliveryStatus,

pkg/reconciler/broker/broker_test.go

@@ -416,6 +416,36 @@ func TestReconcile(t *testing.T) {
 					WithChannelNameAnnotation(triggerChannelName),
 					WithDLSNotConfigured()),
 			}},
+		}, {
+			Name: "Successful Reconciliation with a Channel with Audience",
+			Key:  testKey,
+			Objects: []runtime.Object{
+				NewBroker(brokerName, testNS,
+					WithBrokerClass(eventing.MTChannelBrokerClassValue),
+					WithBrokerConfig(config()),
+					WithInitBrokerConditions),
+				createChannel(withChannelReady, withChannelStatusAudience(channelAudience)),
+				imcConfigMap(),
+				NewEndpoints(filterServiceName, systemNS,
+					WithEndpointsLabels(FilterLabels()),
+					WithEndpointsAddresses(corev1.EndpointAddress{IP: "127.0.0.1"})),
+				NewEndpoints(ingressServiceName, systemNS,
+					WithEndpointsLabels(IngressLabels()),
+					WithEndpointsAddresses(corev1.EndpointAddress{IP: "127.0.0.1"})),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: NewBroker(brokerName, testNS,
+					WithBrokerClass(eventing.MTChannelBrokerClassValue),
+					WithBrokerConfig(config()),
+					WithBrokerReady,
+					WithBrokerAddressURI(brokerAddress),
+					WithChannelAddressAnnotation(triggerChannelURL),
+					WithChannelAudienceAnnotation(channelAudience),
+					WithChannelAPIVersionAnnotation(triggerChannelAPIVersion),
+					WithChannelKindAnnotation(triggerChannelKind),
+					WithChannelNameAnnotation(triggerChannelName),
+					WithDLSNotConfigured()),
+			}},
 		}, {
 			Name: "Successful Reconciliation. Using legacy channel template config element.",
 			Key:  testKey,
@@ -863,6 +893,15 @@ func withChannelStatusCACerts(caCerts string) unstructuredOption {
 	}
 }
 
+func withChannelStatusAudience(aud string) unstructuredOption {
+	return func(channel *unstructured.Unstructured) {
+		if err := unstructured.SetNestedField(channel.Object, aud,
+			"status", "address", "audience"); err != nil {
+			panic(err)
+		}
+	}
+}
+
 func withChannelReady(channel *unstructured.Unstructured) {
 	withChannelStatusAddress(triggerChannelURL)(channel)
 	withChannelStatusDeadLetterSinkURI(dls.URL.String())(channel)
@@ -1018,6 +1057,7 @@ s/wImGnMVk5RzpBVrq2VB9SkX/ThTVYEC/Sd9BQM364MCR+TA1l8/ptaLFLuwyw8
 O2dgzikq8iSy1BlRsVw=
 -----END CERTIFICATE-----
 `
+	channelAudience = "channel-audience"
 )
 
 func makeTLSSecret() *corev1.Secret {

pkg/reconciler/testing/v1/broker.go

@@ -183,6 +183,15 @@ func WithChannelCACertsAnnotation(caCerts string) BrokerOption {
 	}
 }
 
+func WithChannelAudienceAnnotation(audience string) BrokerOption {
+	return func(b *v1.Broker) {
+		if b.Status.Annotations == nil {
+			b.Status.Annotations = make(map[string]string, 1)
+		}
+		b.Status.Annotations[eventing.BrokerChannelAudienceStatusAnnotationKey] = audience
+	}
+}
+
 func WithBrokerStatusDLS(dls duckv1.Addressable) BrokerOption {
 	return func(b *v1.Broker) {
 		b.Status.MarkDeadLetterSinkResolvedSucceeded(eventingv1.NewDeliveryStatusFromAddressable(&dls))