diff options
Diffstat (limited to 'smoketest/scripts/cli')
| -rwxr-xr-x | smoketest/scripts/cli/test_firewall.py | 39 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_policy.py | 17 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_policy_route.py | 35 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_bgp.py | 15 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_service_ssh.py | 49 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_vrf.py | 24 |
6 files changed, 166 insertions, 13 deletions
diff --git a/smoketest/scripts/cli/test_firewall.py b/smoketest/scripts/cli/test_firewall.py index 37e6b9e7a..b8f944575 100755 --- a/smoketest/scripts/cli/test_firewall.py +++ b/smoketest/scripts/cli/test_firewall.py @@ -173,6 +173,45 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): nftables_output = cmd(f'sudo nft list chain {table} {chain}') self.assertTrue('jump VYOS_STATE_POLICY' in nftables_output) + def test_state_and_status_rules(self): + self.cli_set(['firewall', 'name', 'smoketest', 'default-action', 'drop']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'action', 'accept']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'state', 'established', 'enable']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '1', 'state', 'related', 'enable']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'action', 'reject']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '2', 'state', 'invalid', 'enable']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'action', 'accept']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'state', 'new', 'enable']) + + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '3', 'connection-status', 'nat', 'destination']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'action', 'accept']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'state', 'new', 'enable']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'state', 'established', 'enable']) + self.cli_set(['firewall', 'name', 'smoketest', 'rule', '4', 'connection-status', 'nat', 'source']) + + self.cli_set(['interfaces', 'ethernet', 'eth0', 'firewall', 'in', 'name', 'smoketest']) + + self.cli_commit() + + nftables_search = [ + ['iifname "eth0"', 'jump NAME_smoketest'], + ['ct state { established, related }', 'return'], + ['ct state { invalid }', 'reject'], + ['ct state { new }', 'ct status { dnat }', 'return'], + ['ct state { established, new }', 'ct status { snat }', 'return'], + ['smoketest default-action', 'drop'] + ] + + nftables_output = cmd('sudo nft list table ip filter') + + for search in nftables_search: + matched = False + for line in nftables_output.split("\n"): + if all(item in line for item in search): + matched = True + break + self.assertTrue(matched, msg=search) + def test_sysfs(self): for name, conf in sysfs_config.items(): paths = glob(conf['sysfs']) diff --git a/smoketest/scripts/cli/test_policy.py b/smoketest/scripts/cli/test_policy.py index 3e8dd35ae..e8c6ff19b 100755 --- a/smoketest/scripts/cli/test_policy.py +++ b/smoketest/scripts/cli/test_policy.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021 VyOS maintainers and contributors +# Copyright (C) 2021-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 @@ -849,6 +849,13 @@ class TestPolicy(VyOSUnitTestSHIM.TestCase): 'evpn-vni' : '1234', }, }, + '20' : { + 'action' : 'permit', + 'set' : { + 'evpn-gateway-ipv4' : '192.0.2.99', + 'evpn-gateway-ipv6' : '2001:db8:f00::1', + }, + }, }, }, } @@ -996,6 +1003,10 @@ class TestPolicy(VyOSUnitTestSHIM.TestCase): self.cli_set(path + ['rule', rule, 'set', 'tag', rule_config['set']['tag']]) if 'weight' in rule_config['set']: self.cli_set(path + ['rule', rule, 'set', 'weight', rule_config['set']['weight']]) + if 'evpn-gateway-ipv4' in rule_config['set']: + self.cli_set(path + ['rule', rule, 'set', 'evpn', 'gateway', 'ipv4', rule_config['set']['evpn-gateway-ipv4']]) + if 'evpn-gateway-ipv6' in rule_config['set']: + self.cli_set(path + ['rule', rule, 'set', 'evpn', 'gateway', 'ipv6', rule_config['set']['evpn-gateway-ipv6']]) self.cli_commit() @@ -1155,6 +1166,10 @@ class TestPolicy(VyOSUnitTestSHIM.TestCase): tmp += 'tag ' + rule_config['set']['tag'] elif 'weight' in rule_config['set']: tmp += 'weight ' + rule_config['set']['weight'] + elif 'vpn-gateway-ipv4' in rule_config['set']: + tmp += 'evpn gateway ipv4 ' + rule_config['set']['vpn-gateway-ipv4'] + elif 'vpn-gateway-ipv6' in rule_config['set']: + tmp += 'evpn gateway ipv6 ' + rule_config['set']['vpn-gateway-ipv6'] self.assertIn(tmp, config) diff --git a/smoketest/scripts/cli/test_policy_route.py b/smoketest/scripts/cli/test_policy_route.py index 9035f0832..e2d70f289 100755 --- a/smoketest/scripts/cli/test_policy_route.py +++ b/smoketest/scripts/cli/test_policy_route.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021 VyOS maintainers and contributors +# Copyright (C) 2021-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,15 +23,26 @@ from vyos.util import cmd mark = '100' table_mark_offset = 0x7fffffff table_id = '101' +interface = 'eth0' +interface_ip = '172.16.10.1/24' class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): - def setUp(self): - self.cli_set(['interfaces', 'ethernet', 'eth0', 'address', '172.16.10.1/24']) - self.cli_set(['protocols', 'static', 'table', '101', 'route', '0.0.0.0/0', 'interface', 'eth0']) + @classmethod + def setUpClass(cls): + super(TestPolicyRoute, cls).setUpClass() + + cls.cli_set(cls, ['interfaces', 'ethernet', interface, 'address', interface_ip]) + cls.cli_set(cls, ['protocols', 'static', 'table', table_id, 'route', '0.0.0.0/0', 'interface', interface]) + + @classmethod + def tearDownClass(cls): + cls.cli_delete(cls, ['interfaces', 'ethernet', interface, 'address', interface_ip]) + cls.cli_delete(cls, ['protocols', 'static', 'table', table_id]) + + super(TestPolicyRoute, cls).tearDownClass() def tearDown(self): - self.cli_delete(['interfaces', 'ethernet', 'eth0']) - self.cli_delete(['protocols', 'static']) + self.cli_delete(['interfaces', 'ethernet', interface, 'policy']) self.cli_delete(['policy', 'route']) self.cli_delete(['policy', 'route6']) self.cli_commit() @@ -41,14 +52,14 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'destination', 'address', '172.16.10.10']) self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'set', 'mark', mark]) - self.cli_set(['interfaces', 'ethernet', 'eth0', 'policy', 'route', 'smoketest']) + self.cli_set(['interfaces', 'ethernet', interface, 'policy', 'route', 'smoketest']) self.cli_commit() mark_hex = "{0:#010x}".format(int(mark)) nftables_search = [ - ['iifname "eth0"', 'jump VYOS_PBR_smoketest'], + [f'iifname "{interface}"','jump VYOS_PBR_smoketest'], ['ip daddr 172.16.10.10', 'ip saddr 172.16.20.10', 'meta mark set ' + mark_hex], ] @@ -72,8 +83,8 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'destination', 'port', '8888']) self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'set', 'table', table_id]) - self.cli_set(['interfaces', 'ethernet', 'eth0', 'policy', 'route', 'smoketest']) - self.cli_set(['interfaces', 'ethernet', 'eth0', 'policy', 'route6', 'smoketest6']) + self.cli_set(['interfaces', 'ethernet', interface, 'policy', 'route', 'smoketest']) + self.cli_set(['interfaces', 'ethernet', interface, 'policy', 'route6', 'smoketest6']) self.cli_commit() @@ -82,7 +93,7 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): # IPv4 nftables_search = [ - ['iifname "eth0"', 'jump VYOS_PBR_smoketest'], + [f'iifname "{interface}"', 'jump VYOS_PBR_smoketest'], ['tcp flags & (syn | ack) == syn', 'tcp dport { 8888 }', 'meta mark set ' + mark_hex] ] @@ -99,7 +110,7 @@ class TestPolicyRoute(VyOSUnitTestSHIM.TestCase): # IPv6 nftables6_search = [ - ['iifname "eth0"', 'jump VYOS_PBR6_smoketest'], + [f'iifname "{interface}"', 'jump VYOS_PBR6_smoketest'], ['meta l4proto { tcp, udp }', 'th dport { 8888 }', 'meta mark set ' + mark_hex] ] diff --git a/smoketest/scripts/cli/test_protocols_bgp.py b/smoketest/scripts/cli/test_protocols_bgp.py index 6f92457b2..9c0c93779 100755 --- a/smoketest/scripts/cli/test_protocols_bgp.py +++ b/smoketest/scripts/cli/test_protocols_bgp.py @@ -887,6 +887,7 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): remote_asn = str(int(ASN) + 150) neighbor = '192.0.2.1' peer_group = 'bar' + interface = 'eth0' self.cli_set(base_path + ['local-as', ASN]) self.cli_set(base_path + ['neighbor', neighbor, 'remote-as', remote_asn]) @@ -898,6 +899,20 @@ class TestProtocolsBGP(VyOSUnitTestSHIM.TestCase): self.cli_commit() self.cli_delete(base_path + ['neighbor', neighbor, 'remote-as']) + # re-test with interface based peer-group + self.cli_set(base_path + ['neighbor', interface, 'interface', 'peer-group', peer_group]) + self.cli_set(base_path + ['neighbor', interface, 'interface', 'remote-as', 'external']) + with self.assertRaises(ConfigSessionError): + self.cli_commit() + self.cli_delete(base_path + ['neighbor', interface, 'interface', 'remote-as']) + + # re-test with interface based v6only peer-group + self.cli_set(base_path + ['neighbor', interface, 'interface', 'v6only', 'peer-group', peer_group]) + self.cli_set(base_path + ['neighbor', interface, 'interface', 'v6only', 'remote-as', 'external']) + with self.assertRaises(ConfigSessionError): + self.cli_commit() + self.cli_delete(base_path + ['neighbor', interface, 'interface', 'v6only', 'remote-as']) + self.cli_commit() frrconfig = self.getFRRconfig(f'router bgp {ASN}') diff --git a/smoketest/scripts/cli/test_service_ssh.py b/smoketest/scripts/cli/test_service_ssh.py index 77ad5bc0d..0b029dd00 100755 --- a/smoketest/scripts/cli/test_service_ssh.py +++ b/smoketest/scripts/cli/test_service_ssh.py @@ -213,5 +213,54 @@ class TestServiceSSH(VyOSUnitTestSHIM.TestCase): usernames = [x[0] for x in getpwall()] self.assertNotIn(test_user, usernames) + def test_ssh_dynamic_protection(self): + # check sshguard service + + SSHGUARD_CONFIG = '/etc/sshguard/sshguard.conf' + SSHGUARD_WHITELIST = '/etc/sshguard/whitelist' + SSHGUARD_PROCESS = 'sshguard' + block_time = '123' + detect_time = '1804' + port = '22' + threshold = '10' + allow_list = ['192.0.2.0/24', '2001:db8::/48'] + + self.cli_set(base_path + ['dynamic-protection', 'block-time', block_time]) + self.cli_set(base_path + ['dynamic-protection', 'detect-time', detect_time]) + self.cli_set(base_path + ['dynamic-protection', 'threshold', threshold]) + for allow in allow_list: + self.cli_set(base_path + ['dynamic-protection', 'allow-from', allow]) + + # commit changes + self.cli_commit() + + # Check configured port + tmp = get_config_value('Port') + self.assertIn(port, tmp) + + # Check sshgurad service + self.assertTrue(process_named_running(SSHGUARD_PROCESS)) + + sshguard_lines = [ + f'THRESHOLD={threshold}', + f'BLOCK_TIME={block_time}', + f'DETECTION_TIME={detect_time}' + ] + + tmp_sshguard_conf = read_file(SSHGUARD_CONFIG) + for line in sshguard_lines: + self.assertIn(line, tmp_sshguard_conf) + + tmp_whitelist_conf = read_file(SSHGUARD_WHITELIST) + for allow in allow_list: + self.assertIn(allow, tmp_whitelist_conf) + + # Delete service ssh dynamic-protection + # but not service ssh itself + self.cli_delete(base_path + ['dynamic-protection']) + self.cli_commit() + + self.assertFalse(process_named_running(SSHGUARD_PROCESS)) + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_vrf.py b/smoketest/scripts/cli/test_vrf.py index ff18f7261..176c095fb 100755 --- a/smoketest/scripts/cli/test_vrf.py +++ b/smoketest/scripts/cli/test_vrf.py @@ -127,6 +127,9 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): for vrf in vrfs: # Ensure VRF was created self.assertIn(vrf, interfaces()) + # Verify IP forwarding is 1 (enabled) + self.assertEqual(read_file(f'/proc/sys/net/ipv4/conf/{vrf}/forwarding'), '1') + self.assertEqual(read_file(f'/proc/sys/net/ipv6/conf/{vrf}/forwarding'), '1') # Test for proper loopback IP assignment for addr in loopbacks: self.assertTrue(is_intf_addr_assigned(vrf, addr)) @@ -267,5 +270,26 @@ class VRFTest(VyOSUnitTestSHIM.TestCase): self.cli_delete(['interfaces', 'dummy', interface]) self.cli_commit() + def test_vrf_disable_forwarding(self): + table = '2000' + for vrf in vrfs: + base = base_path + ['name', vrf] + self.cli_set(base + ['table', table]) + self.cli_set(base + ['ip', 'disable-forwarding']) + self.cli_set(base + ['ipv6', 'disable-forwarding']) + table = str(int(table) + 1) + + # commit changes + self.cli_commit() + + # Verify VRF configuration + loopbacks = ['127.0.0.1', '::1'] + for vrf in vrfs: + # Ensure VRF was created + self.assertIn(vrf, interfaces()) + # Verify IP forwarding is 0 (disabled) + self.assertEqual(read_file(f'/proc/sys/net/ipv4/conf/{vrf}/forwarding'), '0') + self.assertEqual(read_file(f'/proc/sys/net/ipv6/conf/{vrf}/forwarding'), '0') + if __name__ == '__main__': unittest.main(verbosity=2) |