[Container App]: Add az containerapp ingress access-restriction group

src/containerapp/HISTORY.rst

@@ -3,6 +3,10 @@
 Release History
 ===============
 
+0.3.15
+++++++
+* Add 'az containerapp containerapp ingress ip-restriction' command group to manage IP restrictions on the ingress of a container app.
+
 0.3.14
 ++++++
 * 'az containerapp logs show'/'az containerapp exec': Fix "KeyError" bug

src/containerapp/azext_containerapp/_help.py

@@ -671,6 +671,38 @@
           az containerapp ingress traffic show -n MyContainerapp -g MyResourceGroup
 """
 
+helps['containerapp ingress access-restriction'] = """
+    type: group
+    short-summary: Commands to manage IP access restrictions.
+"""
+
+helps['containerapp ingress access-restriction set'] = """
+    type: command
+    short-summary: Configure IP access restrictions for a container app.
+    examples:
+    - name: Add an allow IP access restriction.
+      text: |
+          az containerapp ingress access-restriction set -n MyContainerapp -g MyResourceGroup --rule-name restrictionName --ip-address 192.168.1.1/32 --description "Restriction description." --action Allow
+"""
+
+helps['containerapp ingress access-restriction remove'] = """
+    type: command
+    short-summary: Remove IP access restrictions from a container app.
+    examples:
+    - name: Remove an IP access restriction.
+      text: |
+          az containerapp ingress access-restriction remove -n MyContainerapp -g MyResourceGroup --rule-name restrictionName
+"""
+
+helps['containerapp ingress access-restriction list'] = """
+    type: command
+    short-summary: List IP access restrictions for a container app.
+    examples:
+    - name: List IP access restrictions.
+      text: |
+          az containerapp ingress access-restriction list -n MyContainerapp -g MyResourceGroup
+"""
+
 # Registry Commands
 helps['containerapp registry'] = """
     type: group

src/containerapp/azext_containerapp/_params.py

@@ -249,6 +249,15 @@ def load_arguments(self, _):
         c.argument('target_port', type=int, validator=validate_target_port, help="The application port used for ingress traffic.")
         c.argument('exposed_port', type=int, help="Additional exposed port. Only supported by tcp transport protocol. Must be unique per environment if the app ingress is external.")
 
+    with self.argument_context('containerapp ingress access-restriction') as c:
+        c.argument('action', arg_type=get_enum_type(['Allow', 'Deny']), help='Whether the IP security restriction allows or denies access. All restrictions must be use the same action. If no restrictions are set, all traffic is allowed.')
+        c.argument('rule_name', help="The IP security restriction name.")
+        c.argument('description', help="The description of the IP security restriction.")
+        c.argument('ip_address', help="The address range of the IP security restriction in IPv4 CIDR notation. (for example, '198.51.100.14/24')")
+
+    with self.argument_context('containerapp ingress access-restriction list') as c:
+        c.argument('name', id_part=None)
+
     with self.argument_context('containerapp ingress traffic') as c:
         c.argument('revision_weights', nargs='+', options_list=['--revision-weight', c.deprecate(target='--traffic-weight', redirect='--revision-weight')], help="A list of revision weight(s) for the container app. Space-separated values in 'revision_name=weight' format. For latest revision, use 'latest=weight'")
         c.argument('label_weights', nargs='+', options_list=['--label-weight'], help="A list of label weight(s) for the container app. Space-separated values in 'label_name=weight' format.")

src/containerapp/azext_containerapp/_utils.py

@@ -1557,6 +1557,26 @@ def list_environment_locations(cmd):
     return res_locations
 
 
+def set_ip_restrictions(ip_restrictions, ip_restriction_name, ip_address_range, description, action):
+    updated = False
+    for e in ip_restrictions:
+        if ip_restriction_name.lower() == e["name"].lower():
+            e["description"] = description
+            e["ipAddressRange"] = ip_address_range
+            e["action"] = action
+            updated = True
+            break
+    if not updated:
+        new_ip_restriction = {
+            "name": ip_restriction_name,
+            "description": description,
+            "ipAddressRange": ip_address_range,
+            "action": action
+        }
+        ip_restrictions.append(new_ip_restriction)
+    return ip_restrictions
+
+
 def _azure_monitor_quickstart(cmd, name, resource_group_name, storage_account, logs_destination):
     if logs_destination != "azure-monitor":
         if storage_account:

src/containerapp/azext_containerapp/commands.py

@@ -120,6 +120,11 @@ def load_command_table(self, _):
         g.custom_command('set', 'set_ingress_traffic', exception_handler=ex_handler_factory())
         g.custom_show_command('show', 'show_ingress_traffic')
 
+    with self.command_group('containerapp ingress access-restriction') as g:
+        g.custom_command('set', 'set_ip_restriction', exception_handler=ex_handler_factory())
+        g.custom_command('remove', 'remove_ip_restriction')
+        g.custom_show_command('list', 'show_ip_restrictions')
+
     with self.command_group('containerapp registry') as g:
         g.custom_command('set', 'set_registry', exception_handler=ex_handler_factory())
         g.custom_show_command('show', 'show_registry')

src/containerapp/azext_containerapp/custom.py

@@ -68,7 +68,8 @@
                      generate_randomized_cert_name, _get_name, load_cert_file, check_cert_name_availability,
                      validate_hostname, patch_new_custom_domain, get_custom_domains, _validate_revision_name, set_managed_identity,
                      create_acrpull_role_assignment, is_registry_msi_system, clean_null_values, _populate_secret_values,
-                     validate_environment_location, safe_set, parse_metadata_flags, parse_auth_flags, _azure_monitor_quickstart)
+                     validate_environment_location, safe_set, parse_metadata_flags, parse_auth_flags, _azure_monitor_quickstart,
+                     set_ip_restrictions)
 from ._validators import validate_create, validate_revision_suffix
 from ._ssh_utils import (SSH_DEFAULT_ENCODING, WebSocketConnection, read_ssh, get_stdin_writer, SSH_CTRL_C_MSG,
                          SSH_BACKUP_ENCODING)
@@ -1951,6 +1952,88 @@ def show_ingress_traffic(cmd, name, resource_group_name):
         raise ValidationError("Ingress must be enabled to show ingress traffic. Try running `az containerapp ingress -h` for more info.") from e
 
 
+def set_ip_restriction(cmd, name, resource_group_name, rule_name, ip_address, action, description=None, no_wait=False):
+    _validate_subscription_registered(cmd, CONTAINER_APPS_RP)
+
+    containerapp_def = None
+    try:
+        containerapp_def = ContainerAppClient.show(cmd=cmd, resource_group_name=resource_group_name, name=name)
+    except:
+        pass
+
+    if not containerapp_def:
+        raise ResourceNotFoundError(f"The containerapp '{name}' does not exist in group '{resource_group_name}'")
+
+    ip_restrictions = safe_get(containerapp_def, "properties", "configuration", "ingress", "ipSecurityRestrictions", default=[])
+
+    ip_security_restrictions = set_ip_restrictions(ip_restrictions, rule_name, ip_address, description, action)
+    containerapp_patch = {}
+    safe_set(containerapp_patch, "properties", "configuration", "ingress", "ipSecurityRestrictions", value=ip_security_restrictions)
+    try:
+        r = ContainerAppClient.update(
+            cmd=cmd, resource_group_name=resource_group_name, name=name, container_app_envelope=containerapp_patch, no_wait=no_wait)
+        return r['properties']['configuration']['ingress']['ipSecurityRestrictions']
+    except Exception as e:
+        handle_raw_exception(e)
+
+
+def remove_ip_restriction(cmd, name, resource_group_name, rule_name, no_wait=False):
+    _validate_subscription_registered(cmd, CONTAINER_APPS_RP)
+
+    containerapp_def = None
+    try:
+        containerapp_def = ContainerAppClient.show(cmd=cmd, resource_group_name=resource_group_name, name=name)
+    except:
+        pass
+
+    if not containerapp_def:
+        raise ResourceNotFoundError("The containerapp '{}' does not exist".format(name))
+
+    ip_restrictions = safe_get(containerapp_def, "properties", "configuration", "ingress", "ipSecurityRestrictions", default=[])
+
+    restriction_removed = False
+    for index, value in enumerate(ip_restrictions):
+        if value["name"].lower() == rule_name.lower():
+            ip_restrictions.pop(index)
+            restriction_removed = True
+            break
+
+    if not restriction_removed:
+        raise ValidationError(f"Ip restriction name '{rule_name}' does not exist.")
+
+    containerapp_patch = {}
+    safe_set(containerapp_patch, "properties", "configuration", "ingress", "ipSecurityRestrictions", value=ip_restrictions)
+    try:
+        r = ContainerAppClient.update(
+            cmd=cmd, resource_group_name=resource_group_name, name=name, container_app_envelope=containerapp_patch, no_wait=no_wait)
+        ip_restrictions = safe_get(r, "properties", "configuration", "ingress", "ipSecurityRestrictions", default=[])
+        return ip_restrictions
+    except Exception as e:
+        handle_raw_exception(e)
+
+
+def show_ip_restrictions(cmd, name, resource_group_name):
+    _validate_subscription_registered(cmd, CONTAINER_APPS_RP)
+
+    containerapp_def = None
+    try:
+        containerapp_def = ContainerAppClient.show(cmd=cmd, resource_group_name=resource_group_name, name=name)
+    except:
+        pass
+
+    if not containerapp_def:
+        raise ResourceNotFoundError("The containerapp '{}' does not exist".format(name))
+
+    try:
+        try:
+            containerapp_def['properties']['configuration']['ingress']
+        except Exception as e:
+            raise ValidationError("Ingress must be enabled to list ip restrictions. Try running `az containerapp ingress -h` for more info.") from e
+        return containerapp_def['properties']['configuration']['ingress']['ipSecurityRestrictions']
+    except Exception as e:
+        return []
+
+
 def show_registry(cmd, name, resource_group_name, server):
     _validate_subscription_registered(cmd, CONTAINER_APPS_RP)