6 from framework import VppTestCase, VppTestRunner
7 from vpp_ip_route import IpRoute, RoutePath, MplsRoute, MplsIpBind
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether
11 from scapy.layers.inet import IP, UDP, ICMP
12 from scapy.layers.inet6 import IPv6
13 from scapy.contrib.mpls import MPLS
15 class TestMPLS(VppTestCase):
16 """ MPLS Test Case """
20 super(TestMPLS, cls).setUpClass()
23 super(TestMPLS, self).setUp()
25 # create 2 pg interfaces
26 self.create_pg_interfaces(range(2))
28 # setup both interfaces
29 # assign them different tables.
32 for i in self.pg_interfaces:
34 i.set_table_ip4(table_id)
35 i.set_table_ip6(table_id)
44 super(TestMPLS, self).tearDown()
46 # the default of 64 matches the IP packet TTL default
47 def create_stream_labelled_ip4(self, src_if, mpls_labels, mpls_ttl=255, ping=0, ip_itf=None):
49 for i in range(0, 257):
50 info = self.create_packet_info(src_if.sw_if_index,
52 payload = self.info_to_payload(info)
53 p = Ether(dst=src_if.local_mac, src=src_if.remote_mac)
55 for ii in range(len(mpls_labels)):
56 if ii == len(mpls_labels) - 1:
57 p = p / MPLS(label=mpls_labels[ii], ttl=mpls_ttl, s=1)
59 p = p / MPLS(label=mpls_labels[ii], ttl=mpls_ttl, s=0)
61 p = (p / IP(src=src_if.local_ip4, dst=src_if.remote_ip4) /
62 UDP(sport=1234, dport=1234) /
65 p = (p / IP(src=ip_itf.remote_ip4,
66 dst=ip_itf.local_ip4) /
73 def create_stream_ip4(self, src_if, dst_ip):
75 for i in range(0, 257):
76 info = self.create_packet_info(src_if.sw_if_index,
78 payload = self.info_to_payload(info)
79 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
80 IP(src=src_if.remote_ip4, dst=dst_ip) /
81 UDP(sport=1234, dport=1234) /
87 def create_stream_labelled_ip6(self, src_if, mpls_label, mpls_ttl):
89 for i in range(0, 257):
90 info = self.create_packet_info(src_if.sw_if_index,
92 payload = self.info_to_payload(info)
93 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
94 MPLS(label=mpls_label, ttl=mpls_ttl) /
95 IPv6(src=src_if.remote_ip6, dst=src_if.remote_ip6) /
96 UDP(sport=1234, dport=1234) /
103 def verify_filter(capture, sent):
104 if not len(capture) == len(sent):
105 # filter out any IPv6 RAs from the capture
111 def verify_capture_ip4(self, src_if, capture, sent, ping_resp=0):
113 capture = self.verify_filter(capture, sent)
115 self.assertEqual(len(capture), len(sent))
117 for i in range(len(capture)):
121 # the rx'd packet has the MPLS label popped
123 self.assertEqual(eth.type, 0x800)
129 self.assertEqual(rx_ip.src, tx_ip.src)
130 self.assertEqual(rx_ip.dst, tx_ip.dst)
131 # IP processing post pop has decremented the TTL
132 self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl)
134 self.assertEqual(rx_ip.src, tx_ip.dst)
135 self.assertEqual(rx_ip.dst, tx_ip.src)
140 def verify_mpls_stack(self, rx, mpls_labels, ttl=255, num=0):
141 # the rx'd packet has the MPLS label popped
143 self.assertEqual(eth.type, 0x8847)
147 for ii in range(len(mpls_labels)):
148 self.assertEqual(rx_mpls.label, mpls_labels[ii])
149 self.assertEqual(rx_mpls.cos, 0)
151 self.assertEqual(rx_mpls.ttl, ttl)
153 self.assertEqual(rx_mpls.ttl, 255)
155 if ii == len(mpls_labels) - 1:
156 self.assertEqual(rx_mpls.s, 1)
159 self.assertEqual(rx_mpls.s, 0)
160 # pop the label to expose the next
161 rx_mpls = rx_mpls[MPLS].payload
163 def verify_capture_labelled_ip4(self, src_if, capture, sent,
166 capture = self.verify_filter(capture, sent)
168 self.assertEqual(len(capture), len(sent))
170 for i in range(len(capture)):
176 # the MPLS TTL is copied from the IP
177 self.verify_mpls_stack(
178 rx, mpls_labels, rx_ip.ttl, len(mpls_labels) - 1)
180 self.assertEqual(rx_ip.src, tx_ip.src)
181 self.assertEqual(rx_ip.dst, tx_ip.dst)
182 # IP processing post pop has decremented the TTL
183 self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl)
188 def verify_capture_tunneled_ip4(self, src_if, capture, sent, mpls_labels):
190 capture = self.verify_filter(capture, sent)
192 self.assertEqual(len(capture), len(sent))
194 for i in range(len(capture)):
200 # the MPLS TTL is 255 since it enters a new tunnel
201 self.verify_mpls_stack(
202 rx, mpls_labels, 255, len(mpls_labels) - 1)
204 self.assertEqual(rx_ip.src, tx_ip.src)
205 self.assertEqual(rx_ip.dst, tx_ip.dst)
206 # IP processing post pop has decremented the TTL
207 self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl)
212 def verify_capture_labelled(self, src_if, capture, sent,
213 mpls_labels, ttl=254, num=0):
215 capture = self.verify_filter(capture, sent)
217 self.assertEqual(len(capture), len(sent))
219 for i in range(len(capture)):
221 self.verify_mpls_stack(rx, mpls_labels, ttl, num)
225 def verify_capture_ip6(self, src_if, capture, sent):
227 self.assertEqual(len(capture), len(sent))
229 for i in range(len(capture)):
233 # the rx'd packet has the MPLS label popped
235 self.assertEqual(eth.type, 0x86DD)
240 self.assertEqual(rx_ip.src, tx_ip.src)
241 self.assertEqual(rx_ip.dst, tx_ip.dst)
242 # IP processing post pop has decremented the TTL
243 self.assertEqual(rx_ip.hlim + 1, tx_ip.hlim)
249 """ MPLS label swap tests """
252 # A simple MPLS xconnect - eos label in label out
254 route_32_eos = MplsRoute(self, 32, 1,
255 [RoutePath(self.pg0.remote_ip4,
256 self.pg0.sw_if_index,
258 route_32_eos.add_vpp_config()
261 # a stream that matches the route for 10.0.0.1
262 # PG0 is in the default table
264 self.vapi.cli("clear trace")
265 tx = self.create_stream_labelled_ip4(self.pg0, [32])
266 self.pg0.add_stream(tx)
268 self.pg_enable_capture(self.pg_interfaces)
271 rx = self.pg0.get_capture()
272 self.verify_capture_labelled_ip4(self.pg0, rx, tx, [33])
275 # A simple MPLS xconnect - non-eos label in label out
277 route_32_neos = MplsRoute(self, 32, 0,
278 [RoutePath(self.pg0.remote_ip4,
279 self.pg0.sw_if_index,
281 route_32_neos.add_vpp_config()
284 # a stream that matches the route for 10.0.0.1
285 # PG0 is in the default table
287 self.vapi.cli("clear trace")
288 tx = self.create_stream_labelled_ip4(self.pg0, [32, 99])
289 self.pg0.add_stream(tx)
291 self.pg_enable_capture(self.pg_interfaces)
294 rx = self.pg0.get_capture()
295 self.verify_capture_labelled(self.pg0, rx, tx, [33, 99])
298 # An MPLS xconnect - EOS label in IP out
300 route_33_eos = MplsRoute(self, 33, 1,
301 [RoutePath(self.pg0.remote_ip4,
302 self.pg0.sw_if_index,
304 route_33_eos.add_vpp_config()
306 self.vapi.cli("clear trace")
307 tx = self.create_stream_labelled_ip4(self.pg0, [33])
308 self.pg0.add_stream(tx)
310 self.pg_enable_capture(self.pg_interfaces)
313 rx = self.pg0.get_capture()
314 self.verify_capture_ip4(self.pg0, rx, tx)
317 # An MPLS xconnect - non-EOS label in IP out - an invalid configuration
318 # so this traffic should be dropped.
320 route_33_neos = MplsRoute(self, 33, 0,
321 [RoutePath(self.pg0.remote_ip4,
322 self.pg0.sw_if_index,
324 route_33_neos.add_vpp_config()
326 self.vapi.cli("clear trace")
327 tx = self.create_stream_labelled_ip4(self.pg0, [33, 99])
328 self.pg0.add_stream(tx)
330 self.pg_enable_capture(self.pg_interfaces)
332 self.pg0.assert_nothing_captured(
333 remark="MPLS non-EOS packets popped and forwarded")
336 # A recursive EOS x-connect, which resolves through another x-connect
338 route_34_eos = MplsRoute(self, 34, 1,
339 [RoutePath("0.0.0.0",
343 route_34_eos.add_vpp_config()
345 self.vapi.cli("clear trace")
346 tx = self.create_stream_labelled_ip4(self.pg0, [34])
347 self.pg0.add_stream(tx)
349 self.pg_enable_capture(self.pg_interfaces)
352 rx = self.pg0.get_capture()
353 self.verify_capture_labelled_ip4(self.pg0, rx, tx, [33, 44, 45])
356 # A recursive non-EOS x-connect, which resolves through another
359 route_34_neos = MplsRoute(self, 34, 0,
360 [RoutePath("0.0.0.0",
364 route_34_neos.add_vpp_config()
366 self.vapi.cli("clear trace")
367 tx = self.create_stream_labelled_ip4(self.pg0, [34, 99])
368 self.pg0.add_stream(tx)
370 self.pg_enable_capture(self.pg_interfaces)
373 rx = self.pg0.get_capture()
374 # it's the 2nd (counting from 0) label in the stack that is swapped
375 self.verify_capture_labelled(self.pg0, rx, tx, [33, 44, 46, 99], num=2)
378 # an recursive IP route that resolves through the recursive non-eos
381 ip_10_0_0_1 = IpRoute(self, "10.0.0.1", 32,
382 [RoutePath("0.0.0.0",
386 ip_10_0_0_1.add_vpp_config()
388 self.vapi.cli("clear trace")
389 tx = self.create_stream_ip4(self.pg0, "10.0.0.1")
390 self.pg0.add_stream(tx)
392 self.pg_enable_capture(self.pg_interfaces)
395 rx = self.pg0.get_capture()
396 self.verify_capture_labelled_ip4(self.pg0, rx, tx, [33, 44, 46, 55])
398 ip_10_0_0_1.remove_vpp_config()
399 route_34_neos.remove_vpp_config()
400 route_34_eos.remove_vpp_config()
401 route_33_neos.remove_vpp_config()
402 route_33_eos.remove_vpp_config()
403 route_32_neos.remove_vpp_config()
404 route_32_eos.remove_vpp_config()
407 """ MPLS Local Label Binding test """
410 # Add a non-recursive route with a single out label
412 route_10_0_0_1 = IpRoute(self, "10.0.0.1", 32,
413 [RoutePath(self.pg0.remote_ip4,
414 self.pg0.sw_if_index,
416 route_10_0_0_1.add_vpp_config()
418 # bind a local label to the route
419 binding = MplsIpBind(self, 44, "10.0.0.1", 32)
420 binding.add_vpp_config()
423 self.vapi.cli("clear trace")
424 tx = self.create_stream_labelled_ip4(self.pg0, [44, 99])
425 self.pg0.add_stream(tx)
427 self.pg_enable_capture(self.pg_interfaces)
430 rx = self.pg0.get_capture()
431 self.verify_capture_labelled(self.pg0, rx, tx, [45, 99])
434 self.vapi.cli("clear trace")
435 tx = self.create_stream_labelled_ip4(self.pg0, [44])
436 self.pg0.add_stream(tx)
438 self.pg_enable_capture(self.pg_interfaces)
441 rx = self.pg0.get_capture()
442 self.verify_capture_labelled(self.pg0, rx, tx, [45])
445 self.vapi.cli("clear trace")
446 tx = self.create_stream_ip4(self.pg0, "10.0.0.1")
447 self.pg0.add_stream(tx)
449 self.pg_enable_capture(self.pg_interfaces)
452 rx = self.pg0.get_capture()
453 self.verify_capture_labelled_ip4(self.pg0, rx, tx, [45])
458 binding.remove_vpp_config()
459 route_10_0_0_1.remove_vpp_config()
461 def test_imposition(self):
462 """ MPLS label imposition test """
465 # Add a non-recursive route with a single out label
467 route_10_0_0_1 = IpRoute(self, "10.0.0.1", 32,
468 [RoutePath(self.pg0.remote_ip4,
469 self.pg0.sw_if_index,
471 route_10_0_0_1.add_vpp_config()
474 # a stream that matches the route for 10.0.0.1
475 # PG0 is in the default table
477 self.vapi.cli("clear trace")
478 tx = self.create_stream_ip4(self.pg0, "10.0.0.1")
479 self.pg0.add_stream(tx)
481 self.pg_enable_capture(self.pg_interfaces)
484 rx = self.pg0.get_capture()
485 self.verify_capture_labelled_ip4(self.pg0, rx, tx, [32])
488 # Add a non-recursive route with a 3 out labels
490 route_10_0_0_2 = IpRoute(self, "10.0.0.2", 32,
491 [RoutePath(self.pg0.remote_ip4,
492 self.pg0.sw_if_index,
493 labels=[32, 33, 34])])
494 route_10_0_0_2.add_vpp_config()
497 # a stream that matches the route for 10.0.0.1
498 # PG0 is in the default table
500 self.vapi.cli("clear trace")
501 tx = self.create_stream_ip4(self.pg0, "10.0.0.2")
502 self.pg0.add_stream(tx)
504 self.pg_enable_capture(self.pg_interfaces)
507 rx = self.pg0.get_capture()
508 self.verify_capture_labelled_ip4(self.pg0, rx, tx, [32, 33, 34])
511 # add a recursive path, with output label, via the 1 label route
513 route_11_0_0_1 = IpRoute(self, "11.0.0.1", 32,
514 [RoutePath("10.0.0.1",
517 route_11_0_0_1.add_vpp_config()
520 # a stream that matches the route for 11.0.0.1, should pick up
521 # the label stack for 11.0.0.1 and 10.0.0.1
523 self.vapi.cli("clear trace")
524 tx = self.create_stream_ip4(self.pg0, "11.0.0.1")
525 self.pg0.add_stream(tx)
527 self.pg_enable_capture(self.pg_interfaces)
530 rx = self.pg0.get_capture()
531 self.verify_capture_labelled_ip4(self.pg0, rx, tx, [32, 44])
534 # add a recursive path, with 2 labels, via the 3 label route
536 route_11_0_0_2 = IpRoute(self, "11.0.0.2", 32,
537 [RoutePath("10.0.0.2",
540 route_11_0_0_2.add_vpp_config()
543 # a stream that matches the route for 11.0.0.1, should pick up
544 # the label stack for 11.0.0.1 and 10.0.0.1
546 self.vapi.cli("clear trace")
547 tx = self.create_stream_ip4(self.pg0, "11.0.0.2")
548 self.pg0.add_stream(tx)
550 self.pg_enable_capture(self.pg_interfaces)
553 rx = self.pg0.get_capture()
554 self.verify_capture_labelled_ip4(
555 self.pg0, rx, tx, [32, 33, 34, 44, 45])
560 route_11_0_0_2.remove_vpp_config()
561 route_11_0_0_1.remove_vpp_config()
562 route_10_0_0_2.remove_vpp_config()
563 route_10_0_0_1.remove_vpp_config()
565 def test_tunnel(self):
566 """ MPLS Tunnel Tests """
569 # Create a tunnel with a single out label
571 nh_addr = socket.inet_pton(socket.AF_INET, self.pg0.remote_ip4)
573 reply = self.vapi.mpls_tunnel_add_del(
574 0xffffffff, # don't know the if index yet
577 self.pg0.sw_if_index,
578 0, # next-hop-table-id
582 self.vapi.sw_interface_set_flags(reply.sw_if_index, admin_up_down=1)
585 # add an unlabelled route through the new tunnel
587 dest_addr = socket.inet_pton(socket.AF_INET, "10.0.0.3")
588 nh_addr = socket.inet_pton(socket.AF_INET, "0.0.0.0")
591 self.vapi.ip_add_del_route(
594 nh_addr, # all zeros next-hop - tunnel is p2p
595 reply.sw_if_index, # sw_if_index of the new tunnel
597 0, # next-hop-table-id
602 self.vapi.cli("clear trace")
603 tx = self.create_stream_ip4(self.pg0, "10.0.0.3")
604 self.pg0.add_stream(tx)
606 self.pg_enable_capture(self.pg_interfaces)
609 rx = self.pg0.get_capture()
610 self.verify_capture_tunneled_ip4(self.pg0, rx, tx, [44, 46])
612 def test_v4_exp_null(self):
613 """ MPLS V4 Explicit NULL test """
616 # The first test case has an MPLS TTL of 0
617 # all packet should be dropped
619 tx = self.create_stream_labelled_ip4(self.pg0, [0], 0)
620 self.pg0.add_stream(tx)
622 self.pg_enable_capture(self.pg_interfaces)
625 self.pg0.assert_nothing_captured(remark="MPLS TTL=0 packets forwarded")
628 # a stream with a non-zero MPLS TTL
629 # PG0 is in the default table
631 self.vapi.cli("clear trace")
632 tx = self.create_stream_labelled_ip4(self.pg0, [0])
633 self.pg0.add_stream(tx)
635 self.pg_enable_capture(self.pg_interfaces)
638 rx = self.pg0.get_capture()
639 self.verify_capture_ip4(self.pg0, rx, tx)
642 # a stream with a non-zero MPLS TTL
644 # we are ensuring the post-pop lookup occurs in the VRF table
646 self.vapi.cli("clear trace")
647 tx = self.create_stream_labelled_ip4(self.pg1, [0])
648 self.pg1.add_stream(tx)
650 self.pg_enable_capture(self.pg_interfaces)
653 rx = self.pg1.get_capture()
654 self.verify_capture_ip4(self.pg0, rx, tx)
656 def test_v6_exp_null(self):
657 """ MPLS V6 Explicit NULL test """
660 # a stream with a non-zero MPLS TTL
661 # PG0 is in the default table
663 self.vapi.cli("clear trace")
664 tx = self.create_stream_labelled_ip6(self.pg0, 2, 2)
665 self.pg0.add_stream(tx)
667 self.pg_enable_capture(self.pg_interfaces)
670 rx = self.pg0.get_capture()
671 self.verify_capture_ip6(self.pg0, rx, tx)
674 # a stream with a non-zero MPLS TTL
676 # we are ensuring the post-pop lookup occurs in the VRF table
678 self.vapi.cli("clear trace")
679 tx = self.create_stream_labelled_ip6(self.pg1, 2, 2)
680 self.pg1.add_stream(tx)
682 self.pg_enable_capture(self.pg_interfaces)
685 rx = self.pg1.get_capture()
686 self.verify_capture_ip6(self.pg0, rx, tx)
692 # A de-agg route - next-hop lookup in default table
694 route_34_eos = MplsRoute(self, 34, 1,
695 [RoutePath("0.0.0.0",
698 route_34_eos.add_vpp_config()
701 # ping an interface in the default table
702 # PG0 is in the default table
704 self.vapi.cli("clear trace")
705 tx = self.create_stream_labelled_ip4(self.pg0, [34], ping=1,
707 self.pg0.add_stream(tx)
709 self.pg_enable_capture(self.pg_interfaces)
712 rx = self.pg0.get_capture()
713 self.verify_capture_ip4(self.pg0, rx, tx, ping_resp=1)
716 # A de-agg route - next-hop lookup in non-default table
718 route_35_eos = MplsRoute(self, 35, 1,
719 [RoutePath("0.0.0.0",
722 route_35_eos.add_vpp_config()
725 # ping an interface in the non-default table
726 # PG0 is in the default table. packet arrive labelled in the
727 # default table and egress unlabelled in the non-default
729 self.vapi.cli("clear trace")
730 tx = self.create_stream_labelled_ip4(self.pg0, [35], ping=1, ip_itf=self.pg1)
731 self.pg0.add_stream(tx)
733 self.pg_enable_capture(self.pg_interfaces)
736 rx = self.pg1.get_capture()
737 self.verify_capture_ip4(self.pg1, rx, tx, ping_resp=1)
739 route_35_eos.remove_vpp_config()
740 route_34_eos.remove_vpp_config()
742 if __name__ == '__main__':
743 unittest.main(testRunner=VppTestRunner)