Add tests for authorization/rbac.go

Signed-off-by: Tayler Geiger <[email protected]>

Author: trgeiger

internal/operator-controller/authorization/rbac_test.go

@@ -0,0 +1,181 @@
+package authorization
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/operator-framework/operator-controller/internal/operator-controller/features"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/api/meta/testrestmapper"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/authentication/user"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+var (
+	testManifest = `apiVersion: v1
+kind: Service
+metadata:
+  name: test-service
+  namespace: test-namespace
+spec:
+  clusterIP: None
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+  name: test-extension-role
+  namespace: test-namespace
+rules:
+- apiGroups: [""]
+  resources: [serviceaccounts]
+  verbs: [watch]
+- apiGroups: ["*"]
+  resources: [services]
+  verbs: [create]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+  name: test-extension-binding
+  namespace: test-namespace
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: Role
+  name: test-extension-role
+subjects:
+- kind: ServiceAccount
+  name: test-serviceaccount
+  namespace: test-namespace
+  `
+
+	saName = "test-serviceaccount"
+	ns     = "test-namespace"
+
+	objects = []client.Object{
+		&corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-namespace",
+			},
+		},
+		&rbacv1.ClusterRoleBinding{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "admin-clusterrole-binding",
+			},
+			Subjects: []rbacv1.Subject{
+				{
+					Kind:      "ServiceAccount",
+					Name:      saName,
+					Namespace: ns,
+				},
+			},
+			RoleRef: rbacv1.RoleRef{
+				Name: "admin-clusterrole",
+				Kind: "ClusterRole",
+			},
+		},
+		&corev1.ServiceAccount{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-serviceaccount",
+				Namespace: "test-namespace",
+			},
+		},
+	}
+
+	privilegedClusterRole = &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "admin-clusterrole",
+		},
+		Rules: []rbacv1.PolicyRule{
+			{
+				APIGroups: []string{"*"},
+				Resources: []string{"*"},
+				Verbs:     []string{"*"},
+			},
+		},
+	}
+
+	limitedClusterRole = &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "admin-clusterrole",
+		},
+		Rules: []rbacv1.PolicyRule{
+			{
+				APIGroups: []string{""},
+				Resources: []string{""},
+				Verbs:     []string{""},
+			},
+		},
+	}
+
+	escalatingClusterRole = &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "admin-clusterrole",
+		},
+		Rules: []rbacv1.PolicyRule{
+			{
+				APIGroups: []string{"*"},
+				Resources: []string{"serviceaccounts"},
+				Verbs:     []string{"*"},
+			},
+			{
+				APIGroups: []string{"rbac.authorization.k8s.io"},
+				Resources: []string{"roles", "clusterroles"},
+				Verbs:     []string{"escalate", "bind"},
+			},
+		},
+	}
+)
+
+func setupFakeClient(role client.Object) client.Client {
+	s := runtime.NewScheme()
+	_ = corev1.AddToScheme(s)
+	_ = rbacv1.AddToScheme(s)
+	restMapper := testrestmapper.TestOnlyStaticRESTMapper(s)
+	// restMapper := meta.NewDefaultRESTMapper(nil)
+	fakeClientBuilder := fake.NewClientBuilder().WithObjects(append(objects, role)...).WithRESTMapper(restMapper)
+	return fakeClientBuilder.Build()
+}
+
+func TestPreAuthorize_Success(t *testing.T) {
+	t.Run("preauthorize succeeds with no missing rbac rules", func(t *testing.T) {
+		featuregatetesting.SetFeatureGateDuringTest(t, features.OperatorControllerFeatureGate, features.PreflightPermissions, true)
+		fakeClient := setupFakeClient(privilegedClusterRole)
+		preAuth := NewRBACPreAuthorizer(fakeClient)
+		testServiceAccount := user.DefaultInfo{Name: fmt.Sprintf("system:serviceaccount:%s:%s", ns, saName)}
+		missingRules, err := preAuth.PreAuthorize(context.TODO(), &testServiceAccount, strings.NewReader(testManifest))
+		require.NoError(t, err)
+		require.Equal(t, map[string][]rbacv1.PolicyRule{}, missingRules)
+	})
+}
+
+func TestPreAuthorize_Failure(t *testing.T) {
+	t.Run("preauthorize failes with missing rbac rules", func(t *testing.T) {
+		featuregatetesting.SetFeatureGateDuringTest(t, features.OperatorControllerFeatureGate, features.PreflightPermissions, true)
+		fakeClient := setupFakeClient(limitedClusterRole)
+		preAuth := NewRBACPreAuthorizer(fakeClient)
+		testServiceAccount := user.DefaultInfo{Name: fmt.Sprintf("system:serviceaccount:%s:%s", ns, saName)}
+		missingRules, err := preAuth.PreAuthorize(context.TODO(), &testServiceAccount, strings.NewReader(testManifest))
+		require.Error(t, err)
+		require.NotEqual(t, map[string][]rbacv1.PolicyRule{}, missingRules)
+	})
+}
+
+func TestPreAuthorize_CheckEscalation(t *testing.T) {
+	t.Run("preauthorize succeeds with no missing rbac rules", func(t *testing.T) {
+		featuregatetesting.SetFeatureGateDuringTest(t, features.OperatorControllerFeatureGate, features.PreflightPermissions, true)
+		fakeClient := setupFakeClient(escalatingClusterRole)
+		preAuth := NewRBACPreAuthorizer(fakeClient)
+		testServiceAccount := user.DefaultInfo{Name: fmt.Sprintf("system:serviceaccount:%s:%s", ns, saName)}
+		missingRules, err := preAuth.PreAuthorize(context.TODO(), &testServiceAccount, strings.NewReader(testManifest))
+		require.NoError(t, err)
+		require.NotEqual(t, map[string][]rbacv1.PolicyRule{}, missingRules)
+	})
+}