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
13 """ Testipip is a subclass of VPPTestCase classes.
20 class TestIPIP(VppTestCase):
21 """ IPIP Test Case """
25 super(TestIPIP, cls).setUpClass()
26 cls.create_pg_interfaces(range(2))
27 cls.interfaces = list(cls.pg_interfaces)
30 super(TestIPIP, self).setUp()
31 for i in self.interfaces:
40 super(TestIPIP, self).tearDown()
42 for i in self.pg_interfaces:
47 def validate(self, rx, expected):
48 self.assertEqual(rx, expected.__class__(expected))
50 def generate_ip4_frags(self, payload_length, fragment_size):
51 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
52 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
53 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
54 outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
56 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
57 frags = fragment(outer_ip4, fragment_size)
58 p4_reply = (p_ip4 / p_payload)
60 return frags, p4_reply
63 """ ip{v4,v6} over ip4 test """
64 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
65 p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
66 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
67 p_payload = UDP(sport=1234, dport=1234)
70 rv = self.vapi.ipip_add_tunnel(
71 src_address=self.pg0.local_ip4n,
72 dst_address=self.pg1.remote_ip4n,
73 is_ipv6=0, tc_tos=0xFF)
74 sw_if_index = rv.sw_if_index
76 # Set interface up and enable IP on it
77 self.vapi.sw_interface_set_flags(sw_if_index, 1)
78 self.vapi.sw_interface_set_unnumbered(
79 sw_if_index=self.pg0.sw_if_index,
80 unnumbered_sw_if_index=sw_if_index)
82 # Add IPv4 and IPv6 routes via tunnel interface
83 ip4_via_tunnel = VppIpRoute(
84 self, "130.67.0.0", 16,
85 [VppRoutePath("0.0.0.0",
87 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
88 ip4_via_tunnel.add_vpp_config()
90 ip6_via_tunnel = VppIpRoute(
94 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
95 ip6_via_tunnel.add_vpp_config()
97 # IPv6 in to IPv4 tunnel
98 p6 = (p_ether / p_ip6 / p_payload)
100 p_inner_ip6.hlim -= 1
101 p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
102 proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
104 rx = self.send_and_expect(self.pg0, p6 * 10, self.pg1)
106 self.validate(p[1], p6_reply)
108 # IPv4 in to IPv4 tunnel
109 p4 = (p_ether / p_ip4 / p_payload)
112 p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
114 p_ip4_inner / p_payload)
117 rx = self.send_and_expect(self.pg0, p4 * 10, self.pg1)
119 self.validate(p[1], p4_reply)
122 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
124 # IPv4 tunnel to IPv4
125 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
126 p4 = (p_ether / IP(src=self.pg1.remote_ip4,
127 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
128 p4_reply = (p_ip4 / p_payload)
130 rx = self.send_and_expect(self.pg1, p4 * 10, self.pg0)
132 self.validate(p[1], p4_reply)
134 err = self.statistics.get_counter(
135 '/err/ipip4-input/packets decapsulated')
136 self.assertEqual(err, 10)
138 # IPv4 tunnel to IPv6
139 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
140 p6 = (p_ether / IP(src=self.pg1.remote_ip4,
141 dst=self.pg0.local_ip4) / p_ip6 / p_payload)
142 p6_reply = (p_ip6 / p_payload)
144 rx = self.send_and_expect(self.pg1, p6 * 10, self.pg0)
146 self.validate(p[1], p6_reply)
148 err = self.statistics.get_counter(
149 '/err/ipip4-input/packets decapsulated')
150 self.assertEqual(err, 20)
153 # Fragmentation / Reassembly and Re-fragmentation
155 rv = self.vapi.ip_reassembly_enable_disable(
156 sw_if_index=self.pg1.sw_if_index,
159 # Send lots of fragments, verify reassembled packet
160 frags, p4_reply = self.generate_ip4_frags(3131, 1400)
162 for i in range(0, 1000):
164 self.pg1.add_stream(f)
165 self.pg_enable_capture()
167 rx = self.pg0.get_capture(1000)
170 self.validate(p[1], p4_reply)
172 err = self.statistics.get_counter(
173 '/err/ipip4-input/packets decapsulated')
174 self.assertEqual(err, 1020)
178 for i in range(1, 90):
179 frags, p4_reply = self.generate_ip4_frags(i * 100, 1000)
182 self.pg_enable_capture()
183 self.pg1.add_stream(f)
185 rx = self.pg0.get_capture(89)
188 self.validate(p[1], r[i])
191 # Now try with re-fragmentation
193 # Send fragments to tunnel head-end, for the tunnel head end
194 # to reassemble and then refragment
196 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
197 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
198 self.pg_enable_capture()
199 self.pg1.add_stream(frags)
201 rx = self.pg0.get_capture(6)
202 reass_pkt = reassemble4(rx)
205 self.validate(reass_pkt, p4_reply)
207 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
208 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
209 self.pg_enable_capture()
210 self.pg1.add_stream(frags)
212 rx = self.pg0.get_capture(2)
213 reass_pkt = reassemble4(rx)
216 self.validate(reass_pkt, p4_reply)
218 def test_ipip_create(self):
219 """ ipip create / delete interface test """
220 rv = self.vapi.ipip_add_tunnel(
221 src_address=inet_pton(AF_INET, '1.2.3.4'),
222 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
223 sw_if_index = rv.sw_if_index
224 self.vapi.ipip_del_tunnel(sw_if_index)
226 def test_ipip_vrf_create(self):
227 """ ipip create / delete interface VRF test """
229 t = VppIpTable(self, 20)
231 rv = self.vapi.ipip_add_tunnel(
232 src_address=inet_pton(AF_INET, '1.2.3.4'),
233 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
235 sw_if_index = rv.sw_if_index
236 self.vapi.ipip_del_tunnel(sw_if_index)
238 def payload(self, len):
242 class TestIPIP6(VppTestCase):
243 """ IPIP6 Test Case """
247 super(TestIPIP6, cls).setUpClass()
248 cls.create_pg_interfaces(range(2))
249 cls.interfaces = list(cls.pg_interfaces)
252 super(TestIPIP6, self).setUp()
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 sw_if_index=self.pg0.sw_if_index,
282 unnumbered_sw_if_index=sw_if_index)
284 # Add IPv4 and IPv6 routes via tunnel interface
285 ip4_via_tunnel = VppIpRoute(
286 self, "130.67.0.0", 16,
287 [VppRoutePath("0.0.0.0",
289 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
290 ip4_via_tunnel.add_vpp_config()
292 ip6_via_tunnel = VppIpRoute(
296 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
297 ip6_via_tunnel.add_vpp_config()
299 self.tunnel_ip6_via_tunnel = ip6_via_tunnel
300 self.tunnel_ip4_via_tunnel = ip4_via_tunnel
302 def destroy_tunnel(self):
304 self.tunnel_ip4_via_tunnel.remove_vpp_config()
305 self.tunnel_ip6_via_tunnel.remove_vpp_config()
307 rv = self.vapi.ipip_del_tunnel(sw_if_index=self.tunnel_if_index)
309 def validate(self, rx, expected):
310 self.assertEqual(rx, expected.__class__(expected))
312 def generate_ip6_frags(self, payload_length, fragment_size):
313 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
314 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
315 p_ip6 = IPv6(src="1::1", dst=self.pg0.remote_ip6)
316 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
317 dst=self.pg0.local_ip6) /
318 IPv6ExtHdrFragment() / p_ip6 / p_payload)
319 frags = fragment6(outer_ip6, fragment_size)
320 p6_reply = (p_ip6 / p_payload)
322 return frags, p6_reply
324 def generate_ip6_hairpin_frags(self, payload_length, fragment_size):
325 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
326 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
327 p_ip6 = IPv6(src="1::1", dst="dead::1")
328 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
329 dst=self.pg0.local_ip6) /
330 IPv6ExtHdrFragment() / p_ip6 / p_payload)
331 frags = fragment6(outer_ip6, fragment_size)
333 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
334 hlim=63) / p_ip6 / p_payload)
336 return frags, p6_reply
338 def test_encap(self):
339 """ ip{v4,v6} over ip6 test encap """
340 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
341 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
342 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
343 p_payload = UDP(sport=1234, dport=1234)
346 # IPv6 in to IPv6 tunnel
347 p6 = (p_ether / p_ip6 / p_payload)
348 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
351 p6_reply[1].hlim -= 1
352 rx = self.send_and_expect(self.pg0, p6 * 11, self.pg1)
354 self.validate(p[1], p6_reply)
356 # IPv4 in to IPv6 tunnel
357 p4 = (p_ether / p_ip4 / p_payload)
358 p4_reply = (IPv6(src=self.pg0.local_ip6,
359 dst=self.pg1.remote_ip6, hlim=64, tc=42) /
362 rx = self.send_and_expect(self.pg0, p4 * 11, self.pg1)
364 self.validate(p[1], p4_reply)
366 def test_decap(self):
367 """ ip{v4,v6} over ip6 test decap """
369 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
370 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
371 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
372 p_payload = UDP(sport=1234, dport=1234)
375 # IPv6 tunnel to IPv4
377 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
378 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
379 p4_reply = (p_ip4 / p_payload)
381 rx = self.send_and_expect(self.pg1, p4 * 11, self.pg0)
383 self.validate(p[1], p4_reply)
385 # IPv6 tunnel to IPv6
386 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
387 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
388 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
389 p6_reply = (p_ip6 / p_payload)
391 rx = self.send_and_expect(self.pg1, p6 * 11, self.pg0)
393 self.validate(p[1], p6_reply)
396 """ ip{v4,v6} over ip6 test frag """
398 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
399 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
400 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
401 p_payload = UDP(sport=1234, dport=1234)
404 # Fragmentation / Reassembly and Re-fragmentation
406 rv = self.vapi.ip_reassembly_enable_disable(
407 sw_if_index=self.pg1.sw_if_index,
410 # Send lots of fragments, verify reassembled packet
411 before_cnt = self.statistics.get_counter(
412 '/err/ipip6-input/packets decapsulated')
413 frags, p6_reply = self.generate_ip6_frags(3131, 1400)
415 for i in range(0, 1000):
417 self.pg1.add_stream(f)
418 self.pg_enable_capture()
420 rx = self.pg0.get_capture(1000)
423 self.validate(p[1], p6_reply)
425 cnt = self.statistics.get_counter(
426 '/err/ipip6-input/packets decapsulated')
427 self.assertEqual(cnt, before_cnt + 1000)
431 # TODO: Check out why reassembly of atomic fragments don't work
432 for i in range(10, 90):
433 frags, p6_reply = self.generate_ip6_frags(i * 100, 1000)
436 self.pg_enable_capture()
437 self.pg1.add_stream(f)
439 rx = self.pg0.get_capture(80)
442 self.validate(p[1], r[i])
445 # Simple fragmentation
446 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
447 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
449 # IPv6 in to IPv6 tunnel
450 p_payload = UDP(sport=1234, dport=1234) / self.payload(1300)
452 p6 = (p_ether / p_ip6 / p_payload)
453 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
456 p6_reply[1].hlim -= 1
457 self.pg_enable_capture()
458 self.pg0.add_stream(p6)
460 rx = self.pg1.get_capture(2)
462 # Scapy defragment doesn't deal well with multiple layers
463 # of same type / Ethernet header first
464 f = [p[1] for p in rx]
465 reass_pkt = defragment6(f)
466 self.validate(reass_pkt, p6_reply)
468 # Now try with re-fragmentation
470 # Send large fragments to tunnel head-end, for the tunnel head end
471 # to reassemble and then refragment out the tunnel again.
474 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
475 frags, p6_reply = self.generate_ip6_hairpin_frags(8000, 1200)
476 self.pg_enable_capture()
477 self.pg1.add_stream(frags)
479 rx = self.pg1.get_capture(7)
480 f = [p[1] for p in rx]
481 reass_pkt = defragment6(f)
483 self.validate(reass_pkt, p6_reply)
485 def test_ipip_create(self):
486 """ ipip create / delete interface test """
487 rv = self.vapi.ipip_add_tunnel(
488 src_address=inet_pton(AF_INET, '1.2.3.4'),
489 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
490 sw_if_index = rv.sw_if_index
491 self.vapi.ipip_del_tunnel(sw_if_index)
493 def test_ipip_vrf_create(self):
494 """ ipip create / delete interface VRF test """
496 t = VppIpTable(self, 20)
498 rv = self.vapi.ipip_add_tunnel(
499 src_address=inet_pton(AF_INET, '1.2.3.4'),
500 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
502 sw_if_index = rv.sw_if_index
503 self.vapi.ipip_del_tunnel(sw_if_index)
505 def payload(self, len):
509 if __name__ == '__main__':
510 unittest.main(testRunner=VppTestRunner)