test/vpp_ip_route.py

   1 """
   2   IP Routes
   3
   4   object abstractions for representing IP routes in VPP
   5 """
   6
   7 from vpp_object import *
   8 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
   9
  10 # from vnet/vnet/mpls/mpls_types.h
  11 MPLS_IETF_MAX_LABEL = 0xfffff
  12 MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
  13
  14
  15 class MRouteItfFlags:
  16     MFIB_ITF_FLAG_NONE = 0
  17     MFIB_ITF_FLAG_NEGATE_SIGNAL = 1
  18     MFIB_ITF_FLAG_ACCEPT = 2
  19     MFIB_ITF_FLAG_FORWARD = 4
  20     MFIB_ITF_FLAG_SIGNAL_PRESENT = 8
  21     MFIB_ITF_FLAG_INTERNAL_COPY = 16
  22
  23
  24 class MRouteEntryFlags:
  25     MFIB_ENTRY_FLAG_NONE = 0
  26     MFIB_ENTRY_FLAG_SIGNAL = 1
  27     MFIB_ENTRY_FLAG_DROP = 2
  28     MFIB_ENTRY_FLAG_CONNECTED = 4
  29     MFIB_ENTRY_FLAG_INHERIT_ACCEPT = 8
  30
  31
  32 def find_route(test, ip_addr, len, table_id=0, inet=AF_INET):
  33     if inet == AF_INET:
  34         s = 4
  35         routes = test.vapi.ip_fib_dump()
  36     else:
  37         s = 16
  38         routes = test.vapi.ip6_fib_dump()
  39
  40     route_addr = inet_pton(inet, ip_addr)
  41     for e in routes:
  42         if route_addr == e.address[:s] \
  43            and len == e.address_length \
  44            and table_id == e.table_id:
  45             return True
  46     return False
  47
  48
  49 class VppRoutePath(object):
  50
  51     def __init__(
  52             self,
  53             nh_addr,
  54             nh_sw_if_index,
  55             nh_table_id=0,
  56             labels=[],
  57             nh_via_label=MPLS_LABEL_INVALID,
  58             is_ip6=0):
  59         self.nh_itf = nh_sw_if_index
  60         self.nh_table_id = nh_table_id
  61         self.nh_via_label = nh_via_label
  62         self.nh_labels = labels
  63         if is_ip6:
  64             self.nh_addr = inet_pton(AF_INET6, nh_addr)
  65         else:
  66             self.nh_addr = inet_pton(AF_INET, nh_addr)
  67
  68
  69 class VppMRoutePath(VppRoutePath):
  70
  71     def __init__(self, nh_sw_if_index, flags):
  72         super(VppMRoutePath, self).__init__("0.0.0.0",
  73                                             nh_sw_if_index)
  74         self.nh_i_flags = flags
  75
  76
  77 class VppIpRoute(VppObject):
  78     """
  79     IP Route
  80     """
  81
  82     def __init__(self, test, dest_addr,
  83                  dest_addr_len, paths, table_id=0, is_ip6=0, is_local=0,
  84                  is_unreach=0, is_prohibit=0):
  85         self._test = test
  86         self.paths = paths
  87         self.dest_addr_len = dest_addr_len
  88         self.table_id = table_id
  89         self.is_ip6 = is_ip6
  90         self.is_local = is_local
  91         self.is_unreach = is_unreach
  92         self.is_prohibit = is_prohibit
  93         self.dest_addr_p = dest_addr
  94         if is_ip6:
  95             self.dest_addr = inet_pton(AF_INET6, dest_addr)
  96         else:
  97             self.dest_addr = inet_pton(AF_INET, dest_addr)
  98
  99     def modify(self, paths, is_local=0,
 100                is_unreach=0, is_prohibit=0):
 101         self.paths = paths
 102         self.is_local = is_local
 103         self.is_unreach = is_unreach
 104         self.is_prohibit = is_prohibit
 105
 106     def add_vpp_config(self):
 107         if self.is_local or self.is_unreach or self.is_prohibit:
 108             self._test.vapi.ip_add_del_route(
 109                 self.dest_addr,
 110                 self.dest_addr_len,
 111                 inet_pton(AF_INET6, "::"),
 112                 0xffffffff,
 113                 is_local=self.is_local,
 114                 is_unreach=self.is_unreach,
 115                 is_prohibit=self.is_prohibit,
 116                 table_id=self.table_id,
 117                 is_ipv6=self.is_ip6)
 118         else:
 119             for path in self.paths:
 120                 self._test.vapi.ip_add_del_route(
 121                     self.dest_addr,
 122                     self.dest_addr_len,
 123                     path.nh_addr,
 124                     path.nh_itf,
 125                     table_id=self.table_id,
 126                     next_hop_out_label_stack=path.nh_labels,
 127                     next_hop_n_out_labels=len(
 128                         path.nh_labels),
 129                     next_hop_via_label=path.nh_via_label,
 130                     is_ipv6=self.is_ip6)
 131         self._test.registry.register(self, self._test.logger)
 132
 133     def remove_vpp_config(self):
 134         if self.is_local or self.is_unreach or self.is_prohibit:
 135             self._test.vapi.ip_add_del_route(
 136                 self.dest_addr,
 137                 self.dest_addr_len,
 138                 inet_pton(AF_INET6, "::"),
 139                 0xffffffff,
 140                 is_local=self.is_local,
 141                 is_unreach=self.is_unreach,
 142                 is_prohibit=self.is_prohibit,
 143                 is_add=0,
 144                 table_id=self.table_id,
 145                 is_ipv6=self.is_ip6)
 146         else:
 147             for path in self.paths:
 148                 self._test.vapi.ip_add_del_route(self.dest_addr,
 149                                                  self.dest_addr_len,
 150                                                  path.nh_addr,
 151                                                  path.nh_itf,
 152                                                  table_id=self.table_id,
 153                                                  is_add=0,
 154                                                  is_ipv6=self.is_ip6)
 155
 156     def query_vpp_config(self):
 157         return find_route(self._test,
 158                           self.dest_addr_p,
 159                           self.dest_addr_len,
 160                           self.table_id,
 161                           inet=AF_INET6 if self.is_ip6 == 1 else AF_INET)
 162
 163     def __str__(self):
 164         return self.object_id()
 165
 166     def object_id(self):
 167         return ("%d:%s/%d"
 168                 % (self.table_id,
 169                    self.dest_addr_p,
 170                    self.dest_addr_len))
 171
 172
 173 class VppIpMRoute(VppObject):
 174     """
 175     IP Multicast Route
 176     """
 177
 178     def __init__(self, test, src_addr, grp_addr,
 179                  grp_addr_len, e_flags, paths, table_id=0, is_ip6=0):
 180         self._test = test
 181         self.paths = paths
 182         self.grp_addr_len = grp_addr_len
 183         self.table_id = table_id
 184         self.e_flags = e_flags
 185         self.is_ip6 = is_ip6
 186
 187         if is_ip6:
 188             self.grp_addr = inet_pton(AF_INET6, grp_addr)
 189             self.src_addr = inet_pton(AF_INET6, src_addr)
 190         else:
 191             self.grp_addr = inet_pton(AF_INET, grp_addr)
 192             self.src_addr = inet_pton(AF_INET, src_addr)
 193
 194     def add_vpp_config(self):
 195         for path in self.paths:
 196             self._test.vapi.ip_mroute_add_del(self.src_addr,
 197                                               self.grp_addr,
 198                                               self.grp_addr_len,
 199                                               self.e_flags,
 200                                               path.nh_itf,
 201                                               path.nh_i_flags,
 202                                               table_id=self.table_id,
 203                                               is_ipv6=self.is_ip6)
 204         self._test.registry.register(self, self._test.logger)
 205
 206     def remove_vpp_config(self):
 207         for path in self.paths:
 208             self._test.vapi.ip_mroute_add_del(self.src_addr,
 209                                               self.grp_addr,
 210                                               self.grp_addr_len,
 211                                               self.e_flags,
 212                                               path.nh_itf,
 213                                               path.nh_i_flags,
 214                                               table_id=self.table_id,
 215                                               is_add=0,
 216                                               is_ipv6=self.is_ip6)
 217
 218     def update_entry_flags(self, flags):
 219         self.e_flags = flags
 220         self._test.vapi.ip_mroute_add_del(self.src_addr,
 221                                           self.grp_addr,
 222                                           self.grp_addr_len,
 223                                           self.e_flags,
 224                                           0xffffffff,
 225                                           0,
 226                                           table_id=self.table_id,
 227                                           is_ipv6=self.is_ip6)
 228
 229     def update_path_flags(self, itf, flags):
 230         for path in self.paths:
 231             if path.nh_itf == itf:
 232                 path.nh_i_flags = flags
 233                 break
 234         self._test.vapi.ip_mroute_add_del(self.src_addr,
 235                                           self.grp_addr,
 236                                           self.grp_addr_len,
 237                                           self.e_flags,
 238                                           path.nh_itf,
 239                                           path.nh_i_flags,
 240                                           table_id=self.table_id,
 241                                           is_ipv6=self.is_ip6)
 242
 243     def query_vpp_config(self):
 244         dump = self._test.vapi.ip_fib_dump()
 245         for e in dump:
 246             if self.grp_addr == e.address \
 247                and self.grp_addr_len == e.address_length \
 248                and self.table_id == e.table_id:
 249                 return True
 250         return False
 251
 252     def __str__(self):
 253         return self.object_id()
 254
 255     def object_id(self):
 256         if self.is_ip6:
 257             return ("%d:(%s,%s/%d)"
 258                     % (self.table_id,
 259                        inet_ntop(AF_INET6, self.src_addr),
 260                        inet_ntop(AF_INET6, self.grp_addr),
 261                        self.grp_addr_len))
 262         else:
 263             return ("%d:(%s,%s/%d)"
 264                     % (self.table_id,
 265                        inet_ntop(AF_INET, self.src_addr),
 266                        inet_ntop(AF_INET, self.grp_addr),
 267                        self.grp_addr_len))
 268
 269
 270 class VppMFibSignal(object):
 271     def __init__(self, test, route, interface, packet):
 272         self.route = route
 273         self.interface = interface
 274         self.packet = packet
 275         self.test = test
 276
 277     def compare(self, signal):
 278         self.test.assertEqual(self.interface, signal.sw_if_index)
 279         self.test.assertEqual(self.route.table_id, signal.table_id)
 280         self.test.assertEqual(self.route.grp_addr_len,
 281                               signal.grp_address_len)
 282         for i in range(self.route.grp_addr_len / 8):
 283             self.test.assertEqual(self.route.grp_addr[i],
 284                                   signal.grp_address[i])
 285         if (self.route.grp_addr_len > 32):
 286             for i in range(4):
 287                 self.test.assertEqual(self.route.src_addr[i],
 288                                       signal.src_address[i])
 289
 290
 291 class VppMplsIpBind(VppObject):
 292     """
 293     MPLS to IP Binding
 294     """
 295
 296     def __init__(self, test, local_label, dest_addr, dest_addr_len,
 297                  table_id=0, ip_table_id=0):
 298         self._test = test
 299         self.dest_addr = inet_pton(AF_INET, dest_addr)
 300         self.dest_addr_len = dest_addr_len
 301         self.local_label = local_label
 302         self.table_id = table_id
 303         self.ip_table_id = ip_table_id
 304
 305     def add_vpp_config(self):
 306         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
 307                                             self.dest_addr,
 308                                             self.dest_addr_len,
 309                                             table_id=self.table_id,
 310                                             ip_table_id=self.ip_table_id)
 311         self._test.registry.register(self, self._test.logger)
 312
 313     def remove_vpp_config(self):
 314         self._test.vapi.mpls_ip_bind_unbind(self.local_label,
 315                                             self.dest_addr,
 316                                             self.dest_addr_len,
 317                                             is_bind=0)
 318
 319     def query_vpp_config(self):
 320         dump = self._test.vapi.mpls_fib_dump()
 321         for e in dump:
 322             if self.local_label == e.label \
 323                and self.eos_bit == e.eos_bit \
 324                and self.table_id == e.table_id:
 325                 return True
 326         return False
 327
 328     def __str__(self):
 329         return self.object_id()
 330
 331     def object_id(self):
 332         return ("%d:%s binds %d:%s/%d"
 333                 % (self.table_id,
 334                    self.local_label,
 335                    self.ip_table_id,
 336                    inet_ntop(AF_INET, self.dest_addr),
 337                    self.dest_addr_len))
 338
 339
 340 class VppMplsRoute(VppObject):
 341     """
 342     MPLS Route/LSP
 343     """
 344
 345     def __init__(self, test, local_label, eos_bit, paths, table_id=0):
 346         self._test = test
 347         self.paths = paths
 348         self.local_label = local_label
 349         self.eos_bit = eos_bit
 350         self.table_id = table_id
 351
 352     def add_vpp_config(self):
 353         for path in self.paths:
 354             self._test.vapi.mpls_route_add_del(
 355                 self.local_label,
 356                 self.eos_bit,
 357                 1,
 358                 path.nh_addr,
 359                 path.nh_itf,
 360                 table_id=self.table_id,
 361                 next_hop_out_label_stack=path.nh_labels,
 362                 next_hop_n_out_labels=len(
 363                     path.nh_labels),
 364                 next_hop_via_label=path.nh_via_label,
 365                 next_hop_table_id=path.nh_table_id)
 366         self._test.registry.register(self, self._test.logger)
 367
 368     def remove_vpp_config(self):
 369         for path in self.paths:
 370             self._test.vapi.mpls_route_add_del(self.local_label,
 371                                                self.eos_bit,
 372                                                1,
 373                                                path.nh_addr,
 374                                                path.nh_itf,
 375                                                table_id=self.table_id,
 376                                                is_add=0)
 377
 378     def query_vpp_config(self):
 379         dump = self._test.vapi.mpls_fib_dump()
 380         for e in dump:
 381             if self.local_label == e.label \
 382                and self.eos_bit == e.eos_bit \
 383                and self.table_id == e.table_id:
 384                 return True
 385         return False
 386
 387     def __str__(self):
 388         return self.object_id()
 389
 390     def object_id(self):
 391         return ("%d:%s/%d"
 392                 % (self.table_id,
 393                    self.local_label,
 394                    20+self.eos_bit))