fix(apiserver): probe SA permissions before loading admission-policy plugins

pkg/apiserver/apiserver.go

@@ -20,6 +20,7 @@ import (
 	kcclient "carvel.dev/kapp-controller/pkg/client/clientset/versioned"
 	"github.com/carvel-dev/semver/v4"
 	"github.com/go-logr/logr"
+	authorizationv1 "k8s.io/api/authorization/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -151,7 +152,7 @@ func NewAPIServer(clientConfig *rest.Config, coreClient kubernetes.Interface, kc
 		return nil, fmt.Errorf("building aggregation client: %v", err)
 	}
 
-	config, caContentProvider, err := newServerConfig(aggClient, opts)
+	config, caContentProvider, err := newServerConfig(aggClient, coreClient, opts)
 	if err != nil {
 		return nil, err
 	}
@@ -235,7 +236,7 @@ func (as *APIServer) isReady() (bool, error) {
 	return false, nil
 }
 
-func newServerConfig(aggClient aggregatorclient.Interface, opts NewAPIServerOpts) (*genericapiserver.RecommendedConfig, *dynamiccertificates.DynamicFileCAContent, error) {
+func newServerConfig(aggClient aggregatorclient.Interface, coreClient kubernetes.Interface, opts NewAPIServerOpts) (*genericapiserver.RecommendedConfig, *dynamiccertificates.DynamicFileCAContent, error) {
 
 	recommendedOptions := genericoptions.NewRecommendedOptions("", Codecs.LegacyCodec(v1alpha1.SchemeGroupVersion))
 	recommendedOptions.Etcd = nil
@@ -281,16 +282,25 @@ func newServerConfig(aggClient aggregatorclient.Interface, opts NewAPIServerOpts
 		recommendedOptions.Features.EnablePriorityAndFairness = false
 	}
 
-	// Always explicitly set the admission plugin order. Only include admission policy
-	// plugins if their backing API resources actually exist in the cluster — some plugins
-	// (MutatingAdmissionPolicy) require a feature gate that is not enabled by default even
-	// on clusters that meet the minimum version, and will spam watch errors if loaded
-	// against a cluster where the resource is absent.
+	// Always explicitly set the admission plugin order. Only include an admission
+	// policy plugin if BOTH of these hold:
+	//   - Its backing API resource is served by the cluster (some plugins are
+	//     gated by feature flags — e.g. MutatingAdmissionPolicy defaults on only
+	//     from k8s 1.36 — and their resources are absent on older clusters).
+	//   - The ServiceAccount kapp-controller runs under can list+watch that
+	//     resource. If discovery says the resource exists but our RBAC does not
+	//     grant list+watch, loading the plugin causes its SharedInformer to
+	//     never sync. WaitForReady() then never returns true and every mutating
+	//     admission decision fails with "not yet ready to handle request
+	//     (reason: Forbidden)" — a very hard failure mode to diagnose because
+	//     the APIService still reports Available=True. Probing RBAC up-front
+	//     lets us skip the plugin cleanly and log an actionable warning
+	//     instead.
 	pluginOrder := []string{lifecycle.PluginName, mutatingwebhook.PluginName, validatingwebhook.PluginName}
-	if serverResourceExists(aggClient.Discovery(), "admissionregistration.k8s.io/v1", "validatingadmissionpolicies") {
+	if canUseAdmissionPolicyPlugin(aggClient.Discovery(), coreClient, "validatingadmissionpolicies", "validatingadmissionpolicybindings", opts.Logger) {
 		pluginOrder = append(pluginOrder, validatingadmissionpolicy.PluginName)
 	}
-	if serverResourceExists(aggClient.Discovery(), "admissionregistration.k8s.io/v1", "mutatingadmissionpolicies") {
+	if canUseAdmissionPolicyPlugin(aggClient.Discovery(), coreClient, "mutatingadmissionpolicies", "mutatingadmissionpolicybindings", opts.Logger) {
 		pluginOrder = append(pluginOrder, mutatingadmissionpolicy.PluginName)
 	}
 	recommendedOptions.Admission.RecommendedPluginOrder = pluginOrder
@@ -323,6 +333,79 @@ func newServerConfig(aggClient aggregatorclient.Interface, opts NewAPIServerOpts
 	return serverConfig, caContentProvider, nil
 }
 
+// admissionPolicyAPIGroupVersion is the API group/version the vendored admission
+// policy plugins operate against. Alpha/beta groupversions are intentionally not
+// probed: the vendored plugin implementations only support the v1 shape, so on
+// clusters where only alpha/beta is served we treat the plugin as unavailable.
+const admissionPolicyAPIGroupVersion = "admissionregistration.k8s.io/v1"
+
+// canUseAdmissionPolicyPlugin reports whether it is safe to load the admission
+// policy plugin identified by (policyResource, bindingResource) into the
+// aggregated apiserver's admission chain.
+//
+// The plugin is safe to load only if:
+//  1. Both resources are present in cluster discovery under
+//     admissionPolicyAPIGroupVersion (skips gracefully on older k8s or when
+//     the plugin's feature gate is disabled).
+//  2. The ServiceAccount this aggregated apiserver runs under is allowed to
+//     list AND watch both resources at cluster scope. This is required
+//     because the plugin's own SharedInformer performs a List+Watch at
+//     startup; if List returns 403 the informer's cache never syncs and
+//     every mutating admission decision that plugin evaluates will
+//     permanently fail with "not yet ready to handle request".
+//
+// Discovery or SubjectAccessReview errors are treated conservatively as
+// "cannot use" and cause the plugin to be skipped with a clear warning log.
+// The aggregated apiserver's own startup is never blocked by this check.
+func canUseAdmissionPolicyPlugin(discoveryClient discovery.DiscoveryInterface, coreClient kubernetes.Interface, policyResource, bindingResource string, logger logr.Logger) bool {
+	for _, resource := range []string{policyResource, bindingResource} {
+		if !serverResourceExists(discoveryClient, admissionPolicyAPIGroupVersion, resource) {
+			logger.Info("Skipping admission-policy plugin: resource not present in cluster discovery",
+				"apiGroupVersion", admissionPolicyAPIGroupVersion, "resource", resource)
+			return false
+		}
+		for _, verb := range []string{"list", "watch"} {
+			allowed, err := selfSubjectAccessAllowed(coreClient, "admissionregistration.k8s.io", resource, verb)
+			if err != nil {
+				logger.Error(err, "Skipping admission-policy plugin: SelfSubjectAccessReview failed",
+					"resource", resource, "verb", verb)
+				return false
+			}
+			if !allowed {
+				logger.Info("Skipping admission-policy plugin: ServiceAccount lacks required permission. "+
+					"Loading this plugin without list+watch permission would cause its informer cache to never sync, "+
+					"resulting in every mutating admission request being rejected with \"not yet ready to handle request\". "+
+					"Grant list+watch on this resource (in group admissionregistration.k8s.io) to the kapp-controller ClusterRole to enable this plugin.",
+					"resource", resource, "verb", verb)
+				return false
+			}
+		}
+	}
+	return true
+}
+
+// selfSubjectAccessAllowed checks whether the caller is permitted to perform
+// verb on the cluster-scoped resource in group. It uses SelfSubjectAccessReview
+// so no additional RBAC is required beyond the subjectaccessreviews/create
+// permission already granted to kapp-controller.
+func selfSubjectAccessAllowed(coreClient kubernetes.Interface, group, resource, verb string) (bool, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	review, err := coreClient.AuthorizationV1().SelfSubjectAccessReviews().Create(ctx, &authorizationv1.SelfSubjectAccessReview{
+		Spec: authorizationv1.SelfSubjectAccessReviewSpec{
+			ResourceAttributes: &authorizationv1.ResourceAttributes{
+				Group:    group,
+				Resource: resource,
+				Verb:     verb,
+			},
+		},
+	}, metav1.CreateOptions{})
+	if err != nil {
+		return false, err
+	}
+	return review.Status.Allowed, nil
+}
+
 // serverResourceExists returns true if the given groupVersion + resource kind is
 // available in the cluster. It is used to guard admission plugins whose backing
 // resources may not be present (e.g. MutatingAdmissionPolicy behind a feature gate).

pkg/apiserver/apiserver_test.go

@@ -6,12 +6,17 @@ package apiserver
 import (
 	"context"
 	"crypto/x509"
+	"fmt"
 	"testing"
 
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/require"
+	authorizationv1 "k8s.io/api/authorization/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/server/dynamiccertificates"
+	fakediscovery "k8s.io/client-go/discovery/fake"
+	fakekube "k8s.io/client-go/kubernetes/fake"
 	clienttesting "k8s.io/client-go/testing"
 	apiregv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
 	fakeaggregator "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/fake"
@@ -92,3 +97,129 @@ func Test_updateAPIService(t *testing.T) {
 		})
 	}
 }
+
+func Test_canUseAdmissionPolicyPlugin(t *testing.T) {
+	const (
+		policyResource  = "mutatingadmissionpolicies"
+		bindingResource = "mutatingadmissionpolicybindings"
+	)
+	fullDiscovery := []*metav1.APIResourceList{{
+		GroupVersion: admissionPolicyAPIGroupVersion,
+		APIResources: []metav1.APIResource{
+			{Name: policyResource, Namespaced: false, Kind: "MutatingAdmissionPolicy"},
+			{Name: bindingResource, Namespaced: false, Kind: "MutatingAdmissionPolicyBinding"},
+		},
+	}}
+
+	// ssarReactor builds a testing reactor that allows the given
+	// (resource, verb) pairs and denies everything else. This lets each
+	// test model a specific RBAC state without a real cluster.
+	ssarReactor := func(allowed map[string]bool) clienttesting.ReactionFunc {
+		return func(action clienttesting.Action) (bool, runtime.Object, error) {
+			create, ok := action.(clienttesting.CreateAction)
+			if !ok {
+				return false, nil, nil
+			}
+			ssar, ok := create.GetObject().(*authorizationv1.SelfSubjectAccessReview)
+			if !ok {
+				return false, nil, nil
+			}
+			attrs := ssar.Spec.ResourceAttributes
+			key := fmt.Sprintf("%s/%s", attrs.Resource, attrs.Verb)
+			ssar.Status.Allowed = allowed[key]
+			return true, ssar, nil
+		}
+	}
+
+	tests := []struct {
+		name       string
+		resources  []*metav1.APIResourceList
+		ssarAllow  map[string]bool
+		ssarError  error
+		wantResult bool
+	}{
+		{
+			name:       "resource absent from discovery -> skip plugin",
+			resources:  nil,
+			wantResult: false,
+		},
+		{
+			name: "policy present but binding missing -> skip plugin",
+			resources: []*metav1.APIResourceList{{
+				GroupVersion: admissionPolicyAPIGroupVersion,
+				APIResources: []metav1.APIResource{
+					{Name: policyResource, Namespaced: false, Kind: "MutatingAdmissionPolicy"},
+				},
+			}},
+			wantResult: false,
+		},
+		{
+			name:      "both resources present, list+watch allowed on both -> enable plugin",
+			resources: fullDiscovery,
+			ssarAllow: map[string]bool{
+				policyResource + "/list":   true,
+				policyResource + "/watch":  true,
+				bindingResource + "/list":  true,
+				bindingResource + "/watch": true,
+			},
+			wantResult: true,
+		},
+		{
+			name:      "list allowed but watch denied on policy -> skip plugin",
+			resources: fullDiscovery,
+			ssarAllow: map[string]bool{
+				policyResource + "/list":   true,
+				policyResource + "/watch":  false,
+				bindingResource + "/list":  true,
+				bindingResource + "/watch": true,
+			},
+			wantResult: false,
+		},
+		{
+			name:      "list+watch denied on policy -> skip plugin",
+			resources: fullDiscovery,
+			ssarAllow: map[string]bool{
+				policyResource + "/list":   false,
+				policyResource + "/watch":  false,
+				bindingResource + "/list":  true,
+				bindingResource + "/watch": true,
+			},
+			wantResult: false,
+		},
+		{
+			name:      "policy allowed but binding denied -> skip plugin",
+			resources: fullDiscovery,
+			ssarAllow: map[string]bool{
+				policyResource + "/list":   true,
+				policyResource + "/watch":  true,
+				bindingResource + "/list":  false,
+				bindingResource + "/watch": false,
+			},
+			wantResult: false,
+		},
+		{
+			name:       "SubjectAccessReview API error -> skip plugin conservatively",
+			resources:  fullDiscovery,
+			ssarError:  fmt.Errorf("simulated api error"),
+			wantResult: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			kubeClient := fakekube.NewSimpleClientset()
+			kubeClient.Discovery().(*fakediscovery.FakeDiscovery).Resources = tc.resources
+
+			if tc.ssarError != nil {
+				kubeClient.PrependReactor("create", "selfsubjectaccessreviews", func(action clienttesting.Action) (bool, runtime.Object, error) {
+					return true, nil, tc.ssarError
+				})
+			} else if tc.ssarAllow != nil {
+				kubeClient.PrependReactor("create", "selfsubjectaccessreviews", ssarReactor(tc.ssarAllow))
+			}
+
+			got := canUseAdmissionPolicyPlugin(kubeClient.Discovery(), kubeClient, policyResource, bindingResource, logr.Discard())
+			require.Equal(t, tc.wantResult, got)
+		})
+	}
+}