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_err_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_err_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 self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=1000,
164 max_reassembly_length=1000,
165 expire_walk_interval_ms=10000,
168 # Send lots of fragments, verify reassembled packet
169 frags, p4_reply = self.generate_ip4_frags(3131, 1400)
171 for i in range(0, 1000):
173 self.pg1.add_stream(f)
174 self.pg_enable_capture()
176 rx = self.pg0.get_capture(1000)
179 self.validate(p[1], p4_reply)
181 err = self.statistics.get_err_counter(
182 '/err/ipip4-input/packets decapsulated')
183 self.assertEqual(err, 1020)
187 for i in range(1, 90):
188 frags, p4_reply = self.generate_ip4_frags(i * 100, 1000)
191 self.pg_enable_capture()
192 self.pg1.add_stream(f)
194 rx = self.pg0.get_capture(89)
197 self.validate(p[1], r[i])
200 # Now try with re-fragmentation
202 # Send fragments to tunnel head-end, for the tunnel head end
203 # to reassemble and then refragment
205 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [576, 0, 0, 0])
206 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
207 self.pg_enable_capture()
208 self.pg1.add_stream(frags)
210 rx = self.pg0.get_capture(6)
211 reass_pkt = reassemble4(rx)
214 self.validate(reass_pkt, p4_reply)
216 self.vapi.sw_interface_set_mtu(self.pg0.sw_if_index, [1600, 0, 0, 0])
217 frags, p4_reply = self.generate_ip4_frags(3123, 1200)
218 self.pg_enable_capture()
219 self.pg1.add_stream(frags)
221 rx = self.pg0.get_capture(2)
222 reass_pkt = reassemble4(rx)
225 self.validate(reass_pkt, p4_reply)
227 def test_ipip_create(self):
228 """ ipip create / delete interface test """
229 rv = self.vapi.ipip_add_tunnel(
230 src_address=inet_pton(AF_INET, '1.2.3.4'),
231 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
232 sw_if_index = rv.sw_if_index
233 self.vapi.ipip_del_tunnel(sw_if_index)
235 def test_ipip_vrf_create(self):
236 """ ipip create / delete interface VRF test """
238 t = VppIpTable(self, 20)
240 rv = self.vapi.ipip_add_tunnel(
241 src_address=inet_pton(AF_INET, '1.2.3.4'),
242 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
244 sw_if_index = rv.sw_if_index
245 self.vapi.ipip_del_tunnel(sw_if_index)
247 def payload(self, len):
251 class TestIPIP6(VppTestCase):
252 """ IPIP6 Test Case """
256 super(TestIPIP6, cls).setUpClass()
257 cls.create_pg_interfaces(range(2))
258 cls.interfaces = list(cls.pg_interfaces)
261 def tearDownClass(cls):
262 super(TestIPIP6, cls).tearDownClass()
265 super(TestIPIP6, self).setUp()
266 for i in self.interfaces:
276 if not self.vpp_dead:
277 self.destroy_tunnel()
278 for i in self.pg_interfaces:
282 super(TestIPIP6, self).tearDown()
284 def setup_tunnel(self):
286 rv = self.vapi.ipip_add_tunnel(
287 src_address=self.pg0.local_ip6n,
288 dst_address=self.pg1.remote_ip6n, tc_tos=255)
290 sw_if_index = rv.sw_if_index
291 self.tunnel_if_index = sw_if_index
292 self.vapi.sw_interface_set_flags(sw_if_index, 1)
293 self.vapi.sw_interface_set_unnumbered(
294 sw_if_index=self.pg0.sw_if_index,
295 unnumbered_sw_if_index=sw_if_index)
297 # Add IPv4 and IPv6 routes via tunnel interface
298 ip4_via_tunnel = VppIpRoute(
299 self, "130.67.0.0", 16,
300 [VppRoutePath("0.0.0.0",
302 proto=DpoProto.DPO_PROTO_IP4)], is_ip6=0)
303 ip4_via_tunnel.add_vpp_config()
305 ip6_via_tunnel = VppIpRoute(
309 proto=DpoProto.DPO_PROTO_IP6)], is_ip6=1)
310 ip6_via_tunnel.add_vpp_config()
312 self.tunnel_ip6_via_tunnel = ip6_via_tunnel
313 self.tunnel_ip4_via_tunnel = ip4_via_tunnel
315 def destroy_tunnel(self):
317 self.tunnel_ip4_via_tunnel.remove_vpp_config()
318 self.tunnel_ip6_via_tunnel.remove_vpp_config()
320 rv = self.vapi.ipip_del_tunnel(sw_if_index=self.tunnel_if_index)
322 def validate(self, rx, expected):
323 self.assertEqual(rx, expected.__class__(expected))
325 def generate_ip6_frags(self, payload_length, fragment_size):
326 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
327 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
328 p_ip6 = IPv6(src="1::1", dst=self.pg0.remote_ip6)
329 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
330 dst=self.pg0.local_ip6) /
331 IPv6ExtHdrFragment() / p_ip6 / p_payload)
332 frags = fragment6(outer_ip6, fragment_size)
333 p6_reply = (p_ip6 / p_payload)
335 return frags, p6_reply
337 def generate_ip6_hairpin_frags(self, payload_length, fragment_size):
338 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
339 p_payload = UDP(sport=1234, dport=1234) / self.payload(payload_length)
340 p_ip6 = IPv6(src="1::1", dst="dead::1")
341 outer_ip6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
342 dst=self.pg0.local_ip6) /
343 IPv6ExtHdrFragment() / p_ip6 / p_payload)
344 frags = fragment6(outer_ip6, fragment_size)
346 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
347 hlim=63) / p_ip6 / p_payload)
349 return frags, p6_reply
351 def test_encap(self):
352 """ ip{v4,v6} over ip6 test encap """
353 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
354 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
355 p_ip4 = IP(src="1.2.3.4", dst="130.67.0.1", tos=42)
356 p_payload = UDP(sport=1234, dport=1234)
359 # IPv6 in to IPv6 tunnel
360 p6 = (p_ether / p_ip6 / p_payload)
361 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
364 p6_reply[1].hlim -= 1
365 rx = self.send_and_expect(self.pg0, p6 * 11, self.pg1)
367 self.validate(p[1], p6_reply)
369 # IPv4 in to IPv6 tunnel
370 p4 = (p_ether / p_ip4 / p_payload)
371 p4_reply = (IPv6(src=self.pg0.local_ip6,
372 dst=self.pg1.remote_ip6, hlim=64, tc=42) /
375 rx = self.send_and_expect(self.pg0, p4 * 11, self.pg1)
377 self.validate(p[1], p4_reply)
379 def test_decap(self):
380 """ ip{v4,v6} over ip6 test decap """
382 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
383 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
384 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
385 p_payload = UDP(sport=1234, dport=1234)
388 # IPv6 tunnel to IPv4
390 p4 = (p_ether / IPv6(src=self.pg1.remote_ip6,
391 dst=self.pg0.local_ip6) / p_ip4 / p_payload)
392 p4_reply = (p_ip4 / p_payload)
394 rx = self.send_and_expect(self.pg1, p4 * 11, self.pg0)
396 self.validate(p[1], p4_reply)
398 # IPv6 tunnel to IPv6
399 p_ip6 = IPv6(src="1:2:3::4", dst=self.pg0.remote_ip6)
400 p6 = (p_ether / IPv6(src=self.pg1.remote_ip6,
401 dst=self.pg0.local_ip6) / p_ip6 / p_payload)
402 p6_reply = (p_ip6 / p_payload)
404 rx = self.send_and_expect(self.pg1, p6 * 11, self.pg0)
406 self.validate(p[1], p6_reply)
409 """ ip{v4,v6} over ip6 test frag """
411 p_ether = Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
412 p_ip6 = IPv6(src="1::1", dst="DEAD::1", tc=42, nh='UDP')
413 p_ip4 = IP(src="1.2.3.4", dst=self.pg0.remote_ip4)
414 p_payload = UDP(sport=1234, dport=1234)
417 # Fragmentation / Reassembly and Re-fragmentation
419 rv = self.vapi.ip_reassembly_enable_disable(
420 sw_if_index=self.pg1.sw_if_index,
423 self.vapi.ip_reassembly_set(timeout_ms=1000, max_reassemblies=1000,
424 max_reassembly_length=1000,
425 expire_walk_interval_ms=10000,
428 # Send lots of fragments, verify reassembled packet
429 before_cnt = self.statistics.get_err_counter(
430 '/err/ipip6-input/packets decapsulated')
431 frags, p6_reply = self.generate_ip6_frags(3131, 1400)
433 for i in range(0, 1000):
435 self.pg1.add_stream(f)
436 self.pg_enable_capture()
438 rx = self.pg0.get_capture(1000)
441 self.validate(p[1], p6_reply)
443 cnt = self.statistics.get_err_counter(
444 '/err/ipip6-input/packets decapsulated')
445 self.assertEqual(cnt, before_cnt + 1000)
449 # TODO: Check out why reassembly of atomic fragments don't work
450 for i in range(10, 90):
451 frags, p6_reply = self.generate_ip6_frags(i * 100, 1000)
454 self.pg_enable_capture()
455 self.pg1.add_stream(f)
457 rx = self.pg0.get_capture(80)
460 self.validate(p[1], r[i])
463 # Simple fragmentation
464 p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
465 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
467 # IPv6 in to IPv6 tunnel
468 p_payload = UDP(sport=1234, dport=1234) / self.payload(1300)
470 p6 = (p_ether / p_ip6 / p_payload)
471 p6_reply = (IPv6(src=self.pg0.local_ip6, dst=self.pg1.remote_ip6,
474 p6_reply[1].hlim -= 1
475 self.pg_enable_capture()
476 self.pg0.add_stream(p6)
478 rx = self.pg1.get_capture(2)
480 # Scapy defragment doesn't deal well with multiple layers
481 # of same type / Ethernet header first
482 f = [p[1] for p in rx]
483 reass_pkt = defragment6(f)
484 self.validate(reass_pkt, p6_reply)
486 # Now try with re-fragmentation
488 # Send large fragments to tunnel head-end, for the tunnel head end
489 # to reassemble and then refragment out the tunnel again.
492 self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
493 frags, p6_reply = self.generate_ip6_hairpin_frags(8000, 1200)
494 self.pg_enable_capture()
495 self.pg1.add_stream(frags)
497 rx = self.pg1.get_capture(7)
498 f = [p[1] for p in rx]
499 reass_pkt = defragment6(f)
501 self.validate(reass_pkt, p6_reply)
503 def test_ipip_create(self):
504 """ ipip create / delete interface test """
505 rv = self.vapi.ipip_add_tunnel(
506 src_address=inet_pton(AF_INET, '1.2.3.4'),
507 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0)
508 sw_if_index = rv.sw_if_index
509 self.vapi.ipip_del_tunnel(sw_if_index)
511 def test_ipip_vrf_create(self):
512 """ ipip create / delete interface VRF test """
514 t = VppIpTable(self, 20)
516 rv = self.vapi.ipip_add_tunnel(
517 src_address=inet_pton(AF_INET, '1.2.3.4'),
518 dst_address=inet_pton(AF_INET, '2.3.4.5'), is_ipv6=0,
520 sw_if_index = rv.sw_if_index
521 self.vapi.ipip_del_tunnel(sw_if_index)
523 def payload(self, len):
527 if __name__ == '__main__':
528 unittest.main(testRunner=VppTestRunner)