support auto generation of SinkBindings identity service account

config/core/roles/webhook-clusterrole.yaml

@@ -109,6 +109,12 @@ rules:
       - "create"
       - "patch"
 
+# For the SinkBinding reconciler adding the OIDC identity service accounts
+  - apiGroups:
+      - ""
+    resources:
+      - "serviceaccounts"
+    verbs: *everything
   # Necessary for conversion webhook. These are copied from the serving
   # TODO: Do we really need all these permissions?
   - apiGroups: ["apiextensions.k8s.io"]

pkg/apis/sources/v1/sinkbinding_lifecycle.go

@@ -35,6 +35,7 @@ import (
 
 var sbCondSet = apis.NewLivingConditionSet(
 	SinkBindingConditionSinkProvided,
+	SinkBindingConditionOIDCIdentityCreated,
 )
 
 // GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
@@ -95,6 +96,22 @@ func (sbs *SinkBindingStatus) MarkSink(addr *duckv1.Addressable) {
 	}
 }
 
+func (sbs *SinkBindingStatus) MarkOIDCIdentityCreatedSucceeded() {
+	sbCondSet.Manage(sbs).MarkTrue(SinkBindingConditionOIDCIdentityCreated)
+}
+
+func (sbs *SinkBindingStatus) MarkOIDCIdentityCreatedSucceededWithReason(reason, messageFormat string, messageA ...interface{}) {
+	sbCondSet.Manage(sbs).MarkTrueWithReason(SinkBindingConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
+func (sbs *SinkBindingStatus) MarkOIDCIdentityCreatedFailed(reason, messageFormat string, messageA ...interface{}) {
+	sbCondSet.Manage(sbs).MarkFalse(SinkBindingConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
+func (sbs *SinkBindingStatus) MarkOIDCIdentityCreatedUnknown(reason, messageFormat string, messageA ...interface{}) {
+	sbCondSet.Manage(sbs).MarkUnknown(SinkBindingConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
 // Do implements psbinding.Bindable
 func (sb *SinkBinding) Do(ctx context.Context, ps *duckv1.WithPod) {
 	// First undo so that we can just unconditionally append below.

pkg/apis/sources/v1/sinkbinding_lifecycle_test.go

@@ -171,9 +171,43 @@ func TestSinkBindingStatusIsReady(t *testing.T) {
 			s.InitializeConditions()
 			s.MarkSink(sink)
 			s.MarkBindingAvailable()
+			s.MarkOIDCIdentityCreatedSucceeded()
 			return s
 		}(),
 		want: true,
+	}, {
+		name: "mark OIDC identity created",
+		s: func() *SinkBindingStatus {
+			s := &SinkBindingStatus{}
+			s.InitializeConditions()
+			s.MarkSink(sink)
+			s.MarkBindingAvailable()
+			s.MarkOIDCIdentityCreatedSucceeded()
+			return s
+		}(),
+		want: true,
+	}, {
+		name: "mark OIDC identity created with reason",
+		s: func() *SinkBindingStatus {
+			s := &SinkBindingStatus{}
+			s.InitializeConditions()
+			s.MarkSink(sink)
+			s.MarkBindingAvailable()
+			s.MarkOIDCIdentityCreatedSucceededWithReason("TheReason", "feature is disabled")
+			return s
+		}(),
+		want: true,
+	}, {
+		name: "mark OIDC identity created failed",
+		s: func() *SinkBindingStatus {
+			s := &SinkBindingStatus{}
+			s.InitializeConditions()
+			s.MarkSink(sink)
+			s.MarkBindingAvailable()
+			s.MarkOIDCIdentityCreatedFailed("TheReason", "this is a message")
+			return s
+		}(),
+		want: false,
 	}}
 
 	for _, test := range tests {

pkg/apis/sources/v1/sinkbinding_types.go

@@ -77,6 +77,10 @@ const (
 	// SinkBindingConditionSinkProvided is configured to indicate whether the
 	// sink has been properly extracted from the resolver.
 	SinkBindingConditionSinkProvided apis.ConditionType = "SinkProvided"
+
+	// SinkBindingConditionOIDCIdentityCreated is configured to indicate whether
+	// the OIDC identity has been created for the sink.
+	SinkBindingConditionOIDCIdentityCreated apis.ConditionType = "OIDCIdentityCreated"
 )
 
 // SinkBindingStatus communicates the observed state of the SinkBinding (from the controller).

pkg/reconciler/sinkbinding/controller.go

@@ -19,24 +19,33 @@ package sinkbinding
 import (
 	"context"
 	"errors"
+	"fmt"
 
+	"knative.dev/eventing/pkg/auth"
 	sbinformer "knative.dev/eventing/pkg/client/injection/informers/sources/v1/sinkbinding"
 	"knative.dev/pkg/client/injection/ducks/duck/v1/podspecable"
 	"knative.dev/pkg/client/injection/kube/informers/core/v1/namespace"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/system"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
+	"knative.dev/eventing/pkg/apis/feature"
 	v1 "knative.dev/eventing/pkg/apis/sources/v1"
 	"knative.dev/pkg/apis/duck"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	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"
@@ -50,8 +59,11 @@ const (
 )
 
 type SinkBindingSubResourcesReconciler struct {
-	res     *resolver.URIResolver
-	tracker tracker.Interface
+	res                  *resolver.URIResolver
+	tracker              tracker.Interface
+	serviceAccountLister corev1listers.ServiceAccountLister
+	kubeclient           kubernetes.Interface
+	featureStore         *feature.Store
 }
 
 // NewController returns a new SinkBinding reconciler.
@@ -65,6 +77,9 @@ func NewController(
 	dc := dynamicclient.Get(ctx)
 	psInformerFactory := podspecable.Get(ctx)
 	namespaceInformer := namespace.Get(ctx)
+	serviceaccountInformer := serviceaccountinformer.Get(ctx)
+	configmapInformer := configmapinformer.Get(ctx)
+
 	c := &psbinding.BaseReconciler{
 		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
 			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
@@ -94,13 +109,22 @@ func NewController(
 		Logger:        logger,
 	})
 
+	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"), func(name string, value interface{}) {
+		impl.GlobalResync(sbInformer.Informer())
+	})
+
+	featureStore.WatchConfigs(cmw)
+
 	sbInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 	namespaceInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	sbResolver := resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 	c.SubResourcesReconciler = &SinkBindingSubResourcesReconciler{
-		res:     sbResolver,
-		tracker: impl.Tracker,
+		res:                  sbResolver,
+		tracker:              impl.Tracker,
+		kubeclient:           kubeclient.Get(ctx),
+		serviceAccountLister: serviceaccountInformer.Lister(),
+		featureStore:         featureStore,
 	}
 
 	c.WithContext = func(ctx context.Context, b psbinding.Bindable) (context.Context, error) {
@@ -114,6 +138,20 @@ func NewController(
 		},
 	}
 
+	// Reconcile SinkBinding when the OIDC service account changes
+	serviceaccountInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterController(&v1.SinkBinding{}),
+		Handler:    controller.HandleAll(impl.EnqueueControllerOf),
+	})
+
+	// reconcile sinkindings on changes on the features configmap
+	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: controller.FilterWithNameAndNamespace(system.Namespace(), feature.FlagsConfigName),
+		Handler: controller.HandleAll(func(i interface{}) {
+			impl.GlobalResync(sbInformer.Informer())
+		}),
+	})
+
 	return impl
 }
 
@@ -161,6 +199,24 @@ func (s *SinkBindingSubResourcesReconciler) Reconcile(ctx context.Context, b psb
 			Name:       sb.Spec.Sink.Ref.Name,
 		}, b)
 	}
+
+	featureFlags := s.featureStore.Load()
+	if featureFlags.IsOIDCAuthentication() {
+		saName := auth.GetOIDCServiceAccountNameForResource(v1.SchemeGroupVersion.WithKind("SinkBinding"), sb.ObjectMeta)
+		sb.Status.Auth = &duckv1.AuthStatus{
+			ServiceAccountName: &saName,
+		}
+
+		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, s.serviceAccountLister, s.kubeclient, v1.SchemeGroupVersion.WithKind("SinkBinding"), sb.ObjectMeta); err != nil {
+			sb.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
+			return err
+		}
+		sb.Status.MarkOIDCIdentityCreatedSucceeded()
+	} else {
+		sb.Status.Auth = nil
+		sb.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
+	}
+
 	addr, err := s.res.AddressableFromDestinationV1(ctx, sb.Spec.Sink, sb)
 	if err != nil {
 		logging.FromContext(ctx).Errorf("Failed to get Addressable from Destination: %w", err)