Add filters to event policy

config/core/resources/eventpolicy.yaml

@@ -110,6 +110,44 @@ spec:
                           description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed.
                           type: object
                           x-kubernetes-preserve-unknown-fields: true
+              filters:
+                description: 'Filters is an array of SubscriptionsAPIFilters that evaluate to true or false. If any filter expression in the array evaluates to false, the event will not continue pass the ingress of the target resources of the policy'
+                type: array
+                items:
+                  type: object
+                  properties:
+                    all:
+                      description: 'All evaluates to true if all the nested expressions evaluate to true. It must contain at least one filter expression'
+                      type: array
+                      items:
+                        type: object
+                        x-kubernetes-preserve-unknown-fields: true
+                    any:
+                      description: 'Any evaluates to true if any of the nested expressions evaluate to true. It must contain at least one filter expression'
+                      type: array
+                      items:
+                        type: object
+                        x-kubernetes-preserve-unknown-fields: true
+                    cesql:
+                      description: 'CESQL is a CloudEvents SQL v1 expression that will evaluate to true or false for each CloudEvent.'
+                      type: string
+                    exact:
+                      description: 'Exact evaluates to true if the values of the matching CloudEvents attributes all exactly match with the associated value string specified (case sensitive)'
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                    not:
+                      description: 'Not evaluates to true if the nested expression evaluates to false.'
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                    prefix:
+                      description: 'Prefix evaluates to true if the values of the matching CloudEvents attributes all start with the associated value string specified (case sensitive)'
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                    suffix:
+                      description: 'Exact evaluates to true if the values of the matching CloudEvents attributes all end with the associated value string specified (case sensitive)'
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+
           status:
             description: Status represents the current state of the EventPolicy. This data may be out of date.
             type: object

docs/eventing-api.md

@@ -2254,7 +2254,7 @@ AppliedEventPoliciesStatus
 <h3 id="eventing.knative.dev/v1.SubscriptionsAPIFilter">SubscriptionsAPIFilter
 </h3>
 <p>
-(<em>Appears on:</em><a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">SubscriptionsAPIFilter</a>, <a href="#eventing.knative.dev/v1.TriggerSpec">TriggerSpec</a>, <a href="#sources.knative.dev/v1.ApiServerSourceSpec">ApiServerSourceSpec</a>)
+(<em>Appears on:</em><a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">SubscriptionsAPIFilter</a>, <a href="#eventing.knative.dev/v1.TriggerSpec">TriggerSpec</a>, <a href="#eventing.knative.dev/v1alpha1.EventPolicySpec">EventPolicySpec</a>, <a href="#sources.knative.dev/v1.ApiServerSourceSpec">ApiServerSourceSpec</a>)
 </p>
 <p>
 <p>SubscriptionsAPIFilter allows defining a filter expression using CloudEvents
@@ -2735,6 +2735,24 @@ An empty list means it applies to all resources in the EventPolicies namespace</
 <p>From is the list of sources or oidc identities, which are allowed to send events to the targets (.spec.to).</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>filters</code><br/>
+<em>
+<a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">
+[]SubscriptionsAPIFilter
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Filters is the list of SubscriptoinsApi filters which determine whether or not the event is accepted.
+It is an array of filter expressions that evaluate to true or false.
+If any filter expression in the array evaluates to false, the event will not
+pass the target resource&rsquo;s ingress. Absence of any filters implies that the filters
+always evaluate to true.</p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -2898,6 +2916,24 @@ An empty list means it applies to all resources in the EventPolicies namespace</
 <p>From is the list of sources or oidc identities, which are allowed to send events to the targets (.spec.to).</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>filters</code><br/>
+<em>
+<a href="#eventing.knative.dev/v1.SubscriptionsAPIFilter">
+[]SubscriptionsAPIFilter
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Filters is the list of SubscriptoinsApi filters which determine whether or not the event is accepted.
+It is an array of filter expressions that evaluate to true or false.
+If any filter expression in the array evaluates to false, the event will not
+pass the target resource&rsquo;s ingress. Absence of any filters implies that the filters
+always evaluate to true.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="eventing.knative.dev/v1alpha1.EventPolicySpecFrom">EventPolicySpecFrom

pkg/adapter/apiserver/adapter.go

@@ -34,7 +34,6 @@ import (
 
 	"knative.dev/eventing/pkg/adapter/v2"
 	v1 "knative.dev/eventing/pkg/apis/sources/v1"
-	brokerfilter "knative.dev/eventing/pkg/broker/filter"
 	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
 )
 
@@ -73,7 +72,7 @@ func (a *apiServerAdapter) start(ctx context.Context, stopCh <-chan struct{}) er
 		logger:              a.logger,
 		ref:                 a.config.EventMode == v1.ReferenceMode,
 		apiServerSourceName: a.name,
-		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(a.logger.Desugar(), a.config.Filters)...),
+		filter:              subscriptionsapi.NewAllFilter(subscriptionsapi.MaterializeFiltersList(a.logger.Desugar(), a.config.Filters)...),
 	}
 	if a.config.ResourceOwner != nil {
 		a.logger.Infow("will be filtered",

pkg/adapter/apiserver/adapter_test.go

@@ -35,7 +35,6 @@ import (
 	kubetesting "k8s.io/client-go/testing"
 	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
-	brokerfilter "knative.dev/eventing/pkg/broker/filter"
 	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
 	rectesting "knative.dev/eventing/pkg/reconciler/testing"
 	"knative.dev/pkg/logging"
@@ -299,7 +298,7 @@ func makeResourceAndTestingClient() (*resourceDelegate, *adaptertest.TestCloudEv
 		source:              "unit-test",
 		apiServerSourceName: apiServerSourceNameTest,
 		logger:              logger,
-		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), []eventingv1.SubscriptionsAPIFilter{})...),
+		filter:              subscriptionsapi.NewAllFilter(subscriptionsapi.MaterializeFiltersList(logger.Desugar(), []eventingv1.SubscriptionsAPIFilter{})...),
 	}, ce
 }
 
@@ -313,6 +312,6 @@ func makeRefAndTestingClient() (*resourceDelegate, *adaptertest.TestCloudEventsC
 		apiServerSourceName: apiServerSourceNameTest,
 		logger:              zap.NewExample().Sugar(),
 		ref:                 true,
-		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), []eventingv1.SubscriptionsAPIFilter{})...),
+		filter:              subscriptionsapi.NewAllFilter(subscriptionsapi.MaterializeFiltersList(logger.Desugar(), []eventingv1.SubscriptionsAPIFilter{})...),
 	}, ce
 }

pkg/adapter/apiserver/delegate_test.go

@@ -22,7 +22,6 @@ import (
 	adaptertest "knative.dev/eventing/pkg/adapter/v2/test"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/sources"
-	brokerfilter "knative.dev/eventing/pkg/broker/filter"
 	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
 )
 
@@ -87,7 +86,7 @@ func TestFilterFails(t *testing.T) {
 		source:              "unit-test",
 		apiServerSourceName: apiServerSourceNameTest,
 		logger:              logger,
-		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), filters)...),
+		filter:              subscriptionsapi.NewAllFilter(subscriptionsapi.MaterializeFiltersList(logger.Desugar(), filters)...),
 	}
 
 	delegate.Update(simplePod("unit", "test"))
@@ -104,7 +103,7 @@ func TestEmptyFiltersList(t *testing.T) {
 		source:              "unit-test",
 		apiServerSourceName: apiServerSourceNameTest,
 		logger:              logger,
-		filter:              subscriptionsapi.NewAllFilter(brokerfilter.MaterializeFiltersList(logger.Desugar(), filters)...),
+		filter:              subscriptionsapi.NewAllFilter(subscriptionsapi.MaterializeFiltersList(logger.Desugar(), filters)...),
 	}
 
 	delegate.Update(simplePod("unit", "test"))

pkg/apis/eventing/v1alpha1/eventpolicy_types.go

@@ -20,9 +20,12 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
+
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 )
 
 // +genclient
@@ -71,6 +74,14 @@ type EventPolicySpec struct {
 
 	// From is the list of sources or oidc identities, which are allowed to send events to the targets (.spec.to).
 	From []EventPolicySpecFrom `json:"from,omitempty"`
+
+	// Filters is the list of SubscriptoinsApi filters which determine whether or not the event is accepted.
+	// It is an array of filter expressions that evaluate to true or false.
+	// If any filter expression in the array evaluates to false, the event will not
+	// pass the target resource's ingress. Absence of any filters implies that the filters
+	// always evaluate to true.
+	// +optional
+	Filters []eventingv1.SubscriptionsAPIFilter `json:"filters,omitempty"`
 }
 
 type EventPolicySpecTo struct {

pkg/apis/eventing/v1alpha1/eventpolicy_validation.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"strings"
 
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 )
@@ -60,6 +61,8 @@ func (ets *EventPolicySpec) Validate(ctx context.Context) *apis.FieldError {
 		}
 	}
 
+	err = err.Also(eventingv1.ValidateSubscriptionAPIFiltersList(ctx, ets.Filters).ViaField("filters"))
+
 	return err
 }
 

pkg/apis/eventing/v1alpha1/eventpolicy_validation_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/ptr"
@@ -294,6 +295,45 @@ func TestEventPolicySpecValidationWithOIDCAuthenticationFeatureFlagEnabled(t *te
 				return nil
 			}(),
 		},
+		{
+			name: "valid, from.sub exactly '*', valid filters",
+			ep: &EventPolicy{
+				Spec: EventPolicySpec{
+					From: []EventPolicySpecFrom{{
+						Sub: ptr.String("*"),
+					}},
+					Filters: []eventingv1.SubscriptionsAPIFilter{
+						{
+							Prefix: map[string]string{"type": "example"},
+						},
+					},
+				},
+			},
+			want: func() *apis.FieldError {
+				return nil
+			}(),
+		},
+		{
+			name: "invalid, from.sub exactly '*', invalid cesql filter",
+			ep: &EventPolicy{
+				Spec: EventPolicySpec{
+					From: []EventPolicySpecFrom{{
+						Sub: ptr.String("*"),
+					}},
+					Filters: []eventingv1.SubscriptionsAPIFilter{
+						{
+							CESQL: "type LIKE id",
+						},
+					},
+				},
+			},
+			want: func() *apis.FieldError {
+
+				return apis.ErrInvalidValue("type LIKE id", "cesql", "parse error: syntax error: |failed to parse LIKE expression: the pattern was not a string literal").
+					ViaFieldIndex("filters", 0).
+					ViaField("spec")
+			}(),
+		},
 	}
 
 	for _, test := range tests {

pkg/auth/event_policy.go

@@ -17,10 +17,16 @@ limitations under the License.
 package auth
 
 import (
+	"context"
 	"fmt"
 	"sort"
 	"strings"
 
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"go.uber.org/zap"
+	"knative.dev/eventing/pkg/eventfilter"
+	"knative.dev/eventing/pkg/eventfilter/subscriptionsapi"
+
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/apis/feature"
 
@@ -210,17 +216,19 @@ func resolveSubjectsFromReference(resolver *resolver.AuthenticatableResolver, re
 	return objFullSANames, nil
 }
 
-// SubjectContained checks if the given sub is contained in the list of allowedSubs
-// or if it matches a prefix pattern in subs (e.g. system:serviceaccounts:my-ns:*)
-func SubjectContained(sub string, allowedSubs []string) bool {
-	for _, s := range allowedSubs {
-		if strings.EqualFold(s, sub) {
-			return true
-		}
+// SubjectAndFiltersPass checks if the given sub is contained in the list of allowedSubs
+// or if it matches a prefix pattern in subs (e.g. system:serviceaccounts:my-ns:*), as
+// well as if the event passes any filters associated with the subjects for an event policy
+func SubjectAndFiltersPass(ctx context.Context, sub string, allowedSubsWithFilters []subjectsWithFilters, event *cloudevents.Event, logger *zap.SugaredLogger) bool {
+	if event == nil {
+		return false
+	}
 
-		if strings.HasSuffix(s, "*") &&
-			strings.HasPrefix(sub, strings.TrimSuffix(s, "*")) {
-			return true
+	for _, swf := range allowedSubsWithFilters {
+		for _, s := range swf.subjects {
+			if strings.EqualFold(s, sub) || (strings.HasSuffix(s, "*") && strings.HasPrefix(sub, strings.TrimSuffix(s, "*"))) {
+				return subscriptionsapi.CreateSubscriptionsAPIFilters(logger.Desugar(), swf.filters).Filter(ctx, *event) != eventfilter.FailFilter
+			}
 		}
 	}