From 733c07f3f9da8bca45f97abbd4521fc61b4a3812 Mon Sep 17 00:00:00 2001 From: Michael Pardatscher Date: Fri, 25 Sep 2026 17:45:49 +0000 Subject: 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 --- python/vyos/template.py | 5 +- python/vyos/utils/network.py | 132 ++++++++++++++++++++++++++++++------------- 2 files changed, 95 insertions(+), 42 deletions(-) (limited to 'python') 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 -- cgit v1.2.3