From 725b6e0454427b8c2b3a507c61d6192ca7522a7e Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Sun, 3 Sep 2023 14:33:22 +0200 Subject: netns: T5241: provide is_netns_interface utility helper --- python/vyos/utils/network.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'python/vyos/utils/network.py') diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py index 2f181d8d9..fa86795eb 100644 --- a/python/vyos/utils/network.py +++ b/python/vyos/utils/network.py @@ -40,9 +40,9 @@ def interface_exists(interface) -> bool: import os return os.path.exists(f'/sys/class/net/{interface}') -def interface_exists_in_netns(interface_name, netns): +def is_netns_interface(interface, netns): from vyos.utils.process import rc_cmd - rc, out = rc_cmd(f'ip netns exec {netns} ip link show dev {interface_name}') + rc, out = rc_cmd(f'ip netns exec {netns} ip link show dev {interface}') if rc == 0: return True return False -- cgit v1.2.3 From 114f8a9a66e49449e09ac3a1721db42626e54212 Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Sun, 3 Sep 2023 14:34:08 +0200 Subject: netns: T5241: use common interface_exists() helper --- python/vyos/utils/network.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) (limited to 'python/vyos/utils/network.py') diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py index fa86795eb..8b4385cfb 100644 --- a/python/vyos/utils/network.py +++ b/python/vyos/utils/network.py @@ -78,8 +78,7 @@ def get_interface_config(interface): """ Returns the used encapsulation protocol for given interface. If interface does not exist, None is returned. """ - import os - if not os.path.exists(f'/sys/class/net/{interface}'): + if not interface_exists(interface): return None from json import loads from vyos.utils.process import cmd @@ -90,8 +89,7 @@ def get_interface_address(interface): """ Returns the used encapsulation protocol for given interface. If interface does not exist, None is returned. """ - import os - if not os.path.exists(f'/sys/class/net/{interface}'): + if not interface_exists(interface): return None from json import loads from vyos.utils.process import cmd @@ -141,8 +139,7 @@ def is_wwan_connected(interface): def get_bridge_fdb(interface): """ Returns the forwarding database entries for a given interface """ - import os - if not os.path.exists(f'/sys/class/net/{interface}'): + if not interface_exists(interface): return None from json import loads from vyos.utils.process import cmd -- cgit v1.2.3 From 00a9f6a39f556ee6eb8ac669dd86f5095c21b22f Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Sun, 3 Sep 2023 14:34:39 +0200 Subject: netns: T5241: improve get_interface_namespace() robustness --- python/vyos/utils/network.py | 78 +++++++++++++++----------------------------- python/vyos/utils/process.py | 2 +- 2 files changed, 28 insertions(+), 52 deletions(-) (limited to 'python/vyos/utils/network.py') diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py index 8b4385cfb..4706dbfb7 100644 --- a/python/vyos/utils/network.py +++ b/python/vyos/utils/network.py @@ -96,23 +96,23 @@ def get_interface_address(interface): tmp = loads(cmd(f'ip --detail --json addr show dev {interface}'))[0] return tmp -def get_interface_namespace(iface): +def get_interface_namespace(interface: str): """ Returns wich netns the interface belongs to """ from json import loads from vyos.utils.process import cmd - # Check if netns exist - tmp = loads(cmd(f'ip --json netns ls')) - if len(tmp) == 0: - return None - for ns in tmp: + # Bail out early if netns does not exist + tmp = cmd(f'ip --json netns ls') + if not tmp: return None + + for ns in loads(tmp): netns = f'{ns["name"]}' # Search interface in each netns data = loads(cmd(f'ip netns exec {netns} ip --json link show')) for tmp in data: - if iface == tmp["ifname"]: + if interface == tmp["ifname"]: return netns def is_wwan_connected(interface): @@ -271,57 +271,33 @@ def is_addr_assigned(ip_address, vrf=None) -> bool: return False -def is_intf_addr_assigned(intf, address) -> bool: +def is_intf_addr_assigned(ifname: str, addr: str, netns: str=None) -> bool: """ Verify if the given IPv4/IPv6 address is assigned to specific interface. It can check both a single IP address (e.g. 192.0.2.1 or a assigned CIDR address 192.0.2.1/24. """ - from vyos.template import is_ipv4 - - from netifaces import ifaddresses - from netifaces import AF_INET - from netifaces import AF_INET6 - - # check if the requested address type is configured at all - # { - # 17: [{'addr': '08:00:27:d9:5b:04', 'broadcast': 'ff:ff:ff:ff:ff:ff'}], - # 2: [{'addr': '10.0.2.15', 'netmask': '255.255.255.0', 'broadcast': '10.0.2.255'}], - # 10: [{'addr': 'fe80::a00:27ff:fed9:5b04%eth0', 'netmask': 'ffff:ffff:ffff:ffff::'}] - # } - try: - addresses = ifaddresses(intf) - except ValueError as e: - print(e) - return False - - # determine IP version (AF_INET or AF_INET6) depending on passed address - addr_type = AF_INET if is_ipv4(address) else AF_INET6 - - # Check every IP address on this interface for a match - netmask = None - if '/' in address: - address, netmask = address.split('/') - for ip in addresses.get(addr_type, []): - # ip can have the interface name in the 'addr' field, we need to remove it - # {'addr': 'fe80::a00:27ff:fec5:f821%eth2', 'netmask': 'ffff:ffff:ffff:ffff::'} - ip_addr = ip['addr'].split('%')[0] - - if not _are_same_ip(address, ip_addr): - continue - - # we do not have a netmask to compare against, they are the same - if not netmask: - return True + import json + import jmespath - prefixlen = '' - if is_ipv4(ip_addr): - prefixlen = sum([bin(int(_)).count('1') for _ in ip['netmask'].split('.')]) - else: - prefixlen = sum([bin(int(_,16)).count('1') for _ in ip['netmask'].split('/')[0].split(':') if _]) + from vyos.utils.process import rc_cmd + from ipaddress import ip_interface - if str(prefixlen) == netmask: - return True + 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: + json_out = json.loads(out) + addresses = jmespath.search("[].addr_info[].{family: family, address: local, prefixlen: prefixlen}", json_out) + for address_info in addresses: + family = address_info['family'] + address = address_info['address'] + prefixlen = address_info['prefixlen'] + # Remove the interface name if present in the given address + if '%' in addr: + addr = addr.split('%')[0] + interface = ip_interface(f"{address}/{prefixlen}") + if ip_interface(addr) == interface or address == addr: + return True return False diff --git a/python/vyos/utils/process.py b/python/vyos/utils/process.py index e09c7d86d..c2ef98140 100644 --- a/python/vyos/utils/process.py +++ b/python/vyos/utils/process.py @@ -170,7 +170,7 @@ def rc_cmd(command, flag='', shell=None, input=None, timeout=None, env=None, (1, 'Device "eth99" does not exist.') """ out, code = popen( - command, flag, + command.lstrip(), flag, stdout=stdout, stderr=stderr, input=input, timeout=timeout, env=env, shell=shell, -- cgit v1.2.3 From 5c2b49092ca1fc56cf198ca89630ec97fddd64b3 Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Tue, 5 Sep 2023 19:37:04 +0200 Subject: smoketest: T5241: re-work netns assertions and provide common utility helper --- python/vyos/utils/network.py | 8 ++- smoketest/scripts/cli/test_interfaces_netns.py | 79 ------------------------ smoketest/scripts/cli/test_netns.py | 83 ++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 80 deletions(-) delete mode 100755 smoketest/scripts/cli/test_interfaces_netns.py create mode 100755 smoketest/scripts/cli/test_netns.py (limited to 'python/vyos/utils/network.py') diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py index 4706dbfb7..abc382766 100644 --- a/python/vyos/utils/network.py +++ b/python/vyos/utils/network.py @@ -42,11 +42,17 @@ def interface_exists(interface) -> bool: def is_netns_interface(interface, netns): from vyos.utils.process import rc_cmd - rc, out = rc_cmd(f'ip netns exec {netns} ip link show dev {interface}') + rc, out = rc_cmd(f'sudo ip netns exec {netns} ip link show dev {interface}') if rc == 0: return True return False +def get_netns_all() -> list: + from json import loads + from vyos.utils.process import cmd + tmp = loads(cmd('ip --json netns ls')) + return [ netns['name'] for netns in tmp ] + def get_vrf_members(vrf: str) -> list: """ Get list of interface VRF members diff --git a/smoketest/scripts/cli/test_interfaces_netns.py b/smoketest/scripts/cli/test_interfaces_netns.py deleted file mode 100755 index b8bebb221..000000000 --- a/smoketest/scripts/cli/test_interfaces_netns.py +++ /dev/null @@ -1,79 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2021 VyOS maintainers and contributors -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 or later as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -import unittest - -from netifaces import interfaces -from base_vyostest_shim import VyOSUnitTestSHIM - -from vyos.configsession import ConfigSession -from vyos.configsession import ConfigSessionError -from vyos.ifconfig import Interface -from vyos.ifconfig import Section -from vyos.utils.process import cmd - -base_path = ['netns'] -namespaces = ['mgmt', 'front', 'back', 'ams-ix'] - -class NETNSTest(VyOSUnitTestSHIM.TestCase): - def setUp(self): - self._interfaces = ['dum10', 'dum12', 'dum50'] - - def test_create_netns(self): - for netns in namespaces: - base = base_path + ['name', netns] - self.cli_set(base) - - # commit changes - self.cli_commit() - - netns_list = cmd('ip netns ls') - - # Verify NETNS configuration - for netns in namespaces: - self.assertTrue(netns in netns_list) - - - def test_netns_assign_interface(self): - netns = 'foo' - self.cli_set(['netns', 'name', netns]) - - # Set - for iface in self._interfaces: - self.cli_set(['interfaces', 'dummy', iface, 'netns', netns]) - - # commit changes - self.cli_commit() - - netns_iface_list = cmd(f'sudo ip netns exec {netns} ip link show') - - for iface in self._interfaces: - self.assertTrue(iface in netns_iface_list) - - # Delete - for iface in self._interfaces: - self.cli_delete(['interfaces', 'dummy', iface, 'netns', netns]) - - # commit changes - self.cli_commit() - - netns_iface_list = cmd(f'sudo ip netns exec {netns} ip link show') - - for iface in self._interfaces: - self.assertNotIn(iface, netns_iface_list) - -if __name__ == '__main__': - unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_netns.py b/smoketest/scripts/cli/test_netns.py new file mode 100755 index 000000000..fd04dd520 --- /dev/null +++ b/smoketest/scripts/cli/test_netns.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2021-2023 VyOS maintainers and contributors +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 or later as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import unittest + +from base_vyostest_shim import VyOSUnitTestSHIM + +from vyos.configsession import ConfigSession +from vyos.configsession import ConfigSessionError +from vyos.ifconfig import Interface +from vyos.ifconfig import Section +from vyos.utils.process import cmd +from vyos.utils.network import is_netns_interface +from vyos.utils.network import get_netns_all + +base_path = ['netns'] +interfaces = ['dum10', 'dum12', 'dum50'] + +class NetNSTest(VyOSUnitTestSHIM.TestCase): + def tearDown(self): + self.cli_delete(base_path) + # commit changes + self.cli_commit() + + # There should be no network namespace remaining + tmp = cmd('ip netns ls') + self.assertFalse(tmp) + + super(NetNSTest, self).tearDown() + + def test_netns_create(self): + namespaces = ['mgmt', 'front', 'back'] + for netns in namespaces: + self.cli_set(base_path + ['name', netns]) + + # commit changes + self.cli_commit() + + # Verify NETNS configuration + for netns in namespaces: + self.assertIn(netns, get_netns_all()) + + def test_netns_interface(self): + netns = 'foo' + self.cli_set(base_path + ['name', netns]) + + # Set + for iface in interfaces: + self.cli_set(['interfaces', 'dummy', iface, 'netns', netns]) + + # commit changes + self.cli_commit() + + for interface in interfaces: + self.assertTrue(is_netns_interface(interface, netns)) + + # Delete + for interface in interfaces: + self.cli_delete(['interfaces', 'dummy', interface]) + + # commit changes + self.cli_commit() + + netns_iface_list = cmd(f'sudo ip netns exec {netns} ip link show') + + for interface in interfaces: + self.assertFalse(is_netns_interface(interface, netns)) + +if __name__ == '__main__': + unittest.main(verbosity=2) -- cgit v1.2.3