| #!/usr/bin/env python3 |
| |
| import unittest |
| |
| from framework import VppTestCase |
| from asfframework import VppTestRunner |
| from vpp_ip_route import ( |
| VppIpRoute, |
| VppRoutePath, |
| VppMplsRoute, |
| VppMplsTable, |
| VppMplsLabel, |
| ) |
| from vpp_mpls_tunnel_interface import VppMPLSTunnelInterface |
| |
| from scapy.packet import Raw |
| from scapy.layers.l2 import Ether |
| from scapy.layers.inet import IP, UDP |
| from scapy.layers.inet6 import IPv6 |
| from scapy.contrib.mpls import MPLS |
| |
| |
| def verify_filter(capture, sent): |
| if not len(capture) == len(sent): |
| # filter out any IPv6 RAs from the capture |
| for p in capture: |
| if p.haslayer(IPv6): |
| capture.remove(p) |
| return capture |
| |
| |
| def verify_mpls_stack(tst, rx, mpls_labels): |
| # the rx'd packet has the MPLS label popped |
| eth = rx[Ether] |
| tst.assertEqual(eth.type, 0x8847) |
| |
| rx_mpls = rx[MPLS] |
| |
| for ii in range(len(mpls_labels)): |
| tst.assertEqual(rx_mpls.label, mpls_labels[ii].value) |
| tst.assertEqual(rx_mpls.cos, mpls_labels[ii].exp) |
| tst.assertEqual(rx_mpls.ttl, mpls_labels[ii].ttl) |
| |
| if ii == len(mpls_labels) - 1: |
| tst.assertEqual(rx_mpls.s, 1) |
| else: |
| # not end of stack |
| tst.assertEqual(rx_mpls.s, 0) |
| # pop the label to expose the next |
| rx_mpls = rx_mpls[MPLS].payload |
| |
| |
| class TestSRMPLS(VppTestCase): |
| """SR-MPLS Test Case""" |
| |
| @classmethod |
| def setUpClass(cls): |
| super(TestSRMPLS, cls).setUpClass() |
| |
| @classmethod |
| def tearDownClass(cls): |
| super(TestSRMPLS, cls).tearDownClass() |
| |
| def setUp(self): |
| super(TestSRMPLS, self).setUp() |
| |
| # create 2 pg interfaces |
| self.create_pg_interfaces(range(4)) |
| |
| # setup both interfaces |
| # assign them different tables. |
| table_id = 0 |
| self.tables = [] |
| |
| tbl = VppMplsTable(self, 0) |
| tbl.add_vpp_config() |
| self.tables.append(tbl) |
| |
| for i in self.pg_interfaces: |
| i.admin_up() |
| i.config_ip4() |
| i.resolve_arp() |
| i.config_ip6() |
| i.resolve_ndp() |
| i.enable_mpls() |
| |
| def tearDown(self): |
| for i in self.pg_interfaces: |
| i.unconfig_ip4() |
| i.unconfig_ip6() |
| i.disable_mpls() |
| i.admin_down() |
| super(TestSRMPLS, self).tearDown() |
| |
| def create_stream_ip4(self, src_if, dst_ip, ip_ttl=64, ip_dscp=0): |
| self.reset_packet_infos() |
| pkts = [] |
| for i in range(0, 257): |
| info = self.create_packet_info(src_if, src_if) |
| payload = self.info_to_payload(info) |
| p = ( |
| Ether(dst=src_if.local_mac, src=src_if.remote_mac) |
| / IP(src=src_if.remote_ip4, dst=dst_ip, ttl=ip_ttl, tos=ip_dscp) |
| / UDP(sport=1234, dport=1234) |
| / Raw(payload) |
| ) |
| info.data = p.copy() |
| pkts.append(p) |
| return pkts |
| |
| def verify_capture_labelled_ip4( |
| self, src_if, capture, sent, mpls_labels, ip_ttl=None |
| ): |
| try: |
| capture = verify_filter(capture, sent) |
| |
| self.assertEqual(len(capture), len(sent)) |
| |
| for i in range(len(capture)): |
| tx = sent[i] |
| rx = capture[i] |
| tx_ip = tx[IP] |
| rx_ip = rx[IP] |
| |
| verify_mpls_stack(self, rx, mpls_labels) |
| |
| self.assertEqual(rx_ip.src, tx_ip.src) |
| self.assertEqual(rx_ip.dst, tx_ip.dst) |
| if not ip_ttl: |
| # IP processing post pop has decremented the TTL |
| self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl) |
| else: |
| self.assertEqual(rx_ip.ttl, ip_ttl) |
| |
| except: |
| raise |
| |
| def verify_capture_tunneled_ip4(self, src_if, capture, sent, mpls_labels): |
| try: |
| capture = verify_filter(capture, sent) |
| |
| self.assertEqual(len(capture), len(sent)) |
| |
| for i in range(len(capture)): |
| tx = sent[i] |
| rx = capture[i] |
| tx_ip = tx[IP] |
| rx_ip = rx[IP] |
| |
| verify_mpls_stack(self, rx, mpls_labels) |
| |
| self.assertEqual(rx_ip.src, tx_ip.src) |
| self.assertEqual(rx_ip.dst, tx_ip.dst) |
| # IP processing post pop has decremented the TTL |
| self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl) |
| |
| except: |
| raise |
| |
| def test_sr_mpls(self): |
| """SR MPLS""" |
| |
| # |
| # A simple MPLS xconnect - neos label in label out |
| # |
| route_32_eos = VppMplsRoute( |
| self, |
| 32, |
| 0, |
| [ |
| VppRoutePath( |
| self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(32)] |
| ) |
| ], |
| ) |
| route_32_eos.add_vpp_config() |
| |
| # |
| # A binding SID with only one label |
| # |
| self.vapi.sr_mpls_policy_add(999, 1, 0, [32]) |
| |
| # |
| # A labeled IP route that resolves thru the binding SID |
| # |
| ip_10_0_0_1 = VppIpRoute( |
| self, |
| "10.0.0.1", |
| 32, |
| [ |
| VppRoutePath( |
| "0.0.0.0", 0xFFFFFFFF, nh_via_label=999, labels=[VppMplsLabel(55)] |
| ) |
| ], |
| ) |
| ip_10_0_0_1.add_vpp_config() |
| |
| tx = self.create_stream_ip4(self.pg1, "10.0.0.1") |
| rx = self.send_and_expect(self.pg1, tx, self.pg0) |
| self.verify_capture_labelled_ip4( |
| self.pg0, rx, tx, [VppMplsLabel(32), VppMplsLabel(55)] |
| ) |
| |
| # |
| # An unlabeled IP route that resolves thru the binding SID |
| # |
| ip_10_0_0_1 = VppIpRoute( |
| self, |
| "10.0.0.2", |
| 32, |
| [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_via_label=999)], |
| ) |
| ip_10_0_0_1.add_vpp_config() |
| |
| tx = self.create_stream_ip4(self.pg1, "10.0.0.2") |
| rx = self.send_and_expect(self.pg1, tx, self.pg0) |
| self.verify_capture_labelled_ip4(self.pg0, rx, tx, [VppMplsLabel(32)]) |
| |
| self.vapi.sr_mpls_policy_del(999) |
| |
| # |
| # this time the SID has many labels pushed |
| # |
| self.vapi.sr_mpls_policy_add(999, 1, 0, [32, 33, 34]) |
| |
| tx = self.create_stream_ip4(self.pg1, "10.0.0.1") |
| rx = self.send_and_expect(self.pg1, tx, self.pg0) |
| self.verify_capture_labelled_ip4( |
| self.pg0, |
| rx, |
| tx, |
| [VppMplsLabel(32), VppMplsLabel(33), VppMplsLabel(34), VppMplsLabel(55)], |
| ) |
| tx = self.create_stream_ip4(self.pg1, "10.0.0.2") |
| rx = self.send_and_expect(self.pg1, tx, self.pg0) |
| self.verify_capture_labelled_ip4( |
| self.pg0, rx, tx, [VppMplsLabel(32), VppMplsLabel(33), VppMplsLabel(34)] |
| ) |
| |
| # |
| # Resolve an MPLS tunnel via the SID |
| # |
| mpls_tun = VppMPLSTunnelInterface( |
| self, |
| [ |
| VppRoutePath( |
| "0.0.0.0", |
| 0xFFFFFFFF, |
| nh_via_label=999, |
| labels=[VppMplsLabel(44), VppMplsLabel(46)], |
| ) |
| ], |
| ) |
| mpls_tun.add_vpp_config() |
| mpls_tun.admin_up() |
| |
| # |
| # add an unlabelled route through the new tunnel |
| # |
| route_10_0_0_3 = VppIpRoute( |
| self, "10.0.0.3", 32, [VppRoutePath("0.0.0.0", mpls_tun._sw_if_index)] |
| ) |
| route_10_0_0_3.add_vpp_config() |
| self.logger.info(self.vapi.cli("sh mpls tun 0")) |
| self.logger.info(self.vapi.cli("sh adj 21")) |
| |
| tx = self.create_stream_ip4(self.pg1, "10.0.0.3") |
| rx = self.send_and_expect(self.pg1, tx, self.pg0) |
| self.verify_capture_tunneled_ip4( |
| self.pg0, |
| rx, |
| tx, |
| [ |
| VppMplsLabel(32), |
| VppMplsLabel(33), |
| VppMplsLabel(34), |
| VppMplsLabel(44), |
| VppMplsLabel(46), |
| ], |
| ) |
| |
| # |
| # add a labelled route through the new tunnel |
| # |
| route_10_0_0_3 = VppIpRoute( |
| self, |
| "10.0.0.4", |
| 32, |
| [VppRoutePath("0.0.0.0", mpls_tun._sw_if_index, labels=[VppMplsLabel(55)])], |
| ) |
| route_10_0_0_3.add_vpp_config() |
| |
| tx = self.create_stream_ip4(self.pg1, "10.0.0.4") |
| rx = self.send_and_expect(self.pg1, tx, self.pg0) |
| self.verify_capture_tunneled_ip4( |
| self.pg0, |
| rx, |
| tx, |
| [ |
| VppMplsLabel(32), |
| VppMplsLabel(33), |
| VppMplsLabel(34), |
| VppMplsLabel(44), |
| VppMplsLabel(46), |
| VppMplsLabel(55), |
| ], |
| ) |
| |
| self.vapi.sr_mpls_policy_del(999) |
| |
| |
| if __name__ == "__main__": |
| unittest.main(testRunner=VppTestRunner) |