#!/usr/bin/env python3 # # Copyright (C) 2020 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 . import unittest import json from vyos.configsession import ConfigSession from vyos.configsession import ConfigSessionError from vyos.util import cmd from base_interfaces_test import BasicInterfaceTest remote_ip4 = '192.0.2.100' remote_ip6 = '2001:db8::ffff' source_if = 'dum2222' mtu = 1476 def tunnel_conf(interface): tmp = cmd(f'ip -d -j link show {interface}') # {'address': '2.2.2.2', # 'broadcast': '192.0.2.10', # 'flags': ['POINTOPOINT', 'NOARP', 'UP', 'LOWER_UP'], # 'group': 'default', # 'gso_max_segs': 65535, # 'gso_max_size': 65536, # 'ifindex': 10, # 'ifname': 'tun10', # 'inet6_addr_gen_mode': 'none', # 'link': None, # 'link_pointtopoint': True, # 'link_type': 'gre', # 'linkinfo': {'info_data': {'local': '2.2.2.2', # 'pmtudisc': True, # 'remote': '192.0.2.10', # 'tos': '0x1', # 'ttl': 255}, # 'info_kind': 'gre'}, # 'linkmode': 'DEFAULT', # 'max_mtu': 65511, # 'min_mtu': 68, # 'mtu': 1476, # 'num_rx_queues': 1, # 'num_tx_queues': 1, # 'operstate': 'UNKNOWN', # 'promiscuity': 0, # 'qdisc': 'noqueue', # 'txqlen': 1000} return json.loads(tmp)[0] class TunnelInterfaceTest(BasicInterfaceTest.BaseTest): def setUp(self): super().setUp() self._base_path = ['interfaces', 'tunnel'] self._test_mtu = True self.local_v4 = '192.0.2.1' self.local_v6 = '2001:db8::1' self.session.set(['interfaces', 'dummy', source_if, 'address', self.local_v4 + '/32']) self.session.set(['interfaces', 'dummy', source_if, 'address', self.local_v6 + '/128']) self._options = { 'tun10': ['encapsulation ipip', 'remote-ip 192.0.2.10', 'local-ip ' + self.local_v4], 'tun20': ['encapsulation gre', 'remote-ip 192.0.2.20', 'local-ip ' + self.local_v4], } self._interfaces = list(self._options) def tearDown(self): self.session.delete(['interfaces', 'dummy', source_if]) super().tearDown() def test_ipip(self): interface = 'tun100' encapsulation = 'ipip' local_if_addr = '10.10.10.1/24' self.session.set(self._base_path + [interface, 'address', local_if_addr]) # Must provide an "encapsulation" for tunnel tun10 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) # Must configure either local-ip or dhcp-interface for tunnel ipip tun100 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v4]) # missing required option remote for ipip with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4]) # Configure Tunnel Source interface self.session.set(self._base_path + [interface, 'source-interface', source_if]) self.session.commit() conf = tunnel_conf(interface) self.assertEqual(interface, conf['ifname']) self.assertEqual(encapsulation, conf['link_type']) self.assertEqual(mtu, conf['mtu']) self.assertEqual(source_if, conf['link']) self.assertEqual(self.local_v4, conf['linkinfo']['info_data']['local']) self.assertEqual(remote_ip4, conf['linkinfo']['info_data']['remote']) def test_ipip6(self): interface = 'tun110' encapsulation = 'ipip6' local_if_addr = '10.10.10.1/24' self.session.set(self._base_path + [interface, 'address', local_if_addr]) # Must provide an "encapsulation" for tunnel tun10 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) # Must configure either local-ip or dhcp-interface for tunnel ipip tun100 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v6]) # missing required option remote for ipip with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6]) # Configure Tunnel Source interface self.session.set(self._base_path + [interface, 'source-interface', source_if]) self.session.commit() conf = tunnel_conf(interface) self.assertEqual(interface, conf['ifname']) self.assertEqual('tunnel6', conf['link_type']) self.assertEqual(mtu, conf['mtu']) self.assertEqual(source_if, conf['link']) self.assertEqual(self.local_v6, conf['linkinfo']['info_data']['local']) self.assertEqual(remote_ip6, conf['linkinfo']['info_data']['remote']) def test_verify_ipv4_local_remote_addr(self): """ When running tests ensure that for certain encapsulation types the local and remote IP address is actually an IPv4 address """ interface = f'tun1000' local_if_addr = f'10.10.200.1/24' for encapsulation in ['ipip', 'sit', 'gre']: self.session.set(self._base_path + [interface, 'address', local_if_addr]) self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) self.session.set(self._base_path + [interface, 'local-ip', self.local_v6]) self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6]) # Encapsulation mode requires IPv4 local-ip with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v4]) # Encapsulation mode requires IPv4 local-ip with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4]) # Check if commit is ok self.session.commit() # cleanup this instance self.session.delete(self._base_path + [interface]) self.session.commit() def test_verify_ipv6_local_remote_addr(self): """ When running tests ensure that for certain encapsulation types the local and remote IP address is actually an IPv6 address """ interface = f'tun1010' local_if_addr = f'10.10.200.1/24' for encapsulation in ['ipip6', 'ip6ip6', 'ip6gre']: self.session.set(self._base_path + [interface, 'address', local_if_addr]) self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) self.session.set(self._base_path + [interface, 'local-ip', self.local_v4]) self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4]) # Encapsulation mode requires IPv6 local-ip with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v6]) # Encapsulation mode requires IPv6 local-ip with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6]) # Check if commit is ok self.session.commit() # cleanup this instance self.session.delete(self._base_path + [interface]) self.session.commit() def test_verify_local_dhcp(self): """ We can not use local-ip and dhcp-interface at the same time """ interface = f'tun1020' local_if_addr = f'10.0.0.1/24' self.session.set(self._base_path + [interface, 'address', local_if_addr]) self.session.set(self._base_path + [interface, 'encapsulation', 'gre']) self.session.set(self._base_path + [interface, 'local-ip', self.local_v4]) self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4]) self.session.set(self._base_path + [interface, 'dhcp-interface', 'eth0']) # local-ip and dhcp-interface can not be used at the same time with self.assertRaises(ConfigSessionError): self.session.commit() self.session.delete(self._base_path + [interface, 'dhcp-interface']) # Check if commit is ok self.session.commit() def test_ip6ip6(self): interface = 'tun120' encapsulation = 'ip6ip6' local_if_addr = '2001:db8:f00::1/24' self.session.set(self._base_path + [interface, 'address', local_if_addr]) # Must provide an "encapsulation" for tunnel tun10 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) # Must configure either local-ip or dhcp-interface for tunnel ipip tun100 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v6]) # missing required option remote for ipip with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6]) # Configure Tunnel Source interface self.session.set(self._base_path + [interface, 'source-interface', source_if]) self.session.commit() conf = tunnel_conf(interface) self.assertEqual(interface, conf['ifname']) self.assertEqual('tunnel6', conf['link_type']) self.assertEqual(mtu, conf['mtu']) self.assertEqual(source_if, conf['link']) self.assertEqual(self.local_v6, conf['linkinfo']['info_data']['local']) self.assertEqual(remote_ip6, conf['linkinfo']['info_data']['remote']) def test_gre_ipv4(self): interface = 'tun200' encapsulation = 'gre' local_if_addr = '172.16.1.1/24' self.session.set(self._base_path + [interface, 'address', local_if_addr]) # Must provide an "encapsulation" for tunnel tun10 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) # Must configure either local-ip or dhcp-interface with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v4]) # No assertion is raised for GRE remote-ip when missing self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4]) # Configure Tunnel Source interface self.session.set(self._base_path + [interface, 'source-interface', source_if]) self.session.commit() conf = tunnel_conf(interface) self.assertEqual(interface, conf['ifname']) self.assertEqual(encapsulation, conf['link_type']) self.assertEqual(mtu, conf['mtu']) self.assertEqual(source_if, conf['link']) self.assertEqual(self.local_v4, conf['linkinfo']['info_data']['local']) self.assertEqual(remote_ip4, conf['linkinfo']['info_data']['remote']) def test_gre_ipv6(self): interface = 'tun210' encapsulation = 'ip6gre' local_if_addr = '2001:db8:f01::1/24' self.session.set(self._base_path + [interface, 'address', local_if_addr]) # Must provide an "encapsulation" for tunnel tun10 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) # Must configure either local-ip or dhcp-interface with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v6]) # No assertion is raised for GRE remote-ip when missing self.session.set(self._base_path + [interface, 'remote-ip', remote_ip6]) # Configure Tunnel Source interface self.session.set(self._base_path + [interface, 'source-interface', source_if]) self.session.commit() conf = tunnel_conf(interface) self.assertEqual(interface, conf['ifname']) self.assertEqual(encapsulation, conf['link_type']) self.assertEqual(mtu, conf['mtu']) self.assertEqual(source_if, conf['link']) self.assertEqual(self.local_v6, conf['linkinfo']['info_data']['local']) self.assertEqual(remote_ip6, conf['linkinfo']['info_data']['remote']) def test_sit(self): interface = 'tun300' encapsulation = 'sit' local_if_addr = '172.16.2.1/24' self.session.set(self._base_path + [interface, 'address', local_if_addr]) # Must provide an "encapsulation" for tunnel tun10 with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'encapsulation', encapsulation]) # Must configure either local-ip or dhcp-interface with self.assertRaises(ConfigSessionError): self.session.commit() self.session.set(self._base_path + [interface, 'local-ip', self.local_v4]) # No assertion is raised for GRE remote-ip when missing self.session.set(self._base_path + [interface, 'remote-ip', remote_ip4]) # Source interface can not be used with sit self.session.set(self._base_path + [interface, 'source-interface', source_if]) with self.assertRaises(ConfigSessionError): self.session.commit() self.session.delete(self._base_path + [interface, 'source-interface']) self.session.commit() conf = tunnel_conf(interface) self.assertEqual(interface, conf['ifname']) self.assertEqual(encapsulation, conf['link_type']) self.assertEqual(mtu, conf['mtu']) self.assertEqual(self.local_v4, conf['linkinfo']['info_data']['local']) self.assertEqual(remote_ip4, conf['linkinfo']['info_data']['remote']) if __name__ == '__main__': unittest.main()