4 object abstractions for representing IP routes in VPP
7 from vpp_object import VppObject
8 from vpp_ip import DpoProto, INVALID_INDEX, VppIpAddressUnion, VppIpMPrefix
9 from ipaddress import ip_network, ip_address, IPv4Network, IPv6Network
10 from vpp_papi_exceptions import UnexpectedApiReturnValueError
12 # from vnet/vnet/mpls/mpls_types.h
13 MPLS_IETF_MAX_LABEL = 0xFFFFF
14 MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
23 FIB_PATH_NH_PROTO_IP4 = 0
24 FIB_PATH_NH_PROTO_IP6 = 1
25 FIB_PATH_NH_PROTO_MPLS = 2
26 FIB_PATH_NH_PROTO_ETHERNET = 3
27 FIB_PATH_NH_PROTO_BIER = 4
28 FIB_PATH_NH_PROTO_NSH = 5
32 FIB_PATH_TYPE_NORMAL = 0
33 FIB_PATH_TYPE_LOCAL = 1
34 FIB_PATH_TYPE_DROP = 2
35 FIB_PATH_TYPE_UDP_ENCAP = 3
36 FIB_PATH_TYPE_BIER_IMP = 4
37 FIB_PATH_TYPE_ICMP_UNREACH = 5
38 FIB_PATH_TYPE_ICMP_PROHIBIT = 6
39 FIB_PATH_TYPE_SOURCE_LOOKUP = 7
41 FIB_PATH_TYPE_INTERFACE_RX = 9
42 FIB_PATH_TYPE_CLASSIFY = 10
46 FIB_PATH_FLAG_NONE = 0
47 FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED = 1
48 FIB_PATH_FLAG_RESOLVE_VIA_HOST = 2
49 FIB_PATH_FLAG_POP_PW_CW = 4
57 def mk_network(addr, len):
58 if ip_address(text_type(addr)).version == 4:
59 return IPv4Network("%s/%d" % (addr, len), strict=False)
61 return IPv6Network("%s/%d" % (addr, len), strict=False)
64 def ip_to_dpo_proto(addr):
66 return DpoProto.DPO_PROTO_IP6
68 return DpoProto.DPO_PROTO_IP4
71 def address_proto(ip_addr):
72 if ip_addr.ip_addr.version == 4:
73 return FibPathProto.FIB_PATH_NH_PROTO_IP4
75 return FibPathProto.FIB_PATH_NH_PROTO_IP6
78 def find_route(test, addr, len, table_id=0, sw_if_index=None):
79 prefix = mk_network(addr, len)
81 if 4 == prefix.version:
82 routes = test.vapi.ip_route_dump(table_id, False)
84 routes = test.vapi.ip_route_dump(table_id, True)
87 if table_id == e.route.table_id and str(e.route.prefix) == str(prefix):
91 # should be only one path if the user is looking
92 # for the interface the route is reachable through
93 if e.route.n_paths != 1:
96 return e.route.paths[0].sw_if_index == sw_if_index
101 def find_route_in_dump(dump, route, table):
103 if table.table_id == r.route.table_id and route.prefix == r.route.prefix:
104 if len(route.paths) == r.route.n_paths:
109 def find_mroute_in_dump(dump, route, table):
111 if table.table_id == r.route.table_id and route.prefix == r.route.prefix:
116 def find_mroute(test, grp_addr, src_addr, grp_addr_len, table_id=0):
117 ip_mprefix = VppIpMPrefix(text_type(src_addr), text_type(grp_addr), grp_addr_len)
119 if 4 == ip_mprefix.version:
120 routes = test.vapi.ip_mroute_dump(table_id, False)
122 routes = test.vapi.ip_mroute_dump(table_id, True)
125 if table_id == e.route.table_id and ip_mprefix == e.route.prefix:
130 def find_mpls_route(test, table_id, label, eos_bit, paths=None):
131 dump = test.vapi.mpls_route_dump(table_id)
134 label == e.mr_route.mr_label
135 and eos_bit == e.mr_route.mr_eos
136 and table_id == e.mr_route.mr_table_id
141 if len(paths) != len(e.mr_route.mr_paths):
143 for i in range(len(paths)):
144 if paths[i] != e.mr_route.mr_paths[i]:
150 def fib_interface_ip_prefix(test, addr, len, sw_if_index):
151 # can't use python net here since we need the host bits in the prefix
152 prefix = "%s/%d" % (addr, len)
153 addrs = test.vapi.ip_address_dump(
154 sw_if_index, is_ipv6=(6 == ip_address(addr).version)
158 if a.sw_if_index == sw_if_index and str(a.prefix) == prefix:
163 class VppIpTable(VppObject):
164 def __init__(self, test, table_id, is_ip6=0, register=True):
166 self.table_id = table_id
168 self.register = register
170 def add_vpp_config(self):
171 self._test.vapi.ip_table_add_del(
172 is_add=1, table={"is_ip6": self.is_ip6, "table_id": self.table_id}
175 self._test.registry.register(self, self._test.logger)
178 def remove_vpp_config(self):
179 self._test.vapi.ip_table_add_del(
180 is_add=0, table={"is_ip6": self.is_ip6, "table_id": self.table_id}
183 def replace_begin(self):
184 self._test.vapi.ip_table_replace_begin(
185 table={"is_ip6": self.is_ip6, "table_id": self.table_id}
188 def replace_end(self):
189 self._test.vapi.ip_table_replace_end(
190 table={"is_ip6": self.is_ip6, "table_id": self.table_id}
194 self._test.vapi.ip_table_flush(
195 table={"is_ip6": self.is_ip6, "table_id": self.table_id}
199 return self._test.vapi.ip_route_dump(self.table_id, self.is_ip6)
202 return self._test.vapi.ip_mroute_dump(self.table_id, self.is_ip6)
204 def query_vpp_config(self):
205 if self.table_id == 0:
206 # the default table always exists
208 # find the default route
210 self._test, "::" if self.is_ip6 else "0.0.0.0", 0, self.table_id
214 return "table-%s-%d" % ("v6" if self.is_ip6 == 1 else "v4", self.table_id)
217 class VppIpInterfaceAddress(VppObject):
218 def __init__(self, test, intf, addr, len, bind=None):
223 self.prefix = "%s/%d" % (addr, len)
224 self.host_len = ip_network(self.prefix, strict=False).max_prefixlen
227 self.table_id = bind.table.table_id
229 def add_vpp_config(self):
230 self._test.vapi.sw_interface_add_del_address(
231 sw_if_index=self.intf.sw_if_index, prefix=self.prefix, is_add=1
233 self._test.registry.register(self, self._test.logger)
236 def remove_vpp_config(self):
237 self._test.vapi.sw_interface_add_del_address(
238 sw_if_index=self.intf.sw_if_index, prefix=self.prefix, is_add=0
241 def query_vpp_config(self):
242 # search for the IP address mapping and the two expected
244 v = ip_address(self.addr).version
246 if (v == 4 and self.len < 31) or (v == 6 and self.len < 127):
248 fib_interface_ip_prefix(
249 self._test, self.addr, self.len, self.intf.sw_if_index
255 table_id=self.table_id,
256 sw_if_index=self.intf.sw_if_index,
262 table_id=self.table_id,
263 sw_if_index=self.intf.sw_if_index,
267 return fib_interface_ip_prefix(
268 self._test, self.addr, self.len, self.intf.sw_if_index
273 table_id=self.table_id,
274 sw_if_index=self.intf.sw_if_index,
278 return "interface-ip-%s-%d-%s" % (self.intf, self.table_id, self.prefix)
281 class VppIp6LinkLocalAddress(VppObject):
282 def __init__(self, test, intf, addr):
287 def add_vpp_config(self):
288 self._test.vapi.sw_interface_ip6_set_link_local_address(
289 sw_if_index=self.intf.sw_if_index, ip=self.addr
291 self._test.registry.register(self, self._test.logger)
294 def remove_vpp_config(self):
295 # link locals can't be removed, only changed
298 def query_vpp_config(self):
303 return "ip6-link-local-%s-%s" % (self.intf, self.addr)
306 class VppIpInterfaceBind(VppObject):
307 def __init__(self, test, intf, table):
312 def add_vpp_config(self):
313 if self.table.is_ip6:
314 self.intf.set_table_ip6(self.table.table_id)
316 self.intf.set_table_ip4(self.table.table_id)
317 self._test.registry.register(self, self._test.logger)
320 def remove_vpp_config(self):
321 if 0 == self.table.table_id:
323 if self.table.is_ip6:
324 self.intf.set_table_ip6(0)
326 self.intf.set_table_ip4(0)
328 def query_vpp_config(self):
329 if 0 == self.table.table_id:
333 self._test.vapi.sw_interface_get_table(
334 self.intf.sw_if_index, self.table.is_ip6
336 == self.table.table_id
338 except UnexpectedApiReturnValueError as e:
339 if e.retval == -2: # INVALID_SW_IF_INDEX
344 return "interface-bind-%s-%s" % (self.intf, self.table)
348 def __init__(self, value, mode=MplsLspMode.PIPE, ttl=64, exp=0):
355 is_uniform = 0 if self.mode is MplsLspMode.PIPE else 1
360 "is_uniform": is_uniform,
363 def __eq__(self, other):
364 if isinstance(other, self.__class__):
366 self.value == other.value
367 and self.ttl == other.ttl
368 and self.exp == other.exp
369 and self.mode == other.mode
371 elif hasattr(other, "label"):
373 self.value == other.label
374 and self.ttl == other.ttl
375 and self.exp == other.exp
376 and (self.mode == MplsLspMode.UNIFORM) == other.is_uniform
381 def __ne__(self, other):
382 return not (self == other)
385 class VppFibPathNextHop:
386 def __init__(self, addr, via_label=MPLS_LABEL_INVALID, next_hop_id=INVALID_INDEX):
387 self.addr = VppIpAddressUnion(addr)
388 self.via_label = via_label
389 self.obj_id = next_hop_id
392 if self.via_label is not MPLS_LABEL_INVALID:
393 return {"via_label": self.via_label}
394 if self.obj_id is not INVALID_INDEX:
395 return {"obj_id": self.obj_id}
397 return {"address": self.addr.encode()}
400 if self.via_label is MPLS_LABEL_INVALID:
401 return address_proto(self.addr)
403 return FibPathProto.FIB_PATH_NH_PROTO_MPLS
405 def __eq__(self, other):
406 if not isinstance(other, self.__class__):
407 # try the other instance's __eq__.
408 return NotImplemented
410 self.addr == other.addr
411 and self.via_label == other.via_label
412 and self.obj_id == other.obj_id
423 nh_via_label=MPLS_LABEL_INVALID,
425 next_hop_id=INVALID_INDEX,
427 flags=FibPathFlags.FIB_PATH_FLAG_NONE,
428 type=FibPathType.FIB_PATH_TYPE_NORMAL,
430 self.nh_itf = nh_sw_if_index
431 self.nh_table_id = nh_table_id
432 self.nh_labels = labels
438 self.nh = VppFibPathNextHop(nh_addr, nh_via_label, next_hop_id)
440 self.proto = self.nh.proto()
443 self.next_hop_id = next_hop_id
445 def encode_labels(self):
447 for l in self.nh_labels:
448 if type(l) == VppMplsLabel:
449 lstack.append(l.encode())
451 lstack.append({"label": l, "ttl": 255})
452 while len(lstack) < 16:
461 "table_id": self.nh_table_id,
462 "nh": self.nh.encode(),
463 "next_hop_id": self.next_hop_id,
464 "sw_if_index": self.nh_itf,
465 "rpf_id": self.rpf_id,
469 "n_labels": len(self.nh_labels),
470 "label_stack": self.encode_labels(),
473 def __eq__(self, other):
474 if isinstance(other, self.__class__):
475 return self.nh == other.nh
476 elif hasattr(other, "sw_if_index"):
478 if len(self.nh_labels) != other.n_labels:
480 for i in range(len(self.nh_labels)):
481 if self.nh_labels[i] != other.label_stack[i]:
483 return self.nh_itf == other.sw_if_index
487 def __ne__(self, other):
488 return not (self == other)
491 class VppMRoutePath(VppRoutePath):
497 proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
498 type=FibPathType.FIB_PATH_TYPE_NORMAL,
499 bier_imp=INVALID_INDEX,
502 nh = "::" if proto is FibPathProto.FIB_PATH_NH_PROTO_IP6 else "0.0.0.0"
503 super(VppMRoutePath, self).__init__(
504 nh, nh_sw_if_index, proto=proto, type=type, next_hop_id=bier_imp
506 self.nh_i_flags = flags
507 self.bier_imp = bier_imp
511 "path": super(VppMRoutePath, self).encode(),
512 "itf_flags": self.nh_i_flags,
516 class VppIpRoute(VppObject):
522 self, test, dest_addr, dest_addr_len, paths, table_id=0, register=True
526 self.table_id = table_id
527 self.prefix = mk_network(dest_addr, dest_addr_len)
528 self.register = register
529 self.stats_index = None
530 self.modified = False
532 self.encoded_paths = []
533 for path in self.paths:
534 self.encoded_paths.append(path.encode())
536 def __eq__(self, other):
537 if self.table_id == other.table_id and self.prefix == other.prefix:
541 def modify(self, paths):
543 self.encoded_paths = []
544 for path in self.paths:
545 self.encoded_paths.append(path.encode())
548 self._test.vapi.ip_route_add_del(
550 "table_id": self.table_id,
551 "prefix": self.prefix,
552 "n_paths": len(self.encoded_paths),
553 "paths": self.encoded_paths,
559 def add_vpp_config(self):
560 r = self._test.vapi.ip_route_add_del(
562 "table_id": self.table_id,
563 "prefix": self.prefix,
564 "n_paths": len(self.encoded_paths),
565 "paths": self.encoded_paths,
570 self.stats_index = r.stats_index
572 self._test.registry.register(self, self._test.logger)
575 def remove_vpp_config(self):
576 # there's no need to issue different deletes for modified routes
577 # we do this only to test the two different ways to delete routes
578 # eiter by passing all the paths to remove and mutlipath=1 or
579 # passing no paths and multipath=0
581 self._test.vapi.ip_route_add_del(
583 "table_id": self.table_id,
584 "prefix": self.prefix,
585 "n_paths": len(self.encoded_paths),
586 "paths": self.encoded_paths,
592 self._test.vapi.ip_route_add_del(
593 route={"table_id": self.table_id, "prefix": self.prefix, "n_paths": 0},
598 def query_vpp_config(self):
601 self.prefix.network_address,
602 self.prefix.prefixlen,
607 return "%s:table-%d-%s" % (
608 "ip6-route" if self.prefix.version == 6 else "ip-route",
613 def get_stats_to(self):
614 c = self._test.statistics.get_counter("/net/route/to")
615 return c[0][self.stats_index]
617 def get_stats_via(self):
618 c = self._test.statistics.get_counter("/net/route/via")
619 return c[0][self.stats_index]
622 class VppIpRouteV2(VppObject):
628 self, test, dest_addr, dest_addr_len, paths, table_id=0, register=True, src=0
632 self.table_id = table_id
633 self.prefix = mk_network(dest_addr, dest_addr_len)
634 self.register = register
635 self.stats_index = None
636 self.modified = False
639 self.encoded_paths = []
640 for path in self.paths:
641 self.encoded_paths.append(path.encode())
643 def __eq__(self, other):
644 if self.table_id == other.table_id and self.prefix == other.prefix:
648 def modify(self, paths):
650 self.encoded_paths = []
651 for path in self.paths:
652 self.encoded_paths.append(path.encode())
655 self._test.vapi.ip_route_add_del_v2(
657 "table_id": self.table_id,
658 "prefix": self.prefix,
660 "n_paths": len(self.encoded_paths),
661 "paths": self.encoded_paths,
667 def add_vpp_config(self):
668 r = self._test.vapi.ip_route_add_del_v2(
670 "table_id": self.table_id,
671 "prefix": self.prefix,
672 "n_paths": len(self.encoded_paths),
673 "paths": self.encoded_paths,
679 self.stats_index = r.stats_index
681 self._test.registry.register(self, self._test.logger)
684 def remove_vpp_config(self):
685 # there's no need to issue different deletes for modified routes
686 # we do this only to test the two different ways to delete routes
687 # eiter by passing all the paths to remove and mutlipath=1 or
688 # passing no paths and multipath=0
690 self._test.vapi.ip_route_add_del_v2(
692 "table_id": self.table_id,
693 "prefix": self.prefix,
695 "n_paths": len(self.encoded_paths),
696 "paths": self.encoded_paths,
702 self._test.vapi.ip_route_add_del_v2(
704 "table_id": self.table_id,
705 "prefix": self.prefix,
713 def query_vpp_config(self):
716 self.prefix.network_address,
717 self.prefix.prefixlen,
722 return "%s:table-%d-%s" % (
723 "ip6-route" if self.prefix.version == 6 else "ip-route",
728 def get_stats_to(self):
729 c = self._test.statistics.get_counter("/net/route/to")
730 return c[0][self.stats_index]
732 def get_stats_via(self):
733 c = self._test.statistics.get_counter("/net/route/via")
734 return c[0][self.stats_index]
737 class VppIpMRoute(VppObject):
755 self.table_id = table_id
756 self.e_flags = e_flags
759 self.prefix = VppIpMPrefix(src_addr, grp_addr, grp_addr_len)
760 self.encoded_paths = []
761 for path in self.paths:
762 self.encoded_paths.append(path.encode())
764 def encode(self, paths=None):
765 _paths = self.encoded_paths if paths is None else paths
767 "table_id": self.table_id,
768 "entry_flags": self.e_flags,
769 "rpf_id": self.rpf_id,
770 "prefix": self.prefix.encode(),
771 "n_paths": len(_paths),
775 def add_vpp_config(self):
776 r = self._test.vapi.ip_mroute_add_del(
777 route=self.encode(), is_multipath=1, is_add=1
779 self.stats_index = r.stats_index
780 self._test.registry.register(self, self._test.logger)
783 def remove_vpp_config(self):
784 self._test.vapi.ip_mroute_add_del(route=self.encode(), is_multipath=1, is_add=0)
786 def update_entry_flags(self, flags):
788 self._test.vapi.ip_mroute_add_del(
789 route=self.encode(paths=[]), is_multipath=1, is_add=1
792 def update_rpf_id(self, rpf_id):
794 self._test.vapi.ip_mroute_add_del(
795 route=self.encode(paths=[]), is_multipath=1, is_add=1
798 def update_path_flags(self, itf, flags):
799 for p in range(len(self.paths)):
800 if self.paths[p].nh_itf == itf:
801 self.paths[p].nh_i_flags = flags
802 self.encoded_paths[p] = self.paths[p].encode()
805 self._test.vapi.ip_mroute_add_del(
806 route=self.encode(paths=[self.encoded_paths[p]]), is_add=1, is_multipath=0
809 def query_vpp_config(self):
819 return "%d:(%s,%s/%d)" % (
827 c = self._test.statistics.get_counter("/net/mroute")
828 return c[0][self.stats_index]
832 def __init__(self, test, route, interface, packet):
834 self.interface = interface
838 def compare(self, signal):
839 self.test.assertEqual(self.interface, signal.sw_if_index)
840 self.test.assertEqual(self.route.table_id, signal.table_id)
841 self.test.assertEqual(self.route.prefix, signal.prefix)
844 class VppMplsIpBind(VppObject):
860 self.dest_addr_len = dest_addr_len
861 self.dest_addr = dest_addr
862 self.ip_addr = ip_address(text_type(dest_addr))
863 self.local_label = local_label
864 self.table_id = table_id
865 self.ip_table_id = ip_table_id
866 self.prefix = mk_network(dest_addr, dest_addr_len)
868 def add_vpp_config(self):
869 self._test.vapi.mpls_ip_bind_unbind(
872 table_id=self.table_id,
873 ip_table_id=self.ip_table_id,
875 self._test.registry.register(self, self._test.logger)
877 def remove_vpp_config(self):
878 self._test.vapi.mpls_ip_bind_unbind(
881 table_id=self.table_id,
882 ip_table_id=self.ip_table_id,
886 def query_vpp_config(self):
887 dump = self._test.vapi.mpls_route_dump(self.table_id)
890 self.local_label == e.mr_route.mr_label
891 and self.table_id == e.mr_route.mr_table_id
897 return "%d:%s binds %d:%s/%d" % (
906 class VppMplsTable(VppObject):
907 def __init__(self, test, table_id):
909 self.table_id = table_id
911 def add_vpp_config(self):
912 self._test.vapi.mpls_table_add_del(self.table_id, is_add=1)
913 self._test.registry.register(self, self._test.logger)
915 def remove_vpp_config(self):
916 self._test.vapi.mpls_table_add_del(self.table_id, is_add=0)
918 def query_vpp_config(self):
919 dump = self._test.vapi.mpls_table_dump()
921 if d.mt_table.mt_table_id == self.table_id:
926 return "table-mpls-%d" % (self.table_id)
929 class VppMplsRoute(VppObject):
942 eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
946 self.local_label = local_label
947 self.eos_bit = eos_bit
948 self.eos_proto = eos_proto
949 self.table_id = table_id
950 self.is_multicast = is_multicast
952 def add_vpp_config(self):
954 for path in self.paths:
955 paths.append(path.encode())
957 r = self._test.vapi.mpls_route_add_del(
967 self.stats_index = r.stats_index
968 self._test.registry.register(self, self._test.logger)
970 def remove_vpp_config(self):
972 for path in self.paths:
973 paths.append(path.encode())
975 self._test.vapi.mpls_route_add_del(
986 def query_vpp_config(self):
987 return find_mpls_route(
988 self._test, self.table_id, self.local_label, self.eos_bit
992 return "mpls-route-%d:%s/%d" % (
998 def get_stats_to(self):
999 c = self._test.statistics.get_counter("/net/route/to")
1000 return c[0][self.stats_index]
1002 def get_stats_via(self):
1003 c = self._test.statistics.get_counter("/net/route/via")
1004 return c[0][self.stats_index]