vyos.utils: T8977: let a caller accept a non-local address as assigned

Whether an address may be bound is not the same question as whether it is
currently configured somewhere: the kernel permits binding a foreign address
once the non-local bind sysctl is set, which is how a failover address is
commonly handed over after the commit. Callers can now opt into that, the
default is unchanged.

A lookup no longer skips the VRF device itself, which carries the addresses
of its own L3 domain, and an interface name can be retrieved for an address
without the caller asking a yes/no question in the first place.

Address and range validation is split into separate checks, so a caller no
longer has to accept ranges where only an address is meaningful. The two
duplicated version checks are dropped in favour of the existing ones.

Co-authored-by: Christian Breunig <christian@breunig.cc>

2 files changed, 95 insertions, 42 deletions

diff --git a/python/vyos/template.py b/python/vyos/template.py
index aee53eb4d..c35a25d21 100755
--- a/python/vyos/template.py
+++ b/python/vyos/template.py
@@ -913,12 +913,13 @@ def range_to_regex(num_range):
 @register_filter('kea_address_json')
 def kea_address_json(addresses):
     from json import dumps
-    from vyos.utils.network import is_addr_assigned
+    from vyos.utils.network import get_interfaces_by_ip
 
     out = []
 
     for address in addresses:
-        ifname = is_addr_assigned(address, return_ifname=True, include_vrf=True)
+        ifaces = get_interfaces_by_ip(address, include_vrf=True)
+        ifname = ifaces[0] if ifaces else None
 
         if not ifname:
             continue
diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py
index e7987c053..e4d98777b 100644
--- a/python/vyos/utils/network.py
+++ b/python/vyos/utils/network.py
@@ -423,10 +423,13 @@ def check_port_availability(address: str = None, port: int = 0,
                 s.bind(sockaddr)
                 return True # port is free to use
         except OSError:
-            return False # port is already in use
+            # This address family is busy, but another might be free.
+            # Continue to the next one.
+            pass
 
-    # if we reach this point, no socket was tested and we assume the port is
-    # already in use - better safe then sorry
+    # If we reach this point, the port is already in use
+    # or no socket was tested, in which case it is safer
+    # to assume the port being in use
     return False
 
 
@@ -470,11 +473,12 @@ def is_listen_port_bind_service(port: int, service: str, address: str = None) ->
         pid_name = process(pid).name()
         pid_port = addr.port
 
-        if has_address:
-            if address == addr.ip and port == pid_port and service == pid_name:
+        # Ignore address check when service listening on "any address",
+        # service config might change from any to limited address scope
+        if port == pid_port and service == pid_name:
+            if not has_address:
                 return True
-        else:
-            if service == pid_name and port == pid_port:
+            if addr.ip in ('::', '0.0.0.0') or address == addr.ip:
                 return True
 
     return False
@@ -490,22 +494,37 @@ def is_ipv6_link_local(addr):
 
     return False
 
-def is_addr_assigned(ip_address, vrf=None, return_ifname=False, include_vrf=False) -> bool | str:
+
+def is_addr_assigned(
+    ip_address: str, vrf=None, include_vrf=False, allow_nonlocal: bool = False
+) -> bool:
     """ Verify if the given IPv4/IPv6 address is assigned to any interface """
     from netifaces import interfaces # pylint: disable = no-name-in-module
     from vyos.utils.network import get_interface_config
     from vyos.utils.dict import dict_search
+    from vyos.utils.system import sysctl_read
+    from vyos.template import is_ipv4
+    from vyos.template import is_ipv6
+
+    if allow_nonlocal:
+        if is_ipv4(ip_address) and \
+           sysctl_read(['net', 'ipv4', 'ip_nonlocal_bind']) == '1':
+            return True
+        if is_ipv6(ip_address) and \
+           sysctl_read(['net', 'ipv6', 'ip_nonlocal_bind']) == '1':
+            return True
 
     for interface in interfaces():
         # Check if interface belongs to the requested VRF, if this is not the
         # case there is no need to proceed with this data set - continue loop
         # with next element
-        tmp = get_interface_config(interface)
-        if dict_search('master', tmp) != vrf and not include_vrf:
-            continue
+        if vrf is not None and not include_vrf and interface != vrf:
+            tmp = get_interface_config(interface)
+            if dict_search('master', tmp) != vrf:
+                continue
 
         if is_intf_addr_assigned(interface, ip_address):
-            return interface if return_ifname else True
+            return True
 
     return False
 
@@ -518,8 +537,13 @@ def is_intf_addr_assigned(ifname: str, addr: str, netns: str=None) -> bool:
     import jmespath
 
     from vyos.utils.process import rc_cmd
+    from vyos.template import is_ipv4
+    from vyos.template import is_ipv6
     from ipaddress import ip_interface
 
+    if not is_ipv4(addr) and not is_ipv6(addr):
+        return False
+
     netns_cmd = f'ip netns exec {netns}' if netns else ''
     rc, out = rc_cmd(f'{netns_cmd} ip --json address show dev {ifname}')
     if rc == 0:
@@ -764,59 +788,87 @@ def get_nft_vrf_zone_mapping() -> dict:
         output.append({'interface' : vrf_name, 'vrf_tableid' : vrf_id})
     return output
 
-def is_valid_ipv4_address_or_range(addr: str) -> bool:
+def is_ipv4_range(addr: str) -> bool:
     """
-    Validates if the provided address is a valid IPv4, CIDR or IPv4 range
+    Validates if the provided address is a valid IPv4 range.
     :param addr: address to test
     :return: bool: True if provided address is valid
     """
-    from ipaddress import ip_network
+    from ipaddress import ip_address
+
+    if addr.count('-') != 1:
+        return False
+    start, end = addr.split('-')
     try:
-        if '-' in addr: # If we are checking a range, validate both address's individually
-            split = addr.split('-')
-            return is_valid_ipv4_address_or_range(split[0]) and is_valid_ipv4_address_or_range(split[1])
-        else:
-            return ip_network(addr).version == 4
-    except Exception:
+        return ip_address(start).version == 4 and ip_address(end).version == 4
+    except ValueError:
         return False
 
-def is_valid_ipv6_address_or_range(addr: str) -> bool:
+
+def is_ipv6_range(addr: str) -> bool:
     """
-    Validates if the provided address is a valid IPv4, CIDR or IPv4 range
+    Validates if the provided address is a valid IPv6 range.
     :param addr: address to test
     :return: bool: True if provided address is valid
     """
-    from ipaddress import ip_network
+    from ipaddress import ip_address
+
+    if addr.count('-') != 1:
+        return False
+    start, end = addr.split('-')
     try:
-        if '-' in addr: # If we are checking a range, validate both address's individually
-            split = addr.split('-')
-            return is_valid_ipv6_address_or_range(split[0]) and is_valid_ipv6_address_or_range(split[1])
-        else:
-            return ip_network(addr).version == 6
-    except Exception:
+        return ip_address(start).version == 6 and ip_address(end).version == 6
+    except ValueError:
         return False
 
 
-def get_interfaces_by_ip(ip_address: str) -> list:
+def is_ipv4_address_or_range(addr: str) -> bool:
+    """
+    Validates if the provided address is a valid IPv4, CIDR or IPv4 range
+    :param addr: address to test
+    :return: bool: True if provided address is valid
+    """
+    from vyos.template import is_ipv4
+
+    return is_ipv4(addr) or is_ipv4_range(addr)
+
+
+def is_ipv6_address_or_range(addr: str) -> bool:
+    """
+    Validates if the provided address is a valid IPv6, CIDR or IPv6 range
+    :param addr: address to test
+    :return: bool: True if provided address is valid
+    """
+    from vyos.template import is_ipv6
+
+    return is_ipv6(addr) or is_ipv6_range(addr)
+
+
+def get_interfaces_by_ip(ip_address: str, vrf=None, include_vrf: bool = False) -> list:
     """
     Return a list of all interface names assigned the given IP address.
     Args:
         ip_address (str): The IP address to search for.
+        vrf (str): If specified, only interfaces in this VRF are considered.
+        include_vrf (bool): If True, interfaces from all VRFs are considered.
     Returns:
         list: List of interface names (str) that have the given IP address assigned.
               Returns an empty list if no interface has the IP assigned.
     """
     import netifaces
-    from vyos.template import is_ipv6
-
-    addr_type = AF_INET
-    if is_ipv6(ip_address):
-        addr_type = AF_INET6
+    from vyos.utils.dict import dict_search
 
     ifaces = []
     for interface in netifaces.interfaces():
-        addresses = netifaces.ifaddresses(interface)
-        for addr_info in addresses.get(addr_type, []):
-            if addr_info.get('addr') == ip_address:
-                ifaces.append(interface)
+        # Check if interface belongs to the requested VRF, if this is not the
+        # case there is no need to proceed with this data set - continue loop
+        # with next element
+        if vrf is not None and not include_vrf and interface != vrf:
+            tmp = get_interface_config(interface)
+            if dict_search('master', tmp) != vrf:
+                continue
+
+        if is_intf_addr_assigned(interface, ip_address):
+            ifaces.append(interface)
+
     return ifaces