[Containerapp] az containerapp create: Support other cloud for acr

src/azure-cli/azure/cli/command_modules/containerapp/_constants.py

@@ -14,6 +14,7 @@
 LONG_POLLING_INTERVAL_SECS = 10
 
 ACR_IMAGE_SUFFIX = ".azurecr.io"
+ACR_IMAGE_SUFFIXES = (".azurecr.io", ".azurecr.cn", ".azurecr.us")
 
 CONTAINER_APPS_SDK_MODELS = "azure.cli.command_modules.containerapp._sdk_models"
 

src/azure-cli/azure/cli/command_modules/containerapp/_up_utils.py

@@ -51,7 +51,9 @@
     format_location,
     is_docker_running,
     get_pack_exec_path,
-    get_latest_buildpack_run_tag
+    get_latest_buildpack_run_tag,
+    is_acr_url,
+    get_acr_name,
 
 )
 
@@ -414,7 +416,7 @@ def create_acr_if_needed(self):
     def create_acr(self):
         registry_rg = self.resource_group
         url = self.registry_server
-        registry_name = url[: url.rindex(ACR_IMAGE_SUFFIX)]
+        registry_name = get_acr_name(url) or url.split('.')[0]
         location = "eastus"
         if self.env.location and self.env.location.lower() != "northcentralusstage":
             location = self.env.location
@@ -548,7 +550,7 @@ def build_container_from_source_with_acr_task(self, image_name, source):
         import os
 
         task_name = "cli_build_containerapp"
-        registry_name = (self.registry_server[: self.registry_server.rindex(ACR_IMAGE_SUFFIX)]).lower()
+        registry_name = (get_acr_name(self.registry_server) or self.registry_server.split('.')[0]).lower()
         if not self.target_port:
             self.target_port = DEFAULT_PORT
         task_content = ACR_TASK_TEMPLATE.replace("{{image_name}}", image_name).replace("{{target_port}}",
@@ -919,7 +921,7 @@ def _get_registry_from_app(app, source):
     containerapp_def = app.get()
     existing_registries = safe_get(containerapp_def, "properties", "configuration", "registries", default=[])
     if source:
-        existing_registries = [r for r in existing_registries if ACR_IMAGE_SUFFIX in r["server"]]
+        existing_registries = [r for r in existing_registries if is_acr_url(r["server"])]
     if containerapp_def:
         if len(existing_registries) == 1:
             app.registry_server = existing_registries[0]["server"]
@@ -931,7 +933,7 @@ def _get_registry_from_app(app, source):
 
 
 def _get_acr_rg(app):
-    registry_name = app.registry_server[: app.registry_server.rindex(ACR_IMAGE_SUFFIX)]
+    registry_name = get_acr_name(app.registry_server) or app.registry_server.split('.')[0]
     client = get_mgmt_service_client(
         app.cmd.cli_ctx, ContainerRegistryManagementClient
     ).registries
@@ -967,7 +969,7 @@ def _get_registry_details(cmd, app: "ContainerApp", source):
     registry_rg = None
     registry_name = None
     if app.registry_server:
-        if "azurecr.io" not in app.registry_server and source:
+        if not is_acr_url(app.registry_server) and source:
             raise ValidationError(
                 "Cannot supply non-Azure registry when using --source."
             )

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

@@ -44,8 +44,9 @@
 from ._clients import ContainerAppClient, ManagedEnvironmentClient, WorkloadProfileClient, ContainerAppsJobClient
 from ._client_factory import handle_raw_exception, providers_client_factory, cf_resource_groups, log_analytics_client_factory, log_analytics_shared_key_client_factory
 from ._constants import (MAXIMUM_CONTAINER_APP_NAME_LENGTH, SHORT_POLLING_INTERVAL_SECS, LONG_POLLING_INTERVAL_SECS,
-                         LOG_ANALYTICS_RP, CONTAINER_APPS_RP, CHECK_CERTIFICATE_NAME_AVAILABILITY_TYPE, ACR_IMAGE_SUFFIX,
-                         LOGS_STRING, PENDING_STATUS, SUCCEEDED_STATUS, UPDATING_STATUS, DEV_SERVICE_LIST)
+                         LOG_ANALYTICS_RP, CONTAINER_APPS_RP, CHECK_CERTIFICATE_NAME_AVAILABILITY_TYPE,
+                         LOGS_STRING, PENDING_STATUS, SUCCEEDED_STATUS, UPDATING_STATUS, DEV_SERVICE_LIST,
+                         ACR_IMAGE_SUFFIXES)
 from ._models import (ContainerAppCustomDomainEnvelope as ContainerAppCustomDomainEnvelopeModel,
                       ManagedCertificateEnvelop as ManagedCertificateEnvelopModel)
 from ._models import OryxMarinerRunImgTagProperty
@@ -59,6 +60,25 @@ class AppType(Enum):
 logger = get_logger(__name__)
 
 
+def is_acr_url(registry_server):
+    """Check if a registry server URL belongs to Azure Container Registry (supports sovereign clouds)."""
+    if not registry_server:
+        return False
+    registry_server_lower = registry_server.lower()
+    return any(registry_server_lower.endswith(suffix) for suffix in ACR_IMAGE_SUFFIXES)
+
+
+def get_acr_name(registry_server):
+    """Extract the ACR registry name from a registry server URL (supports sovereign clouds)."""
+    if not registry_server:
+        return None
+    registry_server_lower = registry_server.lower()
+    for suffix in ACR_IMAGE_SUFFIXES:
+        if registry_server_lower.endswith(suffix):
+            return registry_server[: registry_server_lower.rindex(suffix)]
+    return None
+
+
 def register_provider_if_needed(cmd, rp_name):
     if not _is_resource_provider_registered(cmd, rp_name):
         _register_resource_provider(cmd, rp_name)
@@ -1170,7 +1190,7 @@ def _get_app_from_revision(revision):
 
 def _infer_acr_credentials(cmd, registry_server, disable_warnings=False):
     # If registry is Azure Container Registry, we can try inferring credentials
-    if ACR_IMAGE_SUFFIX not in registry_server:
+    if not is_acr_url(registry_server):
         raise RequiredArgumentMissingError('Registry username and password are required if not using Azure Container Registry.')
     not disable_warnings and logger.warning('No credential was provided to access Azure Container Registry. Trying to look up credentials...')
     parsed = urlparse(registry_server)
@@ -1656,7 +1676,10 @@ def create_acrpull_role_assignment(cmd, registry_server, registry_identity=None,
 
     client = get_mgmt_service_client(cmd.cli_ctx, ContainerRegistryManagementClient).registries
     try:
-        acr_id = acr_show(cmd, client, registry_server[: registry_server.rindex(ACR_IMAGE_SUFFIX)]).id
+        acr_name = get_acr_name(registry_server)
+        if not acr_name:
+            raise RequiredArgumentMissingError(f'Could not parse ACR name from registry server: {registry_server}')
+        acr_id = acr_show(cmd, client, acr_name).id
     except ResourceNotFound as e:
         message = (f"Role assignment failed with error message: \"{' '.join(e.args)}\". \n"
                    f"To add the role assignment manually, please run 'az role assignment create --assignee {sp_id} --scope <container-registry-resource-id> --role acrpull'. \n"

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

@@ -12,8 +12,8 @@
 
 from ._clients import ContainerAppClient, ManagedEnvironmentClient
 from ._ssh_utils import ping_container_app
-from ._utils import safe_get, is_registry_msi_system
-from ._constants import ACR_IMAGE_SUFFIX, LOG_TYPE_SYSTEM, MANAGED_ENVIRONMENT_RESOURCE_TYPE
+from ._utils import safe_get, is_registry_msi_system, is_acr_url
+from ._constants import LOG_TYPE_SYSTEM, MANAGED_ENVIRONMENT_RESOURCE_TYPE
 
 logger = get_logger(__name__)
 
@@ -26,8 +26,8 @@ def validate_create(registry_identity, registry_pass, registry_user, registry_se
         raise MutuallyExclusiveArgumentError("--no-wait is not supported with system registry identity")
     if registry_identity and not is_valid_resource_id(registry_identity) and not is_registry_msi_system(registry_identity):
         raise InvalidArgumentValueError("--registry-identity must be an identity resource ID or 'system'")
-    if registry_identity and ACR_IMAGE_SUFFIX not in (registry_server or ""):
-        raise InvalidArgumentValueError("--registry-identity: expected an ACR registry (*.azurecr.io) for --registry-server")
+    if registry_identity and not is_acr_url(registry_server or ""):
+        raise InvalidArgumentValueError("--registry-identity: expected an ACR registry (*.azurecr.io / *.azurecr.cn / *.azurecr.us) for --registry-server")
 
 
 def _is_number(s):
@@ -106,21 +106,21 @@ def validate_registry_server(namespace):
     if "create" in namespace.command.lower():
         if namespace.registry_server:
             if not namespace.registry_user or not namespace.registry_pass:
-                if ACR_IMAGE_SUFFIX not in namespace.registry_server:
+                if not is_acr_url(namespace.registry_server):
                     raise ValidationError("Usage error: --registry-server, --registry-password and --registry-username are required together if not using Azure Container Registry")
 
 
 def validate_registry_user(namespace):
     if "create" in namespace.command.lower():
         if namespace.registry_user:
-            if not namespace.registry_server or (not namespace.registry_pass and ACR_IMAGE_SUFFIX not in namespace.registry_server):
+            if not namespace.registry_server or (not namespace.registry_pass and not is_acr_url(namespace.registry_server)):
                 raise ValidationError("Usage error: --registry-server, --registry-password and --registry-username are required together if not using Azure Container Registry")
 
 
 def validate_registry_pass(namespace):
     if "create" in namespace.command.lower():
         if namespace.registry_pass:
-            if not namespace.registry_server or (not namespace.registry_user and ACR_IMAGE_SUFFIX not in namespace.registry_server):
+            if not namespace.registry_server or (not namespace.registry_user and not is_acr_url(namespace.registry_server)):
                 raise ValidationError("Usage error: --registry-server, --registry-password and --registry-username are required together if not using Azure Container Registry")
 
 

src/azure-cli/azure/cli/command_modules/containerapp/containerapp_job_registry_decorator.py

@@ -21,7 +21,6 @@
 
 from knack.log import get_logger
 
-from ._constants import ACR_IMAGE_SUFFIX
 from .base_resource import BaseResource
 from ._client_factory import handle_raw_exception, handle_non_404_status_code_exception
 
@@ -33,7 +32,8 @@
                      safe_set,
                      safe_get,
                      _remove_registry_secret,
-                     _get_acr_cred)
+                     _get_acr_cred,
+                     is_acr_url)
 
 logger = get_logger(__name__)
 
@@ -163,7 +163,7 @@ def construct_payload(self):
 
         if (not self.get_argument_username() or not self.get_argument_password()) and not self.get_argument_identity():
             # If registry is Azure Container Registry, we can try inferring credentials
-            if ACR_IMAGE_SUFFIX not in self.get_argument_server():
+            if not is_acr_url(self.get_argument_server()):
                 raise RequiredArgumentMissingError(
                     'Registry username and password are required if you are not using Azure Container Registry.')
             not self.get_argument_disable_warnings() and logger.warning(

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

@@ -82,13 +82,14 @@
                      ensure_workload_profile_supported, _generate_secret_volume_name,
                      trigger_workflow, AppType,
                      format_location, certificate_location_matches, generate_randomized_managed_cert_name,
-                     check_managed_cert_name_availability, prepare_managed_certificate_envelop)
+                     check_managed_cert_name_availability, prepare_managed_certificate_envelop,
+                     is_acr_url)
 from ._validators import validate_revision_suffix
 from ._ssh_utils import (SSH_DEFAULT_ENCODING, WebSocketConnection, read_ssh, get_stdin_writer, SSH_CTRL_C_MSG,
                          SSH_BACKUP_ENCODING)
 from ._constants import (MICROSOFT_SECRET_SETTING_NAME, FACEBOOK_SECRET_SETTING_NAME, GITHUB_SECRET_SETTING_NAME,
                          GOOGLE_SECRET_SETTING_NAME, TWITTER_SECRET_SETTING_NAME, APPLE_SECRET_SETTING_NAME, CONTAINER_APPS_RP,
-                         NAME_INVALID, NAME_ALREADY_EXISTS, ACR_IMAGE_SUFFIX, HELLO_WORLD_IMAGE, LOG_TYPE_SYSTEM, LOG_TYPE_CONSOLE,
+                         NAME_INVALID, NAME_ALREADY_EXISTS, HELLO_WORLD_IMAGE, LOG_TYPE_SYSTEM, LOG_TYPE_CONSOLE,
                          MANAGED_CERTIFICATE_RT, PRIVATE_CERTIFICATE_RT, PENDING_STATUS, SUCCEEDED_STATUS, CONTAINER_APPS_SDK_MODELS,
                          BLOB_STORAGE_TOKEN_STORE_SECRET_SETTING_NAME, DEFAULT_PORT)
 
@@ -665,7 +666,7 @@ def update_containerapp_logic(cmd,
 
         if registry_server:
             if not registry_pass or not registry_user:
-                if ACR_IMAGE_SUFFIX not in registry_server:
+                if not is_acr_url(registry_server):
                     raise RequiredArgumentMissingError('Registry url is required if using Azure Container Registry, otherwise Registry username and password are required if using Dockerhub')
                 logger.warning('No credential was provided to access Azure Container Registry. Trying to look up...')
                 parsed = urlparse(registry_server)
@@ -1401,7 +1402,7 @@ def update_containerappsjob_logic(cmd,
 
         if registry_server:
             if not registry_pass or not registry_user:
-                if ACR_IMAGE_SUFFIX not in registry_server:
+                if not is_acr_url(registry_server):
                     raise RequiredArgumentMissingError('Registry url is required if using Azure Container Registry, otherwise Registry username and password are required if using Dockerhub')
                 logger.warning('No credential was provided to access Azure Container Registry. Trying to look up...')
                 parsed = urlparse(registry_server)
@@ -2010,7 +2011,7 @@ def create_or_update_github_action(cmd,
     # Registry
     if registry_username is None or registry_password is None:
         # If registry is Azure Container Registry, we can try inferring credentials
-        if not registry_url or ACR_IMAGE_SUFFIX not in registry_url:
+        if not registry_url or not is_acr_url(registry_url):
             raise RequiredArgumentMissingError('Registry url is required if using Azure Container Registry, otherwise Registry username and password are required if using Dockerhub')
         logger.warning('No credential was provided to access Azure Container Registry. Trying to look up...')
         parsed = urlparse(registry_url)
@@ -2953,7 +2954,7 @@ def set_registry(cmd, name, resource_group_name, server, username=None, password
 
     if (not username or not password) and not identity:
         # If registry is Azure Container Registry, we can try inferring credentials
-        if ACR_IMAGE_SUFFIX not in server:
+        if not is_acr_url(server):
             raise RequiredArgumentMissingError('Registry username and password are required if you are not using Azure Container Registry.')
         not disable_warnings and logger.warning('No credential was provided to access Azure Container Registry. Trying to look up...')
         parsed = urlparse(server)