[AKS] az aks update: add new parameter --disable-workload-identity to disable Workload Identity

src/aks-preview/HISTORY.rst

@@ -12,6 +12,10 @@ To release a new version, please select a new version number (usually plus 1 to
 Pending
 +++++++
 
+0.5.157
++++++++
+* Add `--disable-workload-identity` to the `az aks update` command.
+
 0.5.156
 +++++++
 * Add `az aks copilot` command to start a chat with the Azure Kubernetes Service expert. API keys for OpenAI or Azure are required.

src/aks-preview/azext_aks_preview/_help.py

@@ -829,6 +829,9 @@
         - name: --enable-workload-identity
           type: bool
           short-summary: (PREVIEW) Enable Workload Identity addon for cluster.
+        - name: --disable-workload-identity
+          type: bool
+          short-summary: (PREVIEW) Disable Workload Identity addon for cluster.
         - name: --enable-secret-rotation
           type: bool
           short-summary: Enable secret rotation. Use with azure-keyvault-secrets-provider addon.

src/aks-preview/azext_aks_preview/_params.py

@@ -409,7 +409,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', arg_type=get_three_state_flag(), is_preview=True)
+        c.argument('enable_workload_identity', action='store_true', is_preview=True)
         c.argument('enable_image_cleaner', action='store_true', is_preview=True)
         c.argument('enable_azure_service_mesh',
                    options_list=["--enable-azure-service-mesh", "--enable-asm"],
@@ -544,7 +544,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', arg_type=get_three_state_flag(), is_preview=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_image_cleaner', action='store_true', is_preview=True)
         c.argument('disable_image_cleaner', action='store_true', validator=validate_image_cleaner_enable_disable_mutually_exclusive, is_preview=True)
         c.argument('image_cleaner_interval_hours', type=int, is_preview=True)

src/aks-preview/azext_aks_preview/custom.py

@@ -554,7 +554,7 @@ def aks_create(
     enable_pod_security_policy=False,
     enable_pod_identity=False,
     enable_pod_identity_with_kubenet=False,
-    enable_workload_identity=None,
+    enable_workload_identity=False,
     enable_image_cleaner=False,
     image_cleaner_interval_hours=None,
     cluster_snapshot_id=None,
@@ -725,7 +725,8 @@ def aks_update(
     enable_pod_identity=False,
     enable_pod_identity_with_kubenet=False,
     disable_pod_identity=False,
-    enable_workload_identity=None,
+    enable_workload_identity=False,
+    disable_workload_identity=False,
     enable_image_cleaner=False,
     disable_image_cleaner=False,
     image_cleaner_interval_hours=None,

src/aks-preview/azext_aks_preview/managed_cluster_decorator.py

@@ -985,11 +985,22 @@ def get_workload_identity_profile(self) -> Optional[ManagedClusterSecurityProfil
         # - 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 enable_workload_identity is None:
+        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
@@ -2510,15 +2521,10 @@ def set_up_workload_identity_profile(self, mc: ManagedCluster) -> ManagedCluster
         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
+        if profile:
+            if mc.security_profile is None:
+                mc.security_profile = self.models.ManagedClusterSecurityProfile()
+            mc.security_profile.workload_identity = profile
 
         return mc
 

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

@@ -4981,7 +4981,7 @@ def test_aks_update_with_workload_identity(self, resource_group, resource_group_
 
         disable_cmd = ' '.join([
             'aks', 'update', '--resource-group={resource_group}', '--name={name}',
-            '--enable-workload-identity', 'False',
+            '--disable-workload-identity',
             '--aks-custom-headers AKSHTTPCustomFeatures=Microsoft.ContainerService/EnableWorkloadIdentityPreview,AKSHTTPCustomFeatures=Microsoft.ContainerService/EnableOIDCIssuerPreview',
         ])
         self.cmd(disable_cmd, checks=[

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

@@ -1172,6 +1172,21 @@ def test_get_workload_identity_profile__update_not_set(self):
         ctx.attach_mc(self.models.ManagedCluster(location="test_location"))
         self.assertIsNone(ctx.get_workload_identity_profile())
 
+    def test_get_workload_identity_profile__update_with_enable_and_disable(self):
+        ctx = AKSPreviewManagedClusterContext(
+            self.cmd,
+            AKSManagedClusterParamDict(
+                {
+                    "enable_workload_identity": True,
+                    "disable_workload_identity": True,
+                }
+            ),
+            self.models, decorator_mode=DecoratorMode.UPDATE
+        )
+        ctx.attach_mc(self.models.ManagedCluster(location="test_location"))
+        with self.assertRaises(MutuallyExclusiveArgumentError):
+            ctx.get_workload_identity_profile()
+
     def test_get_workload_identity_profile__update_with_enable_without_oidc_issuer(self):
         ctx = AKSPreviewManagedClusterContext(
             self.cmd,
@@ -1227,7 +1242,7 @@ def test_get_workload_identity_profile__update_with_disable(self):
                 self.cmd,
                 AKSManagedClusterParamDict(
                     {
-                        "enable_workload_identity": False,
+                        "disable_workload_identity": True,
                     }
                 ),
                 self.models,
@@ -5485,7 +5500,7 @@ def test_update_workload_identity_profile__disabled(self):
             self.cmd,
             self.client,
             {
-                "enable_workload_identity": False,
+                "disable_workload_identity": True,
             },
             CUSTOM_MGMT_AKS_PREVIEW,
         )

src/aks-preview/linter_exclusions.yml

@@ -80,6 +80,9 @@ aks update:
     enable_workload_identity:
       rule_exclusions:
         - option_length_too_long
+    disable_workload_identity:
+      rule_exclusions:
+        - option_length_too_long
     enable_snapshot_controller:
       rule_exclusions:
         - option_length_too_long

src/aks-preview/setup.py

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