Fix PodVolumeBackup list scope during restore

Restrict the listing of PodVolumeBackup resources to the specific
restore namespace in both the core restore controller and the pod
volume restore action plugin. This prevents "Forbidden" errors when
Velero is configured with namespace-scoped minimum privileges,
avoiding the need for cluster-scoped list permissions for
PodVolumeBackups.

Fixes: #9681

Signed-off-by: Adam Zhang <adam.zhang@broadcom.com>

changelogs/unreleased/9696-adam-jian-zhang

@@ -0,0 +1 @@
+Fix issue #9681, fix restores and podvolumerestores list options to only list in installed namespace

pkg/controller/restore_controller.go

@@ -529,6 +529,7 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 		LabelSelector: labels.Set(map[string]string{
 			api.BackupNameLabel: label.GetValidName(restore.Spec.BackupName),
 		}).AsSelector(),
+		Namespace: restore.Namespace,
 	}
 
 	podVolumeBackupList := &api.PodVolumeBackupList{}

pkg/controller/restore_controller_test.go

@@ -238,6 +238,8 @@ func TestRestoreReconcile(t *testing.T) {
 		expectedFinalPhase              string
 		addValidFinalizer               bool
 		emptyVolumeInfo                 bool
+		podVolumeBackups                []*velerov1api.PodVolumeBackup
+		expectedPVBCount                int
 	}{
 		{
 			name:                     "restore with both namespace in both includedNamespaces and excludedNamespaces fails validation",
@@ -357,6 +359,22 @@ func TestRestoreReconcile(t *testing.T) {
 			expectedCompletedTime: &timestamp,
 			expectedRestorerCall:  NewRestore("foo", "bar", "backup-1", "ns-1", "", velerov1api.RestorePhaseInProgress).Result(),
 		},
+		{
+			name:     "valid restore gets executed and only includes pod volume backups from restore namespace",
+			location: defaultStorageLocation,
+			restore:  NewRestore("foo", "bar2", "backup-1", "ns-1", "", velerov1api.RestorePhaseNew).Result(),
+			backup:   defaultBackup().StorageLocation("default").Result(),
+			podVolumeBackups: []*velerov1api.PodVolumeBackup{
+				builder.ForPodVolumeBackup("foo", "pvb-1").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+				builder.ForPodVolumeBackup("other-ns", "pvb-2").ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, "backup-1")).Result(),
+			},
+			expectedPVBCount:      1,
+			expectedErr:           false,
+			expectedPhase:         string(velerov1api.RestorePhaseInProgress),
+			expectedStartTime:     &timestamp,
+			expectedCompletedTime: &timestamp,
+			expectedRestorerCall:  NewRestore("foo", "bar2", "backup-1", "ns-1", "", velerov1api.RestorePhaseInProgress).Result(),
+		},
 		{
 			name:          "restoration of nodes is not supported",
 			location:      defaultStorageLocation,
@@ -501,6 +519,13 @@ func TestRestoreReconcile(t *testing.T) {
 				defaultStorageLocation.ObjectMeta.ResourceVersion = ""
 			}()
 
+			if test.podVolumeBackups != nil {
+				for _, pvb := range test.podVolumeBackups {
+					err := fakeClient.Create(t.Context(), pvb)
+					require.NoError(t, err)
+				}
+			}
+
 			r := NewRestoreReconciler(
 				t.Context(),
 				velerov1api.DefaultNamespace,
@@ -670,6 +695,10 @@ func TestRestoreReconcile(t *testing.T) {
 			// the mock stores the pointer, which gets modified after
 			assert.Equal(t, test.expectedRestorerCall.Spec, restorer.calledWithArg.Spec)
 			assert.Equal(t, test.expectedRestorerCall.Status.Phase, restorer.calledWithArg.Status.Phase)
+
+			if test.podVolumeBackups != nil {
+				assert.Len(t, restorer.calledWithPVBs, test.expectedPVBCount)
+			}
 		})
 	}
 }
@@ -1021,8 +1050,9 @@ func NewRestore(ns, name, backup, includeNS, includeResource string, phase veler
 
 type fakeRestorer struct {
 	mock.Mock
-	calledWithArg velerov1api.Restore
-	kbClient      client.Client
+	calledWithArg  velerov1api.Restore
+	calledWithPVBs []*velerov1api.PodVolumeBackup
+	kbClient       client.Client
 }
 
 func (r *fakeRestorer) Restore(
@@ -1045,6 +1075,7 @@ func (r *fakeRestorer) RestoreWithResolvers(req *pkgrestore.Request,
 		r.kbClient, volumeSnapshotterGetter)
 
 	r.calledWithArg = *req.Restore
+	r.calledWithPVBs = req.PodVolumeBackups
 
 	return res.Get(0).(results.Result), res.Get(1).(results.Result)
 }

pkg/restore/actions/pod_volume_restore_action.go

@@ -101,6 +101,7 @@ func (a *PodVolumeRestoreAction) Execute(input *velero.RestoreItemActionExecuteI
 
 	opts := &ctrlclient.ListOptions{
 		LabelSelector: label.NewSelectorForBackup(input.Restore.Spec.BackupName),
+		Namespace:     input.Restore.Namespace,
 	}
 	podVolumeBackupList := new(velerov1api.PodVolumeBackupList)
 	if err := a.crClient.List(context.TODO(), podVolumeBackupList, opts); err != nil {

pkg/restore/actions/pod_volume_restore_action_test.go

@@ -350,6 +350,28 @@ func TestPodVolumeRestoreActionExecute(t *testing.T) {
 						VolumeMounts(builder.ForVolumeMount("myvol", "/restores/myvol").Result()).
 						Command([]string{"/velero-restore-helper"}).Result()).Result(),
 		},
+		{
+			name: "pod volume backups in a different namespace are ignored when looking for matches due to namespace scoping",
+			pod: builder.ForPod("ns-1", "my-pod").
+				Volumes(
+					builder.ForVolume("myvol").PersistentVolumeClaimSource("pvc-1").Result(),
+				).
+				Result(),
+			podVolumeBackups: []runtime.Object{
+				builder.ForPodVolumeBackup("other-ns", "pvb-1").
+					PodName("my-pod").
+					PodNamespace("ns-1").
+					Volume("myvol").
+					ObjectMeta(builder.WithLabels(velerov1api.BackupNameLabel, backupName)).
+					SnapshotID("foo").
+					Result(),
+			},
+			want: builder.ForPod("ns-1", "my-pod").
+				Volumes(
+					builder.ForVolume("myvol").PersistentVolumeClaimSource("pvc-1").Result(),
+				).
+				Result(),
+		},
 	}
 
 	veleroDeployment := &appsv1api.Deployment{