[Backup] backup protection reconfigure: Add new command to support reconfiguring backup to an alternate vault (#32193)

Co-authored-by: Copilot <[email protected]>

Author: zubairabid

src/azure-cli/azure/cli/command_modules/backup/_help.py

@@ -583,6 +583,43 @@
   - name: Create/Update resource guard mapping of the Recovery Services vault.
     text: az backup vault resource-guard-mapping update --resource-group MyResourceGroup --name MyVault --resource-guard-id MyResourceGuardId
 """
+
+helps['backup protection'] = """
+type: group
+short-summary: Manage protection of items in a Recovery Services vault.
+"""
+
+helps['backup protection reconfigure'] = """
+type: command
+short-summary: Reconfigures backup protection from an old vault to a new vault.
+examples:
+  - name: Reconfigure VM backup from one vault to another
+    text: |
+        az backup protection reconfigure \\
+            --vault-name OldVault \\
+            --resource-group OldVaultRG \\
+            --container-name myVM \\
+            --item-name myVM \\
+            --backup-management-type AzureIaasVM \\
+            --new-vault-name NewVault \\
+            --new-vault-resource-group NewVaultRG \\
+            --new-policy-name DailyPolicy \\
+            --retain-as-per-policy
+  - name: Reconfigure VM backup with cross-tenant MUA scenario
+    text: |
+        az backup protection reconfigure \\
+            --vault-name OldVault \\
+            --resource-group OldVaultRG \\
+            --container-name myVM \\
+            --item-name myVM \\
+            --backup-management-type AzureIaasVM \\
+            --new-vault-name NewVault \\
+            --new-vault-resource-group NewVaultRG \\
+            --new-policy-name DailyPolicy \\
+            --retain-as-per-policy \\
+            --tenant-id 12345678-1234-1234-1234-123456789012
+"""
+
 helps['backup vault resource-guard-mapping show'] = """
 type: command
 short-summary: Get resource guard mapping of the Recovery Services vault.

src/azure-cli/azure/cli/command_modules/backup/_params.py

@@ -326,6 +326,17 @@ def load_arguments(self, _):
         c.argument('workload_type', workload_type)
         c.argument('tenant_id', help='ID of the tenant if the Resource Guard protecting the vault exists in a different tenant.')
 
+    with self.argument_context('backup protection reconfigure') as c:
+        c.argument('container_name', container_name_type, id_part='child_name_2')
+        c.argument('item_name', item_name_type, id_part='child_name_3')
+        c.argument('backup_management_type', backup_management_type)
+        c.argument('workload_type', workload_type)
+        c.argument('new_vault_name', help='Name of the destination Recovery Services vault.')
+        c.argument('new_vault_resource_group', options_list=['--new-vault-resource-group', '--new-rg'], help='Resource group name of the destination Recovery Services vault.')
+        c.argument('new_policy_name', options_list=['--new-policy-name'], help='Name of the backup policy in the destination vault.')
+        c.argument('retain_as_per_policy', arg_type=get_three_state_flag(), help='Retain existing recovery points as per current backup policy when stopping protection in the source vault (the source vault is always the one specified by --vault-name/--resource-group).')
+        c.argument('tenant_id', help='ID of the tenant if the Resource Guard protecting the source vault exists in a different tenant.')
+
     with self.argument_context('backup protection check-vm') as c:
         c.argument('vm_id', help='ID of the virtual machine to be checked for protection.', deprecate_info=c.deprecate(redirect='--vm', hide=True))
 

src/azure-cli/azure/cli/command_modules/backup/_validators.py

@@ -134,3 +134,36 @@ def validate_archive_restore(recovery_point, rehydration_priority):
             rehydration_priority is None):
         raise InvalidArgumentValueError("""The selected recovery point is in archive tier, provide additional
         parameters of rehydration duration and rehydration priority.""")
+
+
+def validate_reconfigure_cli_parameters(source_vault_name, source_vault_resource_group, new_vault_name, new_vault_resource_group,
+                                        backup_management_type, workload_type):
+    """Top-level CLI validation for backup reconfiguration (name / type sanity checks).
+
+    Note: Source vault is always the vault specified by --vault-name / --resource-group in the command context."""
+
+    # Ensure old and new vaults are different
+    if (source_vault_name.lower() == new_vault_name.lower() and
+            source_vault_resource_group.lower() == new_vault_resource_group.lower()):
+        raise InvalidArgumentValueError("Source and destination vaults cannot be the same")
+
+    # Validate workload type is provided for Azure workloads
+    if backup_management_type.lower() == 'azureworkload' and not workload_type:
+        raise RequiredArgumentMissingError("Workload type is required for Azure workload reconfiguration")
+
+    # Validate incompatible parameter combinations
+    if backup_management_type.lower() == 'azureiaasvm' and workload_type:
+        raise MutuallyExclusiveArgumentError("Workload type should not be specified for VM backup reconfiguration")
+
+
+def validate_afs_policy_compatibility(old_policy_type, new_policy_type):
+    """Validate AFS policy type compatibility for reconfiguration"""
+
+    # AFS policy validation: cannot go from vault-based to snapshot-based
+    if old_policy_type == 'vault' and new_policy_type == 'snapshot':
+        raise InvalidArgumentValueError("""
+        Cannot reconfigure from vault-based policy to snapshot-based policy for Azure File Share.
+        This transition is not supported.""")
+
+    # Allow snapshot to vault transition
+    return True

src/azure-cli/azure/cli/command_modules/backup/commands.py

@@ -85,6 +85,7 @@ def load_command_table(self, _):
         g.custom_command('auto-disable-for-azurewl', 'disable_auto_for_azure_wl', client_factory=protection_intent_cf)
         g.custom_command('resume', 'resume_protection')
         g.custom_command('undelete', 'undelete_protection')
+        g.custom_command('reconfigure', 'reconfigure_backup_protection', client_factory=backup_protected_items_cf)
 
     with self.command_group('backup item', backup_custom_base, client_factory=protected_items_cf, exception_handler=backup_exception_handler) as g:
         g.show_command('show', 'show_item', client_factory=backup_protected_items_cf, table_transformer=transform_item)

src/azure-cli/azure/cli/command_modules/backup/custom.py

@@ -1937,3 +1937,80 @@ def _run_client_script_for_linux(client_scripts):
 def _validate_restore_disk_parameters(restore_only_osdisk, diskslist):
     if restore_only_osdisk and diskslist is not None:
         logger.warning("Value of diskslist parameter will be ignored as restore-only-osdisk is set to be true.")
+
+
+def reconfigure_vm_protection(cmd, item, source_vault_name, source_vault_resource_group,
+                              new_vault_name, new_vault_resource_group,
+                              new_policy_name, retain_as_per_policy, tenant_id):
+    """Workload-specific implementation: Reconfigure Azure IaaS VM protection to a new vault.
+
+    Assumes all high-level validations and item retrieval already performed in custom_base.reconfigure_backup_protection.
+    """
+    logger.warning("(VM) Starting backup protection reconfiguration from source vault '%s' to destination vault '%s'...",
+                   source_vault_name, new_vault_name)
+
+    # Step 1: Stop protection in old vault (retain data)
+    logger.warning("Step 1: Stopping protection in old vault...")
+    _disable_protection_in_old_vault(cmd, source_vault_resource_group, source_vault_name,
+                                     item, retain_as_per_policy, tenant_id)
+
+    # Step 2: Enable protection in new vault
+    logger.warning("Step 2: Enabling protection in new vault...")
+    enable_result = _enable_vm_protection_in_new_vault(cmd, new_vault_resource_group, new_vault_name,
+                                                       item, new_policy_name)
+
+    logger.warning("(VM) Backup protection reconfiguration completed successfully.")
+    return enable_result
+
+
+def _disable_protection_in_old_vault(cmd, vault_resource_group, vault_name, item, retain_as_per_policy, tenant_id):
+    """Stop protection in the old vault"""
+    protected_items_client = protected_items_cf(cmd.cli_ctx)
+
+    # Use the existing disable_protection function
+    return disable_protection(cmd, protected_items_client, vault_resource_group, vault_name, item,
+                              retain_as_per_policy, tenant_id)
+
+
+def _enable_vm_protection_in_new_vault(cmd, vault_resource_group, vault_name, old_item, policy_name):
+    """Enable VM protection in new vault"""
+
+    # Extract VM information from the protected item
+    vm_id = _extract_vm_id_from_protected_item(old_item)
+
+    diskslist = _extract_disk_list_from_protected_item(old_item)
+
+    # Use the existing enable_protection_for_vm function
+    protected_items_client = protected_items_cf(cmd.cli_ctx)
+    return enable_protection_for_vm(cmd, protected_items_client, vault_resource_group, vault_name,
+                                    vm_id, policy_name, diskslist)
+
+
+def _extract_vm_id_from_protected_item(protected_item):
+    """Extract VM resource ID from protected item"""
+    # The VM ID is typically in the sourceResourceId property
+    if hasattr(protected_item.properties, 'source_resource_id'):
+        return protected_item.properties.source_resource_id
+
+    # Fallback: try to extract from the virtual machine id property
+    if hasattr(protected_item.properties, 'virtual_machine_id'):
+        return protected_item.properties.virtual_machine_id
+
+    raise CLIError("Could not extract VM resource ID from protected item")
+
+
+def _extract_disk_list_from_protected_item(protected_item):
+    """Extract the list of protected disks from the protected item"""
+    if (hasattr(protected_item.properties, 'extended_info') and
+            protected_item.properties.extended_info and
+            hasattr(protected_item.properties.extended_info, 'disk_exclusion_properties') and
+            protected_item.properties.extended_info.disk_exclusion_properties):
+
+        disk_props = protected_item.properties.extended_info.disk_exclusion_properties
+
+        # Return the list of LUNs that were originally protected
+        if hasattr(disk_props, 'disk_lun_list'):
+            return disk_props.disk_lun_list
+
+    # If we can't extract disk info, return None to protect all disks
+    return None

src/azure-cli/azure/cli/command_modules/backup/custom_afs.py

@@ -18,8 +18,8 @@
 from azure.cli.core.util import CLIError
 from azure.cli.command_modules.backup._client_factory import protection_containers_cf, protectable_containers_cf, \
     protection_policies_cf, backup_protection_containers_cf, backup_protectable_items_cf, \
-    resources_cf, backup_protected_items_cf
-from azure.cli.core.azclierror import ArgumentUsageError
+    resources_cf, backup_protected_items_cf, protected_items_cf
+from azure.cli.core.azclierror import ArgumentUsageError, ValidationError
 
 from azure.mgmt.recoveryservicesbackup.activestamp import RecoveryServicesBackupClient
 from azure.cli.core.commands.client_factory import get_mgmt_service_client
@@ -32,6 +32,55 @@
 workload_type = "AzureFileShare"
 
 
+def reconfigure_afs_protection(cmd, item, source_vault_name, source_vault_rg,
+                               new_vault_name, new_vault_rg,
+                               new_policy_name, retain_as_per_policy, tenant_id):
+    """Reconfigure Azure File Share protection to a new vault and policy.
+
+    Steps:
+    1. Disable protection (retain or stop based on flag) in source vault.
+    2. Unregister storage account container (if no remaining protected items) from source vault.
+    3. Ensure storage account is registered / refreshed in destination vault.
+    4. Enable protection for the same file share name in destination vault with new policy.
+    5. Return the newly protected item from destination vault.
+    """
+    logger.warning("For Storage reconfigure protection, all backup items within the "
+                   "container must have protection disabled first.")
+
+    # 1. Disable in old vault (retain as per policy if requested)
+    items_client = protected_items_cf(cmd.cli_ctx)
+    disable_protection(cmd, items_client, source_vault_rg, source_vault_name, item,
+                       retain_as_per_policy, tenant_id)
+
+    # 2. Unregister container in old vault only if this was the last protected item for that storage account
+    _maybe_unregister_storage_account(cmd, backup_protected_items_cf(cmd.cli_ctx), source_vault_rg, source_vault_name,
+                                      item.properties.container_name)
+
+    # 3. Enable protection in destination vault - also registers storage account in destination vault
+    new_item = enable_for_AzureFileShare(cmd, items_client, new_vault_rg, new_vault_name, item.name,
+                                         item.properties.container_name, new_policy_name)
+    return new_item
+
+
+def _maybe_unregister_storage_account(cmd, client, resource_group_name, vault_name, container_name):
+    """Unregister the storage account container if no more protected items exist in the source vault."""
+    items = common.list_items(cmd, client, resource_group_name, vault_name,
+                              workload_type=workload_type, container_name=container_name,
+                              container_type=backup_management_type)
+    remaining = [pi for pi in items if pi.properties.protection_state.lower() == 'protected']
+    if remaining:
+        raise ValidationError('Cannot unregister container as other items are still protected.')
+
+    # Attempt unregister
+    try:
+        containers_client = protection_containers_cf(cmd.cli_ctx)
+        unregister_afs_container(cmd, containers_client, vault_name, resource_group_name, container_name)
+    except Exception as ex:  # pylint: disable=broad-except
+        logger.warning('Skipping unregister workload container of container %s due to a failure: %s.'
+                       ' Continuing the operation, but if the container is still registered, it may need to be '
+                       'unregistered manually for the operation to succeed.', container_name, str(ex))
+
+
 def enable_for_AzureFileShare(cmd, client, resource_group_name, vault_name, afs_name,
                               storage_account_name, policy_name):
 

src/azure-cli/azure/cli/command_modules/backup/custom_base.py

@@ -6,8 +6,9 @@
 from azure.cli.command_modules.backup import custom
 from azure.cli.command_modules.backup import custom_afs
 from azure.cli.command_modules.backup import custom_help
-import azure.cli.command_modules.backup.custom_common as common
 from azure.cli.command_modules.backup import custom_wl
+import azure.cli.command_modules.backup.custom_common as common
+from azure.cli.command_modules.backup._validators import validate_reconfigure_cli_parameters
 from azure.cli.command_modules.backup._client_factory import protection_policies_cf, backup_protected_items_cf, \
     backup_protection_containers_cf, backup_protectable_items_cf, registered_identities_cf, vaults_cf
 from azure.cli.core.azclierror import ValidationError, RequiredArgumentMissingError, InvalidArgumentValueError, \
@@ -19,6 +20,51 @@
 fabric_name = "Azure"
 
 
+def reconfigure_backup_protection(cmd, client, resource_group_name, vault_name, container_name, item_name,
+                                  new_vault_name, new_vault_resource_group,
+                                  new_policy_name, backup_management_type, workload_type=None,
+                                  retain_as_per_policy=False, tenant_id=None):
+    """Entry point for reconfiguring backup protection across vaults.
+
+    This function performs common validation and dispatches to workload specific implementations
+    in custom.py (VM), custom_afs.py (AFS) and custom_wl.py (Workload)."""
+
+    # Common CLI-level validation (different vaults, workload type rules etc.)
+    validate_reconfigure_cli_parameters(vault_name, resource_group_name,
+                                        new_vault_name, new_vault_resource_group,
+                                        backup_management_type, workload_type)
+
+    # Fetch the protected item from old vault (use existing show_item logic)
+    item = show_item(cmd, client, resource_group_name, vault_name,
+                     container_name, item_name, backup_management_type, workload_type)
+    custom_help.validate_item(item)
+    if isinstance(item, list):  # Ambiguous friendly name
+        raise ValidationError("Multiple items found. Please use native container and item names.")
+
+    # Item-level validation (state, workload specifics)
+    from azure.mgmt.recoveryservicesbackup.activestamp.models import ProtectionState
+    if item.properties.protection_state not in [ProtectionState.protected,
+                                                ProtectionState.protection_stopped]:
+        raise ValidationError(f"Reconfiguration only supported for items in states: Protected or "
+                              f"ProtectionStopped. Current state: "
+                              f"{item.properties.protection_state}")
+
+    # Dispatch by backup management type
+    dispatch_type = backup_management_type.lower()
+    if dispatch_type == 'azureiaasvm':
+        return custom.reconfigure_vm_protection(cmd, item, vault_name, resource_group_name,
+                                                new_vault_name, new_vault_resource_group,
+                                                new_policy_name, retain_as_per_policy, tenant_id)
+    if dispatch_type == 'azurestorage':
+        return custom_afs.reconfigure_afs_protection(cmd, item, vault_name, resource_group_name,
+                                                     new_vault_name, new_vault_resource_group,
+                                                     new_policy_name, retain_as_per_policy, tenant_id)
+    if dispatch_type == 'azureworkload':
+        return custom_wl.reconfigure_wl_protection(cmd, item, vault_name, resource_group_name,
+                                                   new_vault_name, new_vault_resource_group,
+                                                   new_policy_name, workload_type, retain_as_per_policy, tenant_id)
+
+
 def show_container(cmd, client, name, resource_group_name, vault_name, backup_management_type=None,
                    status="Registered", use_secondary_region=None):
     return common.show_container(cmd, client, name, resource_group_name, vault_name, backup_management_type, status,