Merge pull request #10335 from shubham-pampattiwar/cherry-pick/9920-release-1.18

[release-1.18] Support copying namespace-scoped secrets/configmaps for backup and restore PVC provisioning

changelogs/unreleased/10335-shubham-pampattiwar

@@ -0,0 +1 @@
+Support copying namespace-scoped secrets and configmaps for backup and restore PVC provisioning to enable datamover backup/restore of encrypted CSI volumes

config/rbac/role.yaml

@@ -4,6 +4,16 @@ kind: ClusterRole
 metadata:
   name: velero-perms
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - configmaps
+  - secrets
+  verbs:
+  - create
+  - delete
+  - get
+  - list
 - apiGroups:
   - ""
   resources:

pkg/controller/data_download_controller.go

@@ -131,6 +131,7 @@ func NewDataDownloadReconciler(
 // +kubebuilder:rbac:groups="",resources=pods,verbs=get
 // +kubebuilder:rbac:groups="",resources=persistentvolumes,verbs=get
 // +kubebuilder:rbac:groups="",resources=persistentvolumerclaims,verbs=get
+// +kubebuilder:rbac:groups="",resources=secrets;configmaps,verbs=get;list;create;delete
 
 func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	log := r.logger.WithFields(logrus.Fields{

pkg/controller/data_upload_controller.go

@@ -140,6 +140,7 @@ func NewDataUploadReconciler(
 // +kubebuilder:rbac:groups="",resources=pods,verbs=get
 // +kubebuilder:rbac:groups="",resources=persistentvolumes,verbs=get
 // +kubebuilder:rbac:groups="",resources=persistentvolumerclaims,verbs=get
+// +kubebuilder:rbac:groups="",resources=secrets;configmaps,verbs=get;list;create;delete
 
 func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	log := r.logger.WithFields(logrus.Fields{

pkg/exposer/csi_snapshot.go

@@ -41,6 +41,11 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
+// BackupPVCSecretLabel is the label applied to secrets and configmaps copied to the
+// Velero namespace for backup PVC provisioning. The value is the owning DataUpload/DataDownload
+// UID, which is a stable, valid label value (the owner name may exceed the label-value limit).
+const BackupPVCSecretLabel = "velero.io/backup-pvc-secret" //nolint:gosec // not a credential
+
 // CSISnapshotExposeParam define the input param for Expose of CSI snapshots
 type CSISnapshotExposeParam struct {
 	// SnapshotName is the original volume snapshot name
@@ -142,6 +147,28 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O
 
 	curLog.Info("Volumesnapshot is ready")
 
+	// Copy secrets and configmaps from source namespace to Velero namespace if configured.
+	// Done before creating any intermediate objects so failure doesn't require cleanup.
+	// These are needed by CSI drivers that require namespace-scoped resources for volume
+	// provisioning (e.g., encrypted volumes with KMS tokens and tenant Vault configs).
+	if value, exists := csiExposeParam.BackupPVCConfig[csiExposeParam.StorageClass]; exists {
+		copyLabels := map[string]string{BackupPVCSecretLabel: string(ownerObject.UID)}
+		for _, secretName := range value.SecretNames {
+			if copyErr := kube.CopySecret(ctx, e.kubeClient.CoreV1(), secretName,
+				csiExposeParam.SourceNamespace, ownerObject.Namespace, copyLabels, curLog); copyErr != nil {
+				return errors.Wrapf(copyErr, "error copying secret %s from %s to %s",
+					secretName, csiExposeParam.SourceNamespace, ownerObject.Namespace)
+			}
+		}
+		for _, cmName := range value.ConfigMapNames {
+			if copyErr := kube.CopyConfigMap(ctx, e.kubeClient.CoreV1(), cmName,
+				csiExposeParam.SourceNamespace, ownerObject.Namespace, copyLabels, curLog); copyErr != nil {
+				return errors.Wrapf(copyErr, "error copying configmap %s from %s to %s",
+					cmName, csiExposeParam.SourceNamespace, ownerObject.Namespace)
+			}
+		}
+	}
+
 	vsc, err := csi.GetVolumeSnapshotContentForVolumeSnapshot(volumeSnapshot, e.csiSnapshotClient)
 	if err != nil {
 		return errors.Wrap(err, "error to get volume snapshot content")
@@ -446,6 +473,11 @@ func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1api.
 	kube.DeletePodIfAny(ctx, e.kubeClient.CoreV1(), backupPodName, ownerObject.Namespace, e.log)
 	kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), backupPVCName, ownerObject.Namespace, cleanUpTimeout, e.log)
 
+	kube.DeleteSecretsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
+		BackupPVCSecretLabel, string(ownerObject.UID), e.log)
+	kube.DeleteConfigMapsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
+		BackupPVCSecretLabel, string(ownerObject.UID), e.log)
+
 	csi.DeleteVolumeSnapshotIfAny(ctx, e.csiSnapshotClient, backupVSName, ownerObject.Namespace, e.log)
 	csi.DeleteVolumeSnapshotIfAny(ctx, e.csiSnapshotClient, vsName, sourceNamespace, e.log)
 }

pkg/exposer/csi_snapshot_test.go

@@ -1989,3 +1989,186 @@ end diagnose CSI exposer`,
 		})
 	}
 }
+
+func TestExpose_SecretCopy(t *testing.T) {
+	backup := &velerov1.Backup{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: velerov1.SchemeGroupVersion.String(),
+			Kind:       "Backup",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: velerov1.DefaultNamespace,
+			Name:      "fake-backup",
+			UID:       "fake-uid",
+		},
+	}
+
+	ownerObject := corev1api.ObjectReference{
+		Kind:       backup.Kind,
+		Namespace:  backup.Namespace,
+		Name:       backup.Name,
+		UID:        backup.UID,
+		APIVersion: backup.APIVersion,
+	}
+
+	// The secret/configmap copy runs after GetVolumeTopology and WaitVolumeSnapshotReady,
+	// so a StorageClass and a ready VolumeSnapshot are needed to reach the copy block.
+	scObj := &storagev1api.StorageClass{
+		ObjectMeta: metav1.ObjectMeta{Name: "encrypted-sc"},
+	}
+	readyVS := func() *snapshotv1api.VolumeSnapshot {
+		vscName := "fake-vsc"
+		return &snapshotv1api.VolumeSnapshot{
+			ObjectMeta: metav1.ObjectMeta{Name: "fake-vs", Namespace: "app-ns"},
+			Spec: snapshotv1api.VolumeSnapshotSpec{
+				Source: snapshotv1api.VolumeSnapshotSource{VolumeSnapshotContentName: &vscName},
+			},
+			Status: &snapshotv1api.VolumeSnapshotStatus{
+				BoundVolumeSnapshotContentName: &vscName,
+				ReadyToUse:                     boolptr.True(),
+				RestoreSize:                    resource.NewQuantity(1234, ""),
+			},
+		}
+	}
+
+	param := func() *CSISnapshotExposeParam {
+		return &CSISnapshotExposeParam{
+			SourceNamespace:  "app-ns",
+			SourcePVName:     "fake-pv",
+			SnapshotName:     "fake-vs",
+			StorageClass:     "encrypted-sc",
+			OperationTimeout: time.Millisecond,
+			ExposeTimeout:    time.Second,
+		}
+	}
+
+	t.Run("copies secret from source namespace", func(t *testing.T) {
+		srcSecret := &corev1api.Secret{
+			ObjectMeta: metav1.ObjectMeta{Name: "kms-token", Namespace: "app-ns"},
+			Data:       map[string][]byte{"token": []byte("vault-token")},
+			Type:       corev1api.SecretTypeOpaque,
+		}
+		fakeKubeClient := fake.NewSimpleClientset(srcSecret, scObj)
+		fakeSnapshotClient := snapshotFake.NewSimpleClientset(readyVS())
+
+		exposer := csiSnapshotExposer{
+			kubeClient:        fakeKubeClient,
+			csiSnapshotClient: fakeSnapshotClient.SnapshotV1(),
+			log:               velerotest.NewLogger(),
+		}
+
+		p := param()
+		p.BackupPVCConfig = map[string]velerotypes.BackupPVC{
+			"encrypted-sc": {SecretNames: []string{"kms-token"}},
+		}
+
+		// Expose will fail later (no VSC exists), but the secret copy should succeed
+		_ = exposer.Expose(t.Context(), ownerObject, p)
+
+		copied, err := fakeKubeClient.CoreV1().Secrets(ownerObject.Namespace).Get(
+			t.Context(), "kms-token", metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.Equal(t, []byte("vault-token"), copied.Data["token"])
+		assert.Equal(t, string(ownerObject.UID), copied.Labels[BackupPVCSecretLabel])
+	})
+
+	t.Run("copies configmap from source namespace", func(t *testing.T) {
+		srcCM := &corev1api.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{Name: "kms-config", Namespace: "app-ns"},
+			Data:       map[string]string{"vaultAddress": "https://vault.example.com"},
+		}
+		fakeKubeClient := fake.NewSimpleClientset(srcCM, scObj)
+		fakeSnapshotClient := snapshotFake.NewSimpleClientset(readyVS())
+
+		exposer := csiSnapshotExposer{
+			kubeClient:        fakeKubeClient,
+			csiSnapshotClient: fakeSnapshotClient.SnapshotV1(),
+			log:               velerotest.NewLogger(),
+		}
+
+		p := param()
+		p.BackupPVCConfig = map[string]velerotypes.BackupPVC{
+			"encrypted-sc": {ConfigMapNames: []string{"kms-config"}},
+		}
+
+		_ = exposer.Expose(t.Context(), ownerObject, p)
+
+		copied, err := fakeKubeClient.CoreV1().ConfigMaps(ownerObject.Namespace).Get(
+			t.Context(), "kms-config", metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.Equal(t, "https://vault.example.com", copied.Data["vaultAddress"])
+		assert.Equal(t, string(ownerObject.UID), copied.Labels[BackupPVCSecretLabel])
+	})
+
+	t.Run("returns error when source secret missing", func(t *testing.T) {
+		fakeKubeClient := fake.NewSimpleClientset(scObj)
+		fakeSnapshotClient := snapshotFake.NewSimpleClientset(readyVS())
+
+		exposer := csiSnapshotExposer{
+			kubeClient:        fakeKubeClient,
+			csiSnapshotClient: fakeSnapshotClient.SnapshotV1(),
+			log:               velerotest.NewLogger(),
+		}
+
+		p := param()
+		p.BackupPVCConfig = map[string]velerotypes.BackupPVC{
+			"encrypted-sc": {SecretNames: []string{"missing-secret"}},
+		}
+
+		err := exposer.Expose(t.Context(), ownerObject, p)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "error copying secret")
+	})
+}
+
+func TestCleanUp_SecretsAndConfigMaps(t *testing.T) {
+	ownerObject := corev1api.ObjectReference{
+		Kind:       "Backup",
+		Namespace:  "velero",
+		Name:       "du-123",
+		UID:        "fake-uid",
+		APIVersion: "v1",
+	}
+
+	secret := &corev1api.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "kms-token", Namespace: "velero",
+			Labels: map[string]string{BackupPVCSecretLabel: string(ownerObject.UID)},
+			UID:    "secret-uid",
+		},
+	}
+	cm := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "kms-config", Namespace: "velero",
+			Labels: map[string]string{BackupPVCSecretLabel: string(ownerObject.UID)},
+			UID:    "cm-uid",
+		},
+	}
+	unrelatedSecret := &corev1api.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "other-secret", Namespace: "velero",
+			Labels: map[string]string{BackupPVCSecretLabel: "other-owner-uid"},
+			UID:    "other-uid",
+		},
+	}
+
+	fakeKubeClient := fake.NewSimpleClientset(secret, cm, unrelatedSecret)
+	fakeSnapshotClient := snapshotFake.NewSimpleClientset()
+
+	exposer := csiSnapshotExposer{
+		kubeClient:        fakeKubeClient,
+		csiSnapshotClient: fakeSnapshotClient.SnapshotV1(),
+		log:               velerotest.NewLogger(),
+	}
+
+	exposer.CleanUp(t.Context(), ownerObject, "", "app-ns")
+
+	_, err := fakeKubeClient.CoreV1().Secrets("velero").Get(t.Context(), "kms-token", metav1.GetOptions{})
+	require.Error(t, err, "owned secret should be deleted")
+
+	_, err = fakeKubeClient.CoreV1().ConfigMaps("velero").Get(t.Context(), "kms-config", metav1.GetOptions{})
+	require.Error(t, err, "owned configmap should be deleted")
+
+	_, err = fakeKubeClient.CoreV1().Secrets("velero").Get(t.Context(), "other-secret", metav1.GetOptions{})
+	assert.NoError(t, err, "unrelated secret should not be deleted")
+}

pkg/exposer/generic_restore.go

@@ -176,6 +176,36 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
 		}
 	}
 
+	// Copy secrets and configmaps from the target namespace to the Velero namespace if configured.
+	// These are needed by CSI drivers that require namespace-scoped resources for volume
+	// provisioning of the restorePVC (e.g., encrypted volumes with KMS tokens and tenant Vault configs).
+	copyLabels := map[string]string{BackupPVCSecretLabel: string(ownerObject.UID)}
+	for _, secretName := range param.RestorePVCConfig.SecretNames {
+		if copyErr := kube.CopySecret(ctx, e.kubeClient.CoreV1(), secretName,
+			param.TargetNamespace, ownerObject.Namespace, copyLabels, curLog); copyErr != nil {
+			err = errors.Wrapf(copyErr, "error copying secret %s from %s to %s",
+				secretName, param.TargetNamespace, ownerObject.Namespace)
+			return err
+		}
+	}
+	for _, cmName := range param.RestorePVCConfig.ConfigMapNames {
+		if copyErr := kube.CopyConfigMap(ctx, e.kubeClient.CoreV1(), cmName,
+			param.TargetNamespace, ownerObject.Namespace, copyLabels, curLog); copyErr != nil {
+			err = errors.Wrapf(copyErr, "error copying configmap %s from %s to %s",
+				cmName, param.TargetNamespace, ownerObject.Namespace)
+			return err
+		}
+	}
+
+	defer func() {
+		if err != nil {
+			kube.DeleteSecretsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
+				BackupPVCSecretLabel, string(ownerObject.UID), curLog)
+			kube.DeleteConfigMapsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
+				BackupPVCSecretLabel, string(ownerObject.UID), curLog)
+		}
+	}()
+
 	restorePod, err := e.createRestorePod(
 		ctx,
 		ownerObject,
@@ -377,6 +407,11 @@ func (e *genericRestoreExposer) CleanUp(ctx context.Context, ownerObject corev1a
 	kube.DeletePodIfAny(ctx, e.kubeClient.CoreV1(), restorePodName, ownerObject.Namespace, e.log)
 	kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), restorePVCName, ownerObject.Namespace, 0, e.log)
 	kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), cachePVCName, ownerObject.Namespace, 0, e.log)
+
+	kube.DeleteSecretsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
+		BackupPVCSecretLabel, string(ownerObject.UID), e.log)
+	kube.DeleteConfigMapsWithLabel(ctx, e.kubeClient.CoreV1(), ownerObject.Namespace,
+		BackupPVCSecretLabel, string(ownerObject.UID), e.log)
 }
 
 func (e *genericRestoreExposer) RebindVolume(ctx context.Context, ownerObject corev1api.ObjectReference, targetPVCName string, targetNamespace string, timeout time.Duration) error {

pkg/exposer/generic_restore_test.go

@@ -34,6 +34,7 @@ import (
 
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	velerotypes "github.com/vmware-tanzu/velero/pkg/types"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
@@ -306,6 +307,88 @@ func TestRestoreExpose(t *testing.T) {
 	}
 }
 
+func TestRestoreExpose_SecretCopy(t *testing.T) {
+	scName := "fake-sc"
+	restore := &velerov1.Restore{
+		TypeMeta:   metav1.TypeMeta{APIVersion: velerov1.SchemeGroupVersion.String(), Kind: "Restore"},
+		ObjectMeta: metav1.ObjectMeta{Namespace: velerov1.DefaultNamespace, Name: "fake-restore", UID: "fake-uid"},
+	}
+	ownerObject := corev1api.ObjectReference{
+		Kind:       restore.Kind,
+		Namespace:  restore.Namespace,
+		Name:       restore.Name,
+		UID:        restore.UID,
+		APIVersion: restore.APIVersion,
+	}
+	targetPVCObj := &corev1api.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "fake-ns", Name: "fake-target-pvc"},
+		Spec:       corev1api.PersistentVolumeClaimSpec{StorageClassName: &scName},
+	}
+	storageClass := &storagev1api.StorageClass{ObjectMeta: metav1.ObjectMeta{Name: "fake-sc"}}
+	daemonSet := &appsv1api.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "node-agent"},
+		TypeMeta:   metav1.TypeMeta{Kind: "DaemonSet", APIVersion: appsv1api.SchemeGroupVersion.String()},
+		Spec: appsv1api.DaemonSetSpec{
+			Template: corev1api.PodTemplateSpec{
+				Spec: corev1api.PodSpec{Containers: []corev1api.Container{{Image: "fake-image"}}},
+			},
+		},
+	}
+
+	t.Run("copies secret and configmap from target namespace", func(t *testing.T) {
+		srcSecret := &corev1api.Secret{
+			ObjectMeta: metav1.ObjectMeta{Name: "kms-token", Namespace: "fake-ns"},
+			Data:       map[string][]byte{"token": []byte("vault-token")},
+			Type:       corev1api.SecretTypeOpaque,
+		}
+		srcCM := &corev1api.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{Name: "kms-config", Namespace: "fake-ns"},
+			Data:       map[string]string{"vaultAddress": "https://vault.example.com"},
+		}
+		fakeKubeClient := fake.NewSimpleClientset(targetPVCObj, storageClass, daemonSet, srcSecret, srcCM)
+		exposer := genericRestoreExposer{kubeClient: fakeKubeClient, log: velerotest.NewLogger()}
+
+		err := exposer.Expose(t.Context(), ownerObject, GenericRestoreExposeParam{
+			TargetPVCName:    "fake-target-pvc",
+			TargetNamespace:  "fake-ns",
+			HostingPodLabels: map[string]string{},
+			Resources:        corev1api.ResourceRequirements{},
+			ExposeTimeout:    time.Millisecond,
+			RestorePVCConfig: velerotypes.RestorePVC{
+				SecretNames:    []string{"kms-token"},
+				ConfigMapNames: []string{"kms-config"},
+			},
+		})
+		require.NoError(t, err)
+
+		copiedSecret, err := fakeKubeClient.CoreV1().Secrets(ownerObject.Namespace).Get(t.Context(), "kms-token", metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.Equal(t, []byte("vault-token"), copiedSecret.Data["token"])
+		assert.Equal(t, string(ownerObject.UID), copiedSecret.Labels[BackupPVCSecretLabel])
+
+		copiedCM, err := fakeKubeClient.CoreV1().ConfigMaps(ownerObject.Namespace).Get(t.Context(), "kms-config", metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.Equal(t, "https://vault.example.com", copiedCM.Data["vaultAddress"])
+		assert.Equal(t, string(ownerObject.UID), copiedCM.Labels[BackupPVCSecretLabel])
+	})
+
+	t.Run("returns error when source secret missing", func(t *testing.T) {
+		fakeKubeClient := fake.NewSimpleClientset(targetPVCObj, storageClass, daemonSet)
+		exposer := genericRestoreExposer{kubeClient: fakeKubeClient, log: velerotest.NewLogger()}
+
+		err := exposer.Expose(t.Context(), ownerObject, GenericRestoreExposeParam{
+			TargetPVCName:    "fake-target-pvc",
+			TargetNamespace:  "fake-ns",
+			HostingPodLabels: map[string]string{},
+			Resources:        corev1api.ResourceRequirements{},
+			ExposeTimeout:    time.Millisecond,
+			RestorePVCConfig: velerotypes.RestorePVC{SecretNames: []string{"missing-secret"}},
+		})
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "error copying secret")
+	})
+}
+
 func TestRebindVolume(t *testing.T) {
 	restore := &velerov1.Restore{
 		TypeMeta: metav1.TypeMeta{

pkg/types/node_agent.go

@@ -59,11 +59,36 @@ type BackupPVC struct {
 
 	// Annotations permits setting annotations for the backupPVC
 	Annotations map[string]string `json:"annotations,omitempty"`
+
+	// SecretNames is a list of secret names to copy from the source PVC namespace
+	// to the Velero namespace before creating the backupPVC. The secrets are deleted
+	// after the DataUpload completes. This is needed for CSI drivers that require
+	// namespace-scoped secrets for volume provisioning (e.g., encrypted volumes).
+	SecretNames []string `json:"secretNames,omitempty"`
+
+	// ConfigMapNames is a list of configmap names to copy from the source PVC namespace
+	// to the Velero namespace before creating the backupPVC. The configmaps are deleted
+	// after the DataUpload completes. This is needed for CSI drivers that require
+	// namespace-scoped configmaps for volume provisioning (e.g., tenant-specific
+	// Vault connection overrides for encrypted volumes).
+	ConfigMapNames []string `json:"configMapNames,omitempty"`
 }
 
 type RestorePVC struct {
 	// IgnoreDelayBinding indicates to ignore delay binding the restorePVC when it is in WaitForFirstConsumer mode
 	IgnoreDelayBinding bool `json:"ignoreDelayBinding,omitempty"`
+
+	// SecretNames is a list of secret names to copy from the target namespace to the
+	// Velero namespace before creating the restorePVC. The secrets are deleted after the
+	// DataDownload completes. This is needed for CSI drivers that require namespace-scoped
+	// secrets for volume provisioning (e.g., encrypted volumes).
+	SecretNames []string `json:"secretNames,omitempty"`
+
+	// ConfigMapNames is a list of configmap names to copy from the target namespace to the
+	// Velero namespace before creating the restorePVC. The configmaps are deleted after the
+	// DataDownload completes. This is needed for CSI drivers that require namespace-scoped
+	// configmaps for volume provisioning (e.g., tenant-specific Vault connection overrides).
+	ConfigMapNames []string `json:"configMapNames,omitempty"`
 }
 
 type CachePVC struct {

pkg/util/kube/secrets.go

@@ -18,9 +18,14 @@ package kube
 
 import (
 	"context"
+	"reflect"
 
 	"github.com/cockroachdb/errors"
+	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -49,3 +54,151 @@ func GetSecretKey(client kbclient.Client, namespace string, selector *corev1api.
 
 	return key, nil
 }
+
+// ErrSecretCollision is returned when a secret or configmap with the same name but different
+// data already exists in the target namespace, indicating another owner is using it.
+var ErrSecretCollision = errors.New("secret collision: same name exists with different data")
+
+// labelsMatch reports whether all entries in want are present in have with matching values.
+func labelsMatch(have, want map[string]string) bool {
+	for k, v := range want {
+		if have[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
+// CopySecret copies a secret from sourceNamespace to targetNamespace, applying the given labels.
+// If a secret with the same name already exists in the target with identical data and matching
+// labels, it is a no-op. If the data matches but the labels differ, or the data differs, it
+// returns ErrSecretCollision.
+func CopySecret(ctx context.Context, client corev1client.CoreV1Interface, secretName, sourceNamespace, targetNamespace string, labels map[string]string, log logrus.FieldLogger) error {
+	srcSecret, err := client.Secrets(sourceNamespace).Get(ctx, secretName, metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "error getting secret %s/%s", sourceNamespace, secretName)
+	}
+
+	newSecret := &corev1api.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      secretName,
+			Namespace: targetNamespace,
+			Labels:    labels,
+		},
+		Type: srcSecret.Type,
+		Data: srcSecret.Data,
+	}
+
+	_, err = client.Secrets(targetNamespace).Create(ctx, newSecret, metav1.CreateOptions{})
+	if err == nil {
+		log.Infof("Copied secret %s from %s to %s", secretName, sourceNamespace, targetNamespace)
+		return nil
+	}
+
+	if !apierrors.IsAlreadyExists(err) {
+		return errors.Wrapf(err, "error creating secret %s in %s", secretName, targetNamespace)
+	}
+
+	existing, err := client.Secrets(targetNamespace).Get(ctx, secretName, metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "error getting existing secret %s/%s", targetNamespace, secretName)
+	}
+
+	if reflect.DeepEqual(existing.Data, srcSecret.Data) && labelsMatch(existing.Labels, labels) {
+		log.Infof("Secret %s already exists in %s with same data and labels, skipping copy", secretName, targetNamespace)
+		return nil
+	}
+
+	log.Infof("Secret %s already exists in %s owned by a different owner, collision detected", secretName, targetNamespace)
+	return ErrSecretCollision
+}
+
+// DeleteSecretsWithLabel deletes all secrets in a namespace matching a label key=value pair.
+// Uses UID preconditions to avoid deleting a recreated object with the same name.
+func DeleteSecretsWithLabel(ctx context.Context, client corev1client.CoreV1Interface, namespace, labelKey, labelValue string, log logrus.FieldLogger) {
+	secrets, err := client.Secrets(namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: labelKey + "=" + labelValue,
+	})
+	if err != nil {
+		log.WithError(err).Errorf("Failed to list secrets with label %s=%s in %s", labelKey, labelValue, namespace)
+		return
+	}
+
+	for i := range secrets.Items {
+		uid := secrets.Items[i].UID
+		err := client.Secrets(namespace).Delete(ctx, secrets.Items[i].Name, metav1.DeleteOptions{
+			Preconditions: &metav1.Preconditions{UID: &uid},
+		})
+		if err != nil && !apierrors.IsNotFound(err) {
+			log.WithError(err).Errorf("Failed to delete secret %s/%s", namespace, secrets.Items[i].Name)
+		}
+	}
+}
+
+// CopyConfigMap copies a configmap from sourceNamespace to targetNamespace, applying the given
+// labels. If a configmap with the same name already exists in the target with identical data and
+// matching labels, it is a no-op. If the data matches but the labels differ, or the data differs,
+// it returns ErrSecretCollision.
+func CopyConfigMap(ctx context.Context, client corev1client.CoreV1Interface, cmName, sourceNamespace, targetNamespace string, labels map[string]string, log logrus.FieldLogger) error {
+	srcCM, err := client.ConfigMaps(sourceNamespace).Get(ctx, cmName, metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "error getting configmap %s/%s", sourceNamespace, cmName)
+	}
+
+	newCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      cmName,
+			Namespace: targetNamespace,
+			Labels:    labels,
+		},
+		Data:       srcCM.Data,
+		BinaryData: srcCM.BinaryData,
+	}
+
+	_, err = client.ConfigMaps(targetNamespace).Create(ctx, newCM, metav1.CreateOptions{})
+	if err == nil {
+		log.Infof("Copied configmap %s from %s to %s", cmName, sourceNamespace, targetNamespace)
+		return nil
+	}
+
+	if !apierrors.IsAlreadyExists(err) {
+		return errors.Wrapf(err, "error creating configmap %s in %s", cmName, targetNamespace)
+	}
+
+	existing, err := client.ConfigMaps(targetNamespace).Get(ctx, cmName, metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "error getting existing configmap %s/%s", targetNamespace, cmName)
+	}
+
+	if reflect.DeepEqual(existing.Data, srcCM.Data) &&
+		reflect.DeepEqual(existing.BinaryData, srcCM.BinaryData) &&
+		labelsMatch(existing.Labels, labels) {
+		log.Infof("ConfigMap %s already exists in %s with same data and labels, skipping copy", cmName, targetNamespace)
+		return nil
+	}
+
+	log.Infof("ConfigMap %s already exists in %s owned by a different owner, collision detected", cmName, targetNamespace)
+	return ErrSecretCollision
+}
+
+// DeleteConfigMapsWithLabel deletes all configmaps in a namespace matching a label key=value pair.
+// Uses UID preconditions to avoid deleting a recreated object with the same name.
+func DeleteConfigMapsWithLabel(ctx context.Context, client corev1client.CoreV1Interface, namespace, labelKey, labelValue string, log logrus.FieldLogger) {
+	cms, err := client.ConfigMaps(namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: labelKey + "=" + labelValue,
+	})
+	if err != nil {
+		log.WithError(err).Errorf("Failed to list configmaps with label %s=%s in %s", labelKey, labelValue, namespace)
+		return
+	}
+
+	for i := range cms.Items {
+		uid := cms.Items[i].UID
+		err := client.ConfigMaps(namespace).Delete(ctx, cms.Items[i].Name, metav1.DeleteOptions{
+			Preconditions: &metav1.Preconditions{UID: &uid},
+		})
+		if err != nil && !apierrors.IsNotFound(err) {
+			log.WithError(err).Errorf("Failed to delete configmap %s/%s", namespace, cms.Items[i].Name)
+		}
+	}
+}

pkg/util/kube/secrets_copy_test.go

@@ -0,0 +1,362 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kube
+
+import (
+	"context"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+const testCopyLabel = "velero.io/backup-pvc-secret"
+
+func TestCopySecret(t *testing.T) {
+	log := logrus.New()
+
+	tests := []struct {
+		name        string
+		secretName  string
+		sourceNS    string
+		targetNS    string
+		ownerName   string
+		objects     []k8sruntime.Object
+		expectErr   bool
+		errContains string
+	}{
+		{
+			name:       "successfully copies secret to target namespace",
+			secretName: "ceph-csi-kms-token",
+			sourceNS:   "app-ns",
+			targetNS:   "velero",
+			ownerName:  "du-123",
+			objects: []k8sruntime.Object{
+				&corev1api.Secret{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-token", Namespace: "app-ns"},
+					Data:       map[string][]byte{"token": []byte("vault-token-a")},
+					Type:       corev1api.SecretTypeOpaque,
+				},
+			},
+		},
+		{
+			name:        "returns error when source secret does not exist",
+			secretName:  "missing-secret",
+			sourceNS:    "app-ns",
+			targetNS:    "velero",
+			ownerName:   "du-123",
+			objects:     []k8sruntime.Object{},
+			expectErr:   true,
+			errContains: "error getting secret",
+		},
+		{
+			name:       "no-op when target already has secret with same data and same owner",
+			secretName: "ceph-csi-kms-token",
+			sourceNS:   "app-ns",
+			targetNS:   "velero",
+			ownerName:  "du-123",
+			objects: []k8sruntime.Object{
+				&corev1api.Secret{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-token", Namespace: "app-ns"},
+					Data:       map[string][]byte{"token": []byte("same-token")},
+					Type:       corev1api.SecretTypeOpaque,
+				},
+				&corev1api.Secret{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "ceph-csi-kms-token", Namespace: "velero",
+						Labels: map[string]string{testCopyLabel: "du-123"},
+					},
+					Data: map[string][]byte{"token": []byte("same-token")},
+					Type: corev1api.SecretTypeOpaque,
+				},
+			},
+		},
+		{
+			name:       "returns collision when same data but different owner",
+			secretName: "ceph-csi-kms-token",
+			sourceNS:   "app-ns",
+			targetNS:   "velero",
+			ownerName:  "du-456",
+			objects: []k8sruntime.Object{
+				&corev1api.Secret{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-token", Namespace: "app-ns"},
+					Data:       map[string][]byte{"token": []byte("same-token")},
+					Type:       corev1api.SecretTypeOpaque,
+				},
+				&corev1api.Secret{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "ceph-csi-kms-token", Namespace: "velero",
+						Labels: map[string]string{testCopyLabel: "du-123"},
+					},
+					Data: map[string][]byte{"token": []byte("same-token")},
+					Type: corev1api.SecretTypeOpaque,
+				},
+			},
+			expectErr:   true,
+			errContains: "collision",
+		},
+		{
+			name:       "returns collision error when target has secret with different data",
+			secretName: "ceph-csi-kms-token",
+			sourceNS:   "app-ns",
+			targetNS:   "velero",
+			ownerName:  "du-123",
+			objects: []k8sruntime.Object{
+				&corev1api.Secret{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-token", Namespace: "app-ns"},
+					Data:       map[string][]byte{"token": []byte("token-a")},
+					Type:       corev1api.SecretTypeOpaque,
+				},
+				&corev1api.Secret{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-token", Namespace: "velero"},
+					Data:       map[string][]byte{"token": []byte("token-b")},
+					Type:       corev1api.SecretTypeOpaque,
+				},
+			},
+			expectErr:   true,
+			errContains: "secret collision",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fakeClient := fake.NewSimpleClientset(tt.objects...)
+
+			err := CopySecret(context.Background(), fakeClient.CoreV1(),
+				tt.secretName, tt.sourceNS, tt.targetNS,
+				map[string]string{testCopyLabel: tt.ownerName}, log)
+
+			if tt.expectErr {
+				require.Error(t, err)
+				if tt.errContains != "" {
+					assert.Contains(t, err.Error(), tt.errContains)
+				}
+				return
+			}
+
+			require.NoError(t, err)
+
+			copied, getErr := fakeClient.CoreV1().Secrets(tt.targetNS).Get(
+				context.Background(), tt.secretName, metav1.GetOptions{})
+			require.NoError(t, getErr)
+			assert.NotNil(t, copied)
+		})
+	}
+}
+
+func TestDeleteSecretsWithLabel(t *testing.T) {
+	log := logrus.New()
+
+	fakeClient := fake.NewSimpleClientset(
+		&corev1api.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "secret-1", Namespace: "velero",
+				Labels: map[string]string{testCopyLabel: "du-123"},
+			},
+		},
+		&corev1api.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "secret-2", Namespace: "velero",
+				Labels: map[string]string{testCopyLabel: "du-456"},
+			},
+		},
+	)
+
+	DeleteSecretsWithLabel(context.Background(), fakeClient.CoreV1(), "velero",
+		testCopyLabel, "du-123", log)
+
+	_, err := fakeClient.CoreV1().Secrets("velero").Get(
+		context.Background(), "secret-1", metav1.GetOptions{})
+	require.Error(t, err, "secret-1 should be deleted")
+
+	_, err = fakeClient.CoreV1().Secrets("velero").Get(
+		context.Background(), "secret-2", metav1.GetOptions{})
+	assert.NoError(t, err, "secret-2 should still exist")
+}
+
+func TestCopyConfigMap(t *testing.T) {
+	log := logrus.New()
+
+	tests := []struct {
+		name        string
+		cmName      string
+		sourceNS    string
+		targetNS    string
+		ownerName   string
+		objects     []k8sruntime.Object
+		expectErr   bool
+		errContains string
+	}{
+		{
+			name:      "successfully copies configmap to target namespace",
+			cmName:    "ceph-csi-kms-config",
+			sourceNS:  "app-ns",
+			targetNS:  "velero",
+			ownerName: "du-123",
+			objects: []k8sruntime.Object{
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-config", Namespace: "app-ns"},
+					Data:       map[string]string{"vaultAddress": "https://vault.example.com"},
+				},
+			},
+		},
+		{
+			name:        "returns error when source configmap does not exist",
+			cmName:      "missing-cm",
+			sourceNS:    "app-ns",
+			targetNS:    "velero",
+			ownerName:   "du-123",
+			objects:     []k8sruntime.Object{},
+			expectErr:   true,
+			errContains: "error getting configmap",
+		},
+		{
+			name:      "no-op when target already has configmap with same data and same owner",
+			cmName:    "ceph-csi-kms-config",
+			sourceNS:  "app-ns",
+			targetNS:  "velero",
+			ownerName: "du-123",
+			objects: []k8sruntime.Object{
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-config", Namespace: "app-ns"},
+					Data:       map[string]string{"vaultAddress": "https://vault.example.com"},
+				},
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "ceph-csi-kms-config", Namespace: "velero",
+						Labels: map[string]string{testCopyLabel: "du-123"},
+					},
+					Data: map[string]string{"vaultAddress": "https://vault.example.com"},
+				},
+			},
+		},
+		{
+			name:      "returns collision when same data but different owner",
+			cmName:    "ceph-csi-kms-config",
+			sourceNS:  "app-ns",
+			targetNS:  "velero",
+			ownerName: "du-456",
+			objects: []k8sruntime.Object{
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-config", Namespace: "app-ns"},
+					Data:       map[string]string{"vaultAddress": "https://vault.example.com"},
+				},
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "ceph-csi-kms-config", Namespace: "velero",
+						Labels: map[string]string{testCopyLabel: "du-123"},
+					},
+					Data: map[string]string{"vaultAddress": "https://vault.example.com"},
+				},
+			},
+			expectErr:   true,
+			errContains: "collision",
+		},
+		{
+			name:      "copies configmap with BinaryData",
+			cmName:    "ceph-csi-kms-config",
+			sourceNS:  "app-ns",
+			targetNS:  "velero",
+			ownerName: "du-123",
+			objects: []k8sruntime.Object{
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-config", Namespace: "app-ns"},
+					BinaryData: map[string][]byte{"ca.crt": []byte("binary-ca-bundle")},
+				},
+			},
+		},
+		{
+			name:      "returns collision error when target has configmap with different data",
+			cmName:    "ceph-csi-kms-config",
+			sourceNS:  "app-ns",
+			targetNS:  "velero",
+			ownerName: "du-123",
+			objects: []k8sruntime.Object{
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-config", Namespace: "app-ns"},
+					Data:       map[string]string{"vaultAddress": "https://vault-a.example.com"},
+				},
+				&corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{Name: "ceph-csi-kms-config", Namespace: "velero"},
+					Data:       map[string]string{"vaultAddress": "https://vault-b.example.com"},
+				},
+			},
+			expectErr:   true,
+			errContains: "secret collision",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fakeClient := fake.NewSimpleClientset(tt.objects...)
+
+			err := CopyConfigMap(context.Background(), fakeClient.CoreV1(),
+				tt.cmName, tt.sourceNS, tt.targetNS,
+				map[string]string{testCopyLabel: tt.ownerName}, log)
+
+			if tt.expectErr {
+				require.Error(t, err)
+				if tt.errContains != "" {
+					assert.Contains(t, err.Error(), tt.errContains)
+				}
+				return
+			}
+
+			require.NoError(t, err)
+
+			copied, getErr := fakeClient.CoreV1().ConfigMaps(tt.targetNS).Get(
+				context.Background(), tt.cmName, metav1.GetOptions{})
+			require.NoError(t, getErr)
+			assert.NotNil(t, copied)
+		})
+	}
+}
+
+func TestDeleteConfigMapsWithLabel(t *testing.T) {
+	log := logrus.New()
+
+	fakeClient := fake.NewSimpleClientset(
+		&corev1api.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "cm-1", Namespace: "velero",
+				Labels: map[string]string{testCopyLabel: "du-123"},
+			},
+		},
+		&corev1api.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "cm-2", Namespace: "velero",
+				Labels: map[string]string{testCopyLabel: "du-456"},
+			},
+		},
+	)
+
+	DeleteConfigMapsWithLabel(context.Background(), fakeClient.CoreV1(), "velero",
+		testCopyLabel, "du-123", log)
+
+	_, err := fakeClient.CoreV1().ConfigMaps("velero").Get(
+		context.Background(), "cm-1", metav1.GetOptions{})
+	require.Error(t, err, "cm-1 should be deleted")
+
+	_, err = fakeClient.CoreV1().ConfigMaps("velero").Get(
+		context.Background(), "cm-2", metav1.GetOptions{})
+	assert.NoError(t, err, "cm-2 should still exist")
+}