List applying policies in Channel and propagate EventPolicies to unde…

…rlying channel (#8014)

* Move AppliedEventPoliciesStatus to ChannelableStatus to have list of policies in channel resource

* Move functionality to update status with policies to auth package and make generic

* Resync Channels when features CM changes

* Simplify WithEventPolicyToRef

* List applying policies in Channels status

* Add integration test for listing policies

* Copy channels EventPolicies to backing channel

config/core/resources/channel.yaml

@@ -244,6 +244,18 @@ spec:
                   namespace:
                     description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/ This is optional field, it gets defaulted to the object holding it if left out.'
                     type: string
+              policies:
+                description: List of applied EventPolicies
+                type: array
+                items:
+                  type: object
+                  properties:
+                    apiVersion:
+                      description: The API version of the applied EventPolicy. This indicates, which version of EventPolicy is supported by the resource.
+                      type: string
+                    name:
+                      description: The name of the applied EventPolicy
+                      type: string
               conditions:
                 description: Conditions the latest available observations of a resource's current state.
                 type: array

config/core/roles/controller-clusterroles.yaml

@@ -97,6 +97,8 @@ rules:
       - "triggers/status"
       - "eventtypes"
       - "eventtypes/status"
+      - "eventpolicies"
+      - "eventpolicies/status"
     verbs:
       - "get"
       - "list"

docs/eventing-api.md

@@ -49,7 +49,7 @@ Resource Types:
 <h3 id="duck.knative.dev/v1.AppliedEventPoliciesStatus">AppliedEventPoliciesStatus
 </h3>
 <p>
-(<em>Appears on:</em><a href="#eventing.knative.dev/v1.BrokerStatus">BrokerStatus</a>, <a href="#flows.knative.dev/v1.ParallelStatus">ParallelStatus</a>, <a href="#flows.knative.dev/v1.SequenceStatus">SequenceStatus</a>, <a href="#messaging.knative.dev/v1.InMemoryChannelStatus">InMemoryChannelStatus</a>)
+(<em>Appears on:</em><a href="#duck.knative.dev/v1.ChannelableStatus">ChannelableStatus</a>, <a href="#eventing.knative.dev/v1.BrokerStatus">BrokerStatus</a>, <a href="#flows.knative.dev/v1.ParallelStatus">ParallelStatus</a>, <a href="#flows.knative.dev/v1.SequenceStatus">SequenceStatus</a>)
 </p>
 <p>
 <p>AppliedEventPoliciesStatus contains the list of policies which apply to a resource.
@@ -368,6 +368,23 @@ DeliveryStatus
 resolved delivery options.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>AppliedEventPoliciesStatus</code><br/>
+<em>
+<a href="#duck.knative.dev/v1.AppliedEventPoliciesStatus">
+AppliedEventPoliciesStatus
+</a>
+</em>
+</td>
+<td>
+<p>
+(Members of <code>AppliedEventPoliciesStatus</code> are embedded into this type.)
+</p>
+<em>(Optional)</em>
+<p>AppliedEventPoliciesStatus contains the list of EventPolicies which apply to this Channel</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="duck.knative.dev/v1.DeliverySpec">DeliverySpec
@@ -5459,23 +5476,6 @@ ChannelableStatus
 <p>Channel conforms to Duck type ChannelableStatus.</p>
 </td>
 </tr>
-<tr>
-<td>
-<code>AppliedEventPoliciesStatus</code><br/>
-<em>
-<a href="#duck.knative.dev/v1.AppliedEventPoliciesStatus">
-AppliedEventPoliciesStatus
-</a>
-</em>
-</td>
-<td>
-<p>
-(Members of <code>AppliedEventPoliciesStatus</code> are embedded into this type.)
-</p>
-<em>(Optional)</em>
-<p>AppliedEventPoliciesStatus contains the list of EventPolicies which apply to this Broker</p>
-</td>
-</tr>
 </tbody>
 </table>
 <h3 id="messaging.knative.dev/v1.SubscriptionSpec">SubscriptionSpec

pkg/apis/duck/v1/channelable_types.go

@@ -67,6 +67,9 @@ type ChannelableStatus struct {
 	// resolved delivery options.
 	// +optional
 	DeliveryStatus `json:",inline"`
+	// AppliedEventPoliciesStatus contains the list of EventPolicies which apply to this Channel
+	// +optional
+	AppliedEventPoliciesStatus `json:",inline"`
 }
 
 var (

pkg/apis/messaging/v1/channel_lifecycle.go

@@ -29,6 +29,7 @@ var chCondSet = apis.NewLivingConditionSet(
 	ChannelConditionBackingChannelReady,
 	ChannelConditionAddressable,
 	ChannelConditionDeadLetterSinkResolved,
+	ChannelConditionEventPoliciesReady,
 )
 
 const (
@@ -45,6 +46,10 @@ const (
 	// ChannelConditionDeadLetterSinkResolved has status True when there is a Dead Letter Sink ref or URI
 	// defined in the Spec.Delivery, is a valid destination and its correctly resolved into a valid URI
 	ChannelConditionDeadLetterSinkResolved apis.ConditionType = "DeadLetterSinkResolved"
+
+	// ChannelConditionEventPoliciesReady has status True when all the EventPolicies which reference this
+	// Channel are Ready too.
+	ChannelConditionEventPoliciesReady apis.ConditionType = "EventPoliciesReady"
 )
 
 // GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
@@ -146,3 +151,19 @@ func (cs *ChannelStatus) MarkDeadLetterSinkResolvedFailed(reason, messageFormat
 	cs.DeliveryStatus = eventingduck.DeliveryStatus{}
 	chCondSet.Manage(cs).MarkFalse(ChannelConditionDeadLetterSinkResolved, reason, messageFormat, messageA...)
 }
+
+func (cs *ChannelStatus) MarkEventPoliciesFailed(reason, messageFormat string, messageA ...interface{}) {
+	chCondSet.Manage(cs).MarkFalse(ChannelConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}
+
+func (cs *ChannelStatus) MarkEventPoliciesUnknown(reason, messageFormat string, messageA ...interface{}) {
+	chCondSet.Manage(cs).MarkUnknown(ChannelConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}
+
+func (cs *ChannelStatus) MarkEventPoliciesTrue() {
+	chCondSet.Manage(cs).MarkTrue(ChannelConditionEventPoliciesReady)
+}
+
+func (cs *ChannelStatus) MarkEventPoliciesTrueWithReason(reason, messageFormat string, messageA ...interface{}) {
+	chCondSet.Manage(cs).MarkTrueWithReason(ChannelConditionEventPoliciesReady, reason, messageFormat, messageA...)
+}

pkg/apis/messaging/v1/channel_lifecycle_test.go

@@ -108,6 +108,9 @@ func TestChannelInitializeConditions(t *testing.T) {
 					}, {
 						Type:   ChannelConditionDeadLetterSinkResolved,
 						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   ChannelConditionEventPoliciesReady,
+						Status: corev1.ConditionUnknown,
 					}, {
 						Type:   ChannelConditionReady,
 						Status: corev1.ConditionUnknown,
@@ -139,6 +142,9 @@ func TestChannelInitializeConditions(t *testing.T) {
 					}, {
 						Type:   ChannelConditionDeadLetterSinkResolved,
 						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   ChannelConditionEventPoliciesReady,
+						Status: corev1.ConditionUnknown,
 					}, {
 						Type:   ChannelConditionReady,
 						Status: corev1.ConditionUnknown,
@@ -170,6 +176,9 @@ func TestChannelInitializeConditions(t *testing.T) {
 					}, {
 						Type:   ChannelConditionDeadLetterSinkResolved,
 						Status: corev1.ConditionUnknown,
+					}, {
+						Type:   ChannelConditionEventPoliciesReady,
+						Status: corev1.ConditionUnknown,
 					}, {
 						Type:   ChannelConditionReady,
 						Status: corev1.ConditionUnknown,
@@ -198,33 +207,45 @@ func TestChannelConditionStatus(t *testing.T) {
 		address              *duckv1.Addressable
 		backingChannelStatus corev1.ConditionStatus
 		DLSResolved          corev1.ConditionStatus
+		eventPolicyStatus    corev1.ConditionStatus
 		wantConditionStatus  corev1.ConditionStatus
 	}{{
 		name:                 "all happy",
 		address:              validAddress,
 		backingChannelStatus: corev1.ConditionTrue,
 		DLSResolved:          corev1.ConditionTrue,
+		eventPolicyStatus:    corev1.ConditionTrue,
 		wantConditionStatus:  corev1.ConditionTrue,
 	}, {
 		name:                 "address not set",
 		address:              &duckv1.Addressable{},
 		backingChannelStatus: corev1.ConditionTrue,
+		eventPolicyStatus:    corev1.ConditionTrue,
 		wantConditionStatus:  corev1.ConditionFalse,
 	},
 		{
 			name:                 "nil address",
 			address:              nil,
 			backingChannelStatus: corev1.ConditionTrue,
+			eventPolicyStatus:    corev1.ConditionTrue,
 			wantConditionStatus:  corev1.ConditionFalse,
 		}, {
 			name:                 "backing channel with unknown status",
 			address:              validAddress,
 			backingChannelStatus: corev1.ConditionUnknown,
+			eventPolicyStatus:    corev1.ConditionTrue,
 			wantConditionStatus:  corev1.ConditionUnknown,
 		}, {
 			name:                 "backing channel with false status",
 			address:              validAddress,
 			backingChannelStatus: corev1.ConditionFalse,
+			eventPolicyStatus:    corev1.ConditionTrue,
+			wantConditionStatus:  corev1.ConditionFalse,
+		}, {
+			name:                 "EventPolicies not Ready",
+			address:              validAddress,
+			backingChannelStatus: corev1.ConditionTrue,
+			eventPolicyStatus:    corev1.ConditionFalse,
 			wantConditionStatus:  corev1.ConditionFalse,
 		}}
 	for _, test := range tests {
@@ -240,6 +261,14 @@ func TestChannelConditionStatus(t *testing.T) {
 				cs.MarkBackingChannelUnknown("ChannelUnknown", "testing")
 			}
 
+			if test.eventPolicyStatus == corev1.ConditionTrue {
+				cs.MarkEventPoliciesTrue()
+			} else if test.eventPolicyStatus == corev1.ConditionFalse {
+				cs.MarkEventPoliciesFailed("EventPolicyFailure", "testing")
+			} else {
+				cs.MarkEventPoliciesUnknown("EventPolicyFailure", "testing")
+			}
+
 			if test.DLSResolved == corev1.ConditionTrue {
 				cs.MarkDeadLetterSinkResolvedSucceeded(v1.DeliveryStatus{})
 			}
@@ -421,6 +450,7 @@ func TestChannelPropagateStatuses(t *testing.T) {
 			cs := &ChannelStatus{}
 			cs.PropagateStatuses(tc.channelableStatus)
 			cs.MarkDeadLetterSinkNotConfigured()
+			cs.MarkEventPoliciesTrue()
 			got := cs.GetTopLevelCondition().Status
 			if tc.wantConditionStatus != got {
 				t.Errorf("unexpected readiness: want %v, got %v", tc.wantConditionStatus, got)

pkg/apis/messaging/v1/in_memory_channel_types.go

@@ -73,10 +73,6 @@ type InMemoryChannelSpec struct {
 type InMemoryChannelStatus struct {
 	// Channel conforms to Duck type ChannelableStatus.
 	eventingduckv1.ChannelableStatus `json:",inline"`
-
-	// AppliedEventPoliciesStatus contains the list of EventPolicies which apply to this Broker
-	// +optional
-	eventingduckv1.AppliedEventPoliciesStatus `json:",inline"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/auth/event_policy.go

@@ -20,6 +20,9 @@ import (
 	"fmt"
 	"strings"
 
+	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
+	"knative.dev/eventing/pkg/apis/feature"
+
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 
@@ -287,3 +290,52 @@ func EventPolicyEventHandler(indexer cache.Indexer, gk schema.GroupKind, enqueue
 		},
 	}
 }
+
+type EventPolicyStatusMarker interface {
+	MarkEventPoliciesFailed(reason, messageFormat string, messageA ...interface{})
+	MarkEventPoliciesUnknown(reason, messageFormat string, messageA ...interface{})
+	MarkEventPoliciesTrue()
+	MarkEventPoliciesTrueWithReason(reason, messageFormat string, messageA ...interface{})
+}
+
+func UpdateStatusWithEventPolicies(featureFlags feature.Flags, status *eventingduckv1.AppliedEventPoliciesStatus, statusMarker EventPolicyStatusMarker, eventPolicyLister listerseventingv1alpha1.EventPolicyLister, gvk schema.GroupVersionKind, objectMeta metav1.ObjectMeta) error {
+	status.Policies = nil
+
+	applyingEvenPolicies, err := GetEventPoliciesForResource(eventPolicyLister, gvk, objectMeta)
+	if err != nil {
+		statusMarker.MarkEventPoliciesFailed("EventPoliciesGetFailed", "Failed to get applying event policies")
+		return fmt.Errorf("unable to get applying event policies: %w", err)
+	}
+
+	if len(applyingEvenPolicies) > 0 {
+		unreadyEventPolicies := []string{}
+		for _, policy := range applyingEvenPolicies {
+			if !policy.Status.IsReady() {
+				unreadyEventPolicies = append(unreadyEventPolicies, policy.Name)
+			} else {
+				// only add Ready policies to the list
+				status.Policies = append(status.Policies, eventingduckv1.AppliedEventPolicyRef{
+					Name:       policy.Name,
+					APIVersion: v1alpha1.SchemeGroupVersion.String(),
+				})
+			}
+		}
+
+		if len(unreadyEventPolicies) == 0 {
+			statusMarker.MarkEventPoliciesTrue()
+		} else {
+			statusMarker.MarkEventPoliciesFailed("EventPoliciesNotReady", "event policies %s are not ready", strings.Join(unreadyEventPolicies, ", "))
+		}
+	} else {
+		// we have no applying event policy. So we set the EP condition to True
+		if featureFlags.IsOIDCAuthentication() {
+			// in case of OIDC auth, we also set the message with the default authorization mode
+			statusMarker.MarkEventPoliciesTrueWithReason("DefaultAuthorizationMode", "Default authz mode is %q", featureFlags[feature.AuthorizationDefaultMode])
+		} else {
+			// in case OIDC is disabled, we set EP condition to true too, but give some message that authz (EPs) require OIDC
+			statusMarker.MarkEventPoliciesTrueWithReason("OIDCDisabled", "Feature %q must be enabled to support Authorization", feature.OIDCAuthentication)
+		}
+	}
+
+	return nil
+}