[Dataprotection] Add support for automatically granting MSI permission

src/dataprotection/azext_dataprotection/manual/_params.py

@@ -33,7 +33,9 @@
     get_job_operation_values,
     get_datasource_types,
     get_rehydration_priority_values,
-    get_secret_store_type_values
+    get_secret_store_type_values,
+    get_backup_operation_values,
+    get_permission_scope_values
 )
 
 
@@ -79,6 +81,14 @@ def load_arguments(self, _):
         c.argument('subscriptions', type=str, nargs='+', help="List of subscription Ids.")
         c.argument('protection_status', arg_type=get_enum_type(get_protection_status_values()), nargs='+', help="specify protection status.")
         c.argument('datasource_id', type=str, nargs='+', help="specify datasource id filter to apply.")
+
+    with self.argument_context('dataprotection backup-instance update-msi-permissions') as c:
+        c.argument('operation', arg_type=get_enum_type(get_backup_operation_values()), help="List of possible operations")
+        c.argument('datasource_type', arg_type=get_enum_type(get_datasource_types()), help="Specify the datasource type of the resource to be backed up")
+        c.argument('vault_name', type=str, help="Name of the vault.")
+        c.argument('permissions_scope', arg_type=get_enum_type(get_permission_scope_values()), help="Scope for assigning permissions to the backup vault")
+        c.argument('backup_instance', type=validate_file_or_dict, help='Request body for operation Expected value: '
+                   'json-string/@json-file.')
 
     with self.argument_context('dataprotection job list-from-resourcegraph') as c:
         c.argument('subscriptions', type=str, nargs='+', help="List of subscription Ids.")

src/dataprotection/azext_dataprotection/manual/commands.py

@@ -30,6 +30,7 @@ def load_command_table(self, _):
         g.custom_command('restore trigger', 'dataprotection_backup_instance_restore_trigger', supports_no_wait=True)
         g.custom_command('update-policy', "dataprotection_backup_instance_update_policy", supports_no_wait=True)
         g.custom_command('list-from-resourcegraph', 'dataprotection_backup_instance_list_from_resourcegraph', client_factory=cf_resource_graph_client)
+        g.custom_command('update-msi-permissions', 'dataprotection_backup_instance_update_msi_permissions', client_factory=cf_backup_vault)
 
     with self.command_group('dataprotection backup-policy', exception_handler=exception_handler) as g:
         g.custom_command('get-default-policy-template', "dataprotection_backup_policy_get_default_policy_template")

src/dataprotection/azext_dataprotection/manual/custom.py

@@ -139,6 +139,61 @@ def dataprotection_backup_instance_list_from_resourcegraph(client, datasource_ty
     return response.data
 
 
+def dataprotection_backup_instance_update_msi_permissions(cmd, client, resource_group_name, datasource_type, vault_name, operation, permissions_scope, backup_instance=None):
+    backup_vault = client.get(resource_group_name=resource_group_name,
+                vault_name=vault_name)
+    principal_id = backup_vault.identity.principal_id
+
+    role_assignments_arr = []
+
+    if operation == 'Backup' and backup_instance is None:
+        raise CLIError("Backup instance needs to be provided when operation is Backup")
+
+    from azure.cli.command_modules.role.custom import list_role_assignments, create_role_assignment
+
+    if datasource_type == 'AzureDisk':
+        # Create scope variables for role assignment
+        resource_scope_assignment = backup_instance['properties']['data_source_info']['resource_id']
+        snapshot_rg_scope_assignment = backup_instance['properties']['policy_info']['policy_parameters']['data_store_parameters_list'][0]['resource_group_id']
+        resource_scope = resource_scope_assignment
+        snapshot_rg_scope = snapshot_rg_scope_assignment
+
+        if permissions_scope == 'Resource Group':
+            resource_scope_assignment = "/".join(resource_scope_assignment.split("/")[:5]) # Snapshot RG is already in RG scope, so change for resource
+        elif permissions_scope == 'Subscription':
+            # Change scope for both resource and Snapshot RG
+            resource_scope_assignment = "/".join(resource_scope_assignment.split("/")[:3])
+            snapshot_rg_scope_assignment = "/".join(snapshot_rg_scope_assignment.split("/")[:3])
+
+
+        role_assignments = [obj['roleDefinitionName'] for obj in list_role_assignments(cmd, assignee=principal_id, scope=resource_scope, include_inherited=True)]
+        if 'Disk Backup Reader' not in role_assignments:
+            role_assignments_arr.append(create_role_assignment(cmd, role='Disk Backup Reader', assignee=principal_id, scope=resource_scope_assignment))
+
+        role_assignments = [obj['roleDefinitionName'] for obj in list_role_assignments(cmd, assignee=principal_id, scope=snapshot_rg_scope, include_inherited=True)]
+        if 'Disk Snapshot Contributor' not in role_assignments:
+            role_assignments_arr.append(create_role_assignment(cmd, role='Disk Snapshot Contributor', assignee=principal_id, scope=snapshot_rg_scope_assignment))
+
+        return role_assignments_arr
+
+    if datasource_type == 'AzureBlob':
+        storage_account_scope = backup_instance['properties']['data_source_info']['resource_id']
+        storage_account_scope_assignment = storage_account_scope
+
+        if permissions_scope == 'Resource Group':
+            storage_account_scope_assignment = "/".join(storage_account_scope_assignment.split("/")[:5])
+        elif permissions_scope == 'Subscription':
+            storage_account_scope_assignment = "/".join(storage_account_scope_assignment.split("/")[:3])
+
+        role_assignments = [obj['roleDefinitionName'] for obj in list_role_assignments(cmd, assignee=principal_id, scope=storage_account_scope, include_inherited=True)]
+        if 'Storage Account Backup Contributor' not in role_assignments:
+            role_assignments_arr.append(create_role_assignment(cmd, role='Storage Account Backup Contributor', assignee=principal_id, scope=storage_account_scope_assignment))
+
+        return role_assignments_arr
+
+    raise CLIError("Invalid params passed")
+
+
 def dataprotection_job_list_from_resourcegraph(client, datasource_type, resource_groups=None, vaults=None,
                                                subscriptions=None, start_time=None, end_time=None,
                                                status=None, operation=None, datasource_id=None):

src/dataprotection/azext_dataprotection/manual/enums.py

@@ -65,6 +65,13 @@ def get_secret_store_type_values():
     return ['AzureKeyVault']
 
 
+def get_backup_operation_values():
+    return ['Backup']
+
+def get_permission_scope_values():
+    return ['Resource', 'Resource Group', 'Subscription']
+
+
 def get_datasource_types():
     from azext_dataprotection.manual import helpers
     return helpers.get_supported_datasource_types()