[AKS] az aks create/update: Add support for ACNS transit encryption

Add `--acns-transit-encryption-type` parameter to `az aks create` and
`az aks update` commands. This allows users to configure pod-to-pod
transit encryption on Cilium-based AKS clusters as part of the ACNS
security feature suite.

Bump azure-mgmt-containerservice SDK from 40.0.0 to 41.0.0 which
includes the AdvancedNetworkingSecurityTransitEncryption model.

Signed-off-by: Quang Nguyen <nguyenquang@microsoft.com>

Author: nddq

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

@@ -246,6 +246,10 @@
 CONST_NODE_PROVISIONING_DEFAULT_POOLS_NONE = "None"
 CONST_NODE_PROVISIONING_DEFAULT_POOLS_AUTO = "Auto"
 
+# consts for acns transit encryption
+CONST_TRANSIT_ENCRYPTION_WIREGUARD = "WireGuard"
+CONST_TRANSIT_ENCRYPTION_NONE = "None"
+
 
 # consts for decorator pattern
 class DecoratorMode(Enum):

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

@@ -581,6 +581,10 @@
   - name: --disable-acns-security
     type: bool
     short-summary: Used to disable advanced networking security features on a clusters when enabling advanced networking features with "--enable-acns".
+  - name: --acns-transit-encryption-type
+    type: string
+    short-summary: Set transit encryption type for ACNS security.
+    long-summary: Configures pod-to-pod encryption for Cilium-based clusters. Once enabled, all traffic between Cilium managed pods will be encrypted when it leaves the node boundary. Valid values are "WireGuard" and "None". This can only be used with "--enable-acns".
   - name: --nrg-lockdown-restriction-level
     type: string
     short-summary: Restriction level on the managed node resource group.
@@ -1070,6 +1074,10 @@
   - name: --disable-acns-security
     type: bool
     short-summary: Used to disable advanced networking security features on a clusters when enabling advanced networking features with "--enable-acns".
+  - name: --acns-transit-encryption-type
+    type: string
+    short-summary: Set transit encryption type for ACNS security.
+    long-summary: Configures pod-to-pod encryption for Cilium-based clusters. Once enabled, all traffic between Cilium managed pods will be encrypted when it leaves the node boundary. Valid values are "WireGuard" and "None". This can only be used with "--enable-acns".
   - name: --nrg-lockdown-restriction-level
     type: string
     short-summary: Restriction level on the managed node resource group.

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

@@ -64,7 +64,9 @@
     CONST_NODE_PROVISIONING_MODE_MANUAL,
     CONST_NODE_PROVISIONING_MODE_AUTO,
     CONST_NODE_PROVISIONING_DEFAULT_POOLS_NONE,
-    CONST_NODE_PROVISIONING_DEFAULT_POOLS_AUTO)
+    CONST_NODE_PROVISIONING_DEFAULT_POOLS_AUTO,
+    CONST_TRANSIT_ENCRYPTION_WIREGUARD,
+    CONST_TRANSIT_ENCRYPTION_NONE)
 from azure.cli.command_modules.acs.azurecontainerstorage._consts import (
     CONST_ACSTOR_ALL,
     CONST_DISK_TYPE_EPHEMERAL_VOLUME_ONLY,
@@ -209,6 +211,11 @@
     CONST_NODE_PROVISIONING_DEFAULT_POOLS_AUTO,
 ]
 
+transit_encryption_types = [
+    CONST_TRANSIT_ENCRYPTION_WIREGUARD,
+    CONST_TRANSIT_ENCRYPTION_NONE,
+]
+
 dev_space_endpoint_types = ['Public', 'Private', 'None']
 
 keyvault_network_access_types = [CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PUBLIC, CONST_AZURE_KEYVAULT_NETWORK_ACCESS_PRIVATE]
@@ -561,6 +568,7 @@ def load_arguments(self, _):
         c.argument('enable_acns', action='store_true')
         c.argument('disable_acns_observability', action='store_true')
         c.argument('disable_acns_security', action='store_true')
+        c.argument('acns_transit_encryption_type', arg_type=get_enum_type(transit_encryption_types))
         c.argument("if_match")
         c.argument("if_none_match")
         # node provisioning
@@ -617,6 +625,7 @@ def load_arguments(self, _):
         c.argument('disable_acns', action='store_true')
         c.argument('disable_acns_observability', action='store_true')
         c.argument('disable_acns_security', action='store_true')
+        c.argument('acns_transit_encryption_type', arg_type=get_enum_type(transit_encryption_types))
         # private cluster parameters
         c.argument('enable_apiserver_vnet_integration', action='store_true')
         c.argument('apiserver_subnet_id', validator=validate_apiserver_subnet_id)

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

@@ -700,6 +700,7 @@ def aks_create(
     enable_acns=None,
     disable_acns_observability=None,
     disable_acns_security=None,
+    acns_transit_encryption_type=None,
     # network isoalted cluster
     bootstrap_artifact_source=CONST_ARTIFACT_SOURCE_DIRECT,
     bootstrap_container_registry_resource_id=None,
@@ -925,6 +926,7 @@ def aks_update(
     enable_acns=None,
     disable_acns_observability=None,
     disable_acns_security=None,
+    acns_transit_encryption_type=None,
     # network isoalted cluster
     bootstrap_artifact_source=None,
     bootstrap_container_registry_resource_id=None,

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

@@ -73,6 +73,9 @@ aks create:
       disable_acns_observability:
         rule_exclusions:
         - option_length_too_long
+      acns_transit_encryption_type:
+        rule_exclusions:
+        - option_length_too_long
       nrg_lockdown_restriction_level:
         rule_exclusions:
         - option_length_too_long
@@ -182,6 +185,9 @@ aks update:
       disable_acns_observability:
         rule_exclusions:
         - option_length_too_long
+      acns_transit_encryption_type:
+        rule_exclusions:
+        - option_length_too_long
       nrg_lockdown_restriction_level:
         rule_exclusions:
         - option_length_too_long

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

@@ -2580,6 +2580,32 @@ def get_acns_security(self) -> Union[bool, None]:
         disable_acns_security = self.raw_param.get("disable_acns_security")
         return not bool(disable_acns_security) if disable_acns_security is not None else None
 
+    def get_acns_transit_encryption_type(self) -> Union[str, None]:
+        """Get the transit encryption type for acns security.
+
+        :return: str or None
+        """
+        acns_transit_encryption = self.raw_param.get("acns_transit_encryption_type")
+        if acns_transit_encryption is not None:
+            enable_acns = self.raw_param.get("enable_acns")
+            disable_acns = self.raw_param.get("disable_acns")
+            disable_acns_security = self.raw_param.get("disable_acns_security")
+            if disable_acns_security:
+                raise MutuallyExclusiveArgumentError(
+                    "Cannot specify --acns-transit-encryption and "
+                    "--disable-acns-security at the same time."
+                )
+            if disable_acns:
+                raise MutuallyExclusiveArgumentError(
+                    "Cannot specify --acns-transit-encryption and "
+                    "--disable-acns at the same time."
+                )
+            if self.decorator_mode == DecoratorMode.CREATE and not enable_acns:
+                raise MutuallyExclusiveArgumentError(
+                    "--acns-transit-encryption requires --enable-acns."
+                )
+        return acns_transit_encryption
+
     def _get_pod_cidr_and_service_cidr_and_dns_service_ip_and_docker_bridge_address_and_network_policy(
         self, enable_validation: bool = False
     ) -> Tuple[
@@ -6180,6 +6206,7 @@ def set_up_network_profile(self, mc: ManagedCluster) -> ManagedCluster:
         network_dataplane = self.context.get_network_dataplane()
 
         (acns_enabled, acns_observability, acns_security) = self.context.get_acns_enablement()
+        acns_transit_encryption = self.context.get_acns_transit_encryption_type()
         if acns_enabled is not None:
             acns = self.models.AdvancedNetworking(
                 enabled=acns_enabled,
@@ -6192,6 +6219,12 @@ def set_up_network_profile(self, mc: ManagedCluster) -> ManagedCluster:
                 acns.security = self.models.AdvancedNetworkingSecurity(
                     enabled=acns_security,
                 )
+            if acns_transit_encryption is not None:
+                if acns.security is None:
+                    acns.security = self.models.AdvancedNetworkingSecurity()
+                acns.security.transit_encryption = self.models.AdvancedNetworkingSecurityTransitEncryption(
+                    type=acns_transit_encryption,
+                )
 
         if any(
             [
@@ -8091,6 +8124,7 @@ def update_network_profile_advanced_networking(self, mc: ManagedCluster) -> Mana
         """
         self._ensure_mc(mc)
         (acns_enabled, acns_observability, acns_security) = self.context.get_acns_enablement()
+        acns_transit_encryption = self.context.get_acns_transit_encryption_type()
         if acns_enabled is not None:
             acns = self.models.AdvancedNetworking(
                 enabled=acns_enabled,
@@ -8103,8 +8137,23 @@ def update_network_profile_advanced_networking(self, mc: ManagedCluster) -> Mana
                 acns.security = self.models.AdvancedNetworkingSecurity(
                     enabled=acns_security,
                 )
-        if acns_enabled is not None:
+            if acns_transit_encryption is not None:
+                if acns.security is None:
+                    acns.security = self.models.AdvancedNetworkingSecurity()
+                acns.security.transit_encryption = self.models.AdvancedNetworkingSecurityTransitEncryption(
+                    type=acns_transit_encryption,
+                )
             mc.network_profile.advanced_networking = acns
+        elif acns_transit_encryption is not None:
+            if mc.network_profile.advanced_networking is None:
+                mc.network_profile.advanced_networking = self.models.AdvancedNetworking()
+            if mc.network_profile.advanced_networking.security is None:
+                mc.network_profile.advanced_networking.security = self.models.AdvancedNetworkingSecurity()
+            mc.network_profile.advanced_networking.security.transit_encryption = (
+                self.models.AdvancedNetworkingSecurityTransitEncryption(
+                    type=acns_transit_encryption,
+                )
+            )
         return mc
 
     def update_http_proxy_config(self, mc: ManagedCluster) -> ManagedCluster: