{AKS} Replace Workload Identity related functions in aks-preview with azure-cli (#6987)

Author: jiashun0011

src/aks-preview/HISTORY.rst

@@ -11,8 +11,12 @@ To release a new version, please select a new version number (usually plus 1 to
 
 Pending
 +++++++
+
+0.5.174
++++++++
 * Fix the response format for `az aks mesh get-revisions` and `az aks mesh get-upgrades`.
 * Fix for `az aks approuting update` command failing on granting keyvault permissions to managed identity.
+* Replace Workload Identity related functions with stable version.
 
 0.5.173
 +++++++
@@ -1426,4 +1430,4 @@ Pending
 +++++
 
 * new feature `enable-cluster-autoscaler`
-* default agentType is VMSS
+* default agentType is VMSS

src/aks-preview/azext_aks_preview/_params.py

@@ -465,7 +465,7 @@ def load_arguments(self, _):
         c.argument('enable_pod_security_policy', action='store_true', deprecate_info=c.deprecate(target='--enable-pod-security-policy', hide=True))
         c.argument('enable_pod_identity', action='store_true')
         c.argument('enable_pod_identity_with_kubenet', action='store_true')
-        c.argument('enable_workload_identity', action='store_true', is_preview=True)
+        c.argument('enable_workload_identity', action='store_true')
         c.argument('enable_image_cleaner', action='store_true')
         c.argument('enable_azure_service_mesh',
                    options_list=["--enable-azure-service-mesh", "--enable-asm"],
@@ -614,8 +614,8 @@ def load_arguments(self, _):
         c.argument('enable_pod_identity', action='store_true')
         c.argument('enable_pod_identity_with_kubenet', action='store_true')
         c.argument('disable_pod_identity', action='store_true')
-        c.argument('enable_workload_identity', action='store_true', is_preview=True)
-        c.argument('disable_workload_identity', action='store_true', is_preview=True)
+        c.argument('enable_workload_identity', action='store_true')
+        c.argument('disable_workload_identity', action='store_true')
         c.argument('enable_image_cleaner', action='store_true')
         c.argument('disable_image_cleaner', action='store_true', validator=validate_image_cleaner_enable_disable_mutually_exclusive)
         c.argument('image_cleaner_interval_hours', type=int)

src/aks-preview/azext_aks_preview/managed_cluster_decorator.py

@@ -1014,58 +1014,6 @@ def get_enable_pod_identity_with_kubenet(self) -> bool:
         """
         return self._get_enable_pod_identity_with_kubenet(enable_validation=True)
 
-    def get_workload_identity_profile(self) -> Optional[ManagedClusterSecurityProfileWorkloadIdentity]:
-        """Obtrain the value of security_profile.workload_identity.
-
-        :return: Optional[ManagedClusterSecurityProfileWorkloadIdentity]
-        """
-        # NOTE: enable_workload_identity can be one of:
-        #
-        # - True: sets by user, to enable the workload identity feature
-        # - False: sets by user, to disable the workload identity feature
-        # - None: user unspecified, don't set the profile and let server side to backfill
-        enable_workload_identity = self.raw_param.get("enable_workload_identity")
-        disable_workload_identity = self.raw_param.get("disable_workload_identity")
-
-        if not enable_workload_identity and not disable_workload_identity:
-            return None
-
-        if enable_workload_identity and disable_workload_identity:
-            raise MutuallyExclusiveArgumentError(
-                "Cannot specify --enable-workload-identity and "
-                "--disable-workload-identity at the same time."
-            )
-
-        if not hasattr(self.models, "ManagedClusterSecurityProfileWorkloadIdentity"):
-            raise UnknownError("Workload Identity's data model not found")
-
-        profile = self.models.ManagedClusterSecurityProfileWorkloadIdentity()
-
-        if self.decorator_mode == DecoratorMode.UPDATE:
-            if self.mc.security_profile is not None and self.mc.security_profile.workload_identity is not None:
-                # reuse previous profile is has been set
-                profile = self.mc.security_profile.workload_identity
-
-        profile.enabled = bool(enable_workload_identity)
-
-        if profile.enabled:
-            # in enable case, we need to check if OIDC issuer has been enabled
-            oidc_issuer_profile = self.get_oidc_issuer_profile()
-            if self.decorator_mode == DecoratorMode.UPDATE and oidc_issuer_profile is None:
-                # if the cluster has enabled OIDC issuer before, in update call:
-                #
-                #    az aks update --enable-workload-identity
-                #
-                # we need to use previous OIDC issuer profile
-                oidc_issuer_profile = self.mc.oidc_issuer_profile
-            oidc_issuer_enabled = oidc_issuer_profile is not None and oidc_issuer_profile.enabled
-            if not oidc_issuer_enabled:
-                raise RequiredArgumentMissingError(
-                    "Enabling workload identity requires enabling OIDC issuer (--enable-oidc-issuer)."
-                )
-
-        return profile
-
     def get_enable_image_integrity(self) -> bool:
         """Obtain the value of enable_image_integrity.
 
@@ -2684,21 +2632,6 @@ def set_up_pod_identity_profile(self, mc: ManagedCluster) -> ManagedCluster:
         mc.pod_identity_profile = pod_identity_profile
         return mc
 
-    def set_up_workload_identity_profile(self, mc: ManagedCluster) -> ManagedCluster:
-        """Set up workload identity for the ManagedCluster object.
-
-        :return: the ManagedCluster object
-        """
-        self._ensure_mc(mc)
-
-        profile = self.context.get_workload_identity_profile()
-        if profile:
-            if mc.security_profile is None:
-                mc.security_profile = self.models.ManagedClusterSecurityProfile()
-            mc.security_profile.workload_identity = profile
-
-        return mc
-
     def set_up_image_integrity(self, mc: ManagedCluster) -> ManagedCluster:
         """Set up security profile imageIntegrity for the ManagedCluster object.
 
@@ -3697,26 +3630,6 @@ def update_pod_identity_profile(self, mc: ManagedCluster) -> ManagedCluster:
                 mc, enable=False, models=self.models.pod_identity_models)
         return mc
 
-    def update_workload_identity_profile(self, mc: ManagedCluster) -> ManagedCluster:
-        """Update workload identity profile for the ManagedCluster object.
-
-        :return: the ManagedCluster object
-        """
-        self._ensure_mc(mc)
-
-        profile = self.context.get_workload_identity_profile()
-        if profile is None:
-            if mc.security_profile is not None:
-                # set the value to None to let server side to fill in the default value
-                mc.security_profile.workload_identity = None
-            return mc
-
-        if mc.security_profile is None:
-            mc.security_profile = self.models.ManagedClusterSecurityProfile()
-        mc.security_profile.workload_identity = profile
-
-        return mc
-
     def update_k8s_support_plan(self, mc: ManagedCluster) -> ManagedCluster:
         """Update supportPlan for the ManagedCluster object.
         :return: the ManagedCluster object

src/aks-preview/azext_aks_preview/tests/latest/test_managed_cluster_decorator.py

@@ -5924,7 +5924,7 @@ def test_update_workload_identity_profile__default_value_mc_enabled(self):
         )
         dec.context.attach_mc(mc)
         updated_mc = dec.update_workload_identity_profile(mc)
-        self.assertIsNone(updated_mc.security_profile.workload_identity)
+        self.assertIsNotNone(updated_mc.security_profile.workload_identity)
 
     def test_update_workload_identity_profile__enabled(self):
         dec = AKSPreviewManagedClusterUpdateDecorator(

src/aks-preview/setup.py

@@ -9,7 +9,7 @@
 
 from setuptools import setup, find_packages
 
-VERSION = "0.5.173"
+VERSION = "0.5.174"
 
 CLASSIFIERS = [
     "Development Status :: 4 - Beta",