test/test_udp.py

   1 #!/usr/bin/env python
   2
   3 from framework import VppTestCase, VppTestRunner
   4 from vpp_udp_encap import *
   5 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
   6
   7 from scapy.packet import Raw
   8 from scapy.layers.l2 import Ether, ARP
   9 from scapy.layers.inet import IP, UDP
  10 from scapy.layers.inet6 import IPv6
  11 from scapy.contrib.mpls import MPLS
  12
  13
  14 class TestUdpEncap(VppTestCase):
  15     """ UDP Encap Test Case """
  16
  17     def setUp(self):
  18         super(TestUdpEncap, self).setUp()
  19
  20         # create 2 pg interfaces
  21         self.create_pg_interfaces(range(4))
  22
  23         # setup interfaces
  24         # assign them different tables.
  25         table_id = 0
  26         self.tables = []
  27
  28         for i in self.pg_interfaces:
  29             i.admin_up()
  30
  31             if table_id != 0:
  32                 tbl = VppIpTable(self, table_id)
  33                 tbl.add_vpp_config()
  34                 self.tables.append(tbl)
  35                 tbl = VppIpTable(self, table_id, is_ip6=1)
  36                 tbl.add_vpp_config()
  37                 self.tables.append(tbl)
  38
  39             i.set_table_ip4(table_id)
  40             i.set_table_ip6(table_id)
  41             i.config_ip4()
  42             i.resolve_arp()
  43             i.config_ip6()
  44             i.resolve_ndp()
  45             table_id += 1
  46
  47     def tearDown(self):
  48         for i in self.pg_interfaces:
  49             i.unconfig_ip4()
  50             i.unconfig_ip6()
  51             i.ip6_disable()
  52             i.set_table_ip4(0)
  53             i.set_table_ip6(0)
  54             i.admin_down()
  55         super(TestUdpEncap, self).tearDown()
  56
  57     def validate_outer4(self, rx, encap_obj):
  58         self.assertEqual(rx[IP].src, encap_obj.src_ip_s)
  59         self.assertEqual(rx[IP].dst, encap_obj.dst_ip_s)
  60         self.assertEqual(rx[UDP].sport, encap_obj.src_port)
  61         self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
  62
  63     def validate_outer6(self, rx, encap_obj):
  64         self.assertEqual(rx[IPv6].src, encap_obj.src_ip_s)
  65         self.assertEqual(rx[IPv6].dst, encap_obj.dst_ip_s)
  66         self.assertEqual(rx[UDP].sport, encap_obj.src_port)
  67         self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
  68
  69     def validate_inner4(self, rx, tx, ttl=None):
  70         self.assertEqual(rx.src, tx[IP].src)
  71         self.assertEqual(rx.dst, tx[IP].dst)
  72         if ttl:
  73             self.assertEqual(rx.ttl, ttl)
  74         else:
  75             self.assertEqual(rx.ttl, tx[IP].ttl)
  76
  77     def validate_inner6(self, rx, tx):
  78         self.assertEqual(rx.src, tx[IPv6].src)
  79         self.assertEqual(rx.dst, tx[IPv6].dst)
  80         self.assertEqual(rx.hlim, tx[IPv6].hlim)
  81
  82     def send_and_expect(self, input, output, pkts):
  83         self.vapi.cli("clear trace")
  84         input.add_stream(pkts)
  85         self.pg_enable_capture(self.pg_interfaces)
  86         self.pg_start()
  87         rx = output.get_capture(len(pkts))
  88         return rx
  89
  90     def test_udp_encap(self):
  91         """ UDP Encap test
  92         """
  93
  94         #
  95         # construct a UDP encap object through each of the peers
  96         # v4 through the first two peears, v6 through the second.
  97         #
  98         udp_encap_0 = VppUdpEncap(self, 0,
  99                                   self.pg0.local_ip4,
 100                                   self.pg0.remote_ip4,
 101                                   330, 440)
 102         udp_encap_1 = VppUdpEncap(self, 1,
 103                                   self.pg1.local_ip4,
 104                                   self.pg1.remote_ip4,
 105                                   331, 441,
 106                                   table_id=1)
 107         udp_encap_2 = VppUdpEncap(self, 2,
 108                                   self.pg2.local_ip6,
 109                                   self.pg2.remote_ip6,
 110                                   332, 442,
 111                                   table_id=2,
 112                                   is_ip6=1)
 113         udp_encap_3 = VppUdpEncap(self, 3,
 114                                   self.pg3.local_ip6,
 115                                   self.pg3.remote_ip6,
 116                                   333, 443,
 117                                   table_id=3,
 118                                   is_ip6=1)
 119         udp_encap_0.add_vpp_config()
 120         udp_encap_1.add_vpp_config()
 121         udp_encap_2.add_vpp_config()
 122         udp_encap_3.add_vpp_config()
 123
 124         #
 125         # Routes via each UDP encap object - all combinations of v4 and v6.
 126         #
 127         route_4o4 = VppIpRoute(self, "1.1.0.1", 32,
 128                                [VppRoutePath("0.0.0.0",
 129                                              0xFFFFFFFF,
 130                                              is_udp_encap=1,
 131                                              next_hop_id=0)])
 132         route_4o6 = VppIpRoute(self, "1.1.2.1", 32,
 133                                [VppRoutePath("0.0.0.0",
 134                                              0xFFFFFFFF,
 135                                              is_udp_encap=1,
 136                                              next_hop_id=2)])
 137         route_6o4 = VppIpRoute(self, "2001::1", 128,
 138                                [VppRoutePath("0.0.0.0",
 139                                              0xFFFFFFFF,
 140                                              is_udp_encap=1,
 141                                              next_hop_id=1)],
 142                                is_ip6=1)
 143         route_6o6 = VppIpRoute(self, "2001::3", 128,
 144                                [VppRoutePath("0.0.0.0",
 145                                              0xFFFFFFFF,
 146                                              is_udp_encap=1,
 147                                              next_hop_id=3)],
 148                                is_ip6=1)
 149         route_4o4.add_vpp_config()
 150         route_4o6.add_vpp_config()
 151         route_6o6.add_vpp_config()
 152         route_6o4.add_vpp_config()
 153
 154         #
 155         # 4o4 encap
 156         #
 157         p_4o4 = (Ether(src=self.pg0.remote_mac,
 158                        dst=self.pg0.local_mac) /
 159                  IP(src="2.2.2.2", dst="1.1.0.1") /
 160                  UDP(sport=1234, dport=1234) /
 161                  Raw('\xa5' * 100))
 162         rx = self.send_and_expect(self.pg0, self.pg0, p_4o4*65)
 163         for p in rx:
 164             self.validate_outer4(p, udp_encap_0)
 165             p = IP(p["UDP"].payload.load)
 166             self.validate_inner4(p, p_4o4)
 167
 168         #
 169         # 4o6 encap
 170         #
 171         p_4o6 = (Ether(src=self.pg0.remote_mac,
 172                        dst=self.pg0.local_mac) /
 173                  IP(src="2.2.2.2", dst="1.1.2.1") /
 174                  UDP(sport=1234, dport=1234) /
 175                  Raw('\xa5' * 100))
 176         rx = self.send_and_expect(self.pg0, self.pg2, p_4o6*65)
 177         for p in rx:
 178             self.validate_outer6(p, udp_encap_2)
 179             p = IP(p["UDP"].payload.load)
 180             self.validate_inner4(p, p_4o6)
 181
 182         #
 183         # 6o4 encap
 184         #
 185         p_6o4 = (Ether(src=self.pg0.remote_mac,
 186                        dst=self.pg0.local_mac) /
 187                  IPv6(src="2001::100", dst="2001::1") /
 188                  UDP(sport=1234, dport=1234) /
 189                  Raw('\xa5' * 100))
 190         rx = self.send_and_expect(self.pg0, self.pg1, p_6o4*65)
 191         for p in rx:
 192             self.validate_outer4(p, udp_encap_1)
 193             p = IPv6(p["UDP"].payload.load)
 194             self.validate_inner6(p, p_6o4)
 195
 196         #
 197         # 6o6 encap
 198         #
 199         p_6o6 = (Ether(src=self.pg0.remote_mac,
 200                        dst=self.pg0.local_mac) /
 201                  IPv6(src="2001::100", dst="2001::3") /
 202                  UDP(sport=1234, dport=1234) /
 203                  Raw('\xa5' * 100))
 204         rx = self.send_and_expect(self.pg0, self.pg3, p_6o6*65)
 205         for p in rx:
 206             self.validate_outer6(p, udp_encap_3)
 207             p = IPv6(p["UDP"].payload.load)
 208             self.validate_inner6(p, p_6o6)
 209
 210         #
 211         # A route with an output label
 212         # the TTL of the inner packet is decremented on LSP ingress
 213         #
 214         route_4oMPLSo4 = VppIpRoute(self, "1.1.2.22", 32,
 215                                     [VppRoutePath("0.0.0.0",
 216                                                   0xFFFFFFFF,
 217                                                   is_udp_encap=1,
 218                                                   next_hop_id=1,
 219                                                   labels=[66])])
 220         route_4oMPLSo4.add_vpp_config()
 221
 222         p_4omo4 = (Ether(src=self.pg0.remote_mac,
 223                          dst=self.pg0.local_mac) /
 224                    IP(src="2.2.2.2", dst="1.1.2.22") /
 225                    UDP(sport=1234, dport=1234) /
 226                    Raw('\xa5' * 100))
 227         rx = self.send_and_expect(self.pg0, self.pg1, p_4omo4*65)
 228         for p in rx:
 229             self.validate_outer4(p, udp_encap_1)
 230             p = MPLS(p["UDP"].payload.load)
 231             self.validate_inner4(p, p_4omo4, ttl=63)
 232
 233
 234 if __name__ == '__main__':
 235     unittest.main(testRunner=VppTestRunner)