Trigger create and expose OIDC service account

pkg/apis/eventing/v1/trigger_lifecycle.go

@@ -23,7 +23,7 @@ import (
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
-var triggerCondSet = apis.NewLivingConditionSet(TriggerConditionBroker, TriggerConditionSubscribed, TriggerConditionDependency, TriggerConditionSubscriberResolved, TriggerConditionDeadLetterSinkResolved)
+var triggerCondSet = apis.NewLivingConditionSet(TriggerConditionBroker, TriggerConditionSubscribed, TriggerConditionDependency, TriggerConditionSubscriberResolved, TriggerConditionDeadLetterSinkResolved, TriggerConditionOIDCIdentityCreated)
 
 const (
 	// TriggerConditionReady has status True when all subconditions below have been set to True.
@@ -39,6 +39,8 @@ const (
 
 	TriggerConditionDeadLetterSinkResolved apis.ConditionType = "DeadLetterSinkResolved"
 
+	TriggerConditionOIDCIdentityCreated apis.ConditionType = "OIDCIdentityCreated"
+
 	// TriggerAnyFilter Constant to represent that we should allow anything.
 	TriggerAnyFilter = ""
 )
@@ -199,3 +201,19 @@ func (ts *TriggerStatus) PropagateDependencyStatus(ks *duckv1.Source) {
 		ts.MarkDependencyUnknown("DependencyUnknown", "The status of Dependency is invalid: %v", kc.Status)
 	}
 }
+
+func (ts *TriggerStatus) MarkOIDCIdentityCreatedSucceeded() {
+	triggerCondSet.Manage(ts).MarkTrue(TriggerConditionOIDCIdentityCreated)
+}
+
+func (ts *TriggerStatus) MarkOIDCIdentityCreatedSucceededWithReason(reason, messageFormat string, messageA ...interface{}) {
+	triggerCondSet.Manage(ts).MarkTrueWithReason(TriggerConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
+func (ts *TriggerStatus) MarkOIDCIdentityCreatedFailed(reason, messageFormat string, messageA ...interface{}) {
+	triggerCondSet.Manage(ts).MarkFalse(TriggerConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
+func (ts *TriggerStatus) MarkOIDCIdentityCreatedUnknown(reason, messageFormat string, messageA ...interface{}) {
+	triggerCondSet.Manage(ts).MarkUnknown(TriggerConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}

pkg/apis/eventing/v1/trigger_lifecycle_test.go

@@ -153,6 +153,9 @@ func TestTriggerInitializeConditions(t *testing.T) {
 				}, {
 					Type:   TriggerConditionDependency,
 					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionOIDCIdentityCreated,
+					Status: corev1.ConditionUnknown,
 				}, {
 					Type:   TriggerConditionReady,
 					Status: corev1.ConditionUnknown,
@@ -187,17 +190,19 @@ func TestTriggerInitializeConditions(t *testing.T) {
 				}, {
 					Type:   TriggerConditionDependency,
 					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionOIDCIdentityCreated,
+					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionReady,
+					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionSubscriberResolved,
+					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionSubscribed,
+					Status: corev1.ConditionUnknown,
 				},
-					{
-						Type:   TriggerConditionReady,
-						Status: corev1.ConditionUnknown,
-					}, {
-						Type:   TriggerConditionSubscriberResolved,
-						Status: corev1.ConditionUnknown,
-					}, {
-						Type:   TriggerConditionSubscribed,
-						Status: corev1.ConditionUnknown,
-					},
 				},
 			},
 		},
@@ -222,6 +227,9 @@ func TestTriggerInitializeConditions(t *testing.T) {
 				}, {
 					Type:   TriggerConditionDependency,
 					Status: corev1.ConditionUnknown,
+				}, {
+					Type:   TriggerConditionOIDCIdentityCreated,
+					Status: corev1.ConditionUnknown,
 				}, {
 					Type:   TriggerConditionReady,
 					Status: corev1.ConditionUnknown,
@@ -257,6 +265,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		subscriberResolvedStatus    bool
 		dlsResolvedStatus           bool
 		dependencyAnnotationExists  bool
+		oidcServiceAccountStatus    bool
 		dependencyStatus            corev1.ConditionStatus
 		wantConditionStatus         corev1.ConditionStatus
 	}{{
@@ -267,6 +276,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		subscriptionCondition:       TestHelper.ReadySubscriptionCondition(),
 		subscriberResolvedStatus:    true,
 		dlsResolvedStatus:           true,
+		oidcServiceAccountStatus:    true,
 		dependencyAnnotationExists:  false,
 		wantConditionStatus:         corev1.ConditionTrue,
 	}, {
@@ -277,6 +287,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		subscriptionCondition:       TestHelper.ReadySubscriptionCondition(),
 		subscriberResolvedStatus:    true,
 		dlsResolvedStatus:           true,
+		oidcServiceAccountStatus:    true,
 		dependencyAnnotationExists:  false,
 		wantConditionStatus:         corev1.ConditionUnknown,
 	}, {
@@ -286,6 +297,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		markVirtualServiceExists:    true,
 		subscriptionCondition:       TestHelper.ReadySubscriptionCondition(),
 		subscriberResolvedStatus:    true,
+		oidcServiceAccountStatus:    true,
 		dependencyAnnotationExists:  false,
 		wantConditionStatus:         corev1.ConditionFalse,
 	}, {
@@ -295,6 +307,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		markVirtualServiceExists:    true,
 		subscriptionCondition:       TestHelper.FalseSubscriptionCondition(),
 		subscriberResolvedStatus:    true,
+		oidcServiceAccountStatus:    true,
 		dependencyAnnotationExists:  false,
 		wantConditionStatus:         corev1.ConditionFalse,
 	}, {
@@ -304,6 +317,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		markVirtualServiceExists:    true,
 		subscriptionCondition:       TestHelper.ReadySubscriptionCondition(),
 		subscriberResolvedStatus:    false,
+		oidcServiceAccountStatus:    true,
 		dependencyAnnotationExists:  true,
 		dependencyStatus:            corev1.ConditionTrue,
 		wantConditionStatus:         corev1.ConditionFalse,
@@ -316,6 +330,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		subscriberResolvedStatus:    true,
 		dependencyAnnotationExists:  true,
 		dlsResolvedStatus:           true,
+		oidcServiceAccountStatus:    true,
 		dependencyStatus:            corev1.ConditionUnknown,
 		wantConditionStatus:         corev1.ConditionUnknown,
 	}, {
@@ -326,6 +341,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		subscriptionCondition:       TestHelper.ReadySubscriptionCondition(),
 		subscriberResolvedStatus:    true,
 		dependencyAnnotationExists:  true,
+		oidcServiceAccountStatus:    true,
 		dependencyStatus:            corev1.ConditionFalse,
 		wantConditionStatus:         corev1.ConditionFalse,
 	}, {
@@ -335,6 +351,7 @@ func TestTriggerConditionStatus(t *testing.T) {
 		markVirtualServiceExists:    false,
 		subscriptionCondition:       TestHelper.FalseSubscriptionCondition(),
 		subscriberResolvedStatus:    false,
+		oidcServiceAccountStatus:    false,
 		dependencyAnnotationExists:  true,
 		dependencyStatus:            corev1.ConditionFalse,
 		wantConditionStatus:         corev1.ConditionFalse,
@@ -346,9 +363,22 @@ func TestTriggerConditionStatus(t *testing.T) {
 		subscriptionCondition:       TestHelper.ReadySubscriptionCondition(),
 		subscriberResolvedStatus:    true,
 		dependencyAnnotationExists:  true,
+		oidcServiceAccountStatus:    true,
 		dlsResolvedStatus:           false,
 		wantConditionStatus:         corev1.ConditionFalse,
+	}, {
+		name:                        "oidc status false",
+		brokerStatus:                TestHelper.ReadyBrokerStatus(),
+		markKubernetesServiceExists: true,
+		markVirtualServiceExists:    true,
+		subscriptionCondition:       TestHelper.ReadySubscriptionCondition(),
+		subscriberResolvedStatus:    true,
+		dlsResolvedStatus:           true,
+		oidcServiceAccountStatus:    false,
+		dependencyAnnotationExists:  false,
+		wantConditionStatus:         corev1.ConditionFalse,
 	}}
+
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			ts := &TriggerStatus{}
@@ -379,6 +409,11 @@ func TestTriggerConditionStatus(t *testing.T) {
 					ts.MarkDependencyFailed("The status of dependency is false", "The status of dependency is unknown: nil")
 				}
 			}
+			if test.oidcServiceAccountStatus {
+				ts.MarkOIDCIdentityCreatedSucceeded()
+			} else {
+				ts.MarkOIDCIdentityCreatedFailed("Unable to ...", "")
+			}
 			got := ts.GetTopLevelCondition().Status
 			if test.wantConditionStatus != got {
 				t.Errorf("unexpected readiness: want %v, got %v", test.wantConditionStatus, got)

pkg/auth/serviceaccount.go

@@ -17,12 +17,18 @@ limitations under the License.
 package auth
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
+	"go.uber.org/zap"
 	v1 "k8s.io/api/core/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	"knative.dev/pkg/logging"
 	"knative.dev/pkg/ptr"
 )
 
@@ -47,7 +53,7 @@ func GetOIDCServiceAccountForResource(gvk schema.GroupVersionKind, objectMeta me
 					Kind:               gvk.GroupKind().Kind,
 					Name:               objectMeta.GetName(),
 					UID:                objectMeta.GetUID(),
-					Controller:         ptr.Bool(false),
+					Controller:         ptr.Bool(true),
 					BlockOwnerDeletion: ptr.Bool(false),
 				},
 			},
@@ -57,3 +63,34 @@ func GetOIDCServiceAccountForResource(gvk schema.GroupVersionKind, objectMeta me
 		},
 	}
 }
+
+// EnsureOIDCServiceAccountExistsForResource makes sure the given resource has
+// an OIDC service account with an owner reference to the resource set.
+func EnsureOIDCServiceAccountExistsForResource(ctx context.Context, serviceAccountLister corev1listers.ServiceAccountLister, kubeclient kubernetes.Interface, gvk schema.GroupVersionKind, objectMeta metav1.ObjectMeta) error {
+	saName := GetOIDCServiceAccountNameForResource(gvk, objectMeta)
+	sa, err := serviceAccountLister.ServiceAccounts(objectMeta.Namespace).Get(saName)
+
+	// If the resource doesn't exist, we'll create it.
+	if apierrs.IsNotFound(err) {
+		logging.FromContext(ctx).Debugw("Creating OIDC service account", zap.Error(err))
+
+		expected := GetOIDCServiceAccountForResource(gvk, objectMeta)
+
+		_, err = kubeclient.CoreV1().ServiceAccounts(objectMeta.Namespace).Create(ctx, expected, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("could not create OIDC service account %s/%s for %s: %w", objectMeta.Name, objectMeta.Namespace, gvk.Kind, err)
+		}
+
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("could not get OIDC service account %s/%s for %s: %w", objectMeta.Name, objectMeta.Namespace, gvk.Kind, err)
+	}
+
+	if !metav1.IsControlledBy(&sa.ObjectMeta, &objectMeta) {
+		return fmt.Errorf("service account %s not owned by %s %s", sa.Name, gvk.Kind, objectMeta.Name)
+	}
+
+	return nil
+}

pkg/auth/serviceaccount_test.go

@@ -84,7 +84,7 @@ func TestGetOIDCServiceAccountForResource(t *testing.T) {
 					Kind:               "Broker",
 					Name:               "my-broker",
 					UID:                "my-uuid",
-					Controller:         ptr.Bool(false),
+					Controller:         ptr.Bool(true),
 					BlockOwnerDeletion: ptr.Bool(false),
 				},
 			},

pkg/reconciler/broker/trigger/controller.go

@@ -24,6 +24,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"knative.dev/pkg/client/injection/ducks/duck/v1/source"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
+	serviceaccountinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection/clients/dynamicclient"
@@ -43,6 +44,7 @@ import (
 	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	"knative.dev/eventing/pkg/duck"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
 )
 
 // NewController initializes the controller and is called by the generated code
@@ -57,19 +59,22 @@ func NewController(
 	subscriptionInformer := subscriptioninformer.Get(ctx)
 	configmapInformer := configmapinformer.Get(ctx)
 	secretInformer := secretinformer.Get(ctx)
+	serviceaccountInformer := serviceaccountinformer.Get(ctx)
 
 	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"))
 	featureStore.WatchConfigs(cmw)
 
 	triggerLister := triggerInformer.Lister()
 	r := &Reconciler{
-		eventingClientSet:  eventingclient.Get(ctx),
-		dynamicClientSet:   dynamicclient.Get(ctx),
-		subscriptionLister: subscriptionInformer.Lister(),
-		brokerLister:       brokerInformer.Lister(),
-		triggerLister:      triggerLister,
-		configmapLister:    configmapInformer.Lister(),
-		secretLister:       secretInformer.Lister(),
+		eventingClientSet:    eventingclient.Get(ctx),
+		dynamicClientSet:     dynamicclient.Get(ctx),
+		kubeclient:           kubeclient.Get(ctx),
+		subscriptionLister:   subscriptionInformer.Lister(),
+		brokerLister:         brokerInformer.Lister(),
+		triggerLister:        triggerLister,
+		configmapLister:      configmapInformer.Lister(),
+		secretLister:         secretInformer.Lister(),
+		serviceAccountLister: serviceaccountInformer.Lister(),
 	}
 	impl := triggerreconciler.NewImpl(ctx, r, func(impl *controller.Impl) controller.Options {
 		return controller.Options{
@@ -106,6 +111,12 @@ func NewController(
 		Handler:    controller.HandleAll(impl.EnqueueControllerOf),
 	})
 
+	// Reconciler Trigger when the OIDC service account changes
+	serviceaccountInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterController(&eventing.Trigger{}),
+		Handler:    controller.HandleAll(impl.EnqueueControllerOf),
+	})
+
 	return impl
 }
 

pkg/reconciler/broker/trigger/controller_test.go

@@ -44,6 +44,7 @@ import (
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
 	_ "knative.dev/eventing/pkg/client/injection/informers/messaging/v1/subscription/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/serviceaccount/fake"
 )
 
 func TestNew(t *testing.T) {

pkg/reconciler/broker/trigger/trigger.go

@@ -27,6 +27,7 @@ import (
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
 	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/utils/pointer"
 	"knative.dev/pkg/apis"
@@ -44,6 +45,7 @@ import (
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/feature"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	"knative.dev/eventing/pkg/auth"
 	clientset "knative.dev/eventing/pkg/client/clientset/versioned"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
@@ -65,13 +67,15 @@ const (
 type Reconciler struct {
 	eventingClientSet clientset.Interface
 	dynamicClientSet  dynamic.Interface
+	kubeclient        kubernetes.Interface
 
 	// listers index properties about resources
-	subscriptionLister messaginglisters.SubscriptionLister
-	brokerLister       eventinglisters.BrokerLister
-	triggerLister      eventinglisters.TriggerLister
-	configmapLister    corev1listers.ConfigMapLister
-	secretLister       corev1listers.SecretLister
+	subscriptionLister   messaginglisters.SubscriptionLister
+	brokerLister         eventinglisters.BrokerLister
+	triggerLister        eventinglisters.TriggerLister
+	configmapLister      corev1listers.ConfigMapLister
+	secretLister         corev1listers.SecretLister
+	serviceAccountLister corev1listers.ServiceAccountLister
 
 	// Dynamic tracker to track Sources. In particular, it tracks the dependency between Triggers and Sources.
 	sourceTracker duck.ListableTracker
@@ -137,6 +141,23 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, t *eventingv1.Trigger) p
 		return err
 	}
 
+	featureFlags := feature.FromContext(ctx)
+	if featureFlags.IsOIDCAuthentication() {
+		saName := auth.GetOIDCServiceAccountNameForResource(eventingv1.SchemeGroupVersion.WithKind("Trigger"), t.ObjectMeta)
+		t.Status.Auth = &duckv1.AuthStatus{
+			ServiceAccountName: &saName,
+		}
+
+		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeclient, eventingv1.SchemeGroupVersion.WithKind("Trigger"), t.ObjectMeta); err != nil {
+			t.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
+			return err
+		}
+		t.Status.MarkOIDCIdentityCreatedSucceeded()
+	} else {
+		t.Status.Auth = nil
+		t.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	}
+
 	sub, err := r.subscribeToBrokerChannel(ctx, b, t, brokerTrigger)
 	if err != nil {
 		logging.FromContext(ctx).Errorw("Unable to Subscribe", zap.Error(err))