[Backup] az backup vault deleted-vault: Implementing List and Undelete for Deleted Backup Vaults (#32306)

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

Author: 3 people

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

@@ -0,0 +1,53 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import json
+
+from azure.cli.core.util import send_raw_request
+from azure.cli.core.azclierror import HTTPError, AzureResponseError
+
+
+class ARGClient:  # pylint: disable=too-few-public-methods
+    """A lightweight Microsoft ARG API client.
+
+    For what ARG is, please see https://learn.microsoft.com/en-us/azure/governance/resource-graph/overview for details.
+    The reason for directly using this client to request REST is that ARG API does not return "nextLink" data,
+    so the Python SDK "azure-mgmt-resourcegraph" cannot support paging
+
+    """
+
+    def __init__(self, cli_ctx):
+        self._cli_ctx = cli_ctx
+
+        self._endpoint = cli_ctx.cloud.endpoints.resource_manager.rstrip('/')
+        self._resource_provider_uri = 'providers/Microsoft.ResourceGraph/resources'
+        self._api_version = '2021-03-01'
+        self._method = 'post'
+
+    def send(self, query_body):
+        url = f'{self._endpoint}/{self._resource_provider_uri}?api-version={self._api_version}'
+
+        if isinstance(query_body, QueryBody):
+            # Serialize QueryBody object and ignore the None value
+            query_body = json.dumps(query_body,
+                                    default=lambda o: dict((key, value) for key, value in o.__dict__.items() if value))
+
+        try:
+            response = send_raw_request(self._cli_ctx, self._method, url, body=query_body)
+        except HTTPError as ex:
+            raise AzureResponseError(ex.response.json()['error']['message'], ex.response) from ex
+        # Other exceptions like AuthenticationError should not be handled here, so we don't catch CLIError
+
+        if response.text:
+            return response.json()
+
+        return response
+
+
+class QueryBody:  # pylint: disable=too-few-public-methods
+
+    def __init__(self, query, options=None):
+        self.query = query
+        self.options = options

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

@@ -57,6 +57,10 @@ def vaults_cf(cli_ctx, *_):
     return _common_client_factory(cli_ctx).vaults
 
 
+def deleted_vaults_cf(cli_ctx, *_):
+    return _common_client_factory(cli_ctx).deleted_vaults
+
+
 def registered_identities_cf(cli_ctx, *_):
     return _common_client_factory(cli_ctx).registered_identities
 

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

@@ -634,3 +634,42 @@
   - name: Delete resource guard mapping of the Recovery Services vault.
     text: az backup vault resource-guard-mapping delete --resource-group MyResourceGroup --name MyVault
 """
+
+helps['backup deleted-vault'] = """
+type: group
+short-summary: Manage soft-deleted Recovery Services vaults.
+"""
+
+helps['backup deleted-vault list'] = """
+type: command
+short-summary: List soft-deleted Recovery Services vaults.
+examples:
+  - name: List soft-deleted vaults in a specific location under the active subscription.
+    text: az backup deleted-vault list --location eastus
+"""
+
+helps['backup deleted-vault get'] = """
+type: command
+short-summary: Get details of a soft-deleted Recovery Services vault.
+examples:
+  - name: Get details of a soft-deleted vault by name and location.
+    text: az backup deleted-vault get --location eastus --name deletedVaultName
+"""
+
+helps['backup deleted-vault undelete'] = """
+type: command
+short-summary: Restore a soft-deleted Recovery Services vault.
+examples:
+  - name: Restore a soft-deleted vault by name and location.
+    text: az backup deleted-vault undelete --name MyVault --location eastus
+  - name: Restore a soft-deleted vault using its ARM ID.
+    text: az backup deleted-vault undelete --ids /subscriptions/{subscription-id}/locations/{location}/deletedVaults/{deleted-vault-name}
+"""
+
+helps['backup deleted-vault list-containers'] = """
+type: command
+short-summary: List backup containers in a soft-deleted vault.
+examples:
+  - name: List backup containers in a soft-deleted vault.
+    text: az backup deleted-vault list-containers --name MyVault
+"""

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

@@ -108,7 +108,7 @@ def load_arguments(self, _):
         c.argument('azure_monitor_alerts_for_job_failures', options_list=['--job-failure-alerts'], arg_type=get_enum_type(enable_disable_options), help='Use this property to specify whether built-in Azure Monitor alerts should be received for every job failure.')
         c.argument('immutability_state', arg_type=get_enum_type(allowed_immutability_options), help='Use this parameter to configure immutability settings for the vault. By default, immutability is "Disabled" for the vault. "Unlocked" means that immutability is enabled for the vault and can be reversed. "Locked" means that immutability is enabled for the vault and cannot be reversed.')
         c.argument('cross_subscription_restore_state', arg_type=get_enum_type(enable_disable_permadisable_options), help='Use this parameter to configure cross subscription restore settings for the vault. By default, the property is "Enabled" for the vault.')
-        # TODO Re-add once the new SDK is in place
+        # TODO May add the soft_delete_retention_period_in_days parameter later. The other will not be exposed.
         # c.argument('soft_delete_state', options_list=['--soft-delete-state', '--soft-delete-feature-state'], arg_type=get_enum_type(allowed_softdelete_options), help='Set soft-delete feature state for a Recovery Services Vault.')
         # c.argument('soft_delete_retention_period_in_days', type=int, options_list=['--soft-delete-duration'], help='Set soft-delete retention duration time in days for a Recovery Services Vault.')
 
@@ -119,7 +119,7 @@ def load_arguments(self, _):
         c.argument('azure_monitor_alerts_for_job_failures', options_list=['--job-failure-alerts'], arg_type=get_enum_type(enable_disable_options), help='Use this property to specify whether built-in Azure Monitor alerts should be received for every job failure.')
         c.argument('immutability_state', arg_type=get_enum_type(allowed_immutability_options), help='Use this parameter to configure immutability settings for the vault. By default, immutability is "Disabled" for the vault. "Unlocked" means that immutability is enabled for the vault and can be reversed. "Locked" means that immutability is enabled for the vault and cannot be reversed.')
         c.argument('cross_subscription_restore_state', arg_type=get_enum_type(enable_disable_permadisable_options), help='Use this parameter to configure cross subscription restore settings for the vault. By default, the property is "Enabled" for the vault.')
-        # TODO Re-add once the new SDK is in place
+        # TODO Discussion with Rishav once Enhanced Soft Delete is in place. We can only expose the latter, and might have to disable it from vaultconfig API
         # c.argument('soft_delete_state', options_list=['--soft-delete-state', '--soft-delete-feature-state'], arg_type=get_enum_type(allowed_softdelete_options), help='Set soft-delete feature state for a Recovery Services Vault.')
         # c.argument('soft_delete_retention_period_in_days', type=int, options_list=['--soft-delete-duration'], help='Set soft-delete retention duration time in days for a Recovery Services Vault.')
         c.argument('backup_storage_redundancy', arg_type=get_enum_type(['GeoRedundant', 'LocallyRedundant', 'ZoneRedundant']), help='Set backup storage properties for a Recovery Services vault.')
@@ -164,6 +164,15 @@ def load_arguments(self, _):
     with self.argument_context('backup vault encryption show') as c:
         c.argument('vault_name', vault_name_type, options_list=['--name', '-n'], id_part='name')
 
+    # Deleted Vault
+    with self.argument_context('backup deleted-vault') as c:
+        c.argument('location', help='Location of the deleted vault.')
+
+    for command in ['get', 'undelete', 'list-containers']:
+        with self.argument_context('backup deleted-vault ' + command) as c:
+            c.argument('deleted_vault_name', vault_name_type, options_list=['--name', '-n'], help='Name of the deleted vault.')
+            c.argument('deleted_vault_id', options_list=['--ids', '--deleted-vault-id'], help='ID of the deleted vault.')
+
     # Container
     with self.argument_context('backup container') as c:
         c.argument('vault_name', vault_name_type, id_part='name')
@@ -414,7 +423,7 @@ def load_arguments(self, _):
         c.argument('tenant_id', help='ID of the tenant if the Resource Guard protecting the vault exists in a different tenant.')
         c.argument('disk_access_option', arg_type=get_enum_type(allowed_disk_access_options), help='Specify the disk access option for target disks.')
         c.argument('target_disk_access_id', help='Specify the target disk access ID when --disk-access-option is set to EnablePrivateAccessForAllDisks')
-        c.argument('cvm_os_des_id', options_list=['--cvm-os-des-id', '--cvm-os-disk-encryption-set-id'], help='Specify the Disk Encryption Set ID to use for OS disk encryption during restore of a Confidential VM. This is applicable only for Confidential VMs with managed disks. Please ensure that Disk Encryption Set has access to the Key vault.')
+        c.argument('cvm_os_des_id', help='Disk encryption set ID for the OS disk of confidential VMs. This is used to encrypt the OS disk during restore.')
 
     with self.argument_context('backup restore restore-azurefileshare') as c:
         c.argument('resolve_conflict', resolve_conflict_type)

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

@@ -4,12 +4,13 @@
 # --------------------------------------------------------------------------------------------
 
 from azure.cli.core.commands import CliCommandType
-from azure.cli.command_modules.backup._client_factory import vaults_cf, backup_protection_containers_cf, \
-    protection_policies_cf, backup_policies_cf, protected_items_cf, backups_cf, backup_jobs_cf, \
-    job_details_cf, job_cancellations_cf, recovery_points_cf, restores_cf, backup_storage_configs_non_crr_cf, \
-    item_level_recovery_connections_cf, backup_protected_items_cf, backup_protectable_items_cf, \
-    protection_containers_cf, protection_intent_cf, backup_resource_encryption_config_cf, resource_guard_proxy_cf, \
-    deleted_protection_containers_cf  # pylint: disable=unused-variable
+from azure.cli.command_modules.backup._client_factory import (
+    vaults_cf, deleted_vaults_cf, backup_protection_containers_cf, protection_policies_cf,
+    backup_policies_cf, protected_items_cf, backups_cf, backup_jobs_cf, job_details_cf,
+    job_cancellations_cf, recovery_points_cf, restores_cf, backup_storage_configs_non_crr_cf,
+    item_level_recovery_connections_cf, backup_protected_items_cf, backup_protectable_items_cf,
+    protection_containers_cf, protection_intent_cf, backup_resource_encryption_config_cf,
+    resource_guard_proxy_cf, deleted_protection_containers_cf)  # pylint: disable=unused-variable
 from azure.cli.command_modules.backup._exception_handler import backup_exception_handler
 from azure.cli.command_modules.backup._format import (
     transform_container_list, transform_policy_list, transform_item_list, transform_job_list,
@@ -43,6 +44,12 @@ def load_command_table(self, _):
         g.custom_command('encryption update', 'update_encryption')
         g.custom_command('encryption show', 'show_encryption', client_factory=backup_resource_encryption_config_cf)
 
+    with self.command_group('backup deleted-vault', backup_custom, client_factory=deleted_vaults_cf, exception_handler=backup_exception_handler) as g:
+        g.custom_command('list', 'list_deleted_vaults')
+        g.custom_command('get', 'get_deleted_vault')
+        g.custom_command('undelete', 'undelete_vault')
+        g.custom_command('list-containers', 'list_deleted_vault_containers')
+
     with self.command_group('backup vault resource-guard-mapping', backup_custom, client_factory=resource_guard_proxy_cf, exception_handler=backup_exception_handler) as g:
         g.show_command('update', 'update_resource_guard_mapping')
         g.show_command('show', 'show_resource_guard_mapping')