feat(acns): add transit encryption options for az create and update commands

Signed-off-by: Quang Nguyen <[email protected]>

Author: nddq

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
 +++++++
 
+18.0.0b3
++++++++
+* Add option `--acns-transit-encryption-type <None|WireGuard>` to `az aks create/update`
+
 18.0.0b2
 +++++++
 * Vendor new SDK and bump API version to 2025-03-02-preview.

src/aks-preview/azext_aks_preview/_consts.py

@@ -137,6 +137,10 @@
 CONST_ADVANCED_NETWORKPOLICIES_FQDN = "FQDN"
 CONST_ADVANCED_NETWORKPOLICIES_L7 = "L7"
 
+# ACNS transit encryption type
+CONST_TRANSIT_ENCRYPTION_TYPE_NONE = "None"
+CONST_TRANSIT_ENCRYPTION_TYPE_WIREGUARD = "WireGuard"
+
 # network pod ip allocation mode
 CONST_NETWORK_POD_IP_ALLOCATION_MODE_DYNAMIC_INDIVIDUAL = "DynamicIndividual"
 CONST_NETWORK_POD_IP_ALLOCATION_MODE_STATIC_BLOCK = "StaticBlock"

src/aks-preview/azext_aks_preview/_params.py

@@ -135,6 +135,8 @@
     CONST_ADVANCED_NETWORKPOLICIES_NONE,
     CONST_ADVANCED_NETWORKPOLICIES_FQDN,
     CONST_ADVANCED_NETWORKPOLICIES_L7,
+    CONST_TRANSIT_ENCRYPTION_TYPE_NONE,
+    CONST_TRANSIT_ENCRYPTION_TYPE_WIREGUARD
 )
 
 from azext_aks_preview._validators import (
@@ -300,6 +302,10 @@
     CONST_ADVANCED_NETWORKPOLICIES_FQDN,
     CONST_ADVANCED_NETWORKPOLICIES_L7,
 ]
+transit_encryption_types = [
+    CONST_TRANSIT_ENCRYPTION_TYPE_NONE,
+    CONST_TRANSIT_ENCRYPTION_TYPE_WIREGUARD,
+]
 network_dataplanes = [CONST_NETWORK_DATAPLANE_AZURE, CONST_NETWORK_DATAPLANE_CILIUM]
 disk_driver_versions = [CONST_DISK_DRIVER_V1, CONST_DISK_DRIVER_V2]
 outbound_types = [
@@ -846,6 +852,11 @@ def load_arguments(self, _):
             is_preview=True,
             arg_type=get_enum_type(advanced_networkpolicies),
         )
+        c.argument(
+            "acns_transit_encryption_type",
+            is_preview=True,
+            arg_type=get_enum_type(transit_encryption_types),
+        )
         c.argument(
             "enable_retina_flow_logs",
             action="store_true",
@@ -1325,6 +1336,11 @@ def load_arguments(self, _):
             is_preview=True,
             arg_type=get_enum_type(advanced_networkpolicies),
         )
+        c.argument(
+            "acns_transit_encryption_type",
+            is_preview=True,
+            arg_type=get_enum_type(transit_encryption_types),
+        )
         c.argument(
             "enable_retina_flow_logs",
             action="store_true",

src/aks-preview/azext_aks_preview/custom.py

@@ -499,6 +499,7 @@ def aks_create(
     disable_acns_observability=None,
     disable_acns_security=None,
     acns_advanced_networkpolicies=None,
+    acns_transit_encryption_type=None,
     enable_retina_flow_logs=None,
     # nodepool
     crg_id=None,
@@ -730,6 +731,7 @@ def aks_update(
     disable_acns_observability=None,
     disable_acns_security=None,
     acns_advanced_networkpolicies=None,
+    acns_transit_encryption_type=None,
     enable_retina_flow_logs=None,
     disable_retina_flow_logs=None,
     # metrics profile

src/aks-preview/azext_aks_preview/managed_cluster_decorator.py

@@ -781,6 +781,21 @@ def get_acns_advanced_networkpolicies(self) -> Union[str, None]:
                 )
         return self.raw_param.get("acns_advanced_networkpolicies")
 
+    def get_acns_transit_encryption_type(self) -> Union[str, None]:
+        """Get the value of acns_transit_encryption_type
+
+        :return: str or None
+        """
+        disable_acns_security = self.raw_param.get("disable_acns_security")
+        disable_acns = self.raw_param.get("disable_acns")
+        acns_transit_encryption_type = self.raw_param.get("acns_transit_encryption_type")
+        if acns_transit_encryption_type is not None:
+            if disable_acns_security or disable_acns:
+                raise MutuallyExclusiveArgumentError(
+                    "--disable-acns-security and --disable-acns cannot be used with acns_transit_encryption_type."
+                )
+        return self.raw_param.get("acns_transit_encryption_type")
+
     def get_retina_flow_logs(self, mc: ManagedCluster) -> Union[bool, None]:
         """Get the enablement of retina flow logs
 
@@ -2924,6 +2939,7 @@ def set_up_network_profile(self, mc: ManagedCluster) -> ManagedCluster:
         acns = None
         (acns_enabled, acns_observability_enabled, acns_security_enabled) = self.context.get_acns_enablement()
         acns_advanced_networkpolicies = self.context.get_acns_advanced_networkpolicies()
+        acns_transit_encryption_type = self.context.get_acns_transit_encryption_type()
         if acns_enabled is not None:
             acns = self.models.AdvancedNetworking(
                 enabled=acns_enabled,
@@ -2943,6 +2959,13 @@ def set_up_network_profile(self, mc: ManagedCluster) -> ManagedCluster:
                     )
                 else:
                     acns.security.advanced_network_policies = acns_advanced_networkpolicies
+            if acns_transit_encryption_type is not None:
+                if acns.security is None:
+                    acns.security = self.models.AdvancedNetworkingSecurity(
+                        type=acns_transit_encryption_type
+                    )
+                else:
+                    acns.security.type = acns_transit_encryption_type
             network_profile.advanced_networking = acns
         return mc
 
@@ -4020,6 +4043,7 @@ def update_acns_in_network_profile(self, mc: ManagedCluster) -> ManagedCluster:
         acns = None
         (acns_enabled, acns_observability_enabled, acns_security_enabled) = self.context.get_acns_enablement()
         acns_advanced_networkpolicies = self.context.get_acns_advanced_networkpolicies()
+        acns_transit_encryption_type = self.context.get_acns_transit_encryption_type()
         if acns_enabled is not None:
             acns = self.models.AdvancedNetworking(
                 enabled=acns_enabled,
@@ -4039,6 +4063,13 @@ def update_acns_in_network_profile(self, mc: ManagedCluster) -> ManagedCluster:
                     )
                 else:
                     acns.security.advanced_network_policies = acns_advanced_networkpolicies
+            if acns_transit_encryption_type is not None:
+                if acns.security is None:
+                    acns.security = self.models.AdvancedNetworkingSecurity(
+                        type=acns_transit_encryption_type
+                    )
+                else:
+                    acns.security.type = acns_transit_encryption_type
             mc.network_profile.advanced_networking = acns
         return mc