diff options
Diffstat (limited to 'python')
| -rwxr-xr-x | python/vyos/template.py | 5 | ||||
| -rw-r--r-- | python/vyos/utils/network.py | 132 |
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 |