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 From 744cb312dc3b0ed7326b1faee5c4db9e9a74a0df Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Fri, 25 Sep 2026 16:51:37 +0000 Subject: vyos.utils: T8977: do not fail when a listening process vanishes mid-check The list of listening sockets is a snapshot, and the owning process was looked up afterwards. A process exiting in between, or one belonging to another user when we do not run privileged, raised an exception that escaped and aborted the conf-mode script asking whether a port is taken. Neither case can be the service being looked for, so such a socket is skipped now. --- python/vyos/utils/network.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) (limited to 'python') diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py index e4d98777b..e5fa398cf 100644 --- a/python/vyos/utils/network.py +++ b/python/vyos/utils/network.py @@ -450,6 +450,8 @@ def is_listen_port_bind_service(port: int, service: str, address: str = None) -> """ from psutil import net_connections as connections from psutil import Process as process + from psutil import AccessDenied + from psutil import NoSuchProcess from ipaddress import ip_address has_address = bool(address) @@ -470,7 +472,13 @@ def is_listen_port_bind_service(port: int, service: str, address: str = None) -> if pid is None: continue - pid_name = process(pid).name() + # The socket table is a snapshot - the process behind it may be gone + # by now, or belong to another user. Neither can be our service. + try: + pid_name = process(pid).name() + except (NoSuchProcess, AccessDenied): + continue + pid_port = addr.port # Ignore address check when service listening on "any address", -- cgit v1.2.3 From 3a04d88a8c0ebdc356829a512642525885a72582 Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Fri, 25 Sep 2026 17:15:55 +0000 Subject: vyos.utils: T8977: fix regressions in address and port validation Whether an interface belongs to a VRF was only established when a VRF was named explicitly. A lookup for the default VRF - what most callers ask for - thus reported an address living in an arbitrary VRF, or on a VRF device itself, as locally assigned, and accepted a configuration the daemon cannot bind later on. Naming the VRF device remains the way to ask for the addresses it carries. Probing the next address family after a failed bind was unreachable, as the argument is validated as a literal beforehand and then resolves to exactly one family. The early return is restored. An interface address carries host bits and must stay valid, but the range variants classify untrusted input handed over to nftables, which only accepts a prefix in canonical form. Those reject host bits again. All of it is covered by tests. --- python/vyos/utils/network.py | 56 ++++++++++++++++++++++++++++++-------------- src/tests/test_utils.py | 44 ++++++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 18 deletions(-) (limited to 'python') diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py index e5fa398cf..cb5863e51 100644 --- a/python/vyos/utils/network.py +++ b/python/vyos/utils/network.py @@ -423,13 +423,10 @@ def check_port_availability(address: str = None, port: int = 0, s.bind(sockaddr) return True # port is free to use except OSError: - # This address family is busy, but another might be free. - # Continue to the next one. - pass + return False # port is already in use - # 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 + # if we reach this point, no socket was tested and we assume the port is + # already in use - better safe then sorry return False @@ -523,13 +520,16 @@ def is_addr_assigned( 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 - if vrf is not None and not include_vrf and interface != vrf: + # Only an interface of the requested VRF may satisfy the lookup - an + # enslaved interface and a VRF device alike belong to their own L3 + # domain, not to the default one. Naming the VRF device is the way to + # ask for the addresses it carries itself. + if not include_vrf and interface != vrf: tmp = get_interface_config(interface) if dict_search('master', tmp) != vrf: continue + if dict_search('linkinfo.info_kind', tmp) == 'vrf': + continue if is_intf_addr_assigned(interface, ip_address): return True @@ -833,23 +833,40 @@ def is_ipv6_range(addr: str) -> bool: def is_ipv4_address_or_range(addr: str) -> bool: """ Validates if the provided address is a valid IPv4, CIDR or IPv4 range + + A prefix with host bits set (192.0.2.1/24) is rejected - this classifies + untrusted input which is handed over to nftables, where only a prefix in + its canonical form is accepted. :param addr: address to test :return: bool: True if provided address is valid """ - from vyos.template import is_ipv4 + from ipaddress import ip_network - return is_ipv4(addr) or is_ipv4_range(addr) + if '-' in addr: + return is_ipv4_range(addr) + try: + return ip_network(addr).version == 4 + except Exception: + return False def is_ipv6_address_or_range(addr: str) -> bool: """ Validates if the provided address is a valid IPv6, CIDR or IPv6 range + + A prefix with host bits set (2001:db8::1/64) is rejected, see the IPv4 + counterpart above. :param addr: address to test :return: bool: True if provided address is valid """ - from vyos.template import is_ipv6 + from ipaddress import ip_network - return is_ipv6(addr) or is_ipv6_range(addr) + if '-' in addr: + return is_ipv6_range(addr) + try: + return ip_network(addr).version == 6 + except Exception: + return False def get_interfaces_by_ip(ip_address: str, vrf=None, include_vrf: bool = False) -> list: @@ -868,13 +885,16 @@ def get_interfaces_by_ip(ip_address: str, vrf=None, include_vrf: bool = False) - ifaces = [] for interface in netifaces.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 - if vrf is not None and not include_vrf and interface != vrf: + # Only an interface of the requested VRF may satisfy the lookup - an + # enslaved interface and a VRF device alike belong to their own L3 + # domain, not to the default one. Naming the VRF device is the way to + # ask for the addresses it carries itself. + if not include_vrf and interface != vrf: tmp = get_interface_config(interface) if dict_search('master', tmp) != vrf: continue + if dict_search('linkinfo.info_kind', tmp) == 'vrf': + continue if is_intf_addr_assigned(interface, ip_address): ifaces.append(interface) diff --git a/src/tests/test_utils.py b/src/tests/test_utils.py index 8e6f320ee..f641e3e4d 100644 --- a/src/tests/test_utils.py +++ b/src/tests/test_utils.py @@ -192,6 +192,50 @@ class TestVyOSUtils(TestCase): ) mock_is_assigned.assert_called_once_with('eth0', '10.0.0.1') + def test_is_addr_assigned_default_vrf_ignores_vrf_slave(self): + """An address inside a VRF must not satisfy a default VRF lookup.""" + from vyos.utils.network import is_addr_assigned + + with patch('vyos.utils.system.sysctl_read', return_value='0'), patch( + 'netifaces.interfaces', return_value=['eth0'] + ), patch( + 'vyos.utils.network.get_interface_config', return_value={'master': 'red'} + ), patch( + 'vyos.utils.network.is_intf_addr_assigned', return_value=True + ): + self.assertFalse(is_addr_assigned('10.0.0.1')) + + def test_is_addr_assigned_default_vrf_ignores_vrf_device(self): + """An address on a VRF device belongs to that VRF, not to the default one.""" + from vyos.utils.network import is_addr_assigned + + vrf_device = {'linkinfo': {'info_kind': 'vrf'}} + with patch('vyos.utils.system.sysctl_read', return_value='0'), patch( + 'netifaces.interfaces', return_value=['red'] + ), patch( + 'vyos.utils.network.get_interface_config', return_value=vrf_device + ), patch( + 'vyos.utils.network.is_intf_addr_assigned', return_value=True + ): + self.assertFalse(is_addr_assigned('10.0.0.1')) + # naming the VRF device asks for the addresses it carries itself + self.assertTrue(is_addr_assigned('10.0.0.1', vrf='red')) + self.assertTrue(is_addr_assigned('10.0.0.1', include_vrf=True)) + + def test_is_ip_address_or_range_rejects_host_bits(self): + """A prefix with host bits set is not a valid network for nftables.""" + from vyos.utils.network import is_ipv4_address_or_range + from vyos.utils.network import is_ipv6_address_or_range + + self.assertTrue(is_ipv4_address_or_range('192.0.2.1')) + self.assertTrue(is_ipv4_address_or_range('192.0.2.0/24')) + self.assertTrue(is_ipv4_address_or_range('192.0.2.1-192.0.2.9')) + self.assertFalse(is_ipv4_address_or_range('192.0.2.1/24')) + + self.assertTrue(is_ipv6_address_or_range('2001:db8::1')) + self.assertTrue(is_ipv6_address_or_range('2001:db8::/64')) + self.assertFalse(is_ipv6_address_or_range('2001:db8::1/64')) + def test_is_addr_assigned_wrong_vrf(self): """Address on VRF 'red' must not match when querying vrf='blue'.""" from vyos.utils.network import is_addr_assigned -- cgit v1.2.3