Add authorization conformance tests for Sequence and Parallel (#8400)

* Add authorization conformance tests for Sequence and Parallel

* Use fully functional Sequence and Parallel for Authz conformance tests

* Check Sequence addressable

test/rekt/features/parallel/oidc_feature.go

@@ -28,6 +28,7 @@ import (
 	"knative.dev/eventing/pkg/reconciler/parallel/resources"
 	"knative.dev/eventing/test/rekt/features/featureflags"
 	"knative.dev/eventing/test/rekt/resources/addressable"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/eventing/test/rekt/resources/channel_template"
 	"knative.dev/eventing/test/rekt/resources/parallel"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
@@ -177,3 +178,96 @@ func ParallelHasAudienceOfInputChannel(parallelName, parallelNamespace string, c
 
 	return f
 }
+
+func ParallelWithOIDCAudienceForSteps(name string) *feature.Feature {
+	f := feature.NewFeatureNamed("Parallel with OIDC audience for steps")
+
+	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())
+
+	channelTemplate := channel_template.ChannelTemplate{
+		TypeMeta: channel_impl.TypeMeta(),
+		Spec:     map[string]interface{}{},
+	}
+
+	sink := feature.MakeRandomK8sName("sink1")
+	sinkAudience := "sinkAud"
+	subscriber1Audience := "subscriber1Aud"
+	subscriber2Audience := "subscriber2Aud"
+	filter1Audience := "filter1Aud"
+
+	eventBody := `{"msg":"test msg"}`
+	event := test.FullEvent()
+	_ = event.SetData(cloudeventsv2.ApplicationJSON, []byte(eventBody))
+
+	// Construct two branches
+	branch1Num := 0
+	branch2Num := 1
+	subscriber1 := feature.MakeRandomK8sName("subscriber" + strconv.Itoa(branch1Num))
+	subscriber2 := feature.MakeRandomK8sName("subscriber" + strconv.Itoa(branch2Num))
+	filter1 := feature.MakeRandomK8sName("filter" + strconv.Itoa(branch1Num))
+
+	f.Setup("install sink", eventshub.Install(sink,
+		eventshub.OIDCReceiverAudience(sinkAudience),
+		eventshub.StartReceiverTLS))
+
+	// Install Subscribers for both branches.
+	f.Setup("install subscriber1", eventshub.Install(subscriber1,
+		eventshub.ReplyWithAppendedData("appended data 1"),
+		eventshub.OIDCReceiverAudience(subscriber1Audience),
+		eventshub.StartReceiverTLS))
+	f.Setup("install subscriber2", eventshub.Install(subscriber2,
+		eventshub.ReplyWithAppendedData("appended data 2"),
+		eventshub.OIDCReceiverAudience(subscriber2Audience),
+		eventshub.StartReceiverTLS))
+
+	// Install Filter only for first branch.
+	f.Setup("install filter1", eventshub.Install(filter1,
+		eventshub.ReplyWithTransformedEvent(event.Type(), event.Source(), string(event.Data())),
+		eventshub.OIDCReceiverAudience(filter1Audience),
+		eventshub.StartReceiverTLS))
+
+	// Install a Parallel with two branches
+	f.Setup("install Parallel", func(ctx context.Context, t feature.T) {
+		cfg := []manifest.CfgFn{
+			parallel.WithChannelTemplate(channelTemplate),
+			parallel.WithReply(&duckv1.Destination{
+				Ref:      service.AsKReference(sink),
+				Audience: &sinkAudience,
+				CACerts:  eventshub.GetCaCerts(ctx),
+			}),
+			parallel.WithSubscriberAt(branch1Num, &duckv1.Destination{
+				Ref:      service.AsKReference(subscriber1),
+				Audience: &subscriber1Audience,
+				CACerts:  eventshub.GetCaCerts(ctx),
+			}),
+			parallel.WithSubscriberAt(branch2Num, &duckv1.Destination{
+				Ref:      service.AsKReference(subscriber2),
+				Audience: &subscriber2Audience,
+				CACerts:  eventshub.GetCaCerts(ctx),
+			}),
+			parallel.WithFilterAt(branch1Num, &duckv1.Destination{
+				Ref:      service.AsKReference(filter1),
+				Audience: &filter1Audience,
+				CACerts:  eventshub.GetCaCerts(ctx),
+			}),
+			parallel.WithReplyAt(branch1Num, nil),
+			parallel.WithReplyAt(branch2Num, nil),
+
+			// The Reply for second branch is same as global reply.
+			parallel.WithReplyAt(branch2Num, &duckv1.Destination{
+				Ref:      service.AsKReference(sink),
+				Audience: &sinkAudience,
+				CACerts:  eventshub.GetCaCerts(ctx),
+			}),
+		}
+
+		parallel.Install(name, cfg...)(ctx, t)
+	})
+
+	f.Setup("Parallel goes ready", parallel.IsReady(name))
+	f.Setup("Parallel is addressable", parallel.IsAddressable(name))
+
+	return f
+}

test/rekt/features/sequence/feature.go

@@ -72,6 +72,7 @@ func SequenceTest(channelTemplate channel_template.ChannelTemplate) *feature.Fea
 	// Install a Sequence with three steps
 	f.Setup("install Sequence", sequence.Install(sequenceName, cfg...))
 	f.Setup("Sequence goes ready", sequence.IsReady(sequenceName))
+	f.Setup("Sequence is addressable", sequence.IsAddressable(sequenceName))
 
 	eventBody := fmt.Sprintf("TestSequence %s", uuid.New().String())
 	// Install PingSource point to sequence Address with eventBody

test/rekt/features/sequence/oidc_feature.go

@@ -116,6 +116,7 @@ func SequenceSendsEventWithOIDCTokenToSteps() *feature.Feature {
 	})
 
 	f.Setup("Sequence goes ready", sequence.IsReady(sequenceName))
+	f.Setup("Sequence is addressable", sequence.IsAddressable(sequenceName))
 
 	event := test.FullEvent()
 	event.SetData("text/plain", "hello")
@@ -198,6 +199,7 @@ func SequenceSendsEventWithOIDCTokenToReply() *feature.Feature {
 		sequence.Install(sequenceName, cfg...)(ctx, t)
 	})
 	f.Setup("Sequence goes ready", sequence.IsReady(sequenceName))
+	f.Setup("Sequence is addressable", sequence.IsAddressable(sequenceName))
 
 	event := test.FullEvent()
 	event.SetData("text/plain", "hello")
@@ -219,3 +221,52 @@ func SequenceSendsEventWithOIDCTokenToReply() *feature.Feature {
 
 	return f
 }
+
+func SequenceWithOIDCAudienceForSteps(name string) *feature.Feature {
+	f := feature.NewFeatureNamed("Sequence with OIDC audience for steps")
+
+	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())
+
+	channelTemplate := channel_template.ChannelTemplate{
+		TypeMeta: channel_impl.TypeMeta(),
+		Spec:     map[string]interface{}{},
+	}
+
+	step1Name := feature.MakeRandomK8sName("step1")
+	step2Name := feature.MakeRandomK8sName("step2")
+
+	step1Audience := "step1-aud"
+	step2Audience := "step2-aud"
+
+	f.Setup("install step 1", eventshub.Install(step1Name,
+		eventshub.OIDCReceiverAudience(step1Audience),
+		eventshub.StartReceiverTLS))
+	f.Setup("install step 2", eventshub.Install(step2Name,
+		eventshub.OIDCReceiverAudience(step2Audience),
+		eventshub.StartReceiverTLS))
+
+	f.Setup("Install Sequence", func(ctx context.Context, t feature.T) {
+		cfg := []manifest.CfgFn{
+			sequence.WithChannelTemplate(channelTemplate),
+			sequence.WithStepFromDestination(&duckv1.Destination{
+				Ref:      service.AsKReference(step1Name),
+				Audience: &step1Audience,
+				CACerts:  eventshub.GetCaCerts(ctx),
+			}),
+			sequence.WithStepFromDestination(&duckv1.Destination{
+				Ref:      service.AsKReference(step2Name),
+				Audience: &step2Audience,
+				CACerts:  eventshub.GetCaCerts(ctx),
+			}),
+		}
+
+		sequence.Install(name, cfg...)(ctx, t)
+	})
+
+	f.Setup("Sequence goes ready", sequence.IsReady(name))
+	f.Setup("Sequence is addressable", sequence.IsAddressable(name))
+
+	return f
+}

test/rekt/parallel_test.go

@@ -22,7 +22,6 @@ package rekt
 import (
 	"testing"
 
-	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/reconciler-test/pkg/feature"
 
 	"knative.dev/pkg/system"
@@ -31,7 +30,9 @@ import (
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
+	"knative.dev/eventing/test/rekt/features/authz"
 	"knative.dev/eventing/test/rekt/features/parallel"
+	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/eventing/test/rekt/resources/channel_template"
 	parallelresources "knative.dev/eventing/test/rekt/resources/parallel"
 )
@@ -102,3 +103,21 @@ func TestParallelTwoBranchesWithOIDC(t *testing.T) {
 
 	env.Test(ctx, t, parallel.ParallelWithTwoBranchesOIDC(channel_template.ImmemoryChannelTemplate()))
 }
+
+func TestParallelSupportsAuthZ(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),
+	)
+
+	name := feature.MakeRandomK8sName("parallel")
+	env.Prerequisite(ctx, t, parallel.ParallelWithOIDCAudienceForSteps(name))
+
+	env.TestSet(ctx, t, authz.AddressableAuthZConformance(parallelresources.GVR(), "Parallel", name))
+}

test/rekt/sequence_test.go

@@ -24,6 +24,7 @@ import (
 
 	"knative.dev/reconciler-test/pkg/feature"
 
+	"knative.dev/eventing/test/rekt/features/authz"
 	"knative.dev/eventing/test/rekt/features/sequence"
 	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/eventing/test/rekt/resources/channel_template"
@@ -99,3 +100,22 @@ func TestSequenceSendsEventsOIDC(t *testing.T) {
 
 	env.TestSet(ctx, t, sequence.SequenceSendsEventWithOIDC())
 }
+
+func TestSequenceSupportsAuthZ(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),
+	)
+
+	name := feature.MakeRandomK8sName("sequence")
+
+	env.Prerequisite(ctx, t, sequence.SequenceWithOIDCAudienceForSteps(name))
+
+	env.TestSet(ctx, t, authz.AddressableAuthZConformance(sequenceresources.GVR(), "Sequence", name))
+}