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, VppMplsLabel
   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[IP].src, tx[IP].src)
  71         self.assertEqual(rx[IP].dst, tx[IP].dst)
  72         if ttl:
  73             self.assertEqual(rx[IP].ttl, ttl)
  74         else:
  75             self.assertEqual(rx[IP].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 test_udp_encap(self):
  83         """ UDP Encap test
  84         """
  85
  86         #
  87         # construct a UDP encap object through each of the peers
  88         # v4 through the first two peears, v6 through the second.
  89         #
  90         udp_encap_0 = VppUdpEncap(self, 0,
  91                                   self.pg0.local_ip4,
  92                                   self.pg0.remote_ip4,
  93                                   330, 440)
  94         udp_encap_1 = VppUdpEncap(self, 1,
  95                                   self.pg1.local_ip4,
  96                                   self.pg1.remote_ip4,
  97                                   331, 441,
  98                                   table_id=1)
  99         udp_encap_2 = VppUdpEncap(self, 2,
 100                                   self.pg2.local_ip6,
 101                                   self.pg2.remote_ip6,
 102                                   332, 442,
 103                                   table_id=2,
 104                                   is_ip6=1)
 105         udp_encap_3 = VppUdpEncap(self, 3,
 106                                   self.pg3.local_ip6,
 107                                   self.pg3.remote_ip6,
 108                                   333, 443,
 109                                   table_id=3,
 110                                   is_ip6=1)
 111         udp_encap_0.add_vpp_config()
 112         udp_encap_1.add_vpp_config()
 113         udp_encap_2.add_vpp_config()
 114         udp_encap_3.add_vpp_config()
 115
 116         #
 117         # Routes via each UDP encap object - all combinations of v4 and v6.
 118         #
 119         route_4o4 = VppIpRoute(self, "1.1.0.1", 32,
 120                                [VppRoutePath("0.0.0.0",
 121                                              0xFFFFFFFF,
 122                                              is_udp_encap=1,
 123                                              next_hop_id=0)])
 124         route_4o6 = VppIpRoute(self, "1.1.2.1", 32,
 125                                [VppRoutePath("0.0.0.0",
 126                                              0xFFFFFFFF,
 127                                              is_udp_encap=1,
 128                                              next_hop_id=2)])
 129         route_6o4 = VppIpRoute(self, "2001::1", 128,
 130                                [VppRoutePath("0.0.0.0",
 131                                              0xFFFFFFFF,
 132                                              is_udp_encap=1,
 133                                              next_hop_id=1)],
 134                                is_ip6=1)
 135         route_6o6 = VppIpRoute(self, "2001::3", 128,
 136                                [VppRoutePath("0.0.0.0",
 137                                              0xFFFFFFFF,
 138                                              is_udp_encap=1,
 139                                              next_hop_id=3)],
 140                                is_ip6=1)
 141         route_4o4.add_vpp_config()
 142         route_4o6.add_vpp_config()
 143         route_6o6.add_vpp_config()
 144         route_6o4.add_vpp_config()
 145
 146         #
 147         # 4o4 encap
 148         #
 149         p_4o4 = (Ether(src=self.pg0.remote_mac,
 150                        dst=self.pg0.local_mac) /
 151                  IP(src="2.2.2.2", dst="1.1.0.1") /
 152                  UDP(sport=1234, dport=1234) /
 153                  Raw('\xa5' * 100))
 154         rx = self.send_and_expect(self.pg0, p_4o4*65, self.pg0)
 155         for p in rx:
 156             self.validate_outer4(p, udp_encap_0)
 157             p = IP(p["UDP"].payload.load)
 158             self.validate_inner4(p, p_4o4)
 159
 160         #
 161         # 4o6 encap
 162         #
 163         p_4o6 = (Ether(src=self.pg0.remote_mac,
 164                        dst=self.pg0.local_mac) /
 165                  IP(src="2.2.2.2", dst="1.1.2.1") /
 166                  UDP(sport=1234, dport=1234) /
 167                  Raw('\xa5' * 100))
 168         rx = self.send_and_expect(self.pg0, p_4o6*65, self.pg2)
 169         for p in rx:
 170             self.validate_outer6(p, udp_encap_2)
 171             p = IP(p["UDP"].payload.load)
 172             self.validate_inner4(p, p_4o6)
 173
 174         #
 175         # 6o4 encap
 176         #
 177         p_6o4 = (Ether(src=self.pg0.remote_mac,
 178                        dst=self.pg0.local_mac) /
 179                  IPv6(src="2001::100", dst="2001::1") /
 180                  UDP(sport=1234, dport=1234) /
 181                  Raw('\xa5' * 100))
 182         rx = self.send_and_expect(self.pg0, p_6o4*65, self.pg1)
 183         for p in rx:
 184             self.validate_outer4(p, udp_encap_1)
 185             p = IPv6(p["UDP"].payload.load)
 186             self.validate_inner6(p, p_6o4)
 187
 188         #
 189         # 6o6 encap
 190         #
 191         p_6o6 = (Ether(src=self.pg0.remote_mac,
 192                        dst=self.pg0.local_mac) /
 193                  IPv6(src="2001::100", dst="2001::3") /
 194                  UDP(sport=1234, dport=1234) /
 195                  Raw('\xa5' * 100))
 196         rx = self.send_and_expect(self.pg0, p_6o6*65, self.pg3)
 197         for p in rx:
 198             self.validate_outer6(p, udp_encap_3)
 199             p = IPv6(p["UDP"].payload.load)
 200             self.validate_inner6(p, p_6o6)
 201
 202         #
 203         # A route with an output label
 204         # the TTL of the inner packet is decremented on LSP ingress
 205         #
 206         route_4oMPLSo4 = VppIpRoute(self, "1.1.2.22", 32,
 207                                     [VppRoutePath("0.0.0.0",
 208                                                   0xFFFFFFFF,
 209                                                   is_udp_encap=1,
 210                                                   next_hop_id=1,
 211                                                   labels=[VppMplsLabel(66)])])
 212         route_4oMPLSo4.add_vpp_config()
 213
 214         p_4omo4 = (Ether(src=self.pg0.remote_mac,
 215                          dst=self.pg0.local_mac) /
 216                    IP(src="2.2.2.2", dst="1.1.2.22") /
 217                    UDP(sport=1234, dport=1234) /
 218                    Raw('\xa5' * 100))
 219         rx = self.send_and_expect(self.pg0, p_4omo4*65, self.pg1)
 220         for p in rx:
 221             self.validate_outer4(p, udp_encap_1)
 222             p = MPLS(p["UDP"].payload.load)
 223             self.validate_inner4(p, p_4omo4, ttl=63)
 224
 225
 226 if __name__ == '__main__':
 227     unittest.main(testRunner=VppTestRunner)