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 def tearDownClass(cls):
31 super(TestIPIP, cls).tearDownClass()
34 super(TestIPIP, self).setUp()
35 for i in self.interfaces:
44 super(TestIPIP, self).tearDown()
46 for i in self.pg_interfaces:
51 def validate(self, rx, expected):
52 self.assertEqual(rx, expected.__class__(expected))
54 def generate_ip4_frags(self, payload_length, fragment_size):
55 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
56 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
57 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
58 outer_ip4 = (p_ether / IP(src=self.pg1.remote_ip4,
60 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
61 frags = fragment(outer_ip4, fragment_size)
62 p4_reply = (p_ip4 / p_payload)
64 return frags, p4_reply
67 """ ip{v4,v6} over ip4 test """
68 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
69 p_ip6 = IPv6(src="1::1", dst="DEAD::1", nh='UDP', tc=42)
70 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
71 p_payload = UDP(sport=1234, dport=1234)
74 rv = self.vapi.ipip_add_tunnel(
75 src_address=self.pg0.local_ip4n,
76 dst_address=self.pg1.remote_ip4n,
77 is_ipv6=0, tc_tos=0xFF)
78 sw_if_index = rv.sw_if_index
80 # Set interface up and enable IP on it
81 self.vapi.sw_interface_set_flags(sw_if_index, 1)
82 self.vapi.sw_interface_set_unnumbered(
83 sw_if_index=self.pg0.sw_if_index,
84 unnumbered_sw_if_index=sw_if_index)
86 # Add IPv4 and IPv6 routes via tunnel interface
87 ip4_via_tunnel = VppIpRoute(
88 self, "130.67.0.0", 16,
89 [VppRoutePath("0.0.0.0",
91 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
92 ip4_via_tunnel.add_vpp_config()
94 ip6_via_tunnel = VppIpRoute(
98 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
99 ip6_via_tunnel.add_vpp_config()
101 # IPv6 in to IPv4 tunnel
102 p6 = (p_ether / p_ip6 / p_payload)
104 p_inner_ip6.hlim -= 1
105 p6_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
106 proto='ipv6', id=0, tos=42) / p_inner_ip6 / p_payload)
108 rx = self.send_and_expect(self.pg0, p6 * 10, self.pg1)
110 self.validate(p[1], p6_reply)
112 # IPv4 in to IPv4 tunnel
113 p4 = (p_ether / p_ip4 / p_payload)
116 p4_reply = (IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4,
118 p_ip4_inner / p_payload)
121 rx = self.send_and_expect(self.pg0, p4 * 10, self.pg1)
123 self.validate(p[1], p4_reply)
126 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
128 # IPv4 tunnel to IPv4
129 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
130 p4 = (p_ether / IP(src=self.pg1.remote_ip4,
131 dst=self.pg0.local_ip4) / p_ip4 / p_payload)
132 p4_reply = (p_ip4 / p_payload)
134 rx = self.send_and_expect(self.pg1, p4 * 10, self.pg0)
136 self.validate(p[1], p4_reply)
138 err = self.statistics.get_counter(
139 '/err/ipip4-input/packets decapsulated')
140 self.assertEqual(err, 10)
142 # IPv4 tunnel to IPv6
143 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
144 p6 = (p_ether / IP(src=self.pg1.remote_ip4,
145 dst=self.pg0.local_ip4) / p_ip6 / p_payload)
146 p6_reply = (p_ip6 / p_payload)
148 rx = self.send_and_expect(self.pg1, p6 * 10, self.pg0)
150 self.validate(p[1], p6_reply)
152 err = self.statistics.get_counter(
153 '/err/ipip4-input/packets decapsulated')
154 self.assertEqual(err, 20)
157 # Fragmentation / Reassembly and Re-fragmentation
159 rv = self.vapi.ip_reassembly_enable_disable(
160 sw_if_index=self.pg1.sw_if_index,
163 # Send lots of fragments, verify reassembled packet
164 frags, p4_reply = self.generate_ip4_frags(3131, 1400)
166 for i in range(0, 1000):
168 self.pg1.add_stream(f)
169 self.pg_enable_capture()
171 rx = self.pg0.get_capture(1000)
174 self.validate(p[1], p4_reply)
176 err = self.statistics.get_counter(
177 '/err/ipip4-input/packets decapsulated')
178 self.assertEqual(err, 1020)
182 for i in range(1, 90):
183 frags, p4_reply = self.generate_ip4_frags(i * 100, 1000)
186 self.pg_enable_capture()
187 self.pg1.add_stream(f)
189 rx = self.pg0.get_capture(89)
192 self.validate(p[1], r[i])
195 # Now try with re-fragmentation
197 # Send fragments to tunnel head-end, for the tunnel head end
198 # to reassemble and then refragment
200 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
201 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
202 self.pg_enable_capture()
203 self.pg1.add_stream(frags)
205 rx = self.pg0.get_capture(6)
206 reass_pkt = reassemble4(rx)
209 self.validate(reass_pkt, p4_reply)
211 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
212 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
213 self.pg_enable_capture()
214 self.pg1.add_stream(frags)
216 rx = self.pg0.get_capture(2)
217 reass_pkt = reassemble4(rx)
220 self.validate(reass_pkt, p4_reply)
222 def test_ipip_create(self):
223 """ ipip create / delete interface test """
224 rv = self.vapi.ipip_add_tunnel(
225 src_address=inet_pton(AF_INET, '1.2.3.4'),
226 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
227 sw_if_index = rv.sw_if_index
228 self.vapi.ipip_del_tunnel(sw_if_index)
230 def test_ipip_vrf_create(self):
231 """ ipip create / delete interface VRF test """
233 t = VppIpTable(self, 20)
235 rv = self.vapi.ipip_add_tunnel(
236 src_address=inet_pton(AF_INET, '1.2.3.4'),
237 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
239 sw_if_index = rv.sw_if_index
240 self.vapi.ipip_del_tunnel(sw_if_index)
242 def payload(self, len):
246 class TestIPIP6(VppTestCase):
247 """ IPIP6 Test Case """
251 super(TestIPIP6, cls).setUpClass()
252 cls.create_pg_interfaces(range(2))
253 cls.interfaces = list(cls.pg_interfaces)
256 def tearDownClass(cls):
257 super(TestIPIP6, cls).tearDownClass()
260 super(TestIPIP6, self).setUp()
261 for i in self.interfaces:
271 if not self.vpp_dead:
272 self.destroy_tunnel()
273 for i in self.pg_interfaces:
277 super(TestIPIP6, self).tearDown()
279 def setup_tunnel(self):
281 rv = self.vapi.ipip_add_tunnel(
282 src_address=self.pg0.local_ip6n,
283 dst_address=self.pg1.remote_ip6n, tc_tos=255)
285 sw_if_index = rv.sw_if_index
286 self.tunnel_if_index = sw_if_index
287 self.vapi.sw_interface_set_flags(sw_if_index, 1)
288 self.vapi.sw_interface_set_unnumbered(
289 sw_if_index=self.pg0.sw_if_index,
290 unnumbered_sw_if_index=sw_if_index)
292 # Add IPv4 and IPv6 routes via tunnel interface
293 ip4_via_tunnel = VppIpRoute(
294 self, "130.67.0.0", 16,
295 [VppRoutePath("0.0.0.0",
297 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
298 ip4_via_tunnel.add_vpp_config()
300 ip6_via_tunnel = VppIpRoute(
304 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
305 ip6_via_tunnel.add_vpp_config()
307 self.tunnel_ip6_via_tunnel = ip6_via_tunnel
308 self.tunnel_ip4_via_tunnel = ip4_via_tunnel
310 def destroy_tunnel(self):
312 self.tunnel_ip4_via_tunnel.remove_vpp_config()
313 self.tunnel_ip6_via_tunnel.remove_vpp_config()
315 rv = self.vapi.ipip_del_tunnel(sw_if_index=self.tunnel_if_index)
317 def validate(self, rx, expected):
318 self.assertEqual(rx, expected.__class__(expected))
320 def generate_ip6_frags(self, payload_length, fragment_size):
321 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
322 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
323 p_ip6 = IPv6(src="1::1", dst=self.pg0.remote_ip6)
324 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
325 dst=self.pg0.local_ip6) /
326 IPv6ExtHdrFragment() / p_ip6 / p_payload)
327 frags = fragment6(outer_ip6, fragment_size)
328 p6_reply = (p_ip6 / p_payload)
330 return frags, p6_reply
332 def generate_ip6_hairpin_frags(self, payload_length, fragment_size):
333 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
334 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
335 p_ip6 = IPv6(src="1::1", dst="dead::1")
336 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
337 dst=self.pg0.local_ip6) /
338 IPv6ExtHdrFragment() / p_ip6 / p_payload)
339 frags = fragment6(outer_ip6, fragment_size)
341 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
342 hlim=63) / p_ip6 / p_payload)
344 return frags, p6_reply
346 def test_encap(self):
347 """ ip{v4,v6} over ip6 test encap """
348 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
349 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
350 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
351 p_payload = UDP(sport=1234, dport=1234)
354 # IPv6 in to IPv6 tunnel
355 p6 = (p_ether / p_ip6 / p_payload)
356 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
359 p6_reply[1].hlim -= 1
360 rx = self.send_and_expect(self.pg0, p6 * 11, self.pg1)
362 self.validate(p[1], p6_reply)
364 # IPv4 in to IPv6 tunnel
365 p4 = (p_ether / p_ip4 / p_payload)
366 p4_reply = (IPv6(src=self.pg0.local_ip6,
367 dst=self.pg1.remote_ip6, hlim=64, tc=42) /
370 rx = self.send_and_expect(self.pg0, p4 * 11, self.pg1)
372 self.validate(p[1], p4_reply)
374 def test_decap(self):
375 """ ip{v4,v6} over ip6 test decap """
377 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
378 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
379 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
380 p_payload = UDP(sport=1234, dport=1234)
383 # IPv6 tunnel to IPv4
385 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
386 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
387 p4_reply = (p_ip4 / p_payload)
389 rx = self.send_and_expect(self.pg1, p4 * 11, self.pg0)
391 self.validate(p[1], p4_reply)
393 # IPv6 tunnel to IPv6
394 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
395 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
396 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
397 p6_reply = (p_ip6 / p_payload)
399 rx = self.send_and_expect(self.pg1, p6 * 11, self.pg0)
401 self.validate(p[1], p6_reply)
404 """ ip{v4,v6} over ip6 test frag """
406 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
407 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
408 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
409 p_payload = UDP(sport=1234, dport=1234)
412 # Fragmentation / Reassembly and Re-fragmentation
414 rv = self.vapi.ip_reassembly_enable_disable(
415 sw_if_index=self.pg1.sw_if_index,
418 # Send lots of fragments, verify reassembled packet
419 before_cnt = self.statistics.get_counter(
420 '/err/ipip6-input/packets decapsulated')
421 frags, p6_reply = self.generate_ip6_frags(3131, 1400)
423 for i in range(0, 1000):
425 self.pg1.add_stream(f)
426 self.pg_enable_capture()
428 rx = self.pg0.get_capture(1000)
431 self.validate(p[1], p6_reply)
433 cnt = self.statistics.get_counter(
434 '/err/ipip6-input/packets decapsulated')
435 self.assertEqual(cnt, before_cnt + 1000)
439 # TODO: Check out why reassembly of atomic fragments don't work
440 for i in range(10, 90):
441 frags, p6_reply = self.generate_ip6_frags(i * 100, 1000)
444 self.pg_enable_capture()
445 self.pg1.add_stream(f)
447 rx = self.pg0.get_capture(80)
450 self.validate(p[1], r[i])
453 # Simple fragmentation
454 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
455 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
457 # IPv6 in to IPv6 tunnel
458 p_payload = UDP(sport=1234, dport=1234) / self.payload(1300)
460 p6 = (p_ether / p_ip6 / p_payload)
461 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
464 p6_reply[1].hlim -= 1
465 self.pg_enable_capture()
466 self.pg0.add_stream(p6)
468 rx = self.pg1.get_capture(2)
470 # Scapy defragment doesn't deal well with multiple layers
471 # of same type / Ethernet header first
472 f = [p[1] for p in rx]
473 reass_pkt = defragment6(f)
474 self.validate(reass_pkt, p6_reply)
476 # Now try with re-fragmentation
478 # Send large fragments to tunnel head-end, for the tunnel head end
479 # to reassemble and then refragment out the tunnel again.
482 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
483 frags, p6_reply = self.generate_ip6_hairpin_frags(8000, 1200)
484 self.pg_enable_capture()
485 self.pg1.add_stream(frags)
487 rx = self.pg1.get_capture(7)
488 f = [p[1] for p in rx]
489 reass_pkt = defragment6(f)
491 self.validate(reass_pkt, p6_reply)
493 def test_ipip_create(self):
494 """ ipip create / delete interface test """
495 rv = self.vapi.ipip_add_tunnel(
496 src_address=inet_pton(AF_INET, '1.2.3.4'),
497 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
498 sw_if_index = rv.sw_if_index
499 self.vapi.ipip_del_tunnel(sw_if_index)
501 def test_ipip_vrf_create(self):
502 """ ipip create / delete interface VRF test """
504 t = VppIpTable(self, 20)
506 rv = self.vapi.ipip_add_tunnel(
507 src_address=inet_pton(AF_INET, '1.2.3.4'),
508 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
510 sw_if_index = rv.sw_if_index
511 self.vapi.ipip_del_tunnel(sw_if_index)
513 def payload(self, len):
517 if __name__ == '__main__':
518 unittest.main(testRunner=VppTestRunner)