diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/test_ipip.py | 248 | ||||
| -rw-r--r-- | test/test_sixrd.py | 48 | ||||
| -rw-r--r-- | test/vpp_papi_provider.py | 28 |
3 files changed, 299 insertions, 25 deletions
diff --git a/test/test_ipip.py b/test/test_ipip.py new file mode 100644 index 00000000000..1ae4de0adf8 --- /dev/null +++ b/test/test_ipip.py @@ -0,0 +1,248 @@ +# +# IP{4,6} over IP{v,6} tunnel functional tests +# + +import unittest +from scapy.layers.inet import IP, UDP, ICMP +from scapy.layers.inet6 import IPv6 +from scapy.layers.l2 import Ether, GRE +from scapy.packet import Raw + +from framework import VppTestCase +from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto +from util import ppp +from ipaddress import * + +""" Testipip is a subclass of VPPTestCase classes. + +IPIP tests. + +""" + + +class TestIPIP(VppTestCase): + """ IPIP Test Case """ + + @classmethod + def setUpClass(cls): + super(TestIPIP, cls).setUpClass() + try: + cls.create_pg_interfaces(range(2)) + cls.interfaces = list(cls.pg_interfaces) + except Exception: + super(TestIPIP, cls).tearDownClass() + raise + + def setUp(cls): + super(TestIPIP, cls).setUp() + try: + for i in cls.interfaces: + i.admin_up() + i.config_ip4() + i.config_ip6() + i.disable_ipv6_ra() + i.resolve_arp() + i.resolve_ndp() + except Exception: + super(TestIPIP, cls).tearDown() + raise + + def tearDown(self): + super(TestIPIP, self).tearDown() + if not self.vpp_dead: + self.vapi.cli("show hardware") + for i in self.pg_interfaces: + i.unconfig_ip4() + i.unconfig_ip6() + i.admin_down() + self.vapi.cli("show error") + + def validate(self, rx, expected): + expected = expected.__class__(str(expected)) + if rx != expected: + print('RX packet:') + print(rx.show()) + print('EXPECTED packet:') + print(expected.show()) + self.assertDictEqual(rx, expected) + + def payload(self, len): + return 'x' * len + + def test_ipip4(self): + """ ip{v4,v6} over ip4 test """ + p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP') + p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1") + p_payload = UDP(sport=1234, dport=1234) + + # IPv4 transport + rv = self.vapi.ipip_add_tunnel( + src_address=str(ip_address(self.pg0.local_ip4).packed), + dst_address=str(ip_address(self.pg1.remote_ip4).packed), + is_ipv6=0) + self.assertEqual(rv.retval, 0) + sw_if_index = rv.sw_if_index + + # Set interface up and enable IP on it + rv = self.vapi.sw_interface_set_flags(sw_if_index, 1) + self.assertEqual(rv.retval, 0) + rv = self.vapi.sw_interface_set_unnumbered( + ip_sw_if_index=self.pg0.sw_if_index, + sw_if_index=sw_if_index) + self.assertEqual(rv.retval, 0) + + # Add IPv4 and IPv6 routes via tunnel interface + ip4_via_tunnel = VppIpRoute( + self, "130.67.0.0", 16, + [VppRoutePath("0.0.0.0", + sw_if_index, + proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0) + ip4_via_tunnel.add_vpp_config() + + ip6_via_tunnel = VppIpRoute( + self, "dead::", 16, + [VppRoutePath("::", + sw_if_index, + proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1) + ip6_via_tunnel.add_vpp_config() + + # IPv6 in to IPv4 tunnel + p6 = (p_ether / p_ip6 / p_payload) + p_inner_ip6 = p_ip6 + p_inner_ip6.hlim -= 1 + p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4, + proto='ipv6', id=0) / p_inner_ip6 / p_payload) + p6_reply.ttl -= 1 + rx = self.send_and_expect(self.pg0, p6*10, self.pg1) + for p in rx: + self.validate(p[1], p6_reply) + + # IPv4 in to IPv4 tunnel + p4 = (p_ether / p_ip4 / p_payload) + p_ip4_inner = p_ip4 + p_ip4_inner.ttl -= 1 + p4_reply = (IP(src=self.pg0.local_ip4, + dst=self.pg1.remote_ip4) / p_ip4_inner / p_payload) + p4_reply.ttl -= 1 + p4_reply.id = 0 + rx = self.send_and_expect(self.pg0, p4*10, self.pg1) + for p in rx: + self.validate(p[1], p4_reply) + + # Decapsulation + p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + + # IPv4 tunnel to IPv4 + p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4) + p4 = (p_ether / IP(src=self.pg1.remote_ip4, + dst=self.pg0.local_ip4) / p_ip4 / p_payload) + p4_reply = (p_ip4 / p_payload) + p4_reply.ttl -= 1 + rx = self.send_and_expect(self.pg1, p4*10, self.pg0) + for p in rx: + self.validate(p[1], p4_reply) + + # IPv4 tunnel to IPv6 + p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6) + p6 = (p_ether / IP(src=self.pg1.remote_ip4, + dst=self.pg0.local_ip4) / p_ip6 / p_payload) + p6_reply = (p_ip6 / p_payload) + p6_reply.hlim = 63 + rx = self.send_and_expect(self.pg1, p6*10, self.pg0) + for p in rx: + self.validate(p[1], p6_reply) + + def test_ipip6(self): + """ ip{v4,v6} over ip6 test """ + p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP') + p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1") + p_payload = UDP(sport=1234, dport=1234) + + # IPv6 transport + rv = self.vapi.ipip_add_tunnel( + src_address=str(ip_address(self.pg0.local_ip6).packed), + dst_address=str(ip_address(self.pg1.remote_ip6).packed)) + self.assertEqual(rv.retval, 0) + + sw_if_index = rv.sw_if_index + + rv = self.vapi.sw_interface_set_flags(sw_if_index, 1) + self.assertEqual(rv.retval, 0) + rv = self.vapi.sw_interface_set_unnumbered( + ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index) + self.assertEqual(rv.retval, 0) + + # Add IPv4 and IPv6 routes via tunnel interface + ip4_via_tunnel = VppIpRoute( + self, "130.67.0.0", 16, + [VppRoutePath("0.0.0.0", + sw_if_index, + proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0) + ip4_via_tunnel.add_vpp_config() + + ip6_via_tunnel = VppIpRoute( + self, "dead::", 16, + [VppRoutePath("::", + sw_if_index, + proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1) + ip6_via_tunnel.add_vpp_config() + + # Encapsulation + + # IPv6 in to IPv6 tunnel + p6 = (p_ether / p_ip6 / p_payload) + p6_reply = (IPv6(src=self.pg0.local_ip6, + dst=self.pg1.remote_ip6, hlim=63) / p_ip6 / p_payload) + p6_reply[1].hlim -= 1 + rx = self.send_and_expect(self.pg0, p6*10, self.pg1) + for p in rx: + self.validate(p[1], p6_reply) + + # IPv4 in to IPv6 tunnel + p4 = (p_ether / p_ip4 / p_payload) + p4_reply = (IPv6(src=self.pg0.local_ip6, + dst=self.pg1.remote_ip6, hlim=63) / p_ip4 / p_payload) + p4_reply[1].ttl -= 1 + rx = self.send_and_expect(self.pg0, p4*10, self.pg1) + for p in rx: + self.validate(p[1], p4_reply) + + # Decapsulation + + p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) + + # IPv6 tunnel to IPv4 + p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4) + p4 = (p_ether / IPv6(src=self.pg1.remote_ip6, + dst=self.pg0.local_ip6) / p_ip4 / p_payload) + p4_reply = (p_ip4 / p_payload) + p4_reply.ttl -= 1 + rx = self.send_and_expect(self.pg1, p4*10, self.pg0) + for p in rx: + self.validate(p[1], p4_reply) + + # IPv6 tunnel to IPv6 + p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6) + p6 = (p_ether / IPv6(src=self.pg1.remote_ip6, + dst=self.pg0.local_ip6) / p_ip6 / p_payload) + p6_reply = (p_ip6 / p_payload) + p6_reply.hlim = 63 + rx = self.send_and_expect(self.pg1, p6*10, self.pg0) + for p in rx: + self.validate(p[1], p6_reply) + + def test_ipip_create(self): + """ ipip create / delete interface test """ + rv = self.vapi.ipip_add_tunnel( + src_address=str(ip_address('1.2.3.4').packed), + dst_address=str(ip_address('2.3.4.5').packed), is_ipv6=0) + self.assertEqual(rv.retval, 0) + sw_if_index = rv.sw_if_index + rv = self.vapi.ipip_del_tunnel(sw_if_index) + self.assertEqual(rv.retval, 0) + + +if __name__ == '__main__': + unittest.main(testRunner=VppTestRunner) diff --git a/test/test_sixrd.py b/test/test_sixrd.py index 32ebdf4cca0..ac5061474d5 100644 --- a/test/test_sixrd.py +++ b/test/test_sixrd.py @@ -57,9 +57,11 @@ class Test6RD(VppTestCase): i.admin_down() if type(self.tunnel_index) is list: for sw_if_index in self.tunnel_index: - self.vapi.sixrd_del_tunnel(sw_if_index) + rv = self.vapi.ipip_6rd_del_tunnel(sw_if_index) + self.assertEqual(rv.retval, 0) else: - self.vapi.sixrd_del_tunnel(self.tunnel_index) + rv = self.vapi.ipip_6rd_del_tunnel(self.tunnel_index) + self.assertEqual(rv.retval, 0) self.vapi.cli("show error") def validate_6in4(self, rx, expected): @@ -92,13 +94,14 @@ class Test6RD(VppTestCase): p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) p_ip6 = IPv6(src="1::1", dst="2002:AC10:0202::1", nh='UDP') - rv = self.vapi.sixrd_add_tunnel( + rv = self.vapi.ipip_6rd_add_tunnel( 0, str(ip_address('2002::').packed), 16, str(ip_address('0.0.0.0').packed), 0, - str(ip_address(self.pg0.local_ip4).packed), 0, True) + str(ip_address(self.pg0.local_ip4).packed), True) self.assertEqual(rv.retval, 0) self.tunnel_index = rv.sw_if_index + self.vapi.cli("show ip6 fib") p_payload = UDP(sport=1234, dport=1234) p = (p_ether / p_ip6 / p_payload) @@ -124,13 +127,21 @@ class Test6RD(VppTestCase): def test_6rd_ip4_to_ip6(self): """ ip4 -> ip6 (decap) 6rd test """ - rv = self.vapi.sixrd_add_tunnel( + rv = self.vapi.ipip_6rd_add_tunnel( 0, str(ip_address('2002::').packed), 16, str(ip_address('0.0.0.0').packed), - 0, str(ip_address(self.pg0.local_ip4).packed), 0, True) + 0, str(ip_address(self.pg0.local_ip4).packed), True) self.assertEqual(rv.retval, 0) self.tunnel_index = rv.sw_if_index - self.vapi.cli("show ip6 fib") + rv = self.vapi.ipip_6rd_del_tunnel(rv.sw_if_index) + self.assertEqual(rv.retval, 0) + rv = self.vapi.ipip_6rd_add_tunnel( + 0, str(ip_address('2002::').packed), + 16, str(ip_address('0.0.0.0').packed), + 0, str(ip_address(self.pg0.local_ip4).packed), True) + self.tunnel_index = rv.sw_if_index + self.assertEqual(rv.retval, 0) + p_ip6 = (IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(sport=1234, dport=1234)) @@ -149,18 +160,18 @@ class Test6RD(VppTestCase): """ ip4 -> ip6 (decap) 6rd test """ self.tunnel_index = [] - rv = self.vapi.sixrd_add_tunnel( + rv = self.vapi.ipip_6rd_add_tunnel( 0, str(ip_address('2002::').packed), 16, str(ip_address('0.0.0.0').packed), - 0, str(ip_address(self.pg0.local_ip4).packed), 0, True) + 0, str(ip_address(self.pg0.local_ip4).packed), True) self.assertEqual(rv.retval, 0) self.tunnel_index.append(rv.sw_if_index) - rv = self.vapi.sixrd_add_tunnel( + + rv = self.vapi.ipip_6rd_add_tunnel( 0, str(ip_address('2003::').packed), 16, str(ip_address('0.0.0.0').packed), - 0, str(ip_address(self.pg1.local_ip4).packed), 0, True) + 0, str(ip_address(self.pg1.local_ip4).packed), True) self.assertEqual(rv.retval, 0) - self.tunnel_index.append(rv.sw_if_index) self.vapi.cli("show ip6 fib") @@ -184,10 +195,10 @@ class Test6RD(VppTestCase): def test_6rd_ip4_to_ip6_suffix(self): """ ip4 -> ip6 (decap) 6rd test """ - rv = self.vapi.sixrd_add_tunnel( + rv = self.vapi.ipip_6rd_add_tunnel( 0, str(ip_address('2002::').packed), 16, str(ip_address('172.0.0.0').packed), 8, - str(ip_address(self.pg0.local_ip4).packed), 0, True) + str(ip_address(self.pg0.local_ip4).packed), True) self.assertEqual(rv.retval, 0) self.tunnel_index = rv.sw_if_index @@ -206,12 +217,13 @@ class Test6RD(VppTestCase): def test_6rd_ip4_to_ip6_sec_check(self): """ ip4 -> ip6 (decap) security check 6rd test """ - rv = self.vapi.sixrd_add_tunnel( + rv = self.vapi.ipip_6rd_add_tunnel( 0, str(ip_address('2002::').packed), 16, str(ip_address('0.0.0.0').packed), - 0, str(ip_address(self.pg0.local_ip4).packed), 0, True) + 0, str(ip_address(self.pg0.local_ip4).packed), True) self.assertEqual(rv.retval, 0) self.tunnel_index = rv.sw_if_index + self.vapi.cli("show ip6 fib") p_ip6 = (IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(sport=1234, dport=1234)) @@ -238,10 +250,10 @@ class Test6RD(VppTestCase): def test_6rd_bgp_tunnel(self): """ 6rd BGP tunnel """ - rv = self.vapi.sixrd_add_tunnel( + rv = self.vapi.ipip_6rd_add_tunnel( 0, str(ip_address('2002::').packed), 16, str(ip_address('0.0.0.0').packed), - 0, str(ip_address(self.pg0.local_ip4).packed), 0, False) + 0, str(ip_address(self.pg0.local_ip4).packed), False) self.assertEqual(rv.retval, 0) self.tunnel_index = rv.sw_if_index diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py index 18bb1f60d40..22c6a0d4597 100644 --- a/test/vpp_papi_provider.py +++ b/test/vpp_papi_provider.py @@ -3202,21 +3202,35 @@ class VppPapiProvider(object): """ GBP contract Dump """ return self.api(self.papi.gbp_contract_dump, {}) - def sixrd_add_tunnel(self, fib_index, ip6_prefix, ip6_prefix_len, - ip4_prefix, ip4_prefix_len, ip4_src, mtu, - security_check): + def ipip_6rd_add_tunnel(self, fib_index, ip6_prefix, ip6_prefix_len, + ip4_prefix, ip4_prefix_len, ip4_src, + security_check): """ 6RD tunnel Add """ - return self.api(self.papi.sixrd_add_tunnel, + return self.api(self.papi.ipip_6rd_add_tunnel, {'fib_index': fib_index, 'ip6_prefix': ip6_prefix, 'ip6_prefix_len': ip6_prefix_len, 'ip4_prefix': ip4_prefix, 'ip4_prefix_len': ip4_prefix_len, 'ip4_src': ip4_src, - 'mtu': mtu, 'security_check': security_check}) - def sixrd_del_tunnel(self, sw_if_index): + def ipip_6rd_del_tunnel(self, sw_if_index): """ 6RD tunnel Delete """ - return self.api(self.papi.sixrd_del_tunnel, + return self.api(self.papi.ipip_6rd_del_tunnel, + {'sw_if_index': sw_if_index}) + + def ipip_add_tunnel(self, src_address, dst_address, is_ipv6=1, + instance=0xFFFFFFFF, fib_index=0): + """ IPIP tunnel Add/Del """ + return self.api(self.papi.ipip_add_tunnel, + {'is_ipv6': is_ipv6, + 'instance': instance, + 'src_address': src_address, + 'dst_address': dst_address, + 'fib_index': fib_index}) + + def ipip_del_tunnel(self, sw_if_index): + """ IPIP tunnel Delete """ + return self.api(self.papi.ipip_del_tunnel, {'sw_if_index': sw_if_index}) |