Make auth-proxy query EventPolicies dynamically

IntegrationSink was baking EventPolicies into the AUTH_POLICIES env var,
requiring deployment rollouts whenever policies changed. This caused test
failures because old pods with stale policies continued serving traffic
during RollingUpdate.

Change auth-proxy to query EventPolicies dynamically using a namespace-scoped
informer, similar to how Broker and Channel work. This eliminates deployment
rollouts when EventPolicies change.

- Add knative-eventing-eventpolicy-reader ClusterRole
- Create namespace-scoped EventPolicy informer in auth-proxy
- Add parent resource env vars to identify which resource to query policies for
- Create RoleBinding in sink's namespace for EventPolicy access
- Remove AUTH_POLICIES env var from deployment spec
- Add test coverage for OIDC-enabled deployments with RoleBindings

Author: creydr

cmd/auth_proxy/main.go

@@ -22,7 +22,6 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"net"
 	"os"
 
@@ -36,11 +35,16 @@ import (
 
 	"github.com/kelseyhightower/envconfig"
 	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/utils/ptr"
 	cmdbroker "knative.dev/eventing/cmd/broker"
 	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/eventing/pkg/auth"
+	eventingclient "knative.dev/eventing/pkg/client/clientset/versioned"
+	eventinginformers "knative.dev/eventing/pkg/client/informers/externalversions"
+	eventingv1alpha1listers "knative.dev/eventing/pkg/client/listers/eventing/v1alpha1"
 	"knative.dev/eventing/pkg/eventingtls"
 	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/pkg/apis"
@@ -69,7 +73,11 @@ type envConfig struct {
 	SinkNamespace string  `envconfig:"SINK_NAMESPACE" required:"true"`
 	SinkAudience  *string `envconfig:"SINK_AUDIENCE"`
 
-	AuthPolicies string `envconfig:"AUTH_POLICIES"`
+	// Parent resource information for dynamic EventPolicy lookup
+	ParentAPIVersion string `envconfig:"PARENT_API_VERSION" required:"true"`
+	ParentKind       string `envconfig:"PARENT_KIND" required:"true"`
+	ParentName       string `envconfig:"PARENT_NAME" required:"true"`
+	ParentNamespace  string `envconfig:"PARENT_NAMESPACE" required:"true"`
 
 	SinkTLSCertPath *string `envconfig:"SINK_TLS_CERT_FILE"`
 	SinkTLSKeyPath  *string `envconfig:"SINK_TLS_KEY_FILE"`
@@ -79,13 +87,14 @@ type envConfig struct {
 // ProxyHandler handles HTTP requests and performs authentication/authorization
 // before forwarding to the target service
 type ProxyHandler struct {
-	kubeClient   kubernetes.Interface
-	withContext  func(ctx context.Context) context.Context
-	authVerifier *auth.Verifier
-	httpProxy    *httputil.ReverseProxy
-	httpsProxy   *httputil.ReverseProxy
-	config       envConfig
-	authSubjects []auth.SubjectsWithFilters
+	kubeClient        kubernetes.Interface
+	withContext       func(ctx context.Context) context.Context
+	authVerifier      *auth.Verifier
+	httpProxy         *httputil.ReverseProxy
+	httpsProxy        *httputil.ReverseProxy
+	config            envConfig
+	eventPolicyLister eventingv1alpha1listers.EventPolicyLister
+	parentResourceGVK schema.GroupVersionKind
 }
 
 func main() {
@@ -104,7 +113,14 @@ func main() {
 
 	featureStore := setupFeatureStore(ctx, logger, configMapWatcher)
 
-	handler, err := createProxyHandler(ctx, config, logger, featureStore, configMapWatcher)
+	// Create namespace-scoped EventPolicy informer
+	eventPolicyLister, eventPolicyInformer, err := setupEventPolicyInformer(ctx, config, logger)
+	if err != nil {
+		logger.Fatalw("Failed to setup EventPolicy informer", zap.Error(err))
+	}
+	informers = append(informers, eventPolicyInformer)
+
+	handler, err := createProxyHandler(ctx, config, logger, featureStore, configMapWatcher, eventPolicyLister)
 	if err != nil {
 		logger.Fatalw("Failed to create proxy handler", zap.Error(err))
 	}
@@ -166,21 +182,45 @@ func setupFeatureStore(_ context.Context, logger *zap.SugaredLogger, configMapWa
 	return featureStore
 }
 
-// createProxyHandler creates and configures the proxy handler
-func createProxyHandler(ctx context.Context, config envConfig, logger *zap.SugaredLogger, featureStore *feature.Store, configMapWatcher *configmap.InformedWatcher) (*ProxyHandler, error) {
-	var authSubjects []auth.SubjectsWithFilters
+// setupEventPolicyInformer creates a namespace-scoped EventPolicy informer
+func setupEventPolicyInformer(ctx context.Context, config envConfig, logger *zap.SugaredLogger) (eventingv1alpha1listers.EventPolicyLister, controller.Informer, error) {
+	cfg := injection.GetConfig(ctx)
 
-	if len(config.AuthPolicies) > 0 {
-		if err := json.Unmarshal([]byte(config.AuthPolicies), &authSubjects); err != nil {
-			return nil, fmt.Errorf("failed to parse policies: %w", err)
-		}
+	// Create eventing client
+	eventingClient, err := eventingclient.NewForConfig(cfg)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create eventing client: %w", err)
 	}
 
+	// Create namespace-scoped informer factory
+	eventingInformerFactory := eventinginformers.NewSharedInformerFactoryWithOptions(
+		eventingClient,
+		0, // resyncPeriod - 0 means no resync
+		eventinginformers.WithNamespace(config.ParentNamespace),
+	)
+
+	eventPolicyInformer := eventingInformerFactory.Eventing().V1alpha1().EventPolicies()
+
+	logger.Infof("Created namespace-scoped EventPolicy informer for namespace %s", config.ParentNamespace)
+
+	return eventPolicyInformer.Lister(), eventPolicyInformer.Informer(), nil
+}
+
+// createProxyHandler creates and configures the proxy handler
+func createProxyHandler(ctx context.Context, config envConfig, logger *zap.SugaredLogger, featureStore *feature.Store, configMapWatcher *configmap.InformedWatcher, eventPolicyLister eventingv1alpha1listers.EventPolicyLister) (*ProxyHandler, error) {
+	// Parse parent resource GVK
+	gv, err := schema.ParseGroupVersion(config.ParentAPIVersion)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse parent API version %q: %w", config.ParentAPIVersion, err)
+	}
+	parentGVK := gv.WithKind(config.ParentKind)
+
 	handler := &ProxyHandler{
-		kubeClient:   kubeclient.Get(ctx),
-		authVerifier: auth.NewVerifier(ctx, nil, nil, configMapWatcher),
-		config:       config,
-		authSubjects: authSubjects,
+		kubeClient:        kubeclient.Get(ctx),
+		authVerifier:      auth.NewVerifier(ctx, nil, nil, configMapWatcher),
+		config:            config,
+		eventPolicyLister: eventPolicyLister,
+		parentResourceGVK: parentGVK,
 	}
 
 	handler.withContext = func(ctx context.Context) context.Context {
@@ -251,7 +291,15 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	logger.Debugf("Handling request to %s", r.RequestURI)
 
-	err := h.authVerifier.VerifyRequestFromSubjectsWithFilters(ctx, features, h.config.SinkAudience, h.authSubjects, h.config.SinkNamespace, r, w)
+	// Get EventPolicies dynamically for the parent resource
+	authSubjects, err := h.getAuthSubjects(logger)
+	if err != nil {
+		logger.Errorw("Failed to get EventPolicies", zap.Error(err))
+		http.Error(w, "Internal server error", http.StatusInternalServerError)
+		return
+	}
+
+	err = h.authVerifier.VerifyRequestFromSubjectsWithFilters(ctx, features, h.config.SinkAudience, authSubjects, h.config.SinkNamespace, r, w)
 	if err != nil {
 		logger.Debugw("Failed to verify AuthN and AuthZ", zap.Error(err))
 		return
@@ -266,6 +314,35 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// getAuthSubjects retrieves EventPolicies for the parent resource and converts them to SubjectsWithFilters
+func (h *ProxyHandler) getAuthSubjects(logger *zap.SugaredLogger) ([]auth.SubjectsWithFilters, error) {
+	// Get EventPolicies applying to this resource
+	policies, err := auth.GetEventPoliciesForResource(
+		h.eventPolicyLister,
+		h.parentResourceGVK,
+		metav1.ObjectMeta{
+			Name:      h.config.ParentName,
+			Namespace: h.config.ParentNamespace,
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get EventPolicies: %w", err)
+	}
+
+	logger.Debugf("Found %d EventPolicies for %s/%s", len(policies), h.config.ParentNamespace, h.config.ParentName)
+
+	// Convert EventPolicies to SubjectsWithFilters
+	subjectsWithFilters := make([]auth.SubjectsWithFilters, 0, len(policies))
+	for _, policy := range policies {
+		subjectsWithFilters = append(subjectsWithFilters, auth.SubjectsWithFilters{
+			Subjects: policy.Status.From,
+			Filters:  policy.Spec.Filters,
+		})
+	}
+
+	return subjectsWithFilters, nil
+}
+
 // httpReverseProxy creates a reverse proxy for HTTP traffic to the target service
 func httpReverseProxy(config envConfig) (*httputil.ReverseProxy, error) {
 	httpTarget := fmt.Sprintf("http://%s:%d", config.TargetHost, config.TargetHTTPPort)

config/core/roles/clusterrole-namespaced.yaml

@@ -115,3 +115,16 @@ rules:
   - apiGroups: ["eventing.knative.dev", "messaging.knative.dev", "sources.knative.dev", "flows.knative.dev", "bindings.knative.dev", "sinks.knative.dev"]
     resources: ["*"]
     verbs: ["get", "list", "watch"]
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: knative-eventing-eventpolicy-reader
+  labels:
+    rbac.authorization.k8s.io/aggregate-to-view: "true"
+    app.kubernetes.io/version: devel
+    app.kubernetes.io/name: knative-eventing
+rules:
+  - apiGroups: ["eventing.knative.dev"]
+    resources: ["eventpolicies"]
+    verbs: ["get", "list", "watch"]

pkg/reconciler/integration/sink/integrationsink.go

@@ -114,7 +114,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, sink *sinks.IntegrationS
 	}
 
 	logger.Debugw("Reconciling IntegrationSink auth-proxy RBAC")
-	_, err = r.reconcileAuthProxyRBAC(ctx, sink)
+	err = r.reconcileAuthProxyRBAC(ctx, sink)
 	if err != nil {
 		logging.FromContext(ctx).Errorw("Error reconciling auth-proxy RBAC", zap.Error(err))
 		return err
@@ -149,7 +149,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, sink *sinks.IntegrationS
 }
 
 func (r *Reconciler) reconcileDeployment(ctx context.Context, sink *sinks.IntegrationSink, authProxyImage string, featureFlags feature.Flags) (*v1.Deployment, error) {
-	expected, err := resources.MakeDeploymentSpec(sink, authProxyImage, featureFlags, r.trustBundleConfigMapLister, r.eventPolicyLister)
+	expected, err := resources.MakeDeploymentSpec(sink, authProxyImage, featureFlags, r.trustBundleConfigMapLister)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create deployment template: %w", err)
 	}
@@ -434,41 +434,103 @@ func (r *Reconciler) serviceChanged(existingSvc corev1.ServiceSpec, wantSvc core
 	return !equality.Semantic.DeepDerivative(wantSvc, existingSvc)
 }
 
-func (r *Reconciler) reconcileAuthProxyRBAC(ctx context.Context, sink *sinks.IntegrationSink) (*rbacv1.RoleBinding, error) {
+func (r *Reconciler) reconcileAuthProxyRBAC(ctx context.Context, sink *sinks.IntegrationSink) error {
 	features := feature.FromContext(ctx)
 
+	// 1. Reconcile EventPolicy access RBAC (RoleBinding in sink's namespace)
+	// This allows auth-proxy to read EventPolicies to enforce authorization
+	if err := r.reconcileEventPolicyRBAC(ctx, sink, features); err != nil {
+		return fmt.Errorf("failed to reconcile EventPolicy RBAC: %w", err)
+	}
+
+	// 2. Reconcile ConfigMap access RBAC (RoleBinding in knative-eventing namespace)
+	// This allows auth-proxy to read logging configuration from the knative-eventing namespace
+	err := r.reconcileConfigMapAccessRBAC(ctx, sink, features)
+	if err != nil {
+		return fmt.Errorf("failed to reconcile ConfigMap access RBAC: %w", err)
+	}
+
+	return nil
+}
+
+// reconcileEventPolicyRBAC reconciles the namespace-scoped RoleBinding
+// that allows auth-proxy to read EventPolicies in the sink's namespace.
+// The RoleBinding references the knative-eventing-eventpolicy-reader ClusterRole.
+func (r *Reconciler) reconcileEventPolicyRBAC(ctx context.Context, sink *sinks.IntegrationSink, features feature.Flags) error {
+	if !features.IsOIDCAuthentication() {
+		// EventPolicy RoleBinding has OwnerReferences, so it'll be garbage collected
+		// when the sink is deleted. No explicit cleanup needed when OIDC is disabled.
+		return nil
+	}
+
+	if err := r.reconcileEventPolicyRoleBinding(ctx, sink); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// reconcileConfigMapAccessRBAC reconciles the RoleBinding in knative-eventing namespace
+// that allows auth-proxy to read ConfigMaps (for logging configuration).
+func (r *Reconciler) reconcileConfigMapAccessRBAC(ctx context.Context, sink *sinks.IntegrationSink, features feature.Flags) error {
 	expected, err := resources.MakeAuthProxyRoleBindings(sink, r.integrationSinkLister, features)
 	if err != nil {
-		return nil, fmt.Errorf("creating auth proxy rolebinding: %w", err)
+		return fmt.Errorf("creating auth proxy rolebinding: %w", err)
 	}
 
 	if !features.IsOIDCAuthentication() {
-		return nil, r.deleteIntegrationSinkRBAC(ctx, expected)
+		return r.deleteIntegrationSinkRBAC(ctx, expected)
 	}
 
 	rolebinding, err := r.rolebindingLister.RoleBindings(expected.Namespace).Get(expected.Name)
 	if apierrors.IsNotFound(err) {
-		created, err := r.kubeClientSet.RbacV1().RoleBindings(system.Namespace()).Create(ctx, expected, metav1.CreateOptions{})
+		_, err = r.kubeClientSet.RbacV1().RoleBindings(system.Namespace()).Create(ctx, expected, metav1.CreateOptions{})
 		if err != nil {
-			return nil, err
+			return err
 		}
-
-		return created, nil
+		return nil
 	}
 	if err != nil {
-		return nil, fmt.Errorf("failed to get rolebinding %s/%s: %w", expected.GetNamespace(), expected.GetName(), err)
+		return fmt.Errorf("failed to get rolebinding %s/%s: %w", expected.GetNamespace(), expected.GetName(), err)
 	}
 
 	if !r.rolebindingChanged(rolebinding, expected) {
-		return rolebinding, nil
+		return nil
 	}
 
 	expected.ResourceVersion = rolebinding.ResourceVersion
-	updated, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, expected, metav1.UpdateOptions{})
+	_, err = r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, expected, metav1.UpdateOptions{})
 	if err != nil {
-		return nil, err
+		return err
 	}
-	return updated, nil
+	return nil
+}
+
+func (r *Reconciler) reconcileEventPolicyRoleBinding(ctx context.Context, sink *sinks.IntegrationSink) error {
+	expected := resources.MakeEventPolicyRoleBinding(sink)
+
+	rb, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Get(ctx, expected.Name, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) {
+		_, err := r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Create(ctx, expected, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("creating EventPolicy RoleBinding: %w", err)
+		}
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("getting EventPolicy RoleBinding: %w", err)
+	}
+
+	// Update if subjects changed
+	if !equality.Semantic.DeepEqual(rb.Subjects, expected.Subjects) {
+		rb.Subjects = expected.Subjects
+		_, err = r.kubeClientSet.RbacV1().RoleBindings(expected.Namespace).Update(ctx, rb, metav1.UpdateOptions{})
+		if err != nil {
+			return fmt.Errorf("updating EventPolicy RoleBinding: %w", err)
+		}
+	}
+
+	return nil
 }
 
 func (r *Reconciler) deleteIntegrationSinkRBAC(ctx context.Context, rolebinding *rbacv1.RoleBinding) error {
@@ -488,6 +550,10 @@ func (r *Reconciler) deleteIntegrationSinkRBAC(ctx context.Context, rolebinding
 		return fmt.Errorf("failed to delete auth-proxy rolebinding %s/%s: %w", rolebinding.Namespace, rolebinding.Name, err)
 	}
 
+	// Note: EventPolicy Role and RoleBinding have OwnerReferences set to the IntegrationSink,
+	// so they will be automatically garbage collected when the sink is deleted.
+	// We don't need to explicitly delete them here.
+
 	return nil
 }