diff options
Diffstat (limited to 'smoketest/scripts')
| -rwxr-xr-x | smoketest/scripts/cli/test_interfaces_wireguard.py | 108 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_nat.py | 2 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_nhrp.py | 78 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_service_dns_dynamic.py | 19 | ||||
| -rw-r--r-- | smoketest/scripts/cli/test_service_monitoring_network_event.py | 65 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_system_flow-accounting.py | 107 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_system_login.py | 210 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_system_sflow.py | 33 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_vpn_ipsec.py | 46 | ||||
| -rwxr-xr-x | smoketest/scripts/system/test_kernel_options.py | 6 |
10 files changed, 404 insertions, 270 deletions
diff --git a/smoketest/scripts/cli/test_interfaces_wireguard.py b/smoketest/scripts/cli/test_interfaces_wireguard.py index 4b994a659..f8cd18cf2 100755 --- a/smoketest/scripts/cli/test_interfaces_wireguard.py +++ b/smoketest/scripts/cli/test_interfaces_wireguard.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2020-2023 VyOS maintainers and contributors +# Copyright (C) 2020-2025 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 @@ -17,27 +17,33 @@ import os import unittest -from base_vyostest_shim import VyOSUnitTestSHIM +from base_interfaces_test import BasicInterfaceTest from vyos.configsession import ConfigSessionError from vyos.utils.file import read_file from vyos.utils.process import cmd +from vyos.utils.process import is_systemd_service_running base_path = ['interfaces', 'wireguard'] - -class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase): +domain_resolver = 'vyos-domain-resolver.service' +class WireGuardInterfaceTest(BasicInterfaceTest.TestCase): @classmethod def setUpClass(cls): - super(WireGuardInterfaceTest, cls).setUpClass() - - cls._test_addr = ['192.0.2.1/26', '192.0.2.255/31', '192.0.2.64/32', - '2001:db8:1::ffff/64', '2001:db8:101::1/112'] - cls._interfaces = ['wg0', 'wg1'] + cls._base_path = ['interfaces', 'wireguard'] + cls._options = { + 'wg0': ['private-key wBbGJJXYllwDcw63AFjiIR6ZlsvqvAf3eDwog64Dp0Q=', + 'peer RED public-key 6hkkfxN4VUQLu36NLZr47I7ST/FkQl2clPWr+9a6ZH8=', + 'peer RED allowed-ips 169.254.0.0/16', + 'port 5678'], + 'wg1': ['private-key QFwnBHlHYspehvpklBKb7cikM+QMkEy2p6gfsg06S08=', + 'peer BLUE public-key hRJLmP8SVU9/MLmPmYmpOa+RTB4F/zhDqA+/QDuW1Hg=', + 'peer BLUE allowed-ips 169.254.0.0/16', + 'port 4567'], + } + cls._interfaces = list(cls._options) - def tearDown(self): - self.cli_delete(base_path) - self.cli_commit() + super(WireGuardInterfaceTest, cls).setUpClass() - def test_01_wireguard_peer(self): + def test_wireguard_peer(self): # Create WireGuard interfaces with associated peers for intf in self._interfaces: peer = 'foo-' + intf @@ -64,7 +70,7 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase): self.assertTrue(os.path.isdir(f'/sys/class/net/{intf}')) - def test_02_wireguard_add_remove_peer(self): + def test_wireguard_add_remove_peer(self): # T2939: Create WireGuard interfaces with associated peers. # Remove one of the configured peers. # T4774: Test prevention of duplicate peer public keys @@ -102,7 +108,7 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase): self.cli_delete(base_path + [interface, 'peer', 'PEER01']) self.cli_commit() - def test_03_wireguard_same_public_key(self): + def test_wireguard_same_public_key(self): # T5413: Test prevention of equality interface public key and peer's # public key interface = 'wg0' @@ -115,45 +121,40 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase): self.cli_set(base_path + [interface, 'private-key', privkey]) self.cli_set(base_path + [interface, 'peer', 'PEER01', 'public-key', pubkey_fail]) - self.cli_set(base_path + [interface, 'peer', 'PEER01', 'port', port]) self.cli_set(base_path + [interface, 'peer', 'PEER01', 'allowed-ips', '10.205.212.10/32']) self.cli_set(base_path + [interface, 'peer', 'PEER01', 'address', '192.0.2.1']) # The same pubkey as the interface wg0 with self.assertRaises(ConfigSessionError): self.cli_commit() - self.cli_set(base_path + [interface, 'peer', 'PEER01', 'public-key', pubkey_ok]) + # If address is defined for a peer, so must be the peer port + with self.assertRaises(ConfigSessionError): + self.cli_commit() + self.cli_set(base_path + [interface, 'peer', 'PEER01', 'port', port]) + # Commit peers self.cli_commit() self.assertTrue(os.path.isdir(f'/sys/class/net/{interface}')) - def test_04_wireguard_threaded(self): + def test_wireguard_threaded(self): # T5409: Test adding threaded option on interface. - # Test prevention for adding threaded - # if no enabled peer is configured. - interface = 'wg0' - port = '12345' - privkey = 'OOjcXGfgQlAuM6q8Z9aAYduCua7pxf7UKYvIqoUPoGQ=' - pubkey = 'ebFx/1G0ti8tvuZd94sEIosAZZIznX+dBAKG/8DFm0I=' + for intf in self._interfaces: + for option in self._options.get(intf, []): + self.cli_set(self._base_path + [intf] + option.split()) - self.cli_set(base_path + [interface, 'address', '172.16.0.1/24']) - self.cli_set(base_path + [interface, 'private-key', privkey]) - - self.cli_set(base_path + [interface, 'peer', 'PEER01', 'port', port]) - self.cli_set(base_path + [interface, 'peer', 'PEER01', 'public-key', pubkey]) - self.cli_set(base_path + [interface, 'peer', 'PEER01', 'allowed-ips', '10.205.212.10/32']) - self.cli_set(base_path + [interface, 'peer', 'PEER01', 'address', '192.0.2.1']) - self.cli_set(base_path + [interface, 'per-client-thread']) + self.cli_set(base_path + [intf, 'per-client-thread']) # Commit peers self.cli_commit() - tmp = read_file(f'/sys/class/net/{interface}/threaded') - self.assertTrue(tmp, "1") - def test_05_wireguard_peer_pubkey_change(self): + for intf in self._interfaces: + tmp = read_file(f'/sys/class/net/{intf}/threaded') + self.assertTrue(tmp, "1") + + def test_wireguard_peer_pubkey_change(self): # T5707 changing WireGuard CLI public key of a peer - it's not removed def get_peers(interface) -> list: @@ -171,7 +172,6 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase): peers.append(items[0]) return peers - interface = 'wg1337' port = '1337' privkey = 'iJi4lb2HhkLx2KSAGOjji2alKkYsJjSPkHkrcpxgEVU=' @@ -200,5 +200,41 @@ class WireGuardInterfaceTest(VyOSUnitTestSHIM.TestCase): self.assertNotIn(pubkey_1, peers) self.assertIn(pubkey_2, peers) + def test_wireguard_hostname(self): + # T4930: Test dynamic endpoint support + interface = 'wg1234' + port = '54321' + privkey = 'UOWIeZKNzijhgu0bPRy2PB3gnuOBLfQax5GiYfkmU3A=' + pubkey = '4nG5NfhHBQUq/DnwT0RjRoBCqAh3VrRHqdQgzC/xujk=' + + base_interface_path = base_path + [interface] + self.cli_set(base_interface_path + ['address', '172.16.0.1/24']) + self.cli_set(base_interface_path + ['private-key', privkey]) + + peer_base_path = base_interface_path + ['peer', 'dynamic01'] + self.cli_set(peer_base_path + ['port', port]) + self.cli_set(peer_base_path + ['public-key', pubkey]) + self.cli_set(peer_base_path + ['allowed-ips', '169.254.0.0/16']) + self.cli_set(peer_base_path + ['address', '192.0.2.1']) + self.cli_set(peer_base_path + ['host-name', 'wg.vyos.net']) + + # Peer address and host-name are mutually exclusive + with self.assertRaises(ConfigSessionError): + self.cli_commit() + + self.cli_delete(peer_base_path + ['address']) + + # Commit peers + self.cli_commit() + + # Ensure the service is running which checks for DNS changes + self.assertTrue(is_systemd_service_running(domain_resolver)) + + self.cli_delete(base_interface_path) + self.cli_commit() + + # Ensure the service is no longer running after WireGuard interface is deleted + self.assertFalse(is_systemd_service_running(domain_resolver)) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_nat.py b/smoketest/scripts/cli/test_nat.py index 0beafcc6c..b33ef2617 100755 --- a/smoketest/scripts/cli/test_nat.py +++ b/smoketest/scripts/cli/test_nat.py @@ -84,7 +84,7 @@ class TestNAT(VyOSUnitTestSHIM.TestCase): address_group = 'smoketest_addr' address_group_member = '192.0.2.1' interface_group = 'smoketest_ifaces' - interface_group_member = 'bond.99' + interface_group_member = 'eth0' self.cli_set(['firewall', 'group', 'address-group', address_group, 'address', address_group_member]) self.cli_set(['firewall', 'group', 'interface-group', interface_group, 'interface', interface_group_member]) diff --git a/smoketest/scripts/cli/test_protocols_nhrp.py b/smoketest/scripts/cli/test_protocols_nhrp.py index 43ae4abf2..f6d1f1da5 100755 --- a/smoketest/scripts/cli/test_protocols_nhrp.py +++ b/smoketest/scripts/cli/test_protocols_nhrp.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2024 VyOS maintainers and contributors +# Copyright (C) 2021-2025 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 @@ -25,6 +25,7 @@ from vyos.utils.file import read_file tunnel_path = ['interfaces', 'tunnel'] nhrp_path = ['protocols', 'nhrp'] vpn_path = ['vpn', 'ipsec'] +PROCESS_NAME = 'nhrpd' class TestProtocolsNHRP(VyOSUnitTestSHIM.TestCase): @classmethod @@ -41,29 +42,41 @@ class TestProtocolsNHRP(VyOSUnitTestSHIM.TestCase): self.cli_delete(tunnel_path) self.cli_commit() - def test_config(self): + def test_01_nhrp_config(self): tunnel_if = "tun100" - tunnel_source = "192.0.2.1" + tunnel_ip = '172.16.253.134/32' + tunnel_source = "192.0.2.134" tunnel_encapsulation = "gre" esp_group = "ESP-HUB" ike_group = "IKE-HUB" nhrp_secret = "vyos123" nhrp_profile = "NHRPVPN" + nhrp_holdtime = '300' + nhs_tunnelip = '172.16.253.1' + nhs_nbmaip = '192.0.2.1' + map_tunnelip = '172.16.253.135' + map_nbmaip = "192.0.2.135" + nhrp_networkid = '1' ipsec_secret = "secret" - + multicat_log_group = '2' + redirect_log_group = '1' # Tunnel - self.cli_set(tunnel_path + [tunnel_if, "address", "172.16.253.134/29"]) + self.cli_set(tunnel_path + [tunnel_if, "address", tunnel_ip]) self.cli_set(tunnel_path + [tunnel_if, "encapsulation", tunnel_encapsulation]) self.cli_set(tunnel_path + [tunnel_if, "source-address", tunnel_source]) self.cli_set(tunnel_path + [tunnel_if, "enable-multicast"]) self.cli_set(tunnel_path + [tunnel_if, "parameters", "ip", "key", "1"]) # NHRP - self.cli_set(nhrp_path + ["tunnel", tunnel_if, "cisco-authentication", nhrp_secret]) - self.cli_set(nhrp_path + ["tunnel", tunnel_if, "holding-time", "300"]) - self.cli_set(nhrp_path + ["tunnel", tunnel_if, "multicast", "dynamic"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "authentication", nhrp_secret]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "holdtime", nhrp_holdtime]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "multicast", nhs_tunnelip]) self.cli_set(nhrp_path + ["tunnel", tunnel_if, "redirect"]) self.cli_set(nhrp_path + ["tunnel", tunnel_if, "shortcut"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "registration-no-unique"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "network-id", nhrp_networkid]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "nhs", "tunnel-ip", nhs_tunnelip, "nbma", nhs_nbmaip]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "map", "tunnel-ip", map_tunnelip, "nbma", map_nbmaip]) # IKE/ESP Groups self.cli_set(vpn_path + ["esp-group", esp_group, "lifetime", "1800"]) @@ -93,29 +106,40 @@ class TestProtocolsNHRP(VyOSUnitTestSHIM.TestCase): self.cli_commit() - opennhrp_lines = [ - f'interface {tunnel_if} #hub {nhrp_profile}', - f'cisco-authentication {nhrp_secret}', - f'holding-time 300', - f'shortcut', - f'multicast dynamic', - f'redirect' + frrconfig = self.getFRRconfig(f'interface {tunnel_if}', endsection='^exit') + self.assertIn(f'interface {tunnel_if}', frrconfig) + self.assertIn(f' ip nhrp authentication {nhrp_secret}', frrconfig) + self.assertIn(f' ip nhrp holdtime {nhrp_holdtime}', frrconfig) + self.assertIn(f' ip nhrp map multicast {nhs_tunnelip}', frrconfig) + self.assertIn(f' ip nhrp redirect', frrconfig) + self.assertIn(f' ip nhrp registration no-unique', frrconfig) + self.assertIn(f' ip nhrp shortcut', frrconfig) + self.assertIn(f' ip nhrp network-id {nhrp_networkid}', frrconfig) + self.assertIn(f' ip nhrp nhs {nhs_tunnelip} nbma {nhs_nbmaip}', frrconfig) + self.assertIn(f' ip nhrp map {map_tunnelip} {map_nbmaip}', frrconfig) + self.assertIn(f' tunnel protection vici profile dmvpn-{nhrp_profile}-{tunnel_if}-child', + frrconfig) + + nftables_search_multicast = [ + ['chain VYOS_NHRP_MULTICAST_OUTPUT'], + ['type filter hook output priority filter + 10; policy accept;'], + [f'oifname "{tunnel_if}"', 'ip daddr 224.0.0.0/24', 'counter', f'log group {multicat_log_group}'], + [f'oifname "{tunnel_if}"', 'ip daddr 224.0.0.0/24', 'counter', 'drop'], + ['chain VYOS_NHRP_MULTICAST_FORWARD'], + ['type filter hook output priority filter + 10; policy accept;'], + [f'oifname "{tunnel_if}"', 'ip daddr 224.0.0.0/4', 'counter', f'log group {multicat_log_group}'], + [f'oifname "{tunnel_if}"', 'ip daddr 224.0.0.0/4', 'counter', 'drop'] ] - tmp_opennhrp_conf = read_file('/run/opennhrp/opennhrp.conf') - - for line in opennhrp_lines: - self.assertIn(line, tmp_opennhrp_conf) - - firewall_matches = [ - f'ip protocol {tunnel_encapsulation}', - f'ip saddr {tunnel_source}', - f'ip daddr 224.0.0.0/4', - f'comment "VYOS_NHRP_{tunnel_if}"' + nftables_search_redirect = [ + ['chain VYOS_NHRP_REDIRECT_FORWARD'], + ['type filter hook forward priority filter + 10; policy accept;'], + [f'iifname "{tunnel_if}" oifname "{tunnel_if}"', 'meter loglimit-0 size 65535 { ip daddr & 255.255.255.0 . ip saddr & 255.255.255.0 timeout 1m limit rate 4/minute burst 1 packets }', 'counter', f'log group {redirect_log_group}'] ] + self.verify_nftables(nftables_search_multicast, 'ip vyos_nhrp_multicast') + self.verify_nftables(nftables_search_redirect, 'ip vyos_nhrp_redirect') - self.assertTrue(find_nftables_rule('ip vyos_nhrp_filter', 'VYOS_NHRP_OUTPUT', firewall_matches) is not None) - self.assertTrue(process_named_running('opennhrp')) + self.assertTrue(process_named_running(PROCESS_NAME)) if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_service_dns_dynamic.py b/smoketest/scripts/cli/test_service_dns_dynamic.py index 9fbc931de..522102e67 100755 --- a/smoketest/scripts/cli/test_service_dns_dynamic.py +++ b/smoketest/scripts/cli/test_service_dns_dynamic.py @@ -20,8 +20,10 @@ import tempfile from base_vyostest_shim import VyOSUnitTestSHIM from vyos.configsession import ConfigSessionError +from vyos.utils.file import read_file from vyos.utils.process import cmd from vyos.utils.process import process_named_running +from vyos.xml_ref import default_value DDCLIENT_SYSTEMD_UNIT = '/run/systemd/system/ddclient.service.d/override.conf' DDCLIENT_CONF = '/run/ddclient/ddclient.conf' @@ -29,6 +31,7 @@ DDCLIENT_PNAME = 'ddclient' base_path = ['service', 'dns', 'dynamic'] name_path = base_path + ['name'] +default_interval = default_value(base_path + ['interval']) server = 'ddns.vyos.io' hostname = 'test.ddns.vyos.io' zone = 'vyos.io' @@ -95,12 +98,14 @@ class TestServiceDDNS(VyOSUnitTestSHIM.TestCase): # Check the generating config parameters ddclient_conf = cmd(f'sudo cat {DDCLIENT_CONF}') - # default value 300 seconds - self.assertIn(f'daemon=300', ddclient_conf) self.assertIn(f'usev4=ifv4', ddclient_conf) self.assertIn(f'ifv4={interface}', ddclient_conf) self.assertIn(f'password=\'{password}\'', ddclient_conf) + # Check default interval of 300 seconds + systemd_override = read_file(DDCLIENT_SYSTEMD_UNIT) + self.assertIn(f'--daemon {default_interval}', systemd_override) + for opt in details.keys(): if opt == 'username': login = details[opt] @@ -140,7 +145,6 @@ class TestServiceDDNS(VyOSUnitTestSHIM.TestCase): # Check the generating config parameters ddclient_conf = cmd(f'sudo cat {DDCLIENT_CONF}') - self.assertIn(f'daemon={interval}', ddclient_conf) self.assertIn(f'usev6=ifv6', ddclient_conf) self.assertIn(f'ifv6={interface}', ddclient_conf) self.assertIn(f'protocol={proto}', ddclient_conf) @@ -150,6 +154,10 @@ class TestServiceDDNS(VyOSUnitTestSHIM.TestCase): self.assertIn(f'min-interval={wait_time}', ddclient_conf) self.assertIn(f'max-interval={expiry_time_good}', ddclient_conf) + # default value 300 seconds + systemd_override = read_file(DDCLIENT_SYSTEMD_UNIT) + self.assertIn(f'--daemon {interval}', systemd_override) + # IPv4+IPv6 dual DDNS service configuration def test_03_dyndns_service_dual_stack(self): services = {'cloudflare': {'protocol': 'cloudflare', 'zone': zone}, @@ -339,9 +347,10 @@ class TestServiceDDNS(VyOSUnitTestSHIM.TestCase): self.cli_commit() # Check for process in VRF - systemd_override = cmd(f'cat {DDCLIENT_SYSTEMD_UNIT}') + systemd_override = read_file(DDCLIENT_SYSTEMD_UNIT) self.assertIn(f'ExecStart=ip vrf exec {vrf_name} /usr/bin/ddclient ' \ - f'--file {DDCLIENT_CONF} --foreground', systemd_override) + f'--file {DDCLIENT_CONF} --cache {DDCLIENT_CONF.replace("conf", "cache")} ' \ + f'--foreground --daemon {default_interval}', systemd_override) # Check for process in VRF proc = cmd(f'ip vrf pids {vrf_name}') diff --git a/smoketest/scripts/cli/test_service_monitoring_network_event.py b/smoketest/scripts/cli/test_service_monitoring_network_event.py new file mode 100644 index 000000000..3c9b4bf7f --- /dev/null +++ b/smoketest/scripts/cli/test_service_monitoring_network_event.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2024 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 base_vyostest_shim import VyOSUnitTestSHIM +from vyos.utils.file import read_json + + +base_path = ['service', 'monitoring', 'network-event'] + + +def get_logger_config(): + return read_json('/run/vyos-network-event-logger.conf') + + +class TestMonitoringNetworkEvent(VyOSUnitTestSHIM.TestCase): + @classmethod + def setUpClass(cls): + super(TestMonitoringNetworkEvent, 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_network_event_log(self): + expected_config = { + 'event': { + 'route': {}, + 'link': {}, + 'addr': {}, + 'neigh': {}, + 'rule': {}, + }, + 'queue_size': '10000' + } + + self.cli_set(base_path + ['event', 'route']) + self.cli_set(base_path + ['event', 'link']) + self.cli_set(base_path + ['event', 'addr']) + self.cli_set(base_path + ['event', 'neigh']) + self.cli_set(base_path + ['event', 'rule']) + self.cli_set(base_path + ['queue-size', '10000']) + self.cli_commit() + self.assertEqual(expected_config, get_logger_config()) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_system_flow-accounting.py b/smoketest/scripts/cli/test_system_flow-accounting.py index 515134220..9d7942789 100755 --- a/smoketest/scripts/cli/test_system_flow-accounting.py +++ b/smoketest/scripts/cli/test_system_flow-accounting.py @@ -97,111 +97,6 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase): self.assertIn(f'syslog: {syslog}', uacctd) self.assertIn(f'plugins: memory', uacctd) - def test_sflow(self): - sampling_rate = '4000' - source_address = '192.0.2.1' - dummy_if = 'dum3841' - agent_address = '192.0.2.2' - - sflow_server = { - '1.2.3.4' : { }, - '5.6.7.8' : { 'port' : '6000' }, - } - - self.cli_set(['interfaces', 'dummy', dummy_if, 'address', agent_address + '/32']) - self.cli_set(['interfaces', 'dummy', dummy_if, 'address', source_address + '/32']) - self.cli_set(base_path + ['disable-imt']) - - # You need to configure at least one interface for flow-accounting - with self.assertRaises(ConfigSessionError): - self.cli_commit() - for interface in Section.interfaces('ethernet'): - self.cli_set(base_path + ['interface', interface]) - - - # You need to configure at least one sFlow or NetFlow protocol, or not - # set "disable-imt" for flow-accounting - with self.assertRaises(ConfigSessionError): - self.cli_commit() - - self.cli_set(base_path + ['sflow', 'agent-address', agent_address]) - self.cli_set(base_path + ['sflow', 'sampling-rate', sampling_rate]) - self.cli_set(base_path + ['sflow', 'source-address', source_address]) - for server, server_config in sflow_server.items(): - self.cli_set(base_path + ['sflow', 'server', server]) - if 'port' in server_config: - self.cli_set(base_path + ['sflow', 'server', server, 'port', server_config['port']]) - - # commit changes - self.cli_commit() - - uacctd = read_file(uacctd_conf) - - # when 'disable-imt' is not configured on the CLI it must be present - self.assertNotIn(f'imt_path: /tmp/uacctd.pipe', uacctd) - self.assertNotIn(f'imt_mem_pools_number: 169', uacctd) - self.assertNotIn(f'plugins: memory', uacctd) - - for server, server_config in sflow_server.items(): - plugin_name = server.replace('.', '-') - if 'port' in server_config: - self.assertIn(f'sfprobe_receiver[sf_{plugin_name}]: {server}', uacctd) - else: - self.assertIn(f'sfprobe_receiver[sf_{plugin_name}]: {server}:6343', uacctd) - - self.assertIn(f'sfprobe_agentip[sf_{plugin_name}]: {agent_address}', uacctd) - self.assertIn(f'sampling_rate[sf_{plugin_name}]: {sampling_rate}', uacctd) - self.assertIn(f'sfprobe_source_ip[sf_{plugin_name}]: {source_address}', uacctd) - - self.cli_delete(['interfaces', 'dummy', dummy_if]) - - def test_sflow_ipv6(self): - sampling_rate = '100' - sflow_server = { - '2001:db8::1' : { }, - '2001:db8::2' : { 'port' : '6000' }, - } - - self.cli_set(base_path + ['disable-imt']) - - # You need to configure at least one interface for flow-accounting - with self.assertRaises(ConfigSessionError): - self.cli_commit() - for interface in Section.interfaces('ethernet'): - self.cli_set(base_path + ['interface', interface]) - - - # You need to configure at least one sFlow or NetFlow protocol, or not - # set "disable-imt" for flow-accounting - with self.assertRaises(ConfigSessionError): - self.cli_commit() - - self.cli_set(base_path + ['sflow', 'sampling-rate', sampling_rate]) - for server, server_config in sflow_server.items(): - self.cli_set(base_path + ['sflow', 'server', server]) - if 'port' in server_config: - self.cli_set(base_path + ['sflow', 'server', server, 'port', server_config['port']]) - - # commit changes - self.cli_commit() - - uacctd = read_file(uacctd_conf) - - # when 'disable-imt' is not configured on the CLI it must be present - self.assertNotIn(f'imt_path: /tmp/uacctd.pipe', uacctd) - self.assertNotIn(f'imt_mem_pools_number: 169', uacctd) - self.assertNotIn(f'plugins: memory', uacctd) - - for server, server_config in sflow_server.items(): - tmp_srv = server - tmp_srv = tmp_srv.replace(':', '-') - - if 'port' in server_config: - self.assertIn(f'sfprobe_receiver[sf_{tmp_srv}]: {bracketize_ipv6(server)}', uacctd) - else: - self.assertIn(f'sfprobe_receiver[sf_{tmp_srv}]: {bracketize_ipv6(server)}:6343', uacctd) - self.assertIn(f'sampling_rate[sf_{tmp_srv}]: {sampling_rate}', uacctd) - def test_netflow(self): engine_id = '33' max_flows = '667' @@ -288,8 +183,8 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase): self.assertIn(f'nfprobe_timeouts[nf_{tmp_srv}]: expint={tmo_expiry}:general={tmo_flow}:icmp={tmo_icmp}:maxlife={tmo_max}:tcp.fin={tmo_tcp_fin}:tcp={tmo_tcp_generic}:tcp.rst={tmo_tcp_rst}:udp={tmo_udp}', uacctd) - self.cli_delete(['interfaces', 'dummy', dummy_if]) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_system_login.py b/smoketest/scripts/cli/test_system_login.py index f6a2c3cb3..d79f5521c 100755 --- a/smoketest/scripts/cli/test_system_login.py +++ b/smoketest/scripts/cli/test_system_login.py @@ -31,17 +31,19 @@ from subprocess import PIPE from pwd import getpwall from vyos.configsession import ConfigSessionError +from vyos.configquery import ConfigTreeQuery from vyos.utils.auth import get_current_user from vyos.utils.process import cmd -from vyos.utils.process import process_named_running from vyos.utils.file import read_file from vyos.utils.file import write_file from vyos.template import inc_ip +from vyos.template import is_ipv6 +from vyos.xml_ref import default_value base_path = ['system', 'login'] users = ['vyos1', 'vyos-roxx123', 'VyOS-123_super.Nice'] -SSH_PROCESS_NAME = 'sshd' +ssh_test_command = '/opt/vyatta/bin/vyatta-op-cmd-wrapper show version' ssh_pubkey = """ AAAAB3NzaC1yc2EAAAADAQABAAABgQD0NuhUOEtMIKnUVFIHoFatqX/c4mjerXyF @@ -57,7 +59,6 @@ TTSb0X1zPGxPIRFy5GoGtO9Mm5h4OZk= tac_image = 'docker.io/lfkeitel/tacacs_plus:alpine' tac_image_path = '/usr/share/vyos/tacplus-alpine.tar' - TAC_PLUS_TMPL_SRC = """ id = spawnd { debug redirect = /dev/stdout @@ -100,6 +101,25 @@ id = tac_plus { member = admin } } + +""" + +radius_image = 'docker.io/dchidell/radius-web:latest' +radius_image_path = '/usr/share/vyos/radius-latest.tar' +RADIUS_CLIENTS_TMPL_SRC = """ +client SMOKETEST { + secret = {{ radius_key }} + nastype = other + ipaddr = {{ source_address }} +} + +""" +RADIUS_USERS_TMPL_SRC = """ +# User configuration +{{ username }} Cleartext-Password := "{{ password }}" + Service-Type = NAS-Prompt-User, + Cisco-AVPair = "shell:priv-lvl=15" + """ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): @@ -112,16 +132,36 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): cls.cli_delete(cls, base_path + ['radius']) cls.cli_delete(cls, base_path + ['tacacs']) - # Load image for smoketest provided in vyos-1x-smoketest + # Load images for smoketest provided in vyos-1x-smoketest if not os.path.exists(tac_image_path): cls.fail(cls, f'{tac_image} image not available') cmd(f'sudo podman load -i {tac_image_path}') + if not os.path.exists(radius_image_path): + cls.fail(cls, f'{radius_image} image not available') + cmd(f'sudo podman load -i {radius_image_path}') + + cls.ssh_test_command_result = cls.op_mode(cls, ['show', 'version']) + + # Dynamically start SSH service if it's not running + config = ConfigTreeQuery() + cls.is_sshd_pre_test = config.exists(['service', 'sshd']) + if not cls.is_sshd_pre_test: + # Start SSH service + cls.cli_set(cls, ['service', 'ssh']) + @classmethod def tearDownClass(cls): + # Stop SSH service - if it was not running before starting the test + if not cls.is_sshd_pre_test: + cls.cli_set(cls, ['service', 'ssh']) + cls.cli_commit(cls) + super(TestSystemLogin, cls).tearDownClass() - # Cleanup podman image + + # Cleanup container images cmd(f'sudo podman image rm -f {tac_image}') + cmd(f'sudo podman image rm -f {radius_image}') def tearDown(self): # Delete individual users from configuration @@ -152,9 +192,6 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): self.cli_delete(base_path + ['user', system_user]) def test_system_login_user(self): - # Check if user can be created and we can SSH to localhost - self.cli_set(['service', 'ssh', 'port', '22']) - for user in users: name = f'VyOS Roxx {user}' home_dir = f'/tmp/smoketest/{user}' @@ -240,71 +277,71 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): self.assertIn(f'{option}=y', kernel_config) def test_system_login_radius_ipv4(self): - # Verify generated RADIUS configuration files - - radius_key = 'VyOSsecretVyOS' - radius_server = '172.16.100.10' - radius_source = '127.0.0.1' - radius_port = '2000' - radius_timeout = '1' - - self.cli_set(base_path + ['radius', 'server', radius_server, 'key', radius_key]) - self.cli_set(base_path + ['radius', 'server', radius_server, 'port', radius_port]) - self.cli_set(base_path + ['radius', 'server', radius_server, 'timeout', radius_timeout]) - self.cli_set(base_path + ['radius', 'source-address', radius_source]) - self.cli_set(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)]) + radius_servers = ['100.64.0.4', '100.64.0.5'] + radius_source = '100.64.0.1' + self._system_login_radius_test_helper(radius_servers, radius_source) - # check validate() - Only one IPv4 source-address supported - with self.assertRaises(ConfigSessionError): - self.cli_commit() - self.cli_delete(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)]) - - self.cli_commit() + def test_system_login_radius_ipv6(self): + radius_servers = ['2001:db8::4', '2001:db8::5'] + radius_source = '2001:db8::1' + self._system_login_radius_test_helper(radius_servers, radius_source) - # this file must be read with higher permissions - pam_radius_auth_conf = cmd('sudo cat /etc/pam_radius_auth.conf') - tmp = re.findall(r'\n?{}:{}\s+{}\s+{}\s+{}'.format(radius_server, - radius_port, radius_key, radius_timeout, - radius_source), pam_radius_auth_conf) - self.assertTrue(tmp) + def _system_login_radius_test_helper(self, radius_servers: list, radius_source: str): + # Verify generated RADIUS configuration files + radius_key = ''.join(secrets.choice(string.ascii_letters + string.digits) for i in range(10)) - # required, static options - self.assertIn('priv-lvl 15', pam_radius_auth_conf) - self.assertIn('mapped_priv_user radius_priv_user', pam_radius_auth_conf) + default_port = default_value(base_path + ['radius', 'server', radius_servers[0], 'port']) + default_timeout = default_value(base_path + ['radius', 'server', radius_servers[0], 'timeout']) - # PAM - pam_common_account = read_file('/etc/pam.d/common-account') - self.assertIn('pam_radius_auth.so', pam_common_account) + dummy_if = 'dum12760' - pam_common_auth = read_file('/etc/pam.d/common-auth') - self.assertIn('pam_radius_auth.so', pam_common_auth) + # Load container image for FreeRADIUS server + radius_config = '/tmp/smoketest-radius-server' + radius_container_path = ['container', 'name', 'radius-1'] - pam_common_session = read_file('/etc/pam.d/common-session') - self.assertIn('pam_radius_auth.so', pam_common_session) - - pam_common_session_noninteractive = read_file('/etc/pam.d/common-session-noninteractive') - self.assertIn('pam_radius_auth.so', pam_common_session_noninteractive) + # Generate random string with 10 digits + username = 'radius-admin' + password = ''.join(secrets.choice(string.ascii_letters + string.digits) for i in range(10)) + radius_source_mask = '32' + if is_ipv6(radius_source): + radius_source_mask = '128' + radius_test_user = { + 'username' : username, + 'password' : password, + 'radius_key' : radius_key, + 'source_address' : f'{radius_source}/{radius_source_mask}' + } - # NSS - nsswitch_conf = read_file('/etc/nsswitch.conf') - tmp = re.findall(r'passwd:\s+mapuid\s+files\s+mapname', nsswitch_conf) - self.assertTrue(tmp) + tmpl = jinja2.Template(RADIUS_CLIENTS_TMPL_SRC) + write_file(f'{radius_config}/clients.cfg', tmpl.render(radius_test_user)) - tmp = re.findall(r'group:\s+mapname\s+files', nsswitch_conf) - self.assertTrue(tmp) + tmpl = jinja2.Template(RADIUS_USERS_TMPL_SRC) + write_file(f'{radius_config}/users', tmpl.render(radius_test_user)) - def test_system_login_radius_ipv6(self): - # Verify generated RADIUS configuration files + # Start tac_plus container + self.cli_set(radius_container_path + ['allow-host-networks']) + self.cli_set(radius_container_path + ['image', radius_image]) + self.cli_set(radius_container_path + ['volume', 'clients', 'destination', '/etc/raddb/clients.conf']) + self.cli_set(radius_container_path + ['volume', 'clients', 'mode', 'ro']) + self.cli_set(radius_container_path + ['volume', 'clients', 'source', f'{radius_config}/clients.cfg']) + self.cli_set(radius_container_path + ['volume', 'users', 'destination', '/etc/raddb/users']) + self.cli_set(radius_container_path + ['volume', 'users', 'mode', 'ro']) + self.cli_set(radius_container_path + ['volume', 'users', 'source', f'{radius_config}/users']) - radius_key = 'VyOS-VyOS' - radius_server = '2001:db8::1' - radius_source = '::1' - radius_port = '4000' - radius_timeout = '4' + # Start container + self.cli_commit() - self.cli_set(base_path + ['radius', 'server', radius_server, 'key', radius_key]) - self.cli_set(base_path + ['radius', 'server', radius_server, 'port', radius_port]) - self.cli_set(base_path + ['radius', 'server', radius_server, 'timeout', radius_timeout]) + # Deinfine RADIUS servers + for radius_server in radius_servers: + # Use this system as "remote" RADIUS server + dummy_address_mask = '32' + if is_ipv6(radius_server): + dummy_address_mask = '128' + self.cli_set(['interfaces', 'dummy', dummy_if, 'address', f'{radius_server}/{dummy_address_mask}']) + self.cli_set(base_path + ['radius', 'server', radius_server, 'key', radius_key]) + + # Define RADIUS traffic source address + self.cli_set(['interfaces', 'dummy', dummy_if, 'address', f'{radius_source}/{radius_source_mask}']) self.cli_set(base_path + ['radius', 'source-address', radius_source]) self.cli_set(base_path + ['radius', 'source-address', inc_ip(radius_source, 1)]) @@ -317,10 +354,13 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): # this file must be read with higher permissions pam_radius_auth_conf = cmd('sudo cat /etc/pam_radius_auth.conf') - tmp = re.findall(r'\n?\[{}\]:{}\s+{}\s+{}\s+\[{}\]'.format(radius_server, - radius_port, radius_key, radius_timeout, - radius_source), pam_radius_auth_conf) - self.assertTrue(tmp) + + for radius_server in radius_servers: + if is_ipv6(radius_server): + # it is essential to escape the [] brackets when searching with a regex + radius_server = rf'\[{radius_server}\]' + tmp = re.findall(rf'\n?{radius_server}:{default_port}\s+{radius_key}\s+{default_timeout}\s+{radius_source}', pam_radius_auth_conf) + self.assertTrue(tmp) # required, static options self.assertIn('priv-lvl 15', pam_radius_auth_conf) @@ -347,6 +387,27 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): tmp = re.findall(r'group:\s+mapname\s+files', nsswitch_conf) self.assertTrue(tmp) + # Login with proper credentials + out, err = self.ssh_send_cmd(ssh_test_command, username, password) + # verify login + self.assertFalse(err) + self.assertEqual(out, self.ssh_test_command_result) + + # Login with invalid credentials + with self.assertRaises(paramiko.ssh_exception.AuthenticationException): + _, _ = self.ssh_send_cmd(ssh_test_command, username, f'{password}1') + + # Remove RADIUS configuration + self.cli_delete(base_path + ['radius']) + # Remove RADIUS container + self.cli_delete(radius_container_path) + # Remove dummy interface + self.cli_delete(['interfaces', 'dummy', dummy_if]) + self.cli_commit() + + # Remove rendered tac_plus daemon configuration + shutil.rmtree(radius_config) + def test_system_login_max_login_session(self): max_logins = '2' timeout = '600' @@ -390,12 +451,6 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): tmpl = jinja2.Template(TAC_PLUS_TMPL_SRC) write_file(f'{tac_plus_config}/tac_plus.cfg', tmpl.render(tac_test_user)) - # Check if SSH service is running - ssh_running = process_named_running(SSH_PROCESS_NAME) - if not ssh_running: - # Start SSH service - self.cli_set(['service', 'ssh']) - # Start tac_plus container self.cli_set(tac_container_path + ['allow-host-networks']) self.cli_set(tac_container_path + ['image', tac_image]) @@ -450,15 +505,14 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): self.assertIn(f'server={server}', nss_tacacs_conf) # Login with proper credentials - test_command = 'uname -a' - out, err = self.ssh_send_cmd(test_command, username, password) + out, err = self.ssh_send_cmd(ssh_test_command, username, password) # verify login self.assertFalse(err) - self.assertEqual(out, cmd(test_command)) + self.assertEqual(out, self.ssh_test_command_result) # Login with invalid credentials with self.assertRaises(paramiko.ssh_exception.AuthenticationException): - _, _ = self.ssh_send_cmd(test_command, username, f'{password}1') + _, _ = self.ssh_send_cmd(ssh_test_command, username, f'{password}1') # Remove TACACS configuration self.cli_delete(base_path + ['tacacs']) @@ -471,10 +525,6 @@ class TestSystemLogin(VyOSUnitTestSHIM.TestCase): # Remove rendered tac_plus daemon configuration shutil.rmtree(tac_plus_config) - # Stop SSH service if it was not running before - if not ssh_running: - self.cli_delete(['service', 'ssh']) - def test_delete_current_user(self): current_user = get_current_user() diff --git a/smoketest/scripts/cli/test_system_sflow.py b/smoketest/scripts/cli/test_system_sflow.py index 74c065473..700253e2b 100755 --- a/smoketest/scripts/cli/test_system_sflow.py +++ b/smoketest/scripts/cli/test_system_sflow.py @@ -96,6 +96,39 @@ class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase): for interface in Section.interfaces('ethernet'): self.assertIn(f'pcap {{ dev={interface} }}', hsflowd) + def test_sflow_ipv6(self): + sampling_rate = '100' + default_polling = '30' + default_port = '6343' + sflow_server = { + '2001:db8::1': {}, + '2001:db8::2': {'port': '8023'}, + } + + for interface in Section.interfaces('ethernet'): + self.cli_set(base_path + ['interface', interface]) + + self.cli_set(base_path + ['sampling-rate', sampling_rate]) + for server, server_config in sflow_server.items(): + self.cli_set(base_path + ['server', server]) + if 'port' in server_config: + self.cli_set(base_path + ['server', server, 'port', server_config['port']]) + + # commit changes + self.cli_commit() + + # verify configuration + hsflowd = read_file(hsflowd_conf) + + self.assertIn(f'sampling={sampling_rate}', hsflowd) + self.assertIn(f'polling={default_polling}', hsflowd) + + for server, server_config in sflow_server.items(): + if 'port' in server_config: + self.assertIn(f'collector {{ ip = {server} udpport = {server_config["port"]} }}', hsflowd) + else: + self.assertIn(f'collector {{ ip = {server} udpport = {default_port} }}', hsflowd) + def test_vrf(self): interface = 'eth0' server = '192.0.2.1' diff --git a/smoketest/scripts/cli/test_vpn_ipsec.py b/smoketest/scripts/cli/test_vpn_ipsec.py index f2bea58d1..91a76e6f6 100755 --- a/smoketest/scripts/cli/test_vpn_ipsec.py +++ b/smoketest/scripts/cli/test_vpn_ipsec.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2024 VyOS maintainers and contributors +# Copyright (C) 2021-2025 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 @@ -353,24 +353,40 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): def test_dmvpn(self): - tunnel_if = 'tun100' - nhrp_secret = 'secret' ike_lifetime = '3600' esp_lifetime = '1800' + tunnel_if = "tun100" + tunnel_ip = '172.16.253.134/32' + tunnel_source = "192.0.2.134" + tunnel_encapsulation = "gre" + esp_group = "ESP-HUB" + ike_group = "IKE-HUB" + nhrp_secret = "vyos123" + nhrp_holdtime = '300' + nhs_tunnelip = '172.16.253.1' + nhs_nbmaip = '192.0.2.1' + map_tunnelip = '172.16.253.135' + map_nbmaip = "192.0.2.135" + nhrp_networkid = '1' + # Tunnel - self.cli_set(tunnel_path + [tunnel_if, 'address', '172.16.253.134/29']) - self.cli_set(tunnel_path + [tunnel_if, 'encapsulation', 'gre']) - self.cli_set(tunnel_path + [tunnel_if, 'source-address', '192.0.2.1']) - self.cli_set(tunnel_path + [tunnel_if, 'enable-multicast']) - self.cli_set(tunnel_path + [tunnel_if, 'parameters', 'ip', 'key', '1']) + self.cli_set(tunnel_path + [tunnel_if, "address", tunnel_ip]) + self.cli_set(tunnel_path + [tunnel_if, "encapsulation", tunnel_encapsulation]) + self.cli_set(tunnel_path + [tunnel_if, "source-address", tunnel_source]) + self.cli_set(tunnel_path + [tunnel_if, "enable-multicast"]) + self.cli_set(tunnel_path + [tunnel_if, "parameters", "ip", "key", "1"]) # NHRP - self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'cisco-authentication', nhrp_secret]) - self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'holding-time', '300']) - self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'multicast', 'dynamic']) - self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'redirect']) - self.cli_set(nhrp_path + ['tunnel', tunnel_if, 'shortcut']) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "authentication", nhrp_secret]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "holdtime", nhrp_holdtime]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "multicast", nhs_tunnelip]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "redirect"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "shortcut"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "registration-no-unique"]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "network-id", nhrp_networkid]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "nhs", "tunnel-ip", nhs_tunnelip, "nbma", nhs_nbmaip]) + self.cli_set(nhrp_path + ["tunnel", tunnel_if, "map", "tunnel-ip", map_tunnelip, "nbma", map_nbmaip]) # IKE/ESP Groups self.cli_set(base_path + ['esp-group', esp_group, 'lifetime', esp_lifetime]) @@ -399,11 +415,11 @@ class TestVPNIPsec(VyOSUnitTestSHIM.TestCase): swanctl_conf = read_file(swanctl_file) swanctl_lines = [ - f'proposals = aes128-sha1-modp1024,aes256-sha1-prfsha1-modp1024', + f'proposals = aes256-sha1-prfsha1-modp1024', f'version = 1', f'rekey_time = {ike_lifetime}s', f'rekey_time = {esp_lifetime}s', - f'esp_proposals = aes128-sha1-modp1024,aes256-sha1-modp1024,3des-md5-modp1024', + f'esp_proposals = aes256-sha1-modp1024,3des-md5-modp1024', f'local_ts = dynamic[gre]', f'remote_ts = dynamic[gre]', f'mode = transport', diff --git a/smoketest/scripts/system/test_kernel_options.py b/smoketest/scripts/system/test_kernel_options.py index 700e4cec7..b51b0be1d 100755 --- a/smoketest/scripts/system/test_kernel_options.py +++ b/smoketest/scripts/system/test_kernel_options.py @@ -128,5 +128,11 @@ class TestKernelModules(unittest.TestCase): tmp = re.findall(f'{option}=(y|m)', self._config_data) self.assertTrue(tmp) + def test_psample_enabled(self): + # Psample must be enabled in the OS Kernel to enable egress flow for hsflowd + for option in ['CONFIG_PSAMPLE']: + tmp = re.findall(f'{option}=y', self._config_data) + self.assertTrue(tmp) + if __name__ == '__main__': unittest.main(verbosity=2) |