[Storage] az storage account create/update, az storage account network-rule add/remove: IPv6 private preview

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

@@ -428,6 +428,9 @@ def load_arguments(self, _):  # pylint: disable=too-many-locals, too-many-statem
                    help='Allow you to specify the type of endpoint. Set this to AzureDNSZone to create a large number '
                         'of accounts in a single subscription, which creates accounts in an Azure DNS Zone and the '
                         'endpoint URL will have an alphanumeric DNS Zone identifier.')
+        c.argument('publish_ipv6_endpoint', arg_type=get_three_state_flag(), min_api='2023-01-01',
+                   arg_group='IPv6 Endpoint',
+                   help='A boolean flag which indicates whether IPv6 storage endpoints are to be published.')
 
     with self.argument_context('storage account private-endpoint-connection',
                                resource_type=ResourceType.MGMT_STORAGE) as c:
@@ -518,6 +521,9 @@ def load_arguments(self, _):  # pylint: disable=too-many-locals, too-many-statem
         c.argument('upgrade_to_storagev2', arg_type=get_three_state_flag(),
                    help='Upgrade Storage Account Kind to StorageV2.')
         c.argument('yes', options_list=['--yes', '-y'], help='Do not prompt for confirmation.', action='store_true')
+        c.argument('publish_ipv6_endpoint', arg_type=get_three_state_flag(), min_api='2023-01-01',
+                   arg_group='IPv6 Endpoint',
+                   help='A boolean flag which indicates whether IPv6 storage endpoints are to be published.')
 
     for scope in ['storage account create', 'storage account update']:
         with self.argument_context(scope, arg_group='Customer managed key', min_api='2017-06-01',
@@ -633,10 +639,12 @@ def load_arguments(self, _):  # pylint: disable=too-many-locals, too-many-statem
         c.argument('account_name', acct_name_type, id_part=None)
 
     with self.argument_context('storage account network-rule', resource_type=ResourceType.MGMT_STORAGE) as c:
-        from ._validators import validate_ip_address
+        from ._validators import validate_ip_address, validate_ipv6_address
         c.argument('account_name', acct_name_type, id_part=None)
         c.argument('ip_address', nargs='*', help='IPv4 address or CIDR range. Can supply a list: --ip-address ip1 '
                                                  '[ip2]...', validator=validate_ip_address)
+        c.argument('ipv6_address', nargs='*', help='IPv6 address or CIDR range. Can supply a list: --ipv6-address ip1 '
+                                                   '[ip2]...', validator=validate_ipv6_address)
         c.argument('subnet', help='Name or ID of subnet. If name is supplied, `--vnet-name` must be supplied.')
         c.argument('vnet_name', help='Name of a virtual network.', validator=validate_subnet)
         c.argument('action', action_type)

src/azure-cli/azure/cli/command_modules/storage/_validators.py

@@ -2235,22 +2235,32 @@ def validate_fs_file_set_expiry(namespace):
         pass
 
 
-def validate_ip_address(namespace):
-    # if there are overlapping ip ranges, throw an exception
-    ip_address = namespace.ip_address
-
+def _find_ip_address_overlap(ip_address, ipv6=False):
     if not ip_address:
         return
 
     ip_address_networks = [ip_network(ip) for ip in ip_address]
+    error_str = "ipv6 addresses {} and {} provided are overlapping: --ipv6_address ip1 [ip2]..." if ipv6 else \
+        "ip addresses {} and {} provided are overlapping: --ip_address ip1 [ip2]..."
     for idx, ip_address_network in enumerate(ip_address_networks):
         for idx2, ip_address_network2 in enumerate(ip_address_networks):
             if idx == idx2:
                 continue
             if ip_address_network.overlaps(ip_address_network2):
                 from azure.cli.core.azclierror import InvalidArgumentValueError
-                raise InvalidArgumentValueError(f"ip addresses {ip_address_network} and {ip_address_network2} "
-                                                f"provided are overlapping: --ip_address ip1 [ip2]...")
+                raise InvalidArgumentValueError(error_str.format(ip_address_network, ip_address_network2))
+
+
+def validate_ip_address(namespace):
+    # if there are overlapping ip ranges, throw an exception
+    ip_address = namespace.ip_address
+    _find_ip_address_overlap(ip_address=ip_address, ipv6=False)
+
+
+def validate_ipv6_address(namespace):
+    # if there are overlapping ip ranges, throw an exception
+    ipv6_address = namespace.ipv6_address
+    _find_ip_address_overlap(ip_address=ipv6_address, ipv6=True)
 
 
 # pylint: disable=too-few-public-methods

src/azure-cli/azure/cli/command_modules/storage/operations/account.py

@@ -73,7 +73,8 @@ def create_storage_account(cmd, resource_group_name, account_name, sku=None, loc
                            allow_cross_tenant_replication=None, default_share_permission=None,
                            enable_nfs_v3=None, subnet=None, vnet_name=None, action='Allow', enable_alw=None,
                            immutability_period_since_creation_in_days=None, immutability_policy_state=None,
-                           allow_protected_append_writes=None, public_network_access=None, dns_endpoint_type=None):
+                           allow_protected_append_writes=None, public_network_access=None, dns_endpoint_type=None,
+                           publish_ipv6_endpoint=None):
     StorageAccountCreateParameters, Kind, Sku, CustomDomain, AccessTier, Identity, Encryption, NetworkRuleSet = \
         cmd.get_models('StorageAccountCreateParameters', 'Kind', 'Sku', 'CustomDomain', 'AccessTier', 'Identity',
                        'Encryption', 'NetworkRuleSet')
@@ -293,6 +294,12 @@ def create_storage_account(cmd, resource_group_name, account_name, sku=None, loc
     if dns_endpoint_type is not None:
         params.dns_endpoint_type = dns_endpoint_type
 
+    if publish_ipv6_endpoint is not None:
+        DualStackEndpointPreference = cmd.get_models('DualStackEndpointPreference')
+        params.dual_stack_endpoint_preference = DualStackEndpointPreference(
+            publish_ipv6_endpoint=publish_ipv6_endpoint
+        )
+
     return scf.storage_accounts.begin_create(resource_group_name, account_name, params)
 
 
@@ -387,7 +394,7 @@ def update_storage_account(cmd, instance, sku=None, tags=None, custom_domain=Non
                            allow_cross_tenant_replication=None, default_share_permission=None,
                            immutability_period_since_creation_in_days=None, immutability_policy_state=None,
                            allow_protected_append_writes=None, public_network_access=None, upgrade_to_storagev2=None,
-                           yes=None):
+                           yes=None, publish_ipv6_endpoint=None):
     StorageAccountUpdateParameters, Sku, CustomDomain, AccessTier, Identity, Encryption, NetworkRuleSet, Kind = \
         cmd.get_models('StorageAccountUpdateParameters', 'Sku', 'CustomDomain', 'AccessTier', 'Identity', 'Encryption',
                        'NetworkRuleSet', 'Kind')
@@ -671,6 +678,12 @@ def update_storage_account(cmd, instance, sku=None, tags=None, custom_domain=Non
     if enable_local_user is not None:
         params.is_local_user_enabled = enable_local_user
 
+    if publish_ipv6_endpoint is not None:
+        DualStackEndpointPreference = cmd.get_models('DualStackEndpointPreference')
+        params.dual_stack_endpoint_preference = DualStackEndpointPreference(
+            publish_ipv6_endpoint=publish_ipv6_endpoint
+        )
+
     return params
 
 
@@ -683,11 +696,12 @@ def list_network_rules(client, resource_group_name, account_name):
 
 
 def add_network_rule(cmd, client, resource_group_name, account_name, action='Allow', subnet=None,
-                     vnet_name=None, ip_address=None, tenant_id=None, resource_id=None):  # pylint: disable=unused-argument
+                     vnet_name=None, ip_address=None, ipv6_address=None, tenant_id=None, resource_id=None):  # pylint: disable=unused-argument
     sa = client.get_properties(resource_group_name, account_name)
     rules = sa.network_rule_set
-    if not subnet and not ip_address:
+    if not subnet and not ip_address and not ipv6_address:
         logger.warning('No subnet or ip address supplied.')
+
     if subnet:
         from msrestazure.tools import is_valid_resource_id
         if not is_valid_resource_id(subnet):
@@ -698,6 +712,7 @@ def add_network_rule(cmd, client, resource_group_name, account_name, action='All
         rules.virtual_network_rules = [r for r in rules.virtual_network_rules
                                        if r.virtual_network_resource_id.lower() != subnet.lower()]
         rules.virtual_network_rules.append(VirtualNetworkRule(virtual_network_resource_id=subnet, action=action))
+
     if ip_address:
         IpRule = cmd.get_models('IPRule')
         if not rules.ip_rules:
@@ -708,12 +723,30 @@ def add_network_rule(cmd, client, resource_group_name, account_name, action='All
                 existing_ip_network = ip_network(x.ip_address_or_range)
                 new_ip_network = ip_network(ip)
                 if new_ip_network.overlaps(existing_ip_network):
-                    logger.warning("IP/CIDR %s overlaps with %s, which exists already. Not adding duplicates.",
+                    logger.warning("IPv4/CIDR %s overlaps with %s, which exists already. Not adding duplicates.",
                                    ip, x.ip_address_or_range)
                     to_modify = False
                     break
             if to_modify:
                 rules.ip_rules.append(IpRule(ip_address_or_range=ip, action=action))
+
+    if ipv6_address:
+        IpRule = cmd.get_models('IPRule')
+        if not rules.ipv6_rules:
+            rules.ipv6_rules = []
+        for ip in ipv6_address:
+            to_modify = True
+            for x in rules.ipv6_rules:
+                existing_ip_network = ip_network(x.ip_address_or_range)
+                new_ip_network = ip_network(ip)
+                if new_ip_network.overlaps(existing_ip_network):
+                    logger.warning("IPv6/CIDR %s overlaps with %s, which exists already. Not adding duplicates.",
+                                   ip, x.ip_address_or_range)
+                    to_modify = False
+                    break
+            if to_modify:
+                rules.ipv6_rules.append(IpRule(ip_address_or_range=ip, action=action))
+
     if resource_id:
         ResourceAccessRule = cmd.get_models('ResourceAccessRule')
         if not rules.resource_access_rules:
@@ -727,7 +760,7 @@ def add_network_rule(cmd, client, resource_group_name, account_name, action='All
     return client.update(resource_group_name, account_name, params)
 
 
-def remove_network_rule(cmd, client, resource_group_name, account_name, ip_address=None, subnet=None,
+def remove_network_rule(cmd, client, resource_group_name, account_name, ip_address=None, ipv6_address=None, subnet=None,
                         vnet_name=None, tenant_id=None, resource_id=None):  # pylint: disable=unused-argument
     sa = client.get_properties(resource_group_name, account_name)
     rules = sa.network_rule_set
@@ -739,6 +772,11 @@ def remove_network_rule(cmd, client, resource_group_name, account_name, ip_addre
         rules.ip_rules = list(filter(lambda x: all(ip_network(x.ip_address_or_range) != i for i in to_remove),
                                      rules.ip_rules))
 
+    if ipv6_address:
+        to_remove = [ip_network(x) for x in ipv6_address]
+        rules.ipv6_rules = list(filter(lambda x: all(ip_network(x.ip_address_or_range) != i for i in to_remove),
+                                       rules.ipv6_rules))
+
     if resource_id:
         rules.resource_access_rules = [x for x in rules.resource_access_rules if
                                        not (x.tenant_id == tenant_id and x.resource_id == resource_id)]