acr: add notary version compatibility checks (#11792)

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

@@ -67,6 +67,18 @@ def format_error_message(self, *args):
 )
 
 
+# NOTARY ERRORS
+NOTARY_COMMAND_ERROR = ErrorClass(
+    "NOTARY_COMMAND_ERROR",
+    "Please verify if notary is installed."
+)
+
+NOTARY_VERSION_ERROR = ErrorClass(
+    "NOTARY_VERSION_ERROR",
+    "An error occurred while retrieving notary version. Please make sure that you have the latest Azure CLI version, and that you are using the recommended notary version."
+)
+
+
 # CONNECTIVITY ERRORS
 CONNECTIVITY_DNS_ERROR = ErrorClass(
     "CONNECTIVITY_DNS_ERROR",

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

@@ -25,6 +25,8 @@
 MIN_HELM_VERSION = "2.11.0"
 HELM_VERSION_REGEX = re.compile(r'(SemVer|Version):"v([.\d]+)"')
 ACR_CHECK_HEALTH_MSG = "Try running 'az acr check-health -n {} --yes' to diagnose this issue."
+RECOMMENDED_NOTARY_VERSION = "0.6.0"
+NOTARY_VERSION_REGEX = re.compile(r'Version:\s+([.\d]+)')
 
 
 # Utilities functions
@@ -158,7 +160,7 @@ def _get_helm_version(ignore_errors):
     if match_obj:
         output = match_obj.group(2)
 
-    print("Helm version: {}".format(output), file=sys.stderr)
+    logger.warning("Helm version: %s", output)
 
     # Display an error message if the current helm version < min required version
     if match_obj and LooseVersion(output) < LooseVersion(MIN_HELM_VERSION):
@@ -168,6 +170,46 @@ def _get_helm_version(ignore_errors):
         _handle_error(obsolete_ver_error, ignore_errors)
 
 
+def _get_notary_version(ignore_errors):
+    from ._errors import NOTARY_VERSION_ERROR
+    from .notary import get_notary_command
+    from distutils.version import LooseVersion  # pylint: disable=import-error,no-name-in-module
+
+    # Notary command check
+    notary_command, error = get_notary_command(is_diagnostics_context=True)
+
+    if error:
+        _handle_error(error, ignore_errors)
+        return
+
+    # Notary version check
+    output, warning, stderr = _subprocess_communicate([notary_command, "version"])
+
+    if stderr:
+        _handle_error(NOTARY_VERSION_ERROR.append_error_message(stderr), ignore_errors)
+        return
+
+    if warning:
+        logger.warning(warning)
+
+    # Retrieve the notary version if regex pattern is found
+    match_obj = NOTARY_VERSION_REGEX.search(output)
+    if match_obj:
+        output = match_obj.group(1)
+
+    logger.warning("Notary version: %s", output)
+
+    # Display error if the current version does not match the recommended version
+    if match_obj and LooseVersion(output) != LooseVersion(RECOMMENDED_NOTARY_VERSION):
+        version_msg = "upgrade"
+        if LooseVersion(output) > LooseVersion(RECOMMENDED_NOTARY_VERSION):
+            version_msg = "downgrade"
+        obsolete_ver_error = NOTARY_VERSION_ERROR.set_error_message(
+            "Current notary version is not recommended. Please {} your notary client to version {}."
+            .format(version_msg, RECOMMENDED_NOTARY_VERSION))
+        _handle_error(obsolete_ver_error, ignore_errors)
+
+
 # Checks for the connectivity
 # Check DNS lookup and access to challenge endpoint
 def _get_registry_status(login_server, registry_name, ignore_errors):
@@ -289,5 +331,6 @@ def acr_check_health(cmd,  # pylint: disable useless-return
 
     if not in_cloud_console:
         _get_helm_version(ignore_errors)
+        _get_notary_version(ignore_errors)
 
     print(FAQ_MESSAGE, file=sys.stderr)

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

@@ -0,0 +1,36 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from knack.util import CLIError
+from knack.log import get_logger
+
+logger = get_logger(__name__)
+
+
+def get_notary_command(is_diagnostics_context=False):
+    from ._errors import NOTARY_COMMAND_ERROR
+    from subprocess import PIPE, Popen
+    from platform import system
+
+    if system() == "Windows":
+        notary_command = "notary.exe"
+    else:
+        notary_command = "notary"
+
+    try:
+        p = Popen([notary_command, "--help"], stdout=PIPE, stderr=PIPE)
+        _, stderr = p.communicate()
+    except OSError as e:
+        logger.debug("Could not run '%s' command. Exception: %s", notary_command, str(e))
+        if is_diagnostics_context:
+            return None, NOTARY_COMMAND_ERROR
+        raise CLIError(NOTARY_COMMAND_ERROR.get_error_message())
+
+    if stderr:
+        if is_diagnostics_context:
+            return None, NOTARY_COMMAND_ERROR.append_error_message(stderr.decode())
+        raise CLIError(NOTARY_COMMAND_ERROR.append_error_message(stderr.decode()).get_error_message())
+
+    return notary_command, None