test: added rekt test for eventpolicy with filters

Signed-off-by: Calum Murray <cmurray@redhat.com>

test/rekt/features/authz/addressable_authz_conformance.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"time"
 
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/test/rekt/resources/eventpolicy"
 	"knative.dev/eventing/test/rekt/resources/pingsource"
 	"knative.dev/reconciler-test/pkg/environment"
@@ -125,6 +126,73 @@ func addressableRejectsUnauthorizedRequest(gvr schema.GroupVersionResource, kind
 	return f
 }
 
+func addressableRespectsEventPolicyFilters(gvr schema.GroupVersionResource, kind, name string) *feature.Feature {
+	f := feature.NewFeatureNamed(fmt.Sprintf("%s only admits events that pass the event policy filter"))
+
+	f.Prerequisite("OIDC authentication is enabled", featureflags.AuthenticationOIDCEnabled())
+	f.Prerequisite("transport encryption is strict", featureflags.TransportEncryptionStrict())
+	f.Prerequisite("should not run when Istio is enabled", featureflags.IstioDisabled())
+
+	eventPolicy := feature.MakeRandomK8sName("eventpolicy")
+	source1 := feature.MakeRandomK8sName("source")
+	sourceSubject1 := feature.MakeRandomK8sName("source-oidc-identity")
+	source2 := feature.MakeRandomK8sName("source")
+	sourceSubject2 := feature.MakeRandomK8sName("source-oidc-identity")
+
+	event1 := test.FullEvent()
+	event1.SetType("valid.event.type")
+	event1.SetID("1")
+	event2 := test.FullEvent()
+	event2.SetType("invalid.event.type")
+	event2.SetID("2")
+
+	// Install event policy
+	f.Setup("Install the EventPolicy", func(ctx context.Context, t feature.T) {
+		namespace := environment.FromContext(ctx).Namespace()
+		eventpolicy.Install(
+			eventPolicy,
+			eventpolicy.WithToRef(
+				gvr.GroupVersion().WithKind(kind),
+				name),
+			eventpolicy.WithFromSubject(fmt.Sprintf("system:serviceaccount:%s:%s", namespace, sourceSubject1)),
+			eventpolicy.WithFromSubject(fmt.Sprintf("system:serviceaccount:%s:%s", namespace, sourceSubject2)),
+			eventpolicy.WithFilters([]eventingv1.SubscriptionsAPIFilter{
+				{
+					Prefix: map[string]string{
+						"type": "valid",
+					},
+				},
+			}),
+		)(ctx, t)
+	})
+	f.Setup(fmt.Sprintf("EventPolicy for %s %s is ready", kind, name), k8s.IsReady(gvr, name))
+
+	// Install source
+	f.Requirement("install source 1", eventshub.Install(
+		source1,
+		eventshub.StartSenderToResourceTLS(gvr, name, nil),
+		eventshub.InputEvent(event1),
+		eventshub.OIDCSubject(sourceSubject1),
+	))
+
+	f.Requirement("install source 2", eventshub.Install(
+		source2,
+		eventshub.StartSenderToResourceTLS(gvr, name, nil),
+		eventshub.InputEvent(event2),
+		eventshub.OIDCSubject(sourceSubject2),
+	))
+
+	f.Alpha(kind).
+		Must("valid event sent", eventassert.OnStore(source1).MatchSentEvent(test.HasId(event1.ID())).Exact(1)).
+		Must("get 202 on response", eventassert.OnStore(source1).Match(eventassert.MatchStatusCode(202)).AtLeast(1))
+
+	f.Alpha(kind).
+		Must("invalid event sent", eventassert.OnStore(source2).MatchSentEvent(test.HasId(event2.ID())).Exact(1)).
+		Must("get 403 on response", eventassert.OnStore(source2).Match(eventassert.MatchStatusCode(403)).AtLeast(1))
+
+	return f
+}
+
 func addressableBecomesUnreadyOnUnreadyEventPolicy(gvr schema.GroupVersionResource, kind, name string) *feature.Feature {
 	f := feature.NewFeatureNamed(fmt.Sprintf("%s becomes NotReady when EventPolicy is NotReady", kind))
 

test/rekt/resources/eventpolicy/eventpolicy.go

@@ -19,17 +19,22 @@ package eventpolicy
 import (
 	"context"
 	"embed"
+	"encoding/json"
+	"strings"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/manifest"
+	"sig.k8s.io/yaml"
+
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 )
 
 //go:embed *.yaml
-var yaml embed.FS
+var yamlEmbed embed.FS
 
 func GVR() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1alpha1", Resource: "eventpolicies"}
@@ -44,7 +49,7 @@ func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 		fn(cfg)
 	}
 	return func(ctx context.Context, t feature.T) {
-		if _, err := manifest.InstallYamlFS(ctx, yaml, cfg); err != nil {
+		if _, err := manifest.InstallYamlFS(ctx, yamlEmbed, cfg); err != nil {
 			t.Fatal(err)
 		}
 	}
@@ -133,6 +138,30 @@ func WithFromSubject(subject string) manifest.CfgFn {
 	}
 }
 
+func WithFilters(filters []eventingv1.SubscriptionsAPIFilter) manifest.CfgFn {
+	jsonBytes, err := json.Marshal(filters)
+	if err != nil {
+		panic(err)
+	}
+
+	yamlBytes, err := yaml.JSONToYaml(jsonBytes)
+	if err != nil {
+		panic(err)
+	}
+
+	filtersYaml := string(yamlBytes)
+
+	lines := strings.Split(filtersYaml, "\n")
+	out := make([]string, 0, len(lines))
+	for i := range lines {
+		out = append(out, "    "+lines[i])
+	}
+
+	return func(m map[string]interface{}) {
+		m["filters"] = strings.Join(out, "\n")
+	}
+}
+
 // IsReady tests to see if an EventPolicy becomes ready within the time given.
 func IsReady(name string, timing ...time.Duration) feature.StepFn {
 	return k8s.IsReady(GVR(), name, timing...)

test/rekt/resources/eventpolicy/eventpolicy.yaml

@@ -67,3 +67,8 @@ spec:
     - sub: {{ $from.sub }}
     {{ end }}
   {{ end }}
+
+  {{ if .filters }}
+  filters:
+{{ .filters }}
+  {{ .end }}