[AKS] az aks create/update: Add --kms-infrastructure-encryption for KMS platform-managed keys (PMK)

src/azure-cli/azure/cli/command_modules/acs/_consts.py

@@ -222,6 +222,9 @@
 CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PUBLIC = "Public"
 CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE = "Private"
 
+# kms infrastructure encryption (platform-managed keys)
+CONST_KMS_INFRASTRUCTURE_ENCRYPTION_ENABLED = "Enabled"
+
 # app routing nginx config options
 CONST_WEB_APPLICATION_ROUTING_KEY_NAME = "ingress/webApplicationRouting"
 CONST_APP_ROUTING_ANNOTATION_CONTROLLED_NGINX = "AnnotationControlled"

src/azure-cli/azure/cli/command_modules/acs/_help.py

@@ -530,6 +530,10 @@
   - name: --azure-keyvault-kms-key-vault-resource-id
     type: string
     short-summary: Resource ID of Azure Key Vault.
+  - name: --kms-infrastructure-encryption
+    type: string
+    short-summary: Enable encryption at rest of Kubernetes resource objects using service-managed keys.
+    long-summary: Enable infrastructure encryption for Kubernetes resource objects. This feature provides encryption at rest for cluster secrets and configuration using service-managed keys. The only accepted value is "Enabled"; once enabled it cannot be disabled. For more information see https://aka.ms/aks/kubernetesResourceObjectEncryption.
   - name: --enable-image-cleaner
     type: bool
     short-summary: Enable ImageCleaner Service.
@@ -1080,6 +1084,10 @@
   - name: --azure-keyvault-kms-key-vault-resource-id
     type: string
     short-summary: Resource ID of Azure Key Vault.
+  - name: --kms-infrastructure-encryption
+    type: string
+    short-summary: Enable encryption at rest of Kubernetes resource objects using service-managed keys.
+    long-summary: Enable infrastructure encryption for Kubernetes resource objects. This feature provides encryption at rest for cluster secrets and configuration using service-managed keys. The only accepted value is "Enabled"; once enabled it cannot be disabled. For more information see https://aka.ms/aks/kubernetesResourceObjectEncryption.
   - name: --enable-image-cleaner
     type: bool
     short-summary: Enable ImageCleaner Service.

src/azure-cli/azure/cli/command_modules/acs/_params.py

@@ -12,6 +12,7 @@
 from azure.cli.command_modules.acs._consts import (
     CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE,
     CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PUBLIC,
+    CONST_KMS_INFRASTRUCTURE_ENCRYPTION_ENABLED,
     CONST_GPU_INSTANCE_PROFILE_MIG1_G, CONST_GPU_INSTANCE_PROFILE_MIG2_G,
     CONST_GPU_INSTANCE_PROFILE_MIG3_G, CONST_GPU_INSTANCE_PROFILE_MIG4_G,
     CONST_GPU_INSTANCE_PROFILE_MIG7_G, CONST_LOAD_BALANCER_SKU_BASIC,
@@ -262,6 +263,7 @@
 dev_space_endpoint_types = ['Public', 'Private', 'None']
 
 keyvault_network_access_types = [CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PUBLIC, CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE]
+kms_infrastructure_encryption_types = [CONST_KMS_INFRASTRUCTURE_ENCRYPTION_ENABLED]
 
 gpu_instance_profiles = [
     CONST_GPU_INSTANCE_PROFILE_MIG1_G,
@@ -514,6 +516,7 @@ def load_arguments(self, _):
         c.argument('azure_keyvault_kms_key_id', validator=validate_azure_keyvault_kms_key_id)
         c.argument('azure_keyvault_kms_key_vault_network_access', arg_type=get_enum_type(keyvault_network_access_types))
         c.argument('azure_keyvault_kms_key_vault_resource_id', validator=validate_azure_keyvault_kms_key_vault_resource_id)
+        c.argument('kms_infrastructure_encryption', arg_type=get_enum_type(kms_infrastructure_encryption_types))
         c.argument('enable_image_cleaner', action='store_true')
         c.argument('image_cleaner_interval_hours', type=int)
         c.argument('http_proxy_config')
@@ -855,6 +858,7 @@ def load_arguments(self, _):
         c.argument('azure_keyvault_kms_key_id', validator=validate_azure_keyvault_kms_key_id)
         c.argument('azure_keyvault_kms_key_vault_network_access', arg_type=get_enum_type(keyvault_network_access_types))
         c.argument('azure_keyvault_kms_key_vault_resource_id', validator=validate_azure_keyvault_kms_key_vault_resource_id)
+        c.argument('kms_infrastructure_encryption', arg_type=get_enum_type(kms_infrastructure_encryption_types))
         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/azure-cli/azure/cli/command_modules/acs/_validators.py

@@ -870,19 +870,43 @@ def validate_azure_keyvault_kms_key_id(namespace):
         if not key_id.startswith(https_prefix):
             raise InvalidArgumentValueError(err_msg)
 
+        # A versioned key ID has 4 segments ({vault-url}/keys/{key-name}/{key-version}), while a versionless
+        # key ID has 3 segments ({vault-url}/keys/{key-name}). Versionless key IDs are required when KMS
+        # infrastructure encryption (platform-managed keys) is enabled. Any other number of segments is invalid.
         segments = key_id[len(https_prefix):].split("/")
-        if len(segments) != 4 or segments[1] != "keys":
+        if len(segments) not in (3, 4) or segments[1] != "keys":
             raise InvalidArgumentValueError(err_msg)
 
 
 def validate_azure_keyvault_kms_key_vault_resource_id(namespace):
     key_vault_resource_id = namespace.azure_keyvault_kms_key_vault_resource_id
     if key_vault_resource_id is None or key_vault_resource_id == '':
         return
-    from azure.mgmt.core.tools import is_valid_resource_id
+    from azure.mgmt.core.tools import is_valid_resource_id, parse_resource_id
     if not is_valid_resource_id(key_vault_resource_id):
         raise InvalidArgumentValueError("--azure-keyvault-kms-key-vault-resource-id is not a valid Azure resource ID.")
 
+    try:
+        parsed = parse_resource_id(key_vault_resource_id)
+        provider = parsed.get('namespace', '').lower()
+        if provider != 'microsoft.keyvault':
+            raise InvalidArgumentValueError(
+                "--azure-keyvault-kms-key-vault-resource-id must reference a Microsoft.KeyVault resource."
+            )
+        resource_type = parsed.get('type', '').lower()
+        if resource_type not in ['vaults', 'managedhsms']:
+            raise InvalidArgumentValueError(
+                "--azure-keyvault-kms-key-vault-resource-id must reference a Key Vault "
+                "(vaults) or Managed HSM (managedHSMs)."
+            )
+    except InvalidArgumentValueError:
+        # Re-raise our validation errors
+        raise
+    except Exception as ex:
+        raise InvalidArgumentValueError(
+            f"--azure-keyvault-kms-key-vault-resource-id parsing failed: {str(ex)}"
+        )
+
 
 def validate_image_cleaner_enable_disable_mutually_exclusive(namespace):
     enable_image_cleaner = namespace.enable_image_cleaner

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

@@ -938,6 +938,7 @@ def aks_create(
     azure_keyvault_kms_key_id=None,
     azure_keyvault_kms_key_vault_network_access=None,
     azure_keyvault_kms_key_vault_resource_id=None,
+    kms_infrastructure_encryption=None,
     enable_image_cleaner=False,
     image_cleaner_interval_hours=None,
     enable_keda=False,
@@ -1191,6 +1192,7 @@ def aks_update(
     azure_keyvault_kms_key_id=None,
     azure_keyvault_kms_key_vault_network_access=None,
     azure_keyvault_kms_key_vault_resource_id=None,
+    kms_infrastructure_encryption=None,
     enable_image_cleaner=False,
     disable_image_cleaner=False,
     image_cleaner_interval_hours=None,

src/azure-cli/azure/cli/command_modules/acs/linter_exclusions.yml

@@ -127,6 +127,9 @@ aks create:
       enable_upstream_kubescheduler_user_configuration:
         rule_exclusions:
           - option_length_too_long
+      kms_infrastructure_encryption:
+        rule_exclusions:
+          - option_length_too_long
 aks enable-addons:
   parameters:
       appgw_watch_namespace:
@@ -287,6 +290,9 @@ aks update:
       disable_upstream_kubescheduler_user_configuration:
         rule_exclusions:
           - option_length_too_long
+      kms_infrastructure_encryption:
+        rule_exclusions:
+          - option_length_too_long
 aks nodepool add:
   parameters:
       disable_windows_outbound_nat:

src/azure-cli/azure/cli/command_modules/acs/managed_cluster_decorator.py

@@ -33,7 +33,7 @@
     CONST_PRIVATE_DNS_ZONE_NONE,
     CONST_PRIVATE_DNS_ZONE_SYSTEM,
     CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE,
-    CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PUBLIC,
+    CONST_KMS_INFRASTRUCTURE_ENCRYPTION_ENABLED,
     AgentPoolDecoratorMode,
     DecoratorEarlyExitException,
     DecoratorMode,
@@ -5845,6 +5845,28 @@ def get_workload_identity_profile(self) -> Optional[ManagedClusterSecurityProfil
 
         return profile
 
+    def get_kms_infrastructure_encryption(self) -> Union[str, None]:
+        """Obtain the value of kms_infrastructure_encryption.
+
+        :return: string or None
+        """
+        # read the original value passed by the command
+        kms_infrastructure_encryption = self.raw_param.get("kms_infrastructure_encryption")
+
+        # try to read the property value corresponding to the parameter from the `mc` object
+        if kms_infrastructure_encryption is None and self.mc:
+            security_profile = getattr(self.mc, "security_profile", None)
+            if security_profile:
+                encryption_profile = getattr(
+                    security_profile, "kubernetes_resource_object_encryption_profile", None
+                )
+                if encryption_profile:
+                    infrastructure_encryption = getattr(encryption_profile, "infrastructure_encryption", None)
+                    if infrastructure_encryption is not None:
+                        kms_infrastructure_encryption = infrastructure_encryption
+
+        return kms_infrastructure_encryption
+
     def _get_enable_azure_keyvault_kms(self, enable_validation: bool = False) -> bool:
         """Internal function to obtain the value of enable_azure_keyvault_kms.
 
@@ -6047,6 +6069,9 @@ def _get_azure_keyvault_kms_key_vault_resource_id(self, enable_validation: bool
 
         # validation
         if enable_validation:
+            is_pmk_enabled = (
+                self.get_kms_infrastructure_encryption() == CONST_KMS_INFRASTRUCTURE_ENCRYPTION_ENABLED
+            )
             enable_azure_keyvault_kms = self._get_enable_azure_keyvault_kms(
                 enable_validation=False)
             if (
@@ -6060,29 +6085,10 @@ def _get_azure_keyvault_kms_key_vault_resource_id(self, enable_validation: bool
                     '"--azure-keyvault-kms-key-vault-resource-id" requires "--enable-azure-keyvault-kms".'
                 )
 
-            key_vault_network_access = self._get_azure_keyvault_kms_key_vault_network_access(
-                enable_validation=False)
-            if (
-                key_vault_network_access == CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE and
-                (
-                    azure_keyvault_kms_key_vault_resource_id is None or
-                    azure_keyvault_kms_key_vault_resource_id == ""
-                )
-            ):
-                raise ArgumentUsageError(
-                    '"--azure-keyvault-kms-key-vault-resource-id" can not be empty if '
-                    '"--azure-keyvault-kms-key-vault-network-access" is "Private".'
-                )
-            if (
-                key_vault_network_access == CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PUBLIC and
-                (
-                    azure_keyvault_kms_key_vault_resource_id is not None and
-                    azure_keyvault_kms_key_vault_resource_id != ""
-                )
-            ):
-                raise ArgumentUsageError(
-                    '"--azure-keyvault-kms-key-vault-resource-id" must be empty if '
-                    '"--azure-keyvault-kms-key-vault-network-access" is "Public".'
+            if is_pmk_enabled and not azure_keyvault_kms_key_vault_resource_id:
+                raise RequiredArgumentMissingError(
+                    '"--azure-keyvault-kms-key-vault-resource-id" is required when '
+                    '"--kms-infrastructure-encryption" is set to "Enabled".'
                 )
 
         return azure_keyvault_kms_key_vault_resource_id
@@ -7238,12 +7244,25 @@ def set_up_defender(self, mc: ManagedCluster) -> ManagedCluster:
         return mc
 
     def set_up_azure_keyvault_kms(self, mc: ManagedCluster) -> ManagedCluster:
-        """Set up security profile azureKeyVaultKms for the ManagedCluster object.
+        """Set up security profile kubernetesResourceObjectEncryptionProfile and azureKeyVaultKms for
+        the ManagedCluster object.
 
         :return: the ManagedCluster object
         """
         self._ensure_mc(mc)
 
+        kms_infrastructure_encryption = self.context.get_kms_infrastructure_encryption()
+        if kms_infrastructure_encryption:
+            if mc.security_profile is None:
+                mc.security_profile = self.models.ManagedClusterSecurityProfile()
+            if mc.security_profile.kubernetes_resource_object_encryption_profile is None:
+                mc.security_profile.kubernetes_resource_object_encryption_profile = (
+                    self.models.KubernetesResourceObjectEncryptionProfile()
+                )
+            mc.security_profile.kubernetes_resource_object_encryption_profile.infrastructure_encryption = (
+                kms_infrastructure_encryption
+            )
+
         if self.context.get_enable_azure_keyvault_kms():
             key_id = self.context.get_azure_keyvault_kms_key_id()
             if key_id:
@@ -7255,10 +7274,9 @@ def set_up_azure_keyvault_kms(self, mc: ManagedCluster) -> ManagedCluster:
                 )
                 key_vault_network_access = self.context.get_azure_keyvault_kms_key_vault_network_access()
                 mc.security_profile.azure_key_vault_kms.key_vault_network_access = key_vault_network_access
-                if key_vault_network_access == CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE:
-                    mc.security_profile.azure_key_vault_kms.key_vault_resource_id = (
-                        self.context.get_azure_keyvault_kms_key_vault_resource_id()
-                    )
+                mc.security_profile.azure_key_vault_kms.key_vault_resource_id = (
+                    self.context.get_azure_keyvault_kms_key_vault_resource_id()
+                )
 
         return mc
 
@@ -10410,12 +10428,25 @@ def update_k8s_support_plan(self, mc: ManagedCluster) -> ManagedCluster:
         return mc
 
     def update_azure_keyvault_kms(self, mc: ManagedCluster) -> ManagedCluster:
-        """Update security profile azureKeyvaultKms for the ManagedCluster object.
+        """Update security profile kubernetesResourceObjectEncryptionProfile and azureKeyvaultKms for
+        the ManagedCluster object.
 
         :return: the ManagedCluster object
         """
         self._ensure_mc(mc)
 
+        kms_infrastructure_encryption = self.context.get_kms_infrastructure_encryption()
+        if kms_infrastructure_encryption:
+            if mc.security_profile is None:
+                mc.security_profile = self.models.ManagedClusterSecurityProfile()
+            if mc.security_profile.kubernetes_resource_object_encryption_profile is None:
+                mc.security_profile.kubernetes_resource_object_encryption_profile = (
+                    self.models.KubernetesResourceObjectEncryptionProfile()
+                )
+            mc.security_profile.kubernetes_resource_object_encryption_profile.infrastructure_encryption = (
+                kms_infrastructure_encryption
+            )
+
         if self.context.get_enable_azure_keyvault_kms():
             # get kms profile
             if mc.security_profile is None:
@@ -10434,13 +10465,11 @@ def update_azure_keyvault_kms(self, mc: ManagedCluster) -> ManagedCluster:
             azure_key_vault_kms_profile.key_vault_network_access = (
                 self.context.get_azure_keyvault_kms_key_vault_network_access()
             )
-            # set key vault resource id
-            if azure_key_vault_kms_profile.key_vault_network_access == CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE:
-                azure_key_vault_kms_profile.key_vault_resource_id = (
-                    self.context.get_azure_keyvault_kms_key_vault_resource_id()
-                )
-            else:
-                azure_key_vault_kms_profile.key_vault_resource_id = ""
+            # set key vault resource id, it is required when network access is "Private" and when
+            # infrastructure encryption (PMK) is enabled; fall back to "" so that it is cleared otherwise
+            azure_key_vault_kms_profile.key_vault_resource_id = (
+                self.context.get_azure_keyvault_kms_key_vault_resource_id() or ""
+            )
 
         if self.context.get_disable_azure_keyvault_kms():
             # get kms profile