diff options
Diffstat (limited to 'smoketest/scripts')
| -rwxr-xr-x | smoketest/scripts/cli/test_interfaces_bonding.py | 22 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_interfaces_openvpn.py | 38 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_load_balancning_wan.py | 189 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_isis.py | 69 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_service_salt.py | 94 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_service_ssh.py | 73 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_system_ipv6.py | 31 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_system_login.py | 8 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_vrf.py | 141 |
9 files changed, 616 insertions, 49 deletions
diff --git a/smoketest/scripts/cli/test_interfaces_bonding.py b/smoketest/scripts/cli/test_interfaces_bonding.py index 4f2fe979a..9bb561275 100755 --- a/smoketest/scripts/cli/test_interfaces_bonding.py +++ b/smoketest/scripts/cli/test_interfaces_bonding.py @@ -165,5 +165,27 @@ class BondingInterfaceTest(BasicInterfaceTest.TestCase): self.cli_commit() + def test_bonding_uniq_member_description(self): + ethernet_path = ['interfaces', 'ethernet'] + for interface in self._interfaces: + for option in self._options.get(interface, []): + self.cli_set(self._base_path + [interface] + option.split()) + + self.cli_commit() + + # Add any changes on bonding members + # For example add description on separate ethX interfaces + for interface in self._interfaces: + for member in self._members: + self.cli_set(ethernet_path + [member, 'description', member + '_interface']) + + self.cli_commit() + + # verify config + for interface in self._interfaces: + slaves = read_file(f'/sys/class/net/{interface}/bonding/slaves').split() + for member in self._members: + self.assertIn(member, slaves) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_interfaces_openvpn.py b/smoketest/scripts/cli/test_interfaces_openvpn.py index 24df0af4d..55218ecac 100755 --- a/smoketest/scripts/cli/test_interfaces_openvpn.py +++ b/smoketest/scripts/cli/test_interfaces_openvpn.py @@ -607,6 +607,44 @@ class TestInterfacesOpenVPN(VyOSUnitTestSHIM.TestCase): interface = f'vtun{ii}' self.assertNotIn(interface, interfaces()) + def test_openvpn_options(self): + # Ensure OpenVPN process restart on openvpn-option CLI node change + + interface = 'vtun5001' + path = base_path + [interface] + + self.cli_set(path + ['mode', 'site-to-site']) + self.cli_set(path + ['local-address', '10.0.0.2']) + self.cli_set(path + ['remote-address', '192.168.0.3']) + self.cli_set(path + ['shared-secret-key-file', s2s_key]) + + self.cli_commit() + + # Now verify the OpenVPN "raw" option passing. Once an openvpn-option is + # added, modified or deleted from the CLI, OpenVPN daemon must be restarted + cur_pid = process_named_running('openvpn') + self.cli_set(path + ['openvpn-option', '--persist-tun']) + self.cli_commit() + + # PID must be different as OpenVPN Must be restarted + new_pid = process_named_running('openvpn') + self.assertNotEqual(cur_pid, new_pid) + cur_pid = new_pid + + self.cli_set(path + ['openvpn-option', '--persist-key']) + self.cli_commit() + + # PID must be different as OpenVPN Must be restarted + new_pid = process_named_running('openvpn') + self.assertNotEqual(cur_pid, new_pid) + cur_pid = new_pid + + self.cli_delete(path + ['openvpn-option']) + self.cli_commit() + + # PID must be different as OpenVPN Must be restarted + new_pid = process_named_running('openvpn') + self.assertNotEqual(cur_pid, new_pid) if __name__ == '__main__': # Our SSL certificates need a subject ... diff --git a/smoketest/scripts/cli/test_load_balancning_wan.py b/smoketest/scripts/cli/test_load_balancning_wan.py new file mode 100755 index 000000000..edc6deb04 --- /dev/null +++ b/smoketest/scripts/cli/test_load_balancning_wan.py @@ -0,0 +1,189 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2022 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 <http://www.gnu.org/licenses/>. + +import os +import unittest +import time + +from base_vyostest_shim import VyOSUnitTestSHIM +from vyos.configsession import ConfigSessionError +from vyos.ifconfig import Section +from vyos.util import call +from vyos.util import cmd + + +base_path = ['load-balancing'] + + +def create_netns(name): + return call(f'sudo ip netns add {name}') + +def create_veth_pair(local='veth0', peer='ceth0'): + return call(f'sudo ip link add {local} type veth peer name {peer}') + +def move_interface_to_netns(iface, netns_name): + return call(f'sudo ip link set {iface} netns {netns_name}') + +def rename_interface(iface, new_name): + return call(f'sudo ip link set {iface} name {new_name}') + +def cmd_in_netns(netns, cmd): + return call(f'sudo ip netns exec {netns} {cmd}') + +def delete_netns(name): + return call(f'sudo ip netns del {name}') + + +class TestLoadBalancingWan(VyOSUnitTestSHIM.TestCase): + @classmethod + def setUpClass(cls): + super(TestLoadBalancingWan, cls).setUpClass() + + # ensure we can also run this test on a live system - so lets clean + # out the current configuration :) + cls.cli_delete(cls, base_path) + + def tearDown(self): + self.cli_delete(base_path) + self.cli_commit() + + def test_table_routes(self): + + ns1 = 'ns201' + ns2 = 'ns202' + iface1 = 'eth201' + iface2 = 'eth202' + container_iface1 = 'ceth0' + container_iface2 = 'ceth1' + + # Create network namespeces + create_netns(ns1) + create_netns(ns2) + create_veth_pair(iface1, container_iface1) + create_veth_pair(iface2, container_iface2) + move_interface_to_netns(container_iface1, ns1) + move_interface_to_netns(container_iface2, ns2) + call(f'sudo ip a add 203.0.113.10/24 dev {iface1}') + call(f'sudo ip a add 192.0.2.10/24 dev {iface2}') + call(f'sudo ip link set dev {iface1} up') + call(f'sudo ip link set dev {iface2} up') + cmd_in_netns(ns1, f'ip link set {container_iface1} name eth0') + cmd_in_netns(ns2, f'ip link set {container_iface2} name eth0') + cmd_in_netns(ns1, 'ip a add 203.0.113.1/24 dev eth0') + cmd_in_netns(ns2, 'ip a add 192.0.2.1/24 dev eth0') + cmd_in_netns(ns1, 'ip link set dev eth0 up') + cmd_in_netns(ns2, 'ip link set dev eth0 up') + + # Set load-balancing configuration + self.cli_set(base_path + ['wan', 'interface-health', iface1, 'failure-count', '2']) + self.cli_set(base_path + ['wan', 'interface-health', iface1, 'nexthop', '203.0.113.1']) + self.cli_set(base_path + ['wan', 'interface-health', iface1, 'success-count', '1']) + self.cli_set(base_path + ['wan', 'interface-health', iface2, 'failure-count', '2']) + self.cli_set(base_path + ['wan', 'interface-health', iface2, 'nexthop', '192.0.2.1']) + self.cli_set(base_path + ['wan', 'interface-health', iface2, 'success-count', '1']) + + # commit changes + self.cli_commit() + + time.sleep(5) + # Check default routes in tables 201, 202 + # Expected values + original = 'default via 203.0.113.1 dev eth201' + tmp = cmd('sudo ip route show table 201') + self.assertEqual(tmp, original) + + original = 'default via 192.0.2.1 dev eth202' + tmp = cmd('sudo ip route show table 202') + self.assertEqual(tmp, original) + + # Delete veth interfaces and netns + for iface in [iface1, iface2]: + call(f'sudo ip link del dev {iface}') + + delete_netns(ns1) + delete_netns(ns2) + + def test_check_chains(self): + + ns1 = 'nsA' + ns2 = 'nsB' + iface1 = 'veth1' + iface2 = 'veth2' + container_iface1 = 'ceth0' + container_iface2 = 'ceth1' + mangle_isp1 = """table ip mangle { + chain ISP_veth1 { + counter ct mark set 0xc9 + counter meta mark set 0xc9 + counter accept + } +}""" + mangle_isp2 = """table ip mangle { + chain ISP_veth2 { + counter ct mark set 0xca + counter meta mark set 0xca + counter accept + } +}""" + + # Create network namespeces + create_netns(ns1) + create_netns(ns2) + create_veth_pair(iface1, container_iface1) + create_veth_pair(iface2, container_iface2) + move_interface_to_netns(container_iface1, ns1) + move_interface_to_netns(container_iface2, ns2) + call(f'sudo ip a add 203.0.113.10/24 dev {iface1}') + call(f'sudo ip a add 192.0.2.10/24 dev {iface2}') + call(f'sudo ip link set dev {iface1} up') + call(f'sudo ip link set dev {iface2} up') + cmd_in_netns(ns1, f'ip link set {container_iface1} name eth0') + cmd_in_netns(ns2, f'ip link set {container_iface2} name eth0') + cmd_in_netns(ns1, 'ip a add 203.0.113.1/24 dev eth0') + cmd_in_netns(ns2, 'ip a add 192.0.2.1/24 dev eth0') + cmd_in_netns(ns1, 'ip link set dev eth0 up') + cmd_in_netns(ns2, 'ip link set dev eth0 up') + + # Set load-balancing configuration + self.cli_set(base_path + ['wan', 'interface-health', iface1, 'failure-count', '2']) + self.cli_set(base_path + ['wan', 'interface-health', iface1, 'nexthop', '203.0.113.1']) + self.cli_set(base_path + ['wan', 'interface-health', iface1, 'success-count', '1']) + self.cli_set(base_path + ['wan', 'interface-health', iface2, 'failure-count', '2']) + self.cli_set(base_path + ['wan', 'interface-health', iface2, 'nexthop', '192.0.2.1']) + self.cli_set(base_path + ['wan', 'interface-health', iface2, 'success-count', '1']) + + # commit changes + self.cli_commit() + + time.sleep(5) + # Check chains + #call('sudo nft list ruleset') + tmp = cmd(f'sudo nft -s list chain mangle ISP_{iface1}') + self.assertEqual(tmp, mangle_isp1) + + tmp = cmd(f'sudo nft -s list chain mangle ISP_{iface2}') + self.assertEqual(tmp, mangle_isp2) + + # Delete veth interfaces and netns + for iface in [iface1, iface2]: + call(f'sudo ip link del dev {iface}') + + delete_netns(ns1) + delete_netns(ns2) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_protocols_isis.py b/smoketest/scripts/cli/test_protocols_isis.py index 8abdd6d37..167cd05f8 100755 --- a/smoketest/scripts/cli/test_protocols_isis.py +++ b/smoketest/scripts/cli/test_protocols_isis.py @@ -71,13 +71,13 @@ class TestProtocolsISIS(VyOSUnitTestSHIM.TestCase): self.cli_commit() # Verify all changes - tmp = self.getFRRconfig(f'router isis {domain}') + tmp = self.getFRRconfig(f'router isis {domain}', daemon='isisd') self.assertIn(f' net {net}', tmp) self.assertIn(f' log-adjacency-changes', tmp) self.assertIn(f' redistribute ipv4 connected level-2 route-map {route_map}', tmp) for interface in self._interfaces: - tmp = self.getFRRconfig(f'interface {interface}') + tmp = self.getFRRconfig(f'interface {interface}', daemon='isisd') self.assertIn(f' ip router isis {domain}', tmp) self.assertIn(f' ipv6 router isis {domain}', tmp) @@ -93,22 +93,26 @@ class TestProtocolsISIS(VyOSUnitTestSHIM.TestCase): self.isis_base_config() self.cli_set(base_path + ['redistribute', 'ipv4', 'connected', 'level-2', 'route-map', route_map]) self.cli_set(base_path + ['route-map', route_map]) + self.cli_set(base_path + ['level', 'level-2']) # commit changes self.cli_commit() # Verify FRR configuration zebra_route_map = f'ip protocol isis route-map {route_map}' - frrconfig = self.getFRRconfig(zebra_route_map) + frrconfig = self.getFRRconfig(zebra_route_map, daemon='zebra') self.assertIn(zebra_route_map, frrconfig) + tmp = self.getFRRconfig(f'router isis {domain}', daemon='isisd') + self.assertIn(' is-type level-2-only', tmp) + # Remove the route-map again self.cli_delete(base_path + ['route-map']) # commit changes self.cli_commit() # Verify FRR configuration - frrconfig = self.getFRRconfig(zebra_route_map) + frrconfig = self.getFRRconfig(zebra_route_map, daemon='zebra') self.assertNotIn(zebra_route_map, frrconfig) self.cli_delete(['policy', 'route-map', route_map]) @@ -128,7 +132,7 @@ class TestProtocolsISIS(VyOSUnitTestSHIM.TestCase): self.cli_commit() # Verify all changes - tmp = self.getFRRconfig(f'router isis {domain}') + tmp = self.getFRRconfig(f'router isis {domain}', daemon='isisd') self.assertIn(f' net {net}', tmp) for afi in ['ipv4', 'ipv6']: @@ -140,6 +144,8 @@ class TestProtocolsISIS(VyOSUnitTestSHIM.TestCase): password = 'foo' self.isis_base_config() + for interface in self._interfaces: + self.cli_set(base_path + ['interface', interface, 'password', 'plaintext-password', f'{password}-{interface}']) self.cli_set(base_path + ['area-password', 'plaintext-password', password]) self.cli_set(base_path + ['area-password', 'md5', password]) @@ -160,10 +166,61 @@ class TestProtocolsISIS(VyOSUnitTestSHIM.TestCase): self.cli_commit() # Verify all changes - tmp = self.getFRRconfig(f'router isis {domain}') + tmp = self.getFRRconfig(f'router isis {domain}', daemon='isisd') self.assertIn(f' net {net}', tmp) self.assertIn(f' domain-password clear {password}', tmp) self.assertIn(f' area-password clear {password}', tmp) + for interface in self._interfaces: + tmp = self.getFRRconfig(f'interface {interface}', daemon='isisd') + self.assertIn(f' isis password clear {password}-{interface}', tmp) + + def test_isis_06_spf_delay(self): + network = 'point-to-point' + holddown = '10' + init_delay = '50' + long_delay = '200' + short_delay = '100' + time_to_learn = '75' + + self.cli_set(base_path + ['net', net]) + for interface in self._interfaces: + self.cli_set(base_path + ['interface', interface, 'network', network]) + + self.cli_set(base_path + ['spf-delay-ietf', 'holddown', holddown]) + # verify() - All types of spf-delay must be configured + with self.assertRaises(ConfigSessionError): + self.cli_commit() + + self.cli_set(base_path + ['spf-delay-ietf', 'init-delay', init_delay]) + # verify() - All types of spf-delay must be configured + with self.assertRaises(ConfigSessionError): + self.cli_commit() + + self.cli_set(base_path + ['spf-delay-ietf', 'long-delay', long_delay]) + # verify() - All types of spf-delay must be configured + with self.assertRaises(ConfigSessionError): + self.cli_commit() + + self.cli_set(base_path + ['spf-delay-ietf', 'short-delay', short_delay]) + # verify() - All types of spf-delay must be configured + with self.assertRaises(ConfigSessionError): + self.cli_commit() + self.cli_set(base_path + ['spf-delay-ietf', 'time-to-learn', time_to_learn]) + + # Commit all changes + self.cli_commit() + + # Verify all changes + tmp = self.getFRRconfig(f'router isis {domain}', daemon='isisd') + self.assertIn(f' net {net}', tmp) + self.assertIn(f' spf-delay-ietf init-delay {init_delay} short-delay {short_delay} long-delay {long_delay} holddown {holddown} time-to-learn {time_to_learn}', tmp) + + for interface in self._interfaces: + tmp = self.getFRRconfig(f'interface {interface}', daemon='isisd') + self.assertIn(f' ip router isis {domain}', tmp) + self.assertIn(f' ipv6 router isis {domain}', tmp) + self.assertIn(f' isis network {network}', tmp) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_service_salt.py b/smoketest/scripts/cli/test_service_salt.py new file mode 100755 index 000000000..5b328677e --- /dev/null +++ b/smoketest/scripts/cli/test_service_salt.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2022 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 <http://www.gnu.org/licenses/>. + +import unittest + +from socket import gethostname +from base_vyostest_shim import VyOSUnitTestSHIM + +from vyos.util import process_named_running +from vyos.util import read_file +from vyos.util import cmd + +PROCESS_NAME = 'salt-minion' +SALT_CONF = '/etc/salt/minion' +base_path = ['service', 'salt-minion'] + +class TestServiceSALT(VyOSUnitTestSHIM.TestCase): + @classmethod + def setUpClass(cls): + super(TestServiceSALT, cls).setUpClass() + + # ensure we can also run this test on a live system - so lets clean + # out the current configuration :) + cls.cli_delete(cls, base_path) + + def tearDown(self): + # Check for running process + self.assertTrue(process_named_running(PROCESS_NAME)) + + # delete testing SALT config + self.cli_delete(base_path) + self.cli_commit() + + # For an unknown reason on QEMU systems (e.g. where smoketests are executed + # from the CI) salt-minion process is not killed by systemd. Apparently + # no issue on VMWare. + if cmd('systemd-detect-virt') != 'kvm': + self.assertFalse(process_named_running(PROCESS_NAME)) + + def test_default(self): + servers = ['192.0.2.1', '192.0.2.2'] + + for server in servers: + self.cli_set(base_path + ['master', server]) + + self.cli_commit() + + # commiconf = read_file() Check configured port + conf = read_file(SALT_CONF) + self.assertIn(f'- {server}', conf) + + # defaults + hostname = gethostname() + self.assertIn(f'hash_type: sha256', conf) + self.assertIn(f'id: {hostname}', conf) + self.assertIn(f'mine_interval: 60', conf) + + def test_options(self): + server = '192.0.2.3' + hash = 'sha1' + id = 'foo' + interval = '120' + + self.cli_set(base_path + ['master', server]) + self.cli_set(base_path + ['hash', hash]) + self.cli_set(base_path + ['id', id]) + self.cli_set(base_path + ['interval', interval]) + + self.cli_commit() + + # commiconf = read_file() Check configured port + conf = read_file(SALT_CONF) + self.assertIn(f'- {server}', conf) + + # defaults + self.assertIn(f'hash_type: {hash}', conf) + self.assertIn(f'id: {id}', conf) + self.assertIn(f'mine_interval: {interval}', conf) + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_service_ssh.py b/smoketest/scripts/cli/test_service_ssh.py index 6f58ce3d3..49a167e04 100755 --- a/smoketest/scripts/cli/test_service_ssh.py +++ b/smoketest/scripts/cli/test_service_ssh.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2019-2020 VyOS maintainers and contributors +# Copyright (C) 2019-2022 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 @@ -14,14 +14,19 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. +import os +import paramiko import re import os import unittest +from pwd import getpwall + from base_vyostest_shim import VyOSUnitTestSHIM from vyos.configsession import ConfigSessionError from vyos.util import cmd +from vyos.util import is_systemd_service_running from vyos.util import process_named_running from vyos.util import read_file @@ -42,10 +47,18 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase): self.cli_delete(base_path) def tearDown(self): + # Check for running process + self.assertTrue(process_named_running(PROCESS_NAME)) + # delete testing SSH config self.cli_delete(base_path) self.cli_commit() + # Established SSH connections remains running after service is stopped. + # We can not use process_named_running here - we rather need to check + # that the systemd service is no longer running + self.assertFalse(is_systemd_service_running(PROCESS_NAME)) + def test_ssh_default(self): # Check if SSH service runs with default settings - used for checking # behavior of <defaultValue> in XML definition @@ -58,9 +71,6 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase): port = get_config_value('Port')[0] self.assertEqual('22', port) - # Check for running process - self.assertTrue(process_named_running(PROCESS_NAME)) - def test_ssh_single_listen_address(self): # Check if SSH service can be configured and runs self.cli_set(base_path + ['port', '1234']) @@ -97,9 +107,6 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase): keepalive = get_config_value('ClientAliveInterval')[0] self.assertTrue("100" in keepalive) - # Check for running process - self.assertTrue(process_named_running(PROCESS_NAME)) - def test_ssh_multiple_listen_addresses(self): # Check if SSH service can be configured and runs with multiple # listen ports and listen-addresses @@ -124,9 +131,6 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase): for address in addresses: self.assertIn(address, tmp) - # Check for running process - self.assertTrue(process_named_running(PROCESS_NAME)) - def test_ssh_vrf(self): # Check if SSH service can be bound to given VRF port = '22' @@ -146,9 +150,6 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase): tmp = get_config_value('Port') self.assertIn(port, tmp) - # Check for running process - self.assertTrue(process_named_running(PROCESS_NAME)) - # Check for process in VRF tmp = cmd(f'ip vrf pids {vrf}') self.assertIn(PROCESS_NAME, tmp) @@ -156,5 +157,51 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase): # delete VRF self.cli_delete(['vrf', 'name', vrf]) + def test_ssh_login(self): + # Perform SSH login and command execution with a predefined user. The + # result (output of uname -a) must match the output if the command is + # run natively. + # + # We also try to login as an invalid user - this is not allowed to work. + + def ssh_send_cmd(command, username, password, host='localhost'): + """ SSH command execution helper """ + # Try to login via SSH + ssh_client = paramiko.SSHClient() + ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + ssh_client.connect(hostname='localhost', username=username, password=password) + _, stdout, stderr = ssh_client.exec_command(command) + output = stdout.read().decode().strip() + error = stderr.read().decode().strip() + ssh_client.close() + return output, error + + test_user = 'ssh_test' + test_pass = 'v2i57DZs8idUwMN3VC92' + test_command = 'uname -a' + + self.cli_set(base_path) + self.cli_set(['system', 'login', 'user', test_user, 'authentication', 'plaintext-password', test_pass]) + + # commit changes + self.cli_commit() + + # Login with proper credentials + output, error = ssh_send_cmd(test_command, test_user, test_pass) + # verify login + self.assertFalse(error) + self.assertEqual(output, cmd(test_command)) + + # Login with invalid credentials + with self.assertRaises(paramiko.ssh_exception.AuthenticationException): + output, error = ssh_send_cmd(test_command, 'invalid_user', 'invalid_password') + + self.cli_delete(['system', 'login', 'user', test_user]) + self.cli_commit() + + # After deletion the test user is not allowed to remain in /etc/passwd + usernames = [x[0] for x in getpwall()] + self.assertNotIn(test_user, usernames) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_system_ipv6.py b/smoketest/scripts/cli/test_system_ipv6.py index 3112d2e46..837d1dc12 100755 --- a/smoketest/scripts/cli/test_system_ipv6.py +++ b/smoketest/scripts/cli/test_system_ipv6.py @@ -17,7 +17,12 @@ import unittest from base_vyostest_shim import VyOSUnitTestSHIM + +from vyos.template import is_ipv4 from vyos.util import read_file +from vyos.util import is_ipv6_enabled +from vyos.util import get_interface_config +from vyos.validate import is_intf_addr_assigned base_path = ['system', 'ipv6'] @@ -42,6 +47,14 @@ class TestSystemIPv6(VyOSUnitTestSHIM.TestCase): self.assertEqual(read_file(file_forwarding), '0') def test_system_ipv6_disable(self): + # Verify previous "enable" state + self.assertEqual(read_file(file_disable), '0') + self.assertTrue(is_ipv6_enabled()) + + loopbacks = ['127.0.0.1', '::1'] + for addr in loopbacks: + self.assertTrue(is_intf_addr_assigned('lo', addr)) + # Do not assign any IPv6 address on interfaces, this requires a reboot # which can not be tested, but we can read the config file :) self.cli_set(base_path + ['disable']) @@ -49,6 +62,24 @@ class TestSystemIPv6(VyOSUnitTestSHIM.TestCase): # Verify configuration file self.assertEqual(read_file(file_disable), '1') + self.assertFalse(is_ipv6_enabled()) + + for addr in loopbacks: + if is_ipv4(addr): + self.assertTrue(is_intf_addr_assigned('lo', addr)) + else: + self.assertFalse(is_intf_addr_assigned('lo', addr)) + + # T4330: Verify MTU can be changed with IPv6 disabled + mtu = '1600' + eth_if = 'eth0' + self.cli_set(['interfaces', 'ethernet', eth_if, 'mtu', mtu]) + self.cli_commit() + + tmp = get_interface_config(eth_if) + self.assertEqual(tmp['mtu'], int(mtu)) + + self.cli_delete(['interfaces', 'ethernet', eth_if, 'mtu']) def test_system_ipv6_strict_dad(self): # This defaults to 1 diff --git a/smoketest/scripts/cli/test_system_login.py b/smoketest/scripts/cli/test_system_login.py index 69a06eeac..bc76de0ad 100755 --- a/smoketest/scripts/cli/test_system_login.py +++ b/smoketest/scripts/cli/test_system_login.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2019-2020 VyOS maintainers and contributors +# Copyright (C) 2019-2022 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 @@ -23,6 +23,7 @@ from base_vyostest_shim import VyOSUnitTestSHIM from distutils.version import LooseVersion from platform import release as kernel_version from subprocess import Popen, PIPE +from pwd import getpwall from vyos.configsession import ConfigSessionError from vyos.util import cmd @@ -52,6 +53,11 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): self.cli_commit() + # After deletion, a user is not allowed to remain in /etc/passwd + usernames = [x[0] for x in getpwall()] + for user in users: + self.assertNotIn(user, usernames) + def test_add_linux_system_user(self): # We are not allowed to re-use a username already taken by the Linux # base system diff --git a/smoketest/scripts/cli/test_vrf.py b/smoketest/scripts/cli/test_vrf.py index 0f006ca3c..6614aeb06 100755 --- a/smoketest/scripts/cli/test_vrf.py +++ b/smoketest/scripts/cli/test_vrf.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2020-2021 VyOS maintainers and contributors +# Copyright (C) 2020-2022 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 @@ -27,8 +27,10 @@ from vyos.configsession import ConfigSessionError from vyos.ifconfig import Interface from vyos.ifconfig import Section from vyos.template import is_ipv6 +from vyos.template import is_ipv4 from vyos.util import cmd from vyos.util import read_file +from vyos.util import get_interface_config from vyos.validate import is_intf_addr_assigned base_path = ['vrf'] @@ -61,6 +63,8 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): def tearDown(self): # delete all VRFs self.cli_delete(base_path) + self.cli_delete(['interfaces', 'dummy']) + self.cli_delete(['protocols', 'vrf']) self.cli_commit() for vrf in vrfs: self.assertNotIn(vrf, interfaces()) @@ -108,9 +112,14 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): # ... regex = f'{table}\s+{vrf}\s+#\s+{description}' self.assertTrue(re.findall(regex, iproute2_config)) + + tmp = get_interface_config(vrf) + self.assertEqual(int(table), tmp['linkinfo']['info_data']['table']) + + # Increment table ID for the next run table = str(int(table) + 1) - def test_vrf_loopback_ips(self): + def test_vrf_loopbacks_ips(self): table = '2000' for vrf in vrfs: base = base_path + ['name', vrf] @@ -121,10 +130,48 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): self.cli_commit() # Verify VRF configuration + loopbacks = ['127.0.0.1', '::1'] for vrf in vrfs: - self.assertTrue(vrf in interfaces()) - self.assertTrue(is_intf_addr_assigned(vrf, '127.0.0.1')) - self.assertTrue(is_intf_addr_assigned(vrf, '::1')) + # Ensure VRF was created + self.assertIn(vrf, interfaces()) + # Test for proper loopback IP assignment + for addr in loopbacks: + self.assertTrue(is_intf_addr_assigned(vrf, addr)) + + def test_vrf_loopbacks_no_ipv6(self): + table = '2002' + for vrf in vrfs: + base = base_path + ['name', vrf] + self.cli_set(base + ['table', str(table)]) + table = str(int(table) + 1) + + # Globally disable IPv6 - this will remove all IPv6 interface addresses + self.cli_set(['system', 'ipv6', 'disable']) + + # commit changes + self.cli_commit() + + # Verify VRF configuration + table = '2002' + loopbacks = ['127.0.0.1', '::1'] + for vrf in vrfs: + # Ensure VRF was created + self.assertIn(vrf, interfaces()) + + # Verify VRF table ID + tmp = get_interface_config(vrf) + self.assertEqual(int(table), tmp['linkinfo']['info_data']['table']) + + # Test for proper loopback IP assignment + for addr in loopbacks: + if is_ipv4(addr): + self.assertTrue(is_intf_addr_assigned(vrf, addr)) + else: + self.assertFalse(is_intf_addr_assigned(vrf, addr)) + + table = str(int(table) + 1) + + self.cli_delete(['system', 'ipv6']) def test_vrf_bind_all(self): table = '2000' @@ -176,11 +223,11 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): # commit changes self.cli_commit() - # Verify & cleanup + # Verify VRF assignmant for interface in self._interfaces: - # os.readlink resolves to: '../../../../../virtual/net/foovrf' - tmp = os.readlink(f'/sys/class/net/{interface}/master').split('/')[-1] - self.assertEqual(tmp, vrf) + tmp = get_interface_config(interface) + self.assertEqual(vrf, tmp['master']) + # cleanup section = Section.section(interface) self.cli_delete(['interfaces', section, interface, 'vrf']) @@ -204,14 +251,14 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): }, } + # required interface for leaking to default table + self.cli_set(['interfaces', 'ethernet', 'eth0', 'address', '192.0.2.1/24']) + table = '2000' for vrf in vrfs: base = base_path + ['name', vrf] self.cli_set(base + ['table', str(table)]) - # required interface for leaking to default table - self.cli_set(['interfaces', 'ethernet', 'eth0', 'address', '192.0.2.1/24']) - # we also need an interface in "UP" state to install routes self.cli_set(['interfaces', 'dummy', f'dum{table}', 'vrf', vrf]) self.cli_set(['interfaces', 'dummy', f'dum{table}', 'address', '192.0.2.1/24']) @@ -233,28 +280,64 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): self.cli_commit() # Verify routes - table = '2000' for vrf in vrfs: - for route, route_config in routes.items(): - if is_ipv6(route): - tmp = get_vrf_ipv6_routes(vrf) - else: - tmp = get_vrf_ipv4_routes(vrf) + self.assertIn(vrf, interfaces()) + frrconfig = self.getFRRconfig(f'vrf {vrf}') + for prefix, prefix_config in routes.items(): + tmp = 'ip' + if is_ipv6(prefix): + tmp += 'v6' - found = False - for result in tmp: - if 'dst' in result and result['dst'] == route: - if 'gateway' in result and result['gateway'] == route_config['next_hop']: - found = True + tmp += f' route {prefix} {prefix_config["next_hop"]}' + if 'distance' in prefix_config: + tmp += ' ' + prefix_config['distance'] + if 'next_hop_vrf' in prefix_config: + tmp += ' nexthop-vrf ' + prefix_config['next_hop_vrf'] - self.assertTrue(found) + self.assertIn(tmp, frrconfig) - # Cleanup - self.cli_delete(['protocols', 'vrf', vrf]) - self.cli_delete(['interfaces', 'dummy', f'dum{table}']) - self.cli_delete(['interfaces', 'ethernet', 'eth0', 'address', '192.0.2.1/24']) + self.cli_delete(['interfaces', 'ethernet', 'eth0', 'address']) - table = str(int(table) + 1) + + def test_vrf_link_local_ip_addresses(self): + # Testcase for issue T4331 + table = '100' + vrf = 'orange' + interface = 'dum9998' + addresses = ['192.0.2.1/26', '2001:db8:9998::1/64', 'fe80::1/64'] + + for address in addresses: + self.cli_set(['interfaces', 'dummy', interface, 'address', address]) + + # Create dummy interfaces + self.cli_commit() + + # ... and verify IP addresses got assigned + for address in addresses: + self.assertTrue(is_intf_addr_assigned(interface, address)) + + # Move interface to VRF + self.cli_set(base_path + ['name', vrf, 'table', table]) + self.cli_set(['interfaces', 'dummy', interface, 'vrf', vrf]) + + # Apply VRF config + self.cli_commit() + # Ensure VRF got created + self.assertIn(vrf, interfaces()) + # ... and IP addresses are still assigned + for address in addresses: + self.assertTrue(is_intf_addr_assigned(interface, address)) + # Verify VRF table ID + tmp = get_interface_config(vrf) + self.assertEqual(int(table), tmp['linkinfo']['info_data']['table']) + + # Verify interface is assigned to VRF + tmp = get_interface_config(interface) + self.assertEqual(vrf, tmp['master']) + + # Delete Interface + self.cli_delete(['interfaces', 'dummy', interface]) + self.cli_commit() if __name__ == '__main__': unittest.main(verbosity=2) |