[Containerapp] az containerapp create: Set identity only when --system-assigned or --user-assigned is specified. (#8742)

Author: Greedygre

src/containerapp/HISTORY.rst

@@ -5,6 +5,8 @@ Release History
 upcoming
 ++++++
 * 'az containerapp auth update': Support authenticating blob storage token store using managed identity with `--blob-container-uri` and `--blob-container-identity`.
+* 'az containerapp env create': Set identity only when `--mi-system-assigned` or `--mi-user-assigned` is specified.
+* 'az containerapp env create': Set identity only when `--system-assigned` or `--user-assigned` is specified.
 
 1.1.0b4
 ++++++

src/containerapp/azext_containerapp/containerapp_decorator.py

@@ -697,6 +697,34 @@ def set_up_registry_identity(self):
                 if not env_has_managed_identity(self.cmd, managed_env_rg, managed_env_name, identity):
                     set_managed_identity(self.cmd, self.get_argument_resource_group_name(), self.containerapp_def, user_assigned=[identity])
 
+    def set_up_managed_identity(self):
+        if self.get_argument_user_assigned() or self.get_argument_system_assigned():
+            identity_def = deepcopy(ManagedServiceIdentityModel)
+            identity_def["type"] = "None"
+
+            assign_system_identity = self.get_argument_system_assigned()
+            if self.get_argument_user_assigned():
+                assign_user_identities = [x.lower() for x in self.get_argument_user_assigned()]
+            else:
+                assign_user_identities = []
+
+            if assign_system_identity and assign_user_identities:
+                identity_def["type"] = "SystemAssigned, UserAssigned"
+            elif assign_system_identity:
+                identity_def["type"] = "SystemAssigned"
+            elif assign_user_identities:
+                identity_def["type"] = "UserAssigned"
+
+            if assign_user_identities:
+                identity_def["userAssignedIdentities"] = {}
+                subscription_id = get_subscription_id(self.cmd.cli_ctx)
+
+                for r in assign_user_identities:
+                    r = _ensure_identity_resource_id(subscription_id, self.get_argument_resource_group_name(), r)
+                    identity_def["userAssignedIdentities"][r] = {}  # pylint: disable=unsupported-assignment-operation
+
+            self.containerapp_def["identity"] = identity_def
+
     # If --registry-server is ACR, use system-assigned managed identity for image pull by default
     def set_up_system_assigned_identity_as_default_if_using_acr(self):
         registry_server = self.get_argument_registry_server()
@@ -816,29 +844,7 @@ def parent_construct_payload(self):
         config_def["dapr"] = dapr_def
 
         # Identity actions
-        identity_def = deepcopy(ManagedServiceIdentityModel)
-        identity_def["type"] = "None"
-
-        assign_system_identity = self.get_argument_system_assigned()
-        if self.get_argument_user_assigned():
-            assign_user_identities = [x.lower() for x in self.get_argument_user_assigned()]
-        else:
-            assign_user_identities = []
-
-        if assign_system_identity and assign_user_identities:
-            identity_def["type"] = "SystemAssigned, UserAssigned"
-        elif assign_system_identity:
-            identity_def["type"] = "SystemAssigned"
-        elif assign_user_identities:
-            identity_def["type"] = "UserAssigned"
-
-        if assign_user_identities:
-            identity_def["userAssignedIdentities"] = {}
-            subscription_id = get_subscription_id(self.cmd.cli_ctx)
-
-            for r in assign_user_identities:
-                r = _ensure_identity_resource_id(subscription_id, self.get_argument_resource_group_name(), r)
-                identity_def["userAssignedIdentities"][r] = {}  # pylint: disable=unsupported-assignment-operation
+        self.set_up_managed_identity()
 
         scale_def = self.set_up_scale_rule()
 
@@ -885,7 +891,6 @@ def parent_construct_payload(self):
             template_def["terminationGracePeriodSeconds"] = self.get_argument_termination_grace_period()
 
         self.containerapp_def["location"] = location
-        self.containerapp_def["identity"] = identity_def
         self.containerapp_def["properties"]["environmentId"] = self.get_argument_managed_env()
         self.containerapp_def["properties"]["configuration"] = config_def
         self.containerapp_def["properties"]["template"] = template_def

src/containerapp/azext_containerapp/containerapp_env_decorator.py

@@ -83,30 +83,31 @@ def set_up_infrastructure_resource_group(self):
             self.managed_env_def["properties"]["infrastructureResourceGroup"] = self.get_argument_infrastructure_resource_group()
 
     def set_up_managed_identity(self):
-        identity_def = ManagedServiceIdentity
-        identity_def["type"] = "None"
-
-        assign_system_identity = self.get_argument_system_assigned()
-        if self.get_argument_user_assigned():
-            assign_user_identities = [x.lower() for x in self.get_argument_user_assigned()]
-        else:
-            assign_user_identities = []
-
-        if assign_system_identity and assign_user_identities:
-            identity_def["type"] = "SystemAssigned, UserAssigned"
-        elif assign_system_identity:
-            identity_def["type"] = "SystemAssigned"
-        elif assign_user_identities:
-            identity_def["type"] = "UserAssigned"
-
-        if assign_user_identities:
-            identity_def["userAssignedIdentities"] = {}
-            subscription_id = get_subscription_id(self.cmd.cli_ctx)
-
-            for r in assign_user_identities:
-                r = _ensure_identity_resource_id(subscription_id, self.get_argument_resource_group_name(), r)
-                identity_def["userAssignedIdentities"][r] = {}  # pylint: disable=unsupported-assignment-operation
-        self.managed_env_def["identity"] = identity_def
+        if self.get_argument_system_assigned() or self.get_argument_user_assigned():
+            identity_def = ManagedServiceIdentity
+            identity_def["type"] = "None"
+
+            assign_system_identity = self.get_argument_system_assigned()
+            if self.get_argument_user_assigned():
+                assign_user_identities = [x.lower() for x in self.get_argument_user_assigned()]
+            else:
+                assign_user_identities = []
+
+            if assign_system_identity and assign_user_identities:
+                identity_def["type"] = "SystemAssigned, UserAssigned"
+            elif assign_system_identity:
+                identity_def["type"] = "SystemAssigned"
+            elif assign_user_identities:
+                identity_def["type"] = "UserAssigned"
+
+            if assign_user_identities:
+                identity_def["userAssignedIdentities"] = {}
+                subscription_id = get_subscription_id(self.cmd.cli_ctx)
+
+                for r in assign_user_identities:
+                    r = _ensure_identity_resource_id(subscription_id, self.get_argument_resource_group_name(), r)
+                    identity_def["userAssignedIdentities"][r] = {}  # pylint: disable=unsupported-assignment-operation
+            self.managed_env_def["identity"] = identity_def
 
     def set_up_workload_profiles(self):
         if self.get_argument_enable_workload_profiles():