[AKS] Remove validation of --azure-keyvault-kms-key-id when PMK is enabled (#9388)

Author: bingosummer

src/aks-preview/HISTORY.rst

@@ -10,6 +10,8 @@ If there is no rush to release a new version, please just add a description of t
 To release a new version, please select a new version number (usually plus 1 to last patch version, X.Y.Z -> Major.Minor.Patch, more details in `\doc <https://semver.org/>`_), and then add a new section named as the new version number in this file, the content should include the new modifications and everything from the *Pending* section. Finally, update the `VERSION` variable in `setup.py` with this new version number.
 
 Pending
++++++++
+* Remove PMK validation for `--azure-keyvault-kms-key-id` parameter.
 
 19.0.0b10
 +++++++

src/aks-preview/azext_aks_preview/managed_cluster_decorator.py

@@ -1443,30 +1443,6 @@ def _get_azure_keyvault_kms_key_id(self, enable_validation: bool = False) -> Uni
                 raise RequiredArgumentMissingError(
                     '"--azure-keyvault-kms-key-id" requires "--enable-azure-keyvault-kms".')
 
-            # PMK validation logic moved from validate_azure_keyvault_kms_key_id
-            if key_id:
-                # Check if PMK (Platform-Managed Keys) is enabled
-                is_pmk_enabled = self.get_kms_infrastructure_encryption() == "Enabled"
-                segments = key_id[len("https://"):].split("/")
-
-                if is_pmk_enabled:
-                    # PMK enabled (K2P): Only accept versionless key ID (3 segments: vault.net/keys/key-name)
-                    if len(segments) != 3:
-                        err_msg = (
-                            "--azure-keyvault-kms-key-id is not a valid versionless Key Vault key ID for PMK. "
-                            "Valid format is https://{key-vault-url}/keys/{key-name}. "
-                            "See https://docs.microsoft.com/en-us/azure/key-vault/general/about-keys-secrets-certificates#vault-name-and-object-name"  # pylint: disable=line-too-long
-                        )
-                        raise InvalidArgumentValueError(err_msg)
-                else:
-                    # PMK disabled (KMS v2): Accept versioned key ID (4 segments)
-                    if len(segments) != 4:
-                        err_msg = (
-                            "--azure-keyvault-kms-key-id is not a valid Key Vault key ID. "
-                            "See https://docs.microsoft.com/en-us/azure/key-vault/general/about-keys-secrets-certificates#vault-name-and-object-name"  # pylint: disable=line-too-long
-                        )
-                        raise InvalidArgumentValueError(err_msg)
-
         return key_id
 
     def get_azure_keyvault_kms_key_id(self) -> Union[str, None]:

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

@@ -2037,7 +2037,8 @@ def test_get_azure_keyvault_kms_key_id(self):
             ctx_5.get_azure_keyvault_kms_key_id()
 
     def test_get_azure_keyvault_kms_key_id_with_pmk_validation(self):
-        # Test PMK-aware validation in _get_azure_keyvault_kms_key_id method
+        # Test that PMK validation is no longer performed in _get_azure_keyvault_kms_key_id method
+        # Since the validation logic was removed, all key ID formats should be accepted
 
         # PMK enabled (infrastructure encryption = "Enabled") - should accept versionless key ID
         versionless_key_id = "https://fakekeyvault.vault.azure.net/keys/fakekeyname"
@@ -2055,7 +2056,7 @@ def test_get_azure_keyvault_kms_key_id_with_pmk_validation(self):
         )
         self.assertEqual(ctx_pmk_versionless.get_azure_keyvault_kms_key_id(), versionless_key_id)
 
-        # PMK enabled - should reject versioned key ID (4 segments)
+        # PMK enabled - should now accept versioned key ID (4 segments) since validation is removed
         versioned_key_id = "https://fakekeyvault.vault.azure.net/keys/fakekeyname/fakeversion"
         ctx_pmk_versioned = AKSPreviewManagedClusterContext(
             self.cmd,
@@ -2069,9 +2070,8 @@ def test_get_azure_keyvault_kms_key_id_with_pmk_validation(self):
             self.models,
             decorator_mode=DecoratorMode.CREATE,
         )
-        with self.assertRaises(InvalidArgumentValueError) as cm:
-            ctx_pmk_versioned.get_azure_keyvault_kms_key_id()
-        self.assertIn("not a valid versionless Key Vault key ID for PMK", str(cm.exception))
+        # No exception should be raised now that validation is removed
+        self.assertEqual(ctx_pmk_versioned.get_azure_keyvault_kms_key_id(), versioned_key_id)
 
         # PMK disabled - should accept versioned key ID (4 segments)
         ctx_no_pmk_versioned = AKSPreviewManagedClusterContext(
@@ -2088,7 +2088,7 @@ def test_get_azure_keyvault_kms_key_id_with_pmk_validation(self):
         )
         self.assertEqual(ctx_no_pmk_versioned.get_azure_keyvault_kms_key_id(), versioned_key_id)
 
-        # PMK disabled - should reject versionless key ID (3 segments)
+        # PMK disabled - should now accept versionless key ID (3 segments) since validation is removed
         ctx_no_pmk_versionless = AKSPreviewManagedClusterContext(
             self.cmd,
             AKSManagedClusterParamDict(
@@ -2101,9 +2101,8 @@ def test_get_azure_keyvault_kms_key_id_with_pmk_validation(self):
             self.models,
             decorator_mode=DecoratorMode.CREATE,
         )
-        with self.assertRaises(InvalidArgumentValueError) as cm:
-            ctx_no_pmk_versionless.get_azure_keyvault_kms_key_id()
-        self.assertIn("not a valid Key Vault key ID", str(cm.exception))
+        # No exception should be raised now that validation is removed
+        self.assertEqual(ctx_no_pmk_versionless.get_azure_keyvault_kms_key_id(), versionless_key_id)
 
         # Test with existing cluster data (UPDATE mode) - PMK enabled should read from cluster
         ctx_update_pmk = AKSPreviewManagedClusterContext(
@@ -2349,16 +2348,15 @@ def test_get_azure_keyvault_kms_key_id(self):
         })
         self.assertEqual(ctx_pmk_versionless.get_azure_keyvault_kms_key_id(), versionless_key_id)
 
-        # Test 8: PMK enabled - should reject versioned key ID
+        # Test 8: PMK enabled - should now accept versioned key ID (validation removed)
         versioned_key_id = "https://fakekeyvault.vault.azure.net/keys/fakekeyname/fakeversion"
         ctx_pmk_versioned = self._create_kms_context({
             "enable_azure_keyvault_kms": True,
             "azure_keyvault_kms_key_id": versioned_key_id,
             "kms_infrastructure_encryption": "Enabled",
         })
-        with self.assertRaises(InvalidArgumentValueError) as cm:
-            ctx_pmk_versioned.get_azure_keyvault_kms_key_id()
-        self.assertIn("not a valid versionless Key Vault key ID for PMK", str(cm.exception))
+        # No exception should be raised since PMK validation was removed
+        self.assertEqual(ctx_pmk_versioned.get_azure_keyvault_kms_key_id(), versioned_key_id)
 
         # Test 9: PMK disabled - should accept versioned key ID
         ctx_no_pmk_versioned = self._create_kms_context({
@@ -2368,15 +2366,14 @@ def test_get_azure_keyvault_kms_key_id(self):
         })
         self.assertEqual(ctx_no_pmk_versioned.get_azure_keyvault_kms_key_id(), versioned_key_id)
 
-        # Test 10: PMK disabled - should reject versionless key ID
+        # Test 10: PMK disabled - should now accept versionless key ID (validation removed)
         ctx_no_pmk_versionless = self._create_kms_context({
             "enable_azure_keyvault_kms": True,
             "azure_keyvault_kms_key_id": versionless_key_id,
             "kms_infrastructure_encryption": "Disabled",
         })
-        with self.assertRaises(InvalidArgumentValueError) as cm:
-            ctx_no_pmk_versionless.get_azure_keyvault_kms_key_id()
-        self.assertIn("not a valid Key Vault key ID", str(cm.exception))
+        # No exception should be raised since PMK validation was removed
+        self.assertEqual(ctx_no_pmk_versionless.get_azure_keyvault_kms_key_id(), versionless_key_id)
 
         # Test 11: PMK enabled in UPDATE mode - should read PMK status from existing cluster
         ctx_update_pmk = self._create_kms_context({