Add e2e test for Broker authorization (#8132)

* Enable AuthZ tests

* Refactor eventpolicy test rekt resource for easier usage

* Add authorization e2e test for Broker

* Fix style issues

* Don't pass test env namespace and use default

* Run OIDC and AuthZ e2e tests together

test/e2e-rekt-tests.sh

@@ -44,10 +44,10 @@ kubectl apply -Rf "$(dirname "$0")/config-transport-encryption"
 
 go_test_e2e -timeout=1h ./test/rekt -run TLS || fail_test
 
-echo "Running E2E OIDC Reconciler Tests"
+echo "Running E2E Auth Reconciler Tests"
 
 kubectl apply -Rf "$(dirname "$0")/config-authentication-oidc"
 
-go_test_e2e -timeout=1h ./test/rekt -run OIDC || fail_test
+go_test_e2e -timeout=1h ./test/rekt -run "OIDC|AuthZ" || fail_test
 
 success

test/rekt/broker_test.go

@@ -284,3 +284,18 @@ func TestBrokerSendsEventsWithOIDCSupport(t *testing.T) {
 
 	env.TestSet(ctx, t, broker.BrokerSendEventWithOIDC())
 }
+
+func TestBrokerSupportsAuthZ(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+		eventshub.WithTLS(t),
+	)
+
+	env.TestSet(ctx, t, broker.BrokerSupportsAuthZ())
+}

test/rekt/features/broker/authz_feature.go

@@ -0,0 +1,157 @@
+/*
+Copyright 2024 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package broker
+
+import (
+	"context"
+
+	"github.com/cloudevents/sdk-go/v2/test"
+	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
+	"knative.dev/eventing/test/rekt/features/featureflags"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/eventpolicy"
+	"knative.dev/eventing/test/rekt/resources/pingsource"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/eventshub/assert"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/resources/service"
+)
+
+func BrokerSupportsAuthZ() *feature.FeatureSet {
+	return &feature.FeatureSet{
+		Name: "Broker supports authorization",
+		Features: []*feature.Feature{
+			BrokerAcceptsEventsFromAuthorizedSender(),
+			BrokerRejectsEventsFromUnauthorizedSender(),
+		},
+	}
+}
+
+func BrokerAcceptsEventsFromAuthorizedSender() *feature.Feature {
+	f := feature.NewFeatureNamed("Broker accepts events from a authorized sender")
+
+	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())
+
+	source := feature.MakeRandomK8sName("source")
+	brokerName := feature.MakeRandomK8sName("broker")
+	sink := feature.MakeRandomK8sName("sink")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	eventPolicyName := feature.MakeRandomK8sName("eventpolicy")
+
+	// Install the broker
+	f.Setup("Install Broker", broker.Install(brokerName, broker.WithEnvConfig()...))
+	f.Setup("Broker is ready", broker.IsReady(brokerName))
+	f.Setup("Broker is addressable", broker.IsAddressable(brokerName))
+
+	// Install the sink
+	f.Setup("Install Sink", eventshub.Install(
+		sink,
+		eventshub.StartReceiver,
+	))
+
+	f.Setup("Install the Trigger", trigger.Install(triggerName,
+		trigger.WithBrokerName(brokerName),
+		trigger.WithSubscriber(service.AsKReference(sink), "")))
+
+	f.Setup("Install the EventPolicy", eventpolicy.Install(
+		eventPolicyName,
+		eventpolicy.WithToRef(
+			broker.GVR().GroupVersion().WithKind("Broker"),
+			brokerName),
+		eventpolicy.WithFromRef(
+			pingsource.Gvr().GroupVersion().WithKind("PingSource"),
+			source,
+			""),
+	))
+
+	// Install source
+	f.Requirement("Install Pingsource", func(ctx context.Context, t feature.T) {
+		brokeruri, err := broker.Address(ctx, brokerName)
+		if err != nil {
+			t.Error("failed to get address of broker", err)
+		}
+
+		pingsource.Install(source,
+			pingsource.WithSink(&duckv1.Destination{URI: brokeruri.URL, CACerts: brokeruri.CACerts, Audience: brokeruri.Audience}),
+			pingsource.WithData("text/plain", "hello, world!"))(ctx, t)
+	})
+	f.Requirement("PingSource goes ready", pingsource.IsReady(source))
+
+	f.Alpha("Broker").
+		Must("accepts event from valid sender", assert.OnStore(sink).MatchEvent(
+			test.HasType(sourcesv1.PingSourceEventType)).AtLeast(1))
+
+	return f
+}
+
+func BrokerRejectsEventsFromUnauthorizedSender() *feature.Feature {
+	f := feature.NewFeatureNamed("Broker rejects events from an unauthorized sender")
+
+	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())
+
+	source := feature.MakeRandomK8sName("source")
+	brokerName := feature.MakeRandomK8sName("broker")
+	sink := feature.MakeRandomK8sName("sink")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	eventPolicyName := feature.MakeRandomK8sName("eventpolicy")
+
+	event := test.FullEvent()
+
+	// Install the broker
+	f.Setup("Install Broker", broker.Install(brokerName, broker.WithEnvConfig()...))
+	f.Setup("Broker is ready", broker.IsReady(brokerName))
+	f.Setup("Broker is addressable", broker.IsAddressable(brokerName))
+
+	// Install the sink
+	f.Setup("Install Sink", eventshub.Install(
+		sink,
+		eventshub.StartReceiver,
+	))
+
+	f.Setup("Install the Trigger", trigger.Install(triggerName,
+		trigger.WithBrokerName(brokerName),
+		trigger.WithSubscriber(service.AsKReference(sink), "")))
+	f.Setup("Trigger goes ready", trigger.IsReady(triggerName))
+
+	// Install an event policy for Broker allowing from a sample subject, to not fall back to the default-auth-mode
+	f.Setup("Install an EventPolicy", eventpolicy.Install(
+		eventPolicyName,
+		eventpolicy.WithToRef(
+			broker.GVR().GroupVersion().WithKind("Broker"),
+			brokerName),
+		eventpolicy.WithFromSubject("sample-sub")))
+
+	// Send event
+	f.Requirement("Install Source", eventshub.Install(
+		source,
+		eventshub.StartSenderToResourceTLS(broker.GVR(), brokerName, nil),
+		eventshub.InputEvent(event),
+	))
+
+	f.Alpha("Broker").
+		Must("event is sent", assert.OnStore(source).MatchSentEvent(
+			test.HasId(event.ID())).Exact(1)).
+		Must("broker rejects event with a 403 response", assert.OnStore(source).Match(assert.MatchStatusCode(403)).Exact(1))
+
+	return f
+}

test/rekt/resources/eventpolicy/eventpolicy.go

@@ -22,8 +22,6 @@ import (
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
-
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
@@ -52,69 +50,85 @@ func Install(name string, opts ...manifest.CfgFn) feature.StepFn {
 	}
 }
 
-func WithTo(tos ...eventingv1alpha1.EventPolicySpecTo) manifest.CfgFn {
+func WithToRef(gvk schema.GroupVersionKind, name string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["to"]; !set {
+			cfg["to"] = []map[string]interface{}{}
+		}
+
+		res := cfg["to"].([]map[string]interface{})
+
+		to := map[string]interface{}{
+			"ref": map[string]interface{}{
+				"apiVersion": gvk.GroupVersion().String(),
+				"kind":       gvk.Kind,
+				"name":       name,
+			}}
+
+		res = append(res, to)
+
+		cfg["to"] = res
+	}
+}
+
+func WithToSelector(gvk schema.GroupVersionKind, labelSelector *metav1.LabelSelector) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, set := cfg["to"]; !set {
 			cfg["to"] = []map[string]interface{}{}
 		}
 
 		res := cfg["to"].([]map[string]interface{})
-		for _, ref := range tos {
-			to := map[string]interface{}{}
-			if ref.Ref != nil {
-				to = map[string]interface{}{
-					"ref": map[string]interface{}{
-						"apiVersion": ref.Ref.APIVersion,
-						"kind":       ref.Ref.Kind,
-						"name":       ref.Ref.Name,
-					}}
-			}
-
-			if ref.Selector != nil {
-				selector := labelSelectorToStringMap(ref.Selector.LabelSelector)
-				selector["apiVersion"] = ref.Selector.APIVersion
-				selector["kind"] = ref.Selector.Kind
-
-				to = map[string]interface{}{
-					"selector": selector,
-				}
-			}
-
-			res = append(res, to)
+
+		selector := labelSelectorToStringMap(labelSelector)
+		selector["apiVersion"] = gvk.GroupVersion().String()
+		selector["kind"] = gvk.Kind
+
+		to := map[string]interface{}{
+			"selector": selector,
 		}
 
+		res = append(res, to)
+
 		cfg["to"] = res
 	}
 }
 
-func WithFrom(froms ...eventingv1alpha1.EventPolicySpecFrom) manifest.CfgFn {
+func WithFromRef(gvk schema.GroupVersionKind, name, namespace string) manifest.CfgFn {
+	return func(cfg map[string]interface{}) {
+		if _, set := cfg["from"]; !set {
+			cfg["from"] = []map[string]interface{}{}
+		}
+
+		res := cfg["from"].([]map[string]interface{})
+
+		from := map[string]interface{}{
+			"ref": map[string]interface{}{
+				"apiVersion": gvk.GroupVersion().String(),
+				"kind":       gvk.Kind,
+				"name":       name,
+				"namespace":  namespace,
+			}}
+
+		res = append(res, from)
+
+		cfg["from"] = res
+	}
+}
+
+func WithFromSubject(subject string) manifest.CfgFn {
 	return func(cfg map[string]interface{}) {
 		if _, set := cfg["from"]; !set {
 			cfg["from"] = []map[string]interface{}{}
 		}
 
 		res := cfg["from"].([]map[string]interface{})
-		for _, ref := range froms {
-			from := map[string]interface{}{}
-			if ref.Ref != nil {
-				from = map[string]interface{}{
-					"ref": map[string]interface{}{
-						"apiVersion": ref.Ref.APIVersion,
-						"kind":       ref.Ref.Kind,
-						"name":       ref.Ref.Name,
-						"namespace":  ref.Ref.Namespace,
-					}}
-			}
-
-			if ref.Sub != nil && *ref.Sub != "" {
-				from = map[string]interface{}{
-					"sub": *ref.Sub,
-				}
-			}
-
-			res = append(res, from)
+
+		from := map[string]interface{}{
+			"sub": subject,
 		}
 
+		res = append(res, from)
+
 		cfg["from"] = res
 	}
 }
@@ -126,7 +140,7 @@ func IsReady(name string, timing ...time.Duration) feature.StepFn {
 
 func labelSelectorToStringMap(selector *metav1.LabelSelector) map[string]interface{} {
 	if selector == nil {
-		return nil
+		return map[string]interface{}{}
 	}
 
 	r := map[string]interface{}{}