diff options
Diffstat (limited to 'smoketest/scripts/cli')
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_isis.py | 69 | ||||
| -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 | 92 |
5 files changed, 228 insertions, 45 deletions
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_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 5a44fdc66..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,6 +27,7 @@ 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 @@ -62,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()) @@ -109,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] @@ -122,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' @@ -205,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']) @@ -234,28 +280,24 @@ 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 |