List applying EventPolicies in KafkaSink

control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_lifecycle.go

@@ -22,7 +22,8 @@ import (
 )
 
 const (
-	ConditionAddressable apis.ConditionType = "Addressable"
+	ConditionAddressable        apis.ConditionType = "Addressable"
+	ConditionEventPoliciesReady apis.ConditionType = "EventPoliciesReady"
 )
 
 var conditionSet apis.ConditionSet
@@ -54,3 +55,19 @@ func (ks *KafkaSinkStatus) SetAddress(addr *duckv1.Addressable) {
 func (kss *KafkaSinkStatus) InitializeConditions() {
 	kss.GetConditionSet().Manage(kss).InitializeConditions()
 }
+
+func (kss *KafkaSinkStatus) MarkEventPoliciesTrue() {
+	kss.GetConditionSet().Manage(kss).MarkTrue(ConditionEventPoliciesReady)
+}
+
+func (kss *KafkaSinkStatus) MarkEventPoliciesTrueWithReason(reason, messageFormat string, messageA ...interface{}) {
+	kss.GetConditionSet().Manage(kss).MarkTrueWithReason(ConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}
+
+func (kss *KafkaSinkStatus) MarkEventPoliciesFailed(reason, messageFormat string, messageA ...interface{}) {
+	kss.GetConditionSet().Manage(kss).MarkFalse(ConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}
+
+func (kss *KafkaSinkStatus) MarkEventPoliciesUnknown(reason, messageFormat string, messageA ...interface{}) {
+	kss.GetConditionSet().Manage(kss).MarkUnknown(ConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}

control-plane/pkg/reconciler/sink/controller.go

@@ -22,6 +22,8 @@ import (
 	"net"
 	"net/http"
 
+	"knative.dev/eventing/pkg/auth"
+
 	eventpolicyinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1alpha1/eventpolicy"
 
 	"go.uber.org/zap"
@@ -161,5 +163,11 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 		DeleteFunc: reconciler.OnDeleteObserver,
 	})
 
+	sinkGK := eventing.SchemeGroupVersion.WithKind("KafkaSink").GroupKind()
+
+	// Enqueue the KafkaSink, if we have an EventPolicy which was referencing
+	// or got updated and now is referencing the KafkaSink
+	eventPolicyInformer.Informer().AddEventHandler(auth.EventPolicyEventHandler(sinkInformer.Informer().GetIndexer(), sinkGK, impl.EnqueueKey))
+
 	return impl
 }

control-plane/pkg/reconciler/sink/kafka_sink.go

@@ -100,8 +100,6 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		Recorder:   controller.GetEventRecorder(ctx),
 	}
 
-	r.markEventPolicyConditionNotYetSupported(ks)
-
 	if !r.IsReceiverRunning() {
 		return statusConditionManager.DataPlaneNotAvailable()
 	}
@@ -170,6 +168,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 
 	logger.Debug("Got contract config map")
 
+	err = auth.UpdateStatusWithEventPolicies(feature.FromContext(ctx), &ks.Status.AppliedEventPoliciesStatus, &ks.Status, r.EventPolicyLister, eventing.SchemeGroupVersion.WithKind("KafkaSink"), ks.ObjectMeta)
+	if err != nil {
+		return fmt.Errorf("could not update KafkaSinks status with EventPolicies: %v", err)
+	}
+
 	// Get contract data.
 	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
 	if err != nil && ct == nil {
@@ -294,14 +297,6 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 	return nil
 }
 
-func (r *Reconciler) markEventPolicyConditionNotYetSupported(ks *eventing.KafkaSink) {
-	ks.Status.GetConditionSet().Manage(ks.GetStatus()).MarkTrueWithReason(
-		base.ConditionEventPoliciesReady,
-		"AuthzNotSupported",
-		"Authorization not yet supported",
-	)
-}
-
 func (r *Reconciler) FinalizeKind(ctx context.Context, ks *eventing.KafkaSink) reconciler.Event {
 	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
 		return r.finalizeKind(ctx, ks)
@@ -437,8 +432,8 @@ func (r *Reconciler) getSinkContractResource(ctx context.Context, kafkaSink *eve
 		Uid:    string(kafkaSink.UID),
 		Topics: []string{kafkaSink.Spec.Topic},
 		Ingress: &contract.Ingress{
-			Path:                       receiver.PathFromObject(kafkaSink),
-			ContentMode:                coreconfig.ContentModeFromString(*kafkaSink.Spec.ContentMode),
+			Path:        receiver.PathFromObject(kafkaSink),
+			ContentMode: coreconfig.ContentModeFromString(*kafkaSink.Spec.ContentMode),
 		},
 		FeatureFlags: &contract.FeatureFlags{
 			EnableEventTypeAutocreate: features.IsEnabled(feature.EvenTypeAutoCreate),