✨ Make RBACPreAuthorizer clusterCollectionVerbs configurable

Decouple the RBACPreAuthorizer from the hardcoded cluster-scoped
collection verbs (list, watch) that were tightly coupled to the
contentmanager's requirements. The cluster collection verbs are now
configurable via the WithClusterCollectionVerbs functional option.

The helm applier configures the preauthorizer with list and watch
cluster collection verbs (needed by contentmanager), while the
boxcutter applier uses no cluster collection verbs.

Closes: #1911

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Per G. da Silva

cmd/operator-controller/main.go

@@ -724,7 +724,7 @@ func (c *helmReconcilerConfigurator) Configure(ceReconciler *controllers.Cluster
 	// determine if PreAuthorizer should be enabled based on feature gate
 	var preAuth authorization.PreAuthorizer
 	if features.OperatorControllerFeatureGate.Enabled(features.PreflightPermissions) {
-		preAuth = authorization.NewRBACPreAuthorizer(c.mgr.GetClient())
+		preAuth = authorization.NewRBACPreAuthorizer(c.mgr.GetClient(), authorization.WithClusterCollectionVerbs("list", "watch"))
 	}
 
 	cm := contentmanager.NewManager(clientRestConfigMapper, c.mgr.GetConfig(), c.mgr.GetRESTMapper())

internal/operator-controller/authorization/rbac.go

@@ -59,26 +59,35 @@ type ScopedPolicyRules struct {
 
 var objectVerbs = []string{"get", "patch", "update", "delete"}
 
-// Here we are splitting collection verbs based on required scope
-// NB: this split is tightly coupled to the requirements of the contentmanager, specifically
-// its need for cluster-scoped list/watch permissions.
-// TODO: We are accepting this coupling for now, but plan to decouple
-// TODO: link for above https://github.com/operator-framework/operator-controller/issues/1911
 var namespacedCollectionVerbs = []string{"create"}
-var clusterCollectionVerbs = []string{"list", "watch"}
+
+type RBACPreAuthorizerOption func(*rbacPreAuthorizer)
+
+// WithClusterCollectionVerbs configures cluster-scoped collection verbs (e.g. list, watch)
+// that are checked in addition to object and namespaced collection verbs.
+func WithClusterCollectionVerbs(verbs ...string) RBACPreAuthorizerOption {
+	return func(a *rbacPreAuthorizer) {
+		a.clusterCollectionVerbs = verbs
+	}
+}
 
 type rbacPreAuthorizer struct {
-	authorizer   authorizer.Authorizer
-	ruleResolver validation.AuthorizationRuleResolver
-	restMapper   meta.RESTMapper
+	authorizer             authorizer.Authorizer
+	ruleResolver           validation.AuthorizationRuleResolver
+	restMapper             meta.RESTMapper
+	clusterCollectionVerbs []string
 }
 
-func NewRBACPreAuthorizer(cl client.Client) PreAuthorizer {
-	return &rbacPreAuthorizer{
+func NewRBACPreAuthorizer(cl client.Client, opts ...RBACPreAuthorizerOption) PreAuthorizer {
+	a := &rbacPreAuthorizer{
 		authorizer:   newRBACAuthorizer(cl),
 		ruleResolver: newRBACRulesResolver(cl),
 		restMapper:   cl.RESTMapper(),
 	}
+	for _, opt := range opts {
+		opt(a)
+	}
+	return a
 }
 
 func (a *rbacPreAuthorizer) PreAuthorize(ctx context.Context, user user.Info, manifestReader io.Reader, additionalRequiredPerms ...UserAuthorizerAttributesFactory) ([]ScopedPolicyRules, error) {
@@ -88,7 +97,7 @@ func (a *rbacPreAuthorizer) PreAuthorize(ctx context.Context, user user.Info, ma
 	}
 
 	// derive manifest related attributes records
-	attributesRecords := dm.asAuthorizationAttributesRecordsForUser(user)
+	attributesRecords := dm.asAuthorizationAttributesRecordsForUser(user, a.clusterCollectionVerbs)
 
 	// append additional required perms
 	for _, fn := range additionalRequiredPerms {
@@ -324,7 +333,7 @@ func (dm *decodedManifest) rbacObjects() []client.Object {
 	return objects
 }
 
-func (dm *decodedManifest) asAuthorizationAttributesRecordsForUser(manifestManager user.Info) []authorizer.AttributesRecord {
+func (dm *decodedManifest) asAuthorizationAttributesRecordsForUser(manifestManager user.Info, clusterCollectionVerbs []string) []authorizer.AttributesRecord {
 	// Calculate initial capacity as an upper-bound estimate:
 	// - For each key: len(objectVerbs) records (4)
 	// - For unique namespaces: len(namespacedCollectionVerbs) records (1 per unique namespace across all keys in a GVR)
@@ -369,7 +378,7 @@ func (dm *decodedManifest) asAuthorizationAttributesRecordsForUser(manifestManag
 				})
 			}
 		}
-		// generate records for cluster-scoped collection verbs (list, watch) required by contentmanager
+		// generate records for cluster-scoped collection verbs (e.g. list, watch)
 		for _, v := range clusterCollectionVerbs {
 			attributeRecords = append(attributeRecords, authorizer.AttributesRecord{
 				User:            manifestManager,

internal/operator-controller/authorization/rbac_test.go

@@ -396,7 +396,7 @@ func setupFakeClient(role client.Object) client.Client {
 func TestPreAuthorize_Success(t *testing.T) {
 	t.Run("preauthorize succeeds with no missing rbac rules", func(t *testing.T) {
 		fakeClient := setupFakeClient(privilegedClusterRole)
-		preAuth := NewRBACPreAuthorizer(fakeClient)
+		preAuth := NewRBACPreAuthorizer(fakeClient, WithClusterCollectionVerbs("list", "watch"))
 		missingRules, err := preAuth.PreAuthorize(context.TODO(), testUser, strings.NewReader(testManifest))
 		require.NoError(t, err)
 		require.Equal(t, []ScopedPolicyRules{}, missingRules)
@@ -406,7 +406,7 @@ func TestPreAuthorize_Success(t *testing.T) {
 func TestPreAuthorize_MissingRBAC(t *testing.T) {
 	t.Run("preauthorize fails and finds missing rbac rules", func(t *testing.T) {
 		fakeClient := setupFakeClient(limitedClusterRole)
-		preAuth := NewRBACPreAuthorizer(fakeClient)
+		preAuth := NewRBACPreAuthorizer(fakeClient, WithClusterCollectionVerbs("list", "watch"))
 		missingRules, err := preAuth.PreAuthorize(context.TODO(), testUser, strings.NewReader(testManifest))
 		require.NoError(t, err)
 		require.Equal(t, expectedSingleNamespaceMissingRules, missingRules)
@@ -416,7 +416,7 @@ func TestPreAuthorize_MissingRBAC(t *testing.T) {
 func TestPreAuthorizeMultiNamespace_MissingRBAC(t *testing.T) {
 	t.Run("preauthorize fails and finds missing rbac rules in multiple namespaces", func(t *testing.T) {
 		fakeClient := setupFakeClient(limitedClusterRole)
-		preAuth := NewRBACPreAuthorizer(fakeClient)
+		preAuth := NewRBACPreAuthorizer(fakeClient, WithClusterCollectionVerbs("list", "watch"))
 		missingRules, err := preAuth.PreAuthorize(context.TODO(), testUser, strings.NewReader(testManifestMultiNamespace))
 		require.NoError(t, err)
 		require.Equal(t, expectedMultiNamespaceMissingRules, missingRules)
@@ -426,7 +426,7 @@ func TestPreAuthorizeMultiNamespace_MissingRBAC(t *testing.T) {
 func TestPreAuthorize_CheckEscalation(t *testing.T) {
 	t.Run("preauthorize succeeds with no missing rbac rules", func(t *testing.T) {
 		fakeClient := setupFakeClient(escalatingClusterRole)
-		preAuth := NewRBACPreAuthorizer(fakeClient)
+		preAuth := NewRBACPreAuthorizer(fakeClient, WithClusterCollectionVerbs("list", "watch"))
 		missingRules, err := preAuth.PreAuthorize(context.TODO(), testUser, strings.NewReader(testManifest))
 		require.NoError(t, err)
 		require.Equal(t, []ScopedPolicyRules{}, missingRules)
@@ -436,7 +436,7 @@ func TestPreAuthorize_CheckEscalation(t *testing.T) {
 func TestPreAuthorize_AdditionalRequiredPerms_MissingRBAC(t *testing.T) {
 	t.Run("preauthorize fails and finds missing rbac rules coming from the additional required permissions", func(t *testing.T) {
 		fakeClient := setupFakeClient(escalatingClusterRole)
-		preAuth := NewRBACPreAuthorizer(fakeClient)
+		preAuth := NewRBACPreAuthorizer(fakeClient, WithClusterCollectionVerbs("list", "watch"))
 		missingRules, err := preAuth.PreAuthorize(context.TODO(), testUser, strings.NewReader(testManifest), func(user user.Info) []authorizer.AttributesRecord {
 			return []authorizer.AttributesRecord{
 				{