test/test_udp.py

   1 #!/usr/bin/env python3
   2 import unittest
   3 from framework import VppTestCase, VppTestRunner
   4
   5 from vpp_udp_encap import find_udp_encap, VppUdpEncap
   6 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, VppMplsLabel, \
   7     FibPathType
   8
   9 from scapy.packet import Raw
  10 from scapy.layers.l2 import Ether
  11 from scapy.layers.inet import IP, UDP
  12 from scapy.layers.inet6 import IPv6
  13 from scapy.contrib.mpls import MPLS
  14
  15 NUM_PKTS = 67
  16
  17
  18 class TestUdpEncap(VppTestCase):
  19     """ UDP Encap Test Case """
  20
  21     @classmethod
  22     def setUpClass(cls):
  23         super(TestUdpEncap, cls).setUpClass()
  24
  25     @classmethod
  26     def tearDownClass(cls):
  27         super(TestUdpEncap, cls).tearDownClass()
  28
  29     def setUp(self):
  30         super(TestUdpEncap, self).setUp()
  31
  32         # create 2 pg interfaces
  33         self.create_pg_interfaces(range(4))
  34
  35         # setup interfaces
  36         # assign them different tables.
  37         table_id = 0
  38         self.tables = []
  39
  40         for i in self.pg_interfaces:
  41             i.admin_up()
  42
  43             if table_id != 0:
  44                 tbl = VppIpTable(self, table_id)
  45                 tbl.add_vpp_config()
  46                 self.tables.append(tbl)
  47                 tbl = VppIpTable(self, table_id, is_ip6=1)
  48                 tbl.add_vpp_config()
  49                 self.tables.append(tbl)
  50
  51             i.set_table_ip4(table_id)
  52             i.set_table_ip6(table_id)
  53             i.config_ip4()
  54             i.resolve_arp()
  55             i.config_ip6()
  56             i.resolve_ndp()
  57             table_id += 1
  58
  59     def tearDown(self):
  60         for i in self.pg_interfaces:
  61             i.unconfig_ip4()
  62             i.unconfig_ip6()
  63             i.set_table_ip4(0)
  64             i.set_table_ip6(0)
  65             i.admin_down()
  66         super(TestUdpEncap, self).tearDown()
  67
  68     def validate_outer4(self, rx, encap_obj):
  69         self.assertEqual(rx[IP].src, encap_obj.src_ip_s)
  70         self.assertEqual(rx[IP].dst, encap_obj.dst_ip_s)
  71         self.assertEqual(rx[UDP].sport, encap_obj.src_port)
  72         self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
  73
  74     def validate_outer6(self, rx, encap_obj):
  75         self.assertEqual(rx[IPv6].src, encap_obj.src_ip_s)
  76         self.assertEqual(rx[IPv6].dst, encap_obj.dst_ip_s)
  77         self.assertEqual(rx[UDP].sport, encap_obj.src_port)
  78         self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
  79
  80     def validate_inner4(self, rx, tx, ttl=None):
  81         self.assertEqual(rx[IP].src, tx[IP].src)
  82         self.assertEqual(rx[IP].dst, tx[IP].dst)
  83         if ttl:
  84             self.assertEqual(rx[IP].ttl, ttl)
  85         else:
  86             self.assertEqual(rx[IP].ttl, tx[IP].ttl)
  87
  88     def validate_inner6(self, rx, tx):
  89         self.assertEqual(rx.src, tx[IPv6].src)
  90         self.assertEqual(rx.dst, tx[IPv6].dst)
  91         self.assertEqual(rx.hlim, tx[IPv6].hlim)
  92
  93     def test_udp_encap(self):
  94         """ UDP Encap test
  95         """
  96
  97         #
  98         # construct a UDP encap object through each of the peers
  99         # v4 through the first two peers, v6 through the second.
 100         #
 101         udp_encap_0 = VppUdpEncap(self,
 102                                   self.pg0.local_ip4,
 103                                   self.pg0.remote_ip4,
 104                                   330, 440)
 105         udp_encap_1 = VppUdpEncap(self,
 106                                   self.pg1.local_ip4,
 107                                   self.pg1.remote_ip4,
 108                                   331, 441,
 109                                   table_id=1)
 110         udp_encap_2 = VppUdpEncap(self,
 111                                   self.pg2.local_ip6,
 112                                   self.pg2.remote_ip6,
 113                                   332, 442,
 114                                   table_id=2)
 115         udp_encap_3 = VppUdpEncap(self,
 116                                   self.pg3.local_ip6,
 117                                   self.pg3.remote_ip6,
 118                                   333, 443,
 119                                   table_id=3)
 120         udp_encap_0.add_vpp_config()
 121         udp_encap_1.add_vpp_config()
 122         udp_encap_2.add_vpp_config()
 123         udp_encap_3.add_vpp_config()
 124
 125         self.logger.info(self.vapi.cli("sh udp encap"))
 126
 127         self.assertTrue(find_udp_encap(self, udp_encap_2))
 128         self.assertTrue(find_udp_encap(self, udp_encap_3))
 129         self.assertTrue(find_udp_encap(self, udp_encap_0))
 130         self.assertTrue(find_udp_encap(self, udp_encap_1))
 131
 132         #
 133         # Routes via each UDP encap object - all combinations of v4 and v6.
 134         #
 135         route_4o4 = VppIpRoute(
 136             self, "1.1.0.1", 32,
 137             [VppRoutePath("0.0.0.0",
 138                           0xFFFFFFFF,
 139                           type=FibPathType.FIB_PATH_TYPE_UDP_ENCAP,
 140                           next_hop_id=udp_encap_0.id)])
 141         route_4o6 = VppIpRoute(
 142             self, "1.1.2.1", 32,
 143             [VppRoutePath("0.0.0.0",
 144                           0xFFFFFFFF,
 145                           type=FibPathType.FIB_PATH_TYPE_UDP_ENCAP,
 146                           next_hop_id=udp_encap_2.id)])
 147         route_6o4 = VppIpRoute(
 148             self, "2001::1", 128,
 149             [VppRoutePath("0.0.0.0",
 150                           0xFFFFFFFF,
 151                           type=FibPathType.FIB_PATH_TYPE_UDP_ENCAP,
 152                           next_hop_id=udp_encap_1.id)])
 153         route_6o6 = VppIpRoute(
 154             self, "2001::3", 128,
 155             [VppRoutePath("0.0.0.0",
 156                           0xFFFFFFFF,
 157                           type=FibPathType.FIB_PATH_TYPE_UDP_ENCAP,
 158                           next_hop_id=udp_encap_3.id)])
 159         route_4o6.add_vpp_config()
 160         route_6o6.add_vpp_config()
 161         route_6o4.add_vpp_config()
 162         route_4o4.add_vpp_config()
 163
 164         #
 165         # 4o4 encap
 166         #
 167         p_4o4 = (Ether(src=self.pg0.remote_mac,
 168                        dst=self.pg0.local_mac) /
 169                  IP(src="2.2.2.2", dst="1.1.0.1") /
 170                  UDP(sport=1234, dport=1234) /
 171                  Raw(b'\xa5' * 100))
 172         rx = self.send_and_expect(self.pg0, p_4o4*NUM_PKTS, self.pg0)
 173         for p in rx:
 174             self.validate_outer4(p, udp_encap_0)
 175             p = IP(p["UDP"].payload.load)
 176             self.validate_inner4(p, p_4o4)
 177         self.assertEqual(udp_encap_0.get_stats()['packets'], NUM_PKTS)
 178
 179         #
 180         # 4o6 encap
 181         #
 182         p_4o6 = (Ether(src=self.pg0.remote_mac,
 183                        dst=self.pg0.local_mac) /
 184                  IP(src="2.2.2.2", dst="1.1.2.1") /
 185                  UDP(sport=1234, dport=1234) /
 186                  Raw(b'\xa5' * 100))
 187         rx = self.send_and_expect(self.pg0, p_4o6*NUM_PKTS, self.pg2)
 188         for p in rx:
 189             self.validate_outer6(p, udp_encap_2)
 190             p = IP(p["UDP"].payload.load)
 191             self.validate_inner4(p, p_4o6)
 192         self.assertEqual(udp_encap_2.get_stats()['packets'], NUM_PKTS)
 193
 194         #
 195         # 6o4 encap
 196         #
 197         p_6o4 = (Ether(src=self.pg0.remote_mac,
 198                        dst=self.pg0.local_mac) /
 199                  IPv6(src="2001::100", dst="2001::1") /
 200                  UDP(sport=1234, dport=1234) /
 201                  Raw(b'\xa5' * 100))
 202         rx = self.send_and_expect(self.pg0, p_6o4*NUM_PKTS, self.pg1)
 203         for p in rx:
 204             self.validate_outer4(p, udp_encap_1)
 205             p = IPv6(p["UDP"].payload.load)
 206             self.validate_inner6(p, p_6o4)
 207         self.assertEqual(udp_encap_1.get_stats()['packets'], NUM_PKTS)
 208
 209         #
 210         # 6o6 encap
 211         #
 212         p_6o6 = (Ether(src=self.pg0.remote_mac,
 213                        dst=self.pg0.local_mac) /
 214                  IPv6(src="2001::100", dst="2001::3") /
 215                  UDP(sport=1234, dport=1234) /
 216                  Raw(b'\xa5' * 100))
 217         rx = self.send_and_expect(self.pg0, p_6o6*NUM_PKTS, self.pg3)
 218         for p in rx:
 219             self.validate_outer6(p, udp_encap_3)
 220             p = IPv6(p["UDP"].payload.load)
 221             self.validate_inner6(p, p_6o6)
 222         self.assertEqual(udp_encap_3.get_stats()['packets'], NUM_PKTS)
 223
 224         #
 225         # A route with an output label
 226         # the TTL of the inner packet is decremented on LSP ingress
 227         #
 228         route_4oMPLSo4 = VppIpRoute(
 229             self, "1.1.2.22", 32,
 230             [VppRoutePath("0.0.0.0",
 231                           0xFFFFFFFF,
 232                           type=FibPathType.FIB_PATH_TYPE_UDP_ENCAP,
 233                           next_hop_id=1,
 234                           labels=[VppMplsLabel(66)])])
 235         route_4oMPLSo4.add_vpp_config()
 236
 237         p_4omo4 = (Ether(src=self.pg0.remote_mac,
 238                          dst=self.pg0.local_mac) /
 239                    IP(src="2.2.2.2", dst="1.1.2.22") /
 240                    UDP(sport=1234, dport=1234) /
 241                    Raw(b'\xa5' * 100))
 242         rx = self.send_and_expect(self.pg0, p_4omo4*NUM_PKTS, self.pg1)
 243         for p in rx:
 244             self.validate_outer4(p, udp_encap_1)
 245             p = MPLS(p["UDP"].payload.load)
 246             self.validate_inner4(p, p_4omo4, ttl=63)
 247         self.assertEqual(udp_encap_1.get_stats()['packets'], 2*NUM_PKTS)
 248
 249
 250 class TestUDP(VppTestCase):
 251     """ UDP Test Case """
 252
 253     @classmethod
 254     def setUpClass(cls):
 255         super(TestUDP, cls).setUpClass()
 256
 257     @classmethod
 258     def tearDownClass(cls):
 259         super(TestUDP, cls).tearDownClass()
 260
 261     def setUp(self):
 262         super(TestUDP, self).setUp()
 263         self.vapi.session_enable_disable(is_enabled=1)
 264         self.create_loopback_interfaces(2)
 265
 266         table_id = 0
 267
 268         for i in self.lo_interfaces:
 269             i.admin_up()
 270
 271             if table_id != 0:
 272                 tbl = VppIpTable(self, table_id)
 273                 tbl.add_vpp_config()
 274
 275             i.set_table_ip4(table_id)
 276             i.config_ip4()
 277             table_id += 1
 278
 279         # Configure namespaces
 280         self.vapi.app_namespace_add_del(namespace_id="0".encode('ascii'),
 281                                         sw_if_index=self.loop0.sw_if_index)
 282         self.vapi.app_namespace_add_del(namespace_id="1".encode('ascii'),
 283                                         sw_if_index=self.loop1.sw_if_index)
 284
 285     def tearDown(self):
 286         for i in self.lo_interfaces:
 287             i.unconfig_ip4()
 288             i.set_table_ip4(0)
 289             i.admin_down()
 290         self.vapi.session_enable_disable(is_enabled=0)
 291         super(TestUDP, self).tearDown()
 292
 293     def test_udp_transfer(self):
 294         """ UDP echo client/server transfer """
 295
 296         # Add inter-table routes
 297         ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
 298                             [VppRoutePath("0.0.0.0",
 299                                           0xffffffff,
 300                                           nh_table_id=1)])
 301         ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
 302                             [VppRoutePath("0.0.0.0",
 303                                           0xffffffff,
 304                                           nh_table_id=0)], table_id=1)
 305         ip_t01.add_vpp_config()
 306         ip_t10.add_vpp_config()
 307
 308         # Start builtin server and client
 309         uri = "udp://" + self.loop0.local_ip4 + "/1234"
 310         error = self.vapi.cli("test echo server appns 0 fifo-size 4 no-echo" +
 311                               "uri " + uri)
 312         if error:
 313             self.logger.critical(error)
 314             self.assertNotIn("failed", error)
 315
 316         error = self.vapi.cli("test echo client mbytes 10 appns 1 " +
 317                               "fifo-size 4 no-output test-bytes " +
 318                               "syn-timeout 2 no-return uri " + uri)
 319         if error:
 320             self.logger.critical(error)
 321             self.assertNotIn("failed", error)
 322
 323         # Delete inter-table routes
 324         ip_t01.remove_vpp_config()
 325         ip_t10.remove_vpp_config()
 326
 327
 328 if __name__ == '__main__':
 329     unittest.main(testRunner=VppTestRunner)