From 935b1d42d0cf1a7d9a9b55139e25fe2342c78b49 Mon Sep 17 00:00:00 2001 From: Xing <675230+mikespook@users.noreply.github.com> Date: Wed, 5 Aug 2026 12:10:41 +1200 Subject: [PATCH] test(kubernetes): add rbac test coverage for pv listing [C9S-343] (#3350) --- api/kubernetes/cli/persistent_volumes_test.go | 242 ++++++++++++++++++ 1 file changed, 242 insertions(+) diff --git a/api/kubernetes/cli/persistent_volumes_test.go b/api/kubernetes/cli/persistent_volumes_test.go index 49c951fa53..d0d56abeb2 100644 --- a/api/kubernetes/cli/persistent_volumes_test.go +++ b/api/kubernetes/cli/persistent_volumes_test.go @@ -105,6 +105,248 @@ func TestGetPersistentVolumes(t *testing.T) { assert.Equal(t, corev1.VolumeBound, got.Status) assert.Equal(t, []string{"RWO"}, got.AccessModes) }) + + t.Run("non-admin sees only PVs bound to their allowed namespaces", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"nonAdmin"}, + } + + allowedPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-allowed"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "nonAdmin", Name: "pvc-allowed"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), allowedPV, metav1.CreateOptions{}) + require.NoError(t, err) + + pvs, err := kcl.GetPersistentVolumes() + require.NoError(t, err) + require.Len(t, pvs, 1) + assert.Equal(t, "pv-allowed", pvs[0].Name) + }) + + t.Run("non-admin does not see PVs bound to another tenant's namespace", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"nonAdmin"}, + } + + otherTenantPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-other-tenant"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "otherTenant", Name: "pvc-other-tenant"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), otherTenantPV, metav1.CreateOptions{}) + require.NoError(t, err) + + pvs, err := kcl.GetPersistentVolumes() + require.NoError(t, err) + assert.Empty(t, pvs) + }) + + t.Run("non-admin does not see PVs bound to a system namespace even if granted", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"portainer"}, + } + + systemPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-system"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "portainer", Name: "pvc-system"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), systemPV, metav1.CreateOptions{}) + require.NoError(t, err) + + pvs, err := kcl.GetPersistentVolumes() + require.NoError(t, err) + assert.Empty(t, pvs) + }) + + t.Run("non-admin does not see unbound PVs", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"nonAdmin"}, + } + + unboundPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-unbound"}, + Spec: corev1.PersistentVolumeSpec{}, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), unboundPV, metav1.CreateOptions{}) + require.NoError(t, err) + + pvs, err := kcl.GetPersistentVolumes() + require.NoError(t, err) + assert.Empty(t, pvs) + }) + + t.Run("non-admin with multiple allowed namespaces sees PVs from all of them", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"nonAdminOne", "nonAdminTwo"}, + } + + pvOne := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-one"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "nonAdminOne", Name: "pvc-one"}, + }, + } + pvTwo := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-two"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "nonAdminTwo", Name: "pvc-two"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pvOne, metav1.CreateOptions{}) + require.NoError(t, err) + _, err = kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pvTwo, metav1.CreateOptions{}) + require.NoError(t, err) + + pvs, err := kcl.GetPersistentVolumes() + require.NoError(t, err) + require.Len(t, pvs, 2) + + names := []string{pvs[0].Name, pvs[1].Name} + assert.ElementsMatch(t, []string{"pv-one", "pv-two"}, names) + }) + + t.Run("non-admin with no allowed namespaces gets empty list", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{}, + } + + pv := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-somewhere"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "somewhere", Name: "pvc-somewhere"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{}) + require.NoError(t, err) + + pvs, err := kcl.GetPersistentVolumes() + require.NoError(t, err) + assert.Empty(t, pvs) + }) +} + +func TestGetPersistentVolume(t *testing.T) { + t.Parallel() + + t.Run("non-admin can get a PV bound to an allowed namespace", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"nonAdmin"}, + } + + pv := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-allowed"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "nonAdmin", Name: "pvc-allowed"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{}) + require.NoError(t, err) + + got, err := kcl.GetPersistentVolume("pv-allowed") + require.NoError(t, err) + assert.Equal(t, "pv-allowed", got.Name) + }) + + t.Run("non-admin is denied access to a PV bound to another tenant's namespace", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"nonAdmin"}, + } + + pv := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-other-tenant"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "otherTenant", Name: "pvc-other-tenant"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{}) + require.NoError(t, err) + + _, err = kcl.GetPersistentVolume("pv-other-tenant") + require.Error(t, err) + assert.ErrorIs(t, err, ErrUnauthorized) + }) + + t.Run("non-admin is denied access to a PV bound to a system namespace", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"portainer"}, + } + + pv := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-system"}, + Spec: corev1.PersistentVolumeSpec{ + ClaimRef: &corev1.ObjectReference{Namespace: "portainer", Name: "pvc-system"}, + }, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{}) + require.NoError(t, err) + + _, err = kcl.GetPersistentVolume("pv-system") + require.Error(t, err) + assert.ErrorIs(t, err, ErrUnauthorized) + }) + + t.Run("non-admin is denied access to an unbound PV", func(t *testing.T) { + t.Parallel() + kcl := &KubeClient{ + cli: kfake.NewSimpleClientset(), + instanceID: "instance", + isKubeAdmin: false, + nonAdminNamespaces: []string{"nonAdmin"}, + } + + pv := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "pv-unbound"}, + Spec: corev1.PersistentVolumeSpec{}, + } + _, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{}) + require.NoError(t, err) + + _, err = kcl.GetPersistentVolume("pv-unbound") + require.Error(t, err) + assert.ErrorIs(t, err, ErrUnauthorized) + }) } func TestDeletePersistentVolumes(t *testing.T) {