summaryrefslogtreecommitdiff
path: root/smoketest
diff options
context:
space:
mode:
Diffstat (limited to 'smoketest')
-rwxr-xr-xsmoketest/scripts/cli/test_service_ssh.py73
-rwxr-xr-xsmoketest/scripts/cli/test_system_ipv6.py31
-rwxr-xr-xsmoketest/scripts/cli/test_system_login.py8
-rwxr-xr-xsmoketest/scripts/cli/test_vrf.py93
4 files changed, 166 insertions, 39 deletions
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..75be9e4f8 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'
@@ -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,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)
if __name__ == '__main__':
unittest.main(verbosity=2)
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-22 19:58:54 +0000