{ACR} add step in "check-health" command to ensure the identity used for CMK does exist

src/azure-cli/azure/cli/command_modules/acr/_errors.py

@@ -139,3 +139,8 @@ def format_error_message(self, *args):
     "UNEXPECTED_ERROR",
     "An unexpected error occurred."
 )
+
+CMK_MANAGED_IDENTITY_ERROR = ErrorClass(
+    "CMK_ERROR",
+    "The identity used for registry '{}' encryption doesn't exist."
+)

src/azure-cli/azure/cli/command_modules/acr/_utils.py

@@ -562,3 +562,13 @@ def get_scope_map_from_id(cmd, scope_map_id):
     scope_map_name = scope_info[9]
     return acr_scope_map_show(cmd, scope_map_client, registry_name, scope_map_name, resource_group_name)
 # endregion
+
+
+def resolve_identity_client_id(cli_ctx, managed_identity_resource_id):
+    from azure.mgmt.msi import ManagedServiceIdentityClient
+    from azure.cli.core.commands.client_factory import get_mgmt_service_client
+    from msrestazure.tools import parse_resource_id
+
+    res = parse_resource_id(managed_identity_resource_id)
+    client = get_mgmt_service_client(cli_ctx, ManagedServiceIdentityClient, subscription_id=res['subscription'])
+    return client.user_assigned_identities.get(res['resource_group'], res['name']).client_id

src/azure-cli/azure/cli/command_modules/acr/check_health.py

@@ -11,7 +11,7 @@
 from .custom import get_docker_command
 from ._docker_utils import _get_aad_token
 from .helm import get_helm_command
-from ._utils import get_registry_by_name
+from ._utils import get_registry_by_name, resolve_identity_client_id
 from ._errors import ErrorClass
 
 logger = get_logger(__name__)
@@ -295,11 +295,12 @@ def _get_endpoint_and_token_status(cmd, login_server, ignore_errors):
     print_pass("Fetch access token for registry '{}'".format(login_server))
 
 
-def _check_health_connectivity(cmd, registry_name, ignore_errors):
+def _check_registry_health(cmd, registry_name, ignore_errors):
     if registry_name is None:
         logger.warning("Registry name must be provided to check connectivity.")
         return
 
+    # Connectivity
     try:
         registry, _ = get_registry_by_name(cmd.cli_ctx, registry_name)
         login_server = registry.login_server.rstrip('/')
@@ -318,6 +319,24 @@ def _check_health_connectivity(cmd, registry_name, ignore_errors):
     if status_validated:
         _get_endpoint_and_token_status(cmd, login_server, ignore_errors)
 
+    # CMK settings
+    if registry.encryption and registry.encryption.key_vault_properties:  # pylint: disable=too-many-nested-blocks
+        client_id = registry.encryption.key_vault_properties.identity
+        valid_identity = False
+        if registry.identity:
+            valid_identity = (client_id == 'system') and bool(registry.identity.principal_id)  # use system identity?
+            if not valid_identity and registry.identity.user_assigned_identities:
+                for k, v in registry.identity.user_assigned_identities.items():
+                    if v.client_id == client_id:
+                        from msrestazure.azure_exceptions import CloudError
+                        try:
+                            valid_identity = (resolve_identity_client_id(cmd.cli_ctx, k) == client_id)
+                        except CloudError:
+                            pass
+        if not valid_identity:
+            from ._errors import CMK_MANAGED_IDENTITY_ERROR
+            _handle_error(CMK_MANAGED_IDENTITY_ERROR.format_error_message(registry_name), ignore_errors)
+
 
 # General command
 def acr_check_health(cmd,  # pylint: disable useless-return
@@ -332,7 +351,7 @@ def acr_check_health(cmd,  # pylint: disable useless-return
         _get_docker_status_and_version(ignore_errors, yes)
         _get_cli_version()
 
-    _check_health_connectivity(cmd, registry_name, ignore_errors)
+    _check_registry_health(cmd, registry_name, ignore_errors)
 
     if not in_cloud_console:
         _get_helm_version(ignore_errors)

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

@@ -12,7 +12,8 @@
     validate_managed_registry,
     validate_sku_update,
     get_resource_group_name_by_registry_name,
-    user_confirmation
+    user_confirmation,
+    resolve_identity_client_id
 )
 from ._docker_utils import get_login_credentials, EMPTY_GUID
 from .network_rule import NETWORK_RULE_NOT_SUPPORTED
@@ -403,7 +404,7 @@ def _configure_cmk(cmd, registry, resource_group_name, identity, key_encryption_
                                             resource_group=resource_group_name,
                                             resource=identity)
 
-    identity_client_id = _resolve_identity_client_id(cmd.cli_ctx, identity)
+    identity_client_id = resolve_identity_client_id(cmd.cli_ctx, identity)
 
     KeyVaultProperties, EncryptionProperty = cmd.get_models('KeyVaultProperties', 'EncryptionProperty')
     registry.encryption = EncryptionProperty(status='enabled', key_vault_properties=KeyVaultProperties(
@@ -524,7 +525,7 @@ def rotate_key(cmd, client, registry_name, identity=None, key_encryption_key=Non
                 identity = _ensure_identity_resource_id(subscription_id=get_subscription_id(cmd.cli_ctx),
                                                         resource_group=resource_group_name,
                                                         resource=identity)
-                client_id = _resolve_identity_client_id(cmd.cli_ctx, identity)
+                client_id = resolve_identity_client_id(cmd.cli_ctx, identity)
 
         registry.encryption.key_vault_properties.identity = client_id
 
@@ -547,16 +548,6 @@ def _ensure_identity_resource_id(subscription_id, resource_group, resource):
                        name=resource)
 
 
-def _resolve_identity_client_id(cli_ctx, managed_identity_resource_id):
-    from azure.mgmt.msi import ManagedServiceIdentityClient
-    from azure.cli.core.commands.client_factory import get_mgmt_service_client
-    from msrestazure.tools import parse_resource_id
-
-    res = parse_resource_id(managed_identity_resource_id)
-    client = get_mgmt_service_client(cli_ctx, ManagedServiceIdentityClient, subscription_id=res['subscription'])
-    return client.user_assigned_identities.get(res['resource_group'], res['name']).client_id
-
-
 def list_private_link_resources(cmd, client, registry_name, resource_group_name=None):
     resource_group_name = get_resource_group_name_by_registry_name(cmd.cli_ctx, registry_name, resource_group_name)
     return client.list_private_link_resources(resource_group_name, registry_name)