2 """IP{4,6} over IP{v,6} tunnel functional tests"""
5 from scapy.layers.inet6 import IPv6, Ether, IP, UDP, IPv6ExtHdrFragment
6 from scapy.all import fragment, fragment6, RandShort, defragment6
7 from framework import VppTestCase, VppTestRunner
8 from vpp_ip import DpoProto
9 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
10 from socket import AF_INET, AF_INET6, inet_pton
11 from util import reassemble4
14 """ Testipip is a subclass of VPPTestCase classes.
21 class TestIPIP(VppTestCase):
22 """ IPIP Test Case """
26 super(TestIPIP, cls).setUpClass()
27 cls.create_pg_interfaces(range(2))
28 cls.interfaces = list(cls.pg_interfaces)
31 super(TestIPIP, self).setUp()
32 for i in self.interfaces:
41 super(TestIPIP, self).tearDown()
43 for i in self.pg_interfaces:
48 def validate(self, rx, expected):
49 self.assertEqual(rx, expected.__class__(expected))
51 def generate_ip4_frags(self, payload_length, fragment_size):
52 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
53 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
54 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
55 outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
57 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
58 frags = fragment(outer_ip4, fragment_size)
59 p4_reply = (p_ip4 / p_payload)
61 return frags, p4_reply
64 """ ip{v4,v6} over ip4 test """
65 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
66 p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
67 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
68 p_payload = UDP(sport=1234, dport=1234)
71 rv = self.vapi.ipip_add_tunnel(
72 src_address=self.pg0.local_ip4n,
73 dst_address=self.pg1.remote_ip4n,
74 is_ipv6=0, tc_tos=0xFF)
75 sw_if_index = rv.sw_if_index
77 # Set interface up and enable IP on it
78 self.vapi.sw_interface_set_flags(sw_if_index, 1)
79 self.vapi.sw_interface_set_unnumbered(
80 ip_sw_if_index=self.pg0.sw_if_index,
81 sw_if_index=sw_if_index)
83 # Add IPv4 and IPv6 routes via tunnel interface
84 ip4_via_tunnel = VppIpRoute(
85 self, "130.67.0.0", 16,
86 [VppRoutePath("0.0.0.0",
88 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
89 ip4_via_tunnel.add_vpp_config()
91 ip6_via_tunnel = VppIpRoute(
95 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
96 ip6_via_tunnel.add_vpp_config()
98 # IPv6 in to IPv4 tunnel
99 p6 = (p_ether / p_ip6 / p_payload)
101 p_inner_ip6.hlim -= 1
102 p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
103 proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
105 rx = self.send_and_expect(self.pg0, p6*10, self.pg1)
107 self.validate(p[1], p6_reply)
109 # IPv4 in to IPv4 tunnel
110 p4 = (p_ether / p_ip4 / p_payload)
113 p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
115 p_ip4_inner / p_payload)
118 rx = self.send_and_expect(self.pg0, p4*10, self.pg1)
120 self.validate(p[1], p4_reply)
123 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
125 # IPv4 tunnel to IPv4
126 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
127 p4 = (p_ether / IP(src=self.pg1.remote_ip4,
128 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
129 p4_reply = (p_ip4 / p_payload)
131 rx = self.send_and_expect(self.pg1, p4*10, self.pg0)
133 self.validate(p[1], p4_reply)
135 err = self.statistics.get_counter(
136 '/err/ipip4-input/packets decapsulated')
137 self.assertEqual(err, 10)
139 # IPv4 tunnel to IPv6
140 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
141 p6 = (p_ether / IP(src=self.pg1.remote_ip4,
142 dst=self.pg0.local_ip4) / p_ip6 / p_payload)
143 p6_reply = (p_ip6 / p_payload)
145 rx = self.send_and_expect(self.pg1, p6*10, self.pg0)
147 self.validate(p[1], p6_reply)
149 err = self.statistics.get_counter(
150 '/err/ipip4-input/packets decapsulated')
151 self.assertEqual(err, 20)
154 # Fragmentation / Reassembly and Re-fragmentation
156 rv = self.vapi.ip_reassembly_enable_disable(
157 sw_if_index=self.pg1.sw_if_index,
160 # Send lots of fragments, verify reassembled packet
161 frags, p4_reply = self.generate_ip4_frags(3131, 1400)
163 for i in range(0, 1000):
165 self.pg1.add_stream(f)
166 self.pg_enable_capture()
168 rx = self.pg0.get_capture(1000)
171 self.validate(p[1], p4_reply)
173 err = self.statistics.get_counter(
174 '/err/ipip4-input/packets decapsulated')
175 self.assertEqual(err, 1020)
179 for i in range(1, 90):
180 frags, p4_reply = self.generate_ip4_frags(i * 100, 1000)
183 self.pg_enable_capture()
184 self.pg1.add_stream(f)
186 rx = self.pg0.get_capture(89)
189 self.validate(p[1], r[i])
192 # Now try with re-fragmentation
194 # Send fragments to tunnel head-end, for the tunnel head end
195 # to reassemble and then refragment
197 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
198 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
199 self.pg_enable_capture()
200 self.pg1.add_stream(frags)
202 rx = self.pg0.get_capture(6)
203 reass_pkt = reassemble4(rx)
206 self.validate(reass_pkt, p4_reply)
208 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
209 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
210 self.pg_enable_capture()
211 self.pg1.add_stream(frags)
213 rx = self.pg0.get_capture(2)
214 reass_pkt = reassemble4(rx)
217 self.validate(reass_pkt, p4_reply)
219 def test_ipip_create(self):
220 """ ipip create / delete interface test """
221 rv = self.vapi.ipip_add_tunnel(
222 src_address=inet_pton(AF_INET, '1.2.3.4'),
223 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
224 sw_if_index = rv.sw_if_index
225 self.vapi.ipip_del_tunnel(sw_if_index)
227 def test_ipip_vrf_create(self):
228 """ ipip create / delete interface VRF test """
230 t = VppIpTable(self, 20)
232 rv = self.vapi.ipip_add_tunnel(
233 src_address=inet_pton(AF_INET, '1.2.3.4'),
234 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
236 sw_if_index = rv.sw_if_index
237 self.vapi.ipip_del_tunnel(sw_if_index)
239 def payload(self, len):
243 class TestIPIP6(VppTestCase):
244 """ IPIP6 Test Case """
248 super(TestIPIP6, cls).setUpClass()
249 cls.create_pg_interfaces(range(2))
250 cls.interfaces = list(cls.pg_interfaces)
253 for i in self.interfaces:
263 if not self.vpp_dead:
264 self.destroy_tunnel()
265 for i in self.pg_interfaces:
269 super(TestIPIP6, self).tearDown()
271 def setup_tunnel(self):
273 rv = self.vapi.ipip_add_tunnel(
274 src_address=self.pg0.local_ip6n,
275 dst_address=self.pg1.remote_ip6n, tc_tos=255)
277 sw_if_index = rv.sw_if_index
278 self.tunnel_if_index = sw_if_index
279 self.vapi.sw_interface_set_flags(sw_if_index, 1)
280 self.vapi.sw_interface_set_unnumbered(
281 ip_sw_if_index=self.pg0.sw_if_index, sw_if_index=sw_if_index)
283 # Add IPv4 and IPv6 routes via tunnel interface
284 ip4_via_tunnel = VppIpRoute(
285 self, "130.67.0.0", 16,
286 [VppRoutePath("0.0.0.0",
288 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
289 ip4_via_tunnel.add_vpp_config()
291 ip6_via_tunnel = VppIpRoute(
295 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
296 ip6_via_tunnel.add_vpp_config()
298 self.tunnel_ip6_via_tunnel = ip6_via_tunnel
299 self.tunnel_ip4_via_tunnel = ip4_via_tunnel
301 def destroy_tunnel(self):
303 self.tunnel_ip4_via_tunnel.remove_vpp_config()
304 self.tunnel_ip6_via_tunnel.remove_vpp_config()
306 rv = self.vapi.ipip_del_tunnel(sw_if_index=self.tunnel_if_index)
308 def validate(self, rx, expected):
309 self.assertEqual(rx, expected.__class__(expected))
311 def generate_ip6_frags(self, payload_length, fragment_size):
312 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
313 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
314 p_ip6 = IPv6(src="1::1", dst=self.pg0.remote_ip6)
315 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
316 dst=self.pg0.local_ip6) /
317 IPv6ExtHdrFragment() / p_ip6 / p_payload)
318 frags = fragment6(outer_ip6, fragment_size)
319 p6_reply = (p_ip6 / p_payload)
321 return frags, p6_reply
323 def generate_ip6_hairpin_frags(self, payload_length, fragment_size):
324 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
325 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
326 p_ip6 = IPv6(src="1::1", dst="dead::1")
327 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
328 dst=self.pg0.local_ip6) /
329 IPv6ExtHdrFragment() / p_ip6 / p_payload)
330 frags = fragment6(outer_ip6, fragment_size)
332 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
333 hlim=63) / p_ip6 / p_payload)
335 return frags, p6_reply
337 def test_encap(self):
338 """ ip{v4,v6} over ip6 test encap """
339 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
340 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
341 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
342 p_payload = UDP(sport=1234, dport=1234)
345 # IPv6 in to IPv6 tunnel
346 p6 = (p_ether / p_ip6 / p_payload)
347 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
350 p6_reply[1].hlim -= 1
351 rx = self.send_and_expect(self.pg0, p6*11, self.pg1)
353 self.validate(p[1], p6_reply)
355 # IPv4 in to IPv6 tunnel
356 p4 = (p_ether / p_ip4 / p_payload)
357 p4_reply = (IPv6(src=self.pg0.local_ip6,
358 dst=self.pg1.remote_ip6, hlim=64, tc=42) /
361 rx = self.send_and_expect(self.pg0, p4*11, self.pg1)
363 self.validate(p[1], p4_reply)
365 def test_decap(self):
366 """ ip{v4,v6} over ip6 test decap """
368 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
369 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
370 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
371 p_payload = UDP(sport=1234, dport=1234)
374 # IPv6 tunnel to IPv4
376 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
377 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
378 p4_reply = (p_ip4 / p_payload)
380 rx = self.send_and_expect(self.pg1, p4*11, self.pg0)
382 self.validate(p[1], p4_reply)
384 # IPv6 tunnel to IPv6
385 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
386 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
387 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
388 p6_reply = (p_ip6 / p_payload)
390 rx = self.send_and_expect(self.pg1, p6*11, self.pg0)
392 self.validate(p[1], p6_reply)
395 """ ip{v4,v6} over ip6 test frag """
397 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
398 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
399 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
400 p_payload = UDP(sport=1234, dport=1234)
403 # Fragmentation / Reassembly and Re-fragmentation
405 rv = self.vapi.ip_reassembly_enable_disable(
406 sw_if_index=self.pg1.sw_if_index,
409 # Send lots of fragments, verify reassembled packet
410 before_cnt = self.statistics.get_counter(
411 '/err/ipip6-input/packets decapsulated')
412 frags, p6_reply = self.generate_ip6_frags(3131, 1400)
414 for i in range(0, 1000):
416 self.pg1.add_stream(f)
417 self.pg_enable_capture()
419 rx = self.pg0.get_capture(1000)
422 self.validate(p[1], p6_reply)
424 cnt = self.statistics.get_counter(
425 '/err/ipip6-input/packets decapsulated')
426 self.assertEqual(cnt, before_cnt + 1000)
430 # TODO: Check out why reassembly of atomic fragments don't work
431 for i in range(10, 90):
432 frags, p6_reply = self.generate_ip6_frags(i * 100, 1000)
435 self.pg_enable_capture()
436 self.pg1.add_stream(f)
438 rx = self.pg0.get_capture(80)
441 self.validate(p[1], r[i])
444 # Simple fragmentation
445 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
446 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
448 # IPv6 in to IPv6 tunnel
449 p_payload = UDP(sport=1234, dport=1234) / self.payload(1300)
451 p6 = (p_ether / p_ip6 / p_payload)
452 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
455 p6_reply[1].hlim -= 1
456 self.pg_enable_capture()
457 self.pg0.add_stream(p6)
459 rx = self.pg1.get_capture(2)
461 # Scapy defragment doesn't deal well with multiple layers
462 # of samy type / Ethernet header first
463 f = [p[1] for p in rx]
464 reass_pkt = defragment6(f)
465 self.validate(reass_pkt, p6_reply)
467 # Now try with re-fragmentation
469 # Send large fragments to tunnel head-end, for the tunnel head end
470 # to reassemble and then refragment out the tunnel again.
473 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
474 frags, p6_reply = self.generate_ip6_hairpin_frags(8000, 1200)
475 self.pg_enable_capture()
476 self.pg1.add_stream(frags)
478 rx = self.pg1.get_capture(7)
479 f = [p[1] for p in rx]
480 reass_pkt = defragment6(f)
482 self.validate(reass_pkt, p6_reply)
484 def test_ipip_create(self):
485 """ ipip create / delete interface test """
486 rv = self.vapi.ipip_add_tunnel(
487 src_address=inet_pton(AF_INET, '1.2.3.4'),
488 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
489 sw_if_index = rv.sw_if_index
490 self.vapi.ipip_del_tunnel(sw_if_index)
492 def test_ipip_vrf_create(self):
493 """ ipip create / delete interface VRF test """
495 t = VppIpTable(self, 20)
497 rv = self.vapi.ipip_add_tunnel(
498 src_address=inet_pton(AF_INET, '1.2.3.4'),
499 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
501 sw_if_index = rv.sw_if_index
502 self.vapi.ipip_del_tunnel(sw_if_index)
504 def payload(self, len):
508 if __name__ == '__main__':
509 unittest.main(testRunner=VppTestRunner)