180279b912
Drop comes before lookup when enabled. is_first_or_last is not required when setting a feature, the anchor is added in find_config_with_features(). Don't make the PG interfaces automatically L3 enabled, this way we can have tests that check the L3 protocol disbaled behaviour. Change-Id: Icef22a920b27ff9cec6ab2da6b05f05c532cb60f Signed-off-by: Neale Ranns <nranns@cisco.com>
395 lines
14 KiB
Python
395 lines
14 KiB
Python
"""
|
|
IP Routes
|
|
|
|
object abstractions for representing IP routes in VPP
|
|
"""
|
|
|
|
from vpp_object import *
|
|
from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
|
|
|
|
# from vnet/vnet/mpls/mpls_types.h
|
|
MPLS_IETF_MAX_LABEL = 0xfffff
|
|
MPLS_LABEL_INVALID = MPLS_IETF_MAX_LABEL + 1
|
|
|
|
|
|
class MRouteItfFlags:
|
|
MFIB_ITF_FLAG_NONE = 0
|
|
MFIB_ITF_FLAG_NEGATE_SIGNAL = 1
|
|
MFIB_ITF_FLAG_ACCEPT = 2
|
|
MFIB_ITF_FLAG_FORWARD = 4
|
|
MFIB_ITF_FLAG_SIGNAL_PRESENT = 8
|
|
MFIB_ITF_FLAG_INTERNAL_COPY = 16
|
|
|
|
|
|
class MRouteEntryFlags:
|
|
MFIB_ENTRY_FLAG_NONE = 0
|
|
MFIB_ENTRY_FLAG_SIGNAL = 1
|
|
MFIB_ENTRY_FLAG_DROP = 2
|
|
MFIB_ENTRY_FLAG_CONNECTED = 4
|
|
MFIB_ENTRY_FLAG_INHERIT_ACCEPT = 8
|
|
|
|
|
|
def find_route(test, ip_addr, len, table_id=0, inet=AF_INET):
|
|
if inet == AF_INET:
|
|
s = 4
|
|
routes = test.vapi.ip_fib_dump()
|
|
else:
|
|
s = 16
|
|
routes = test.vapi.ip6_fib_dump()
|
|
|
|
route_addr = inet_pton(inet, ip_addr)
|
|
for e in routes:
|
|
if route_addr == e.address[:s] \
|
|
and len == e.address_length \
|
|
and table_id == e.table_id:
|
|
return True
|
|
return False
|
|
|
|
|
|
class VppRoutePath(object):
|
|
|
|
def __init__(
|
|
self,
|
|
nh_addr,
|
|
nh_sw_if_index,
|
|
nh_table_id=0,
|
|
labels=[],
|
|
nh_via_label=MPLS_LABEL_INVALID,
|
|
is_ip6=0):
|
|
self.nh_itf = nh_sw_if_index
|
|
self.nh_table_id = nh_table_id
|
|
self.nh_via_label = nh_via_label
|
|
self.nh_labels = labels
|
|
if is_ip6:
|
|
self.nh_addr = inet_pton(AF_INET6, nh_addr)
|
|
else:
|
|
self.nh_addr = inet_pton(AF_INET, nh_addr)
|
|
|
|
|
|
class VppMRoutePath(VppRoutePath):
|
|
|
|
def __init__(self, nh_sw_if_index, flags):
|
|
super(VppMRoutePath, self).__init__("0.0.0.0",
|
|
nh_sw_if_index)
|
|
self.nh_i_flags = flags
|
|
|
|
|
|
class VppIpRoute(VppObject):
|
|
"""
|
|
IP Route
|
|
"""
|
|
|
|
def __init__(self, test, dest_addr,
|
|
dest_addr_len, paths, table_id=0, is_ip6=0, is_local=0,
|
|
is_unreach=0, is_prohibit=0):
|
|
self._test = test
|
|
self.paths = paths
|
|
self.dest_addr_len = dest_addr_len
|
|
self.table_id = table_id
|
|
self.is_ip6 = is_ip6
|
|
self.is_local = is_local
|
|
self.is_unreach = is_unreach
|
|
self.is_prohibit = is_prohibit
|
|
self.dest_addr_p = dest_addr
|
|
if is_ip6:
|
|
self.dest_addr = inet_pton(AF_INET6, dest_addr)
|
|
else:
|
|
self.dest_addr = inet_pton(AF_INET, dest_addr)
|
|
|
|
def modify(self, paths, is_local=0,
|
|
is_unreach=0, is_prohibit=0):
|
|
self.paths = paths
|
|
self.is_local = is_local
|
|
self.is_unreach = is_unreach
|
|
self.is_prohibit = is_prohibit
|
|
|
|
def add_vpp_config(self):
|
|
if self.is_local or self.is_unreach or self.is_prohibit:
|
|
self._test.vapi.ip_add_del_route(
|
|
self.dest_addr,
|
|
self.dest_addr_len,
|
|
inet_pton(AF_INET6, "::"),
|
|
0xffffffff,
|
|
is_local=self.is_local,
|
|
is_unreach=self.is_unreach,
|
|
is_prohibit=self.is_prohibit,
|
|
table_id=self.table_id,
|
|
is_ipv6=self.is_ip6)
|
|
else:
|
|
for path in self.paths:
|
|
self._test.vapi.ip_add_del_route(
|
|
self.dest_addr,
|
|
self.dest_addr_len,
|
|
path.nh_addr,
|
|
path.nh_itf,
|
|
table_id=self.table_id,
|
|
next_hop_out_label_stack=path.nh_labels,
|
|
next_hop_n_out_labels=len(
|
|
path.nh_labels),
|
|
next_hop_via_label=path.nh_via_label,
|
|
is_ipv6=self.is_ip6)
|
|
self._test.registry.register(self, self._test.logger)
|
|
|
|
def remove_vpp_config(self):
|
|
if self.is_local or self.is_unreach or self.is_prohibit:
|
|
self._test.vapi.ip_add_del_route(
|
|
self.dest_addr,
|
|
self.dest_addr_len,
|
|
inet_pton(AF_INET6, "::"),
|
|
0xffffffff,
|
|
is_local=self.is_local,
|
|
is_unreach=self.is_unreach,
|
|
is_prohibit=self.is_prohibit,
|
|
is_add=0,
|
|
table_id=self.table_id,
|
|
is_ipv6=self.is_ip6)
|
|
else:
|
|
for path in self.paths:
|
|
self._test.vapi.ip_add_del_route(self.dest_addr,
|
|
self.dest_addr_len,
|
|
path.nh_addr,
|
|
path.nh_itf,
|
|
table_id=self.table_id,
|
|
is_add=0,
|
|
is_ipv6=self.is_ip6)
|
|
|
|
def query_vpp_config(self):
|
|
return find_route(self._test,
|
|
self.dest_addr_p,
|
|
self.dest_addr_len,
|
|
self.table_id,
|
|
inet=AF_INET6 if self.is_ip6 == 1 else AF_INET)
|
|
|
|
def __str__(self):
|
|
return self.object_id()
|
|
|
|
def object_id(self):
|
|
return ("%d:%s/%d"
|
|
% (self.table_id,
|
|
self.dest_addr_p,
|
|
self.dest_addr_len))
|
|
|
|
|
|
class VppIpMRoute(VppObject):
|
|
"""
|
|
IP Multicast Route
|
|
"""
|
|
|
|
def __init__(self, test, src_addr, grp_addr,
|
|
grp_addr_len, e_flags, paths, table_id=0, is_ip6=0):
|
|
self._test = test
|
|
self.paths = paths
|
|
self.grp_addr_len = grp_addr_len
|
|
self.table_id = table_id
|
|
self.e_flags = e_flags
|
|
self.is_ip6 = is_ip6
|
|
|
|
if is_ip6:
|
|
self.grp_addr = inet_pton(AF_INET6, grp_addr)
|
|
self.src_addr = inet_pton(AF_INET6, src_addr)
|
|
else:
|
|
self.grp_addr = inet_pton(AF_INET, grp_addr)
|
|
self.src_addr = inet_pton(AF_INET, src_addr)
|
|
|
|
def add_vpp_config(self):
|
|
for path in self.paths:
|
|
self._test.vapi.ip_mroute_add_del(self.src_addr,
|
|
self.grp_addr,
|
|
self.grp_addr_len,
|
|
self.e_flags,
|
|
path.nh_itf,
|
|
path.nh_i_flags,
|
|
table_id=self.table_id,
|
|
is_ipv6=self.is_ip6)
|
|
self._test.registry.register(self, self._test.logger)
|
|
|
|
def remove_vpp_config(self):
|
|
for path in self.paths:
|
|
self._test.vapi.ip_mroute_add_del(self.src_addr,
|
|
self.grp_addr,
|
|
self.grp_addr_len,
|
|
self.e_flags,
|
|
path.nh_itf,
|
|
path.nh_i_flags,
|
|
table_id=self.table_id,
|
|
is_add=0,
|
|
is_ipv6=self.is_ip6)
|
|
|
|
def update_entry_flags(self, flags):
|
|
self.e_flags = flags
|
|
self._test.vapi.ip_mroute_add_del(self.src_addr,
|
|
self.grp_addr,
|
|
self.grp_addr_len,
|
|
self.e_flags,
|
|
0xffffffff,
|
|
0,
|
|
table_id=self.table_id,
|
|
is_ipv6=self.is_ip6)
|
|
|
|
def update_path_flags(self, itf, flags):
|
|
for path in self.paths:
|
|
if path.nh_itf == itf:
|
|
path.nh_i_flags = flags
|
|
break
|
|
self._test.vapi.ip_mroute_add_del(self.src_addr,
|
|
self.grp_addr,
|
|
self.grp_addr_len,
|
|
self.e_flags,
|
|
path.nh_itf,
|
|
path.nh_i_flags,
|
|
table_id=self.table_id,
|
|
is_ipv6=self.is_ip6)
|
|
|
|
def query_vpp_config(self):
|
|
dump = self._test.vapi.ip_fib_dump()
|
|
for e in dump:
|
|
if self.grp_addr == e.address \
|
|
and self.grp_addr_len == e.address_length \
|
|
and self.table_id == e.table_id:
|
|
return True
|
|
return False
|
|
|
|
def __str__(self):
|
|
return self.object_id()
|
|
|
|
def object_id(self):
|
|
if self.is_ip6:
|
|
return ("%d:(%s,%s/%d)"
|
|
% (self.table_id,
|
|
inet_ntop(AF_INET6, self.src_addr),
|
|
inet_ntop(AF_INET6, self.grp_addr),
|
|
self.grp_addr_len))
|
|
else:
|
|
return ("%d:(%s,%s/%d)"
|
|
% (self.table_id,
|
|
inet_ntop(AF_INET, self.src_addr),
|
|
inet_ntop(AF_INET, self.grp_addr),
|
|
self.grp_addr_len))
|
|
|
|
|
|
class VppMFibSignal(object):
|
|
def __init__(self, test, route, interface, packet):
|
|
self.route = route
|
|
self.interface = interface
|
|
self.packet = packet
|
|
self.test = test
|
|
|
|
def compare(self, signal):
|
|
self.test.assertEqual(self.interface, signal.sw_if_index)
|
|
self.test.assertEqual(self.route.table_id, signal.table_id)
|
|
self.test.assertEqual(self.route.grp_addr_len,
|
|
signal.grp_address_len)
|
|
for i in range(self.route.grp_addr_len / 8):
|
|
self.test.assertEqual(self.route.grp_addr[i],
|
|
signal.grp_address[i])
|
|
if (self.route.grp_addr_len > 32):
|
|
for i in range(4):
|
|
self.test.assertEqual(self.route.src_addr[i],
|
|
signal.src_address[i])
|
|
|
|
|
|
class VppMplsIpBind(VppObject):
|
|
"""
|
|
MPLS to IP Binding
|
|
"""
|
|
|
|
def __init__(self, test, local_label, dest_addr, dest_addr_len,
|
|
table_id=0, ip_table_id=0):
|
|
self._test = test
|
|
self.dest_addr = inet_pton(AF_INET, dest_addr)
|
|
self.dest_addr_len = dest_addr_len
|
|
self.local_label = local_label
|
|
self.table_id = table_id
|
|
self.ip_table_id = ip_table_id
|
|
|
|
def add_vpp_config(self):
|
|
self._test.vapi.mpls_ip_bind_unbind(self.local_label,
|
|
self.dest_addr,
|
|
self.dest_addr_len,
|
|
table_id=self.table_id,
|
|
ip_table_id=self.ip_table_id)
|
|
self._test.registry.register(self, self._test.logger)
|
|
|
|
def remove_vpp_config(self):
|
|
self._test.vapi.mpls_ip_bind_unbind(self.local_label,
|
|
self.dest_addr,
|
|
self.dest_addr_len,
|
|
is_bind=0)
|
|
|
|
def query_vpp_config(self):
|
|
dump = self._test.vapi.mpls_fib_dump()
|
|
for e in dump:
|
|
if self.local_label == e.label \
|
|
and self.eos_bit == e.eos_bit \
|
|
and self.table_id == e.table_id:
|
|
return True
|
|
return False
|
|
|
|
def __str__(self):
|
|
return self.object_id()
|
|
|
|
def object_id(self):
|
|
return ("%d:%s binds %d:%s/%d"
|
|
% (self.table_id,
|
|
self.local_label,
|
|
self.ip_table_id,
|
|
inet_ntop(AF_INET, self.dest_addr),
|
|
self.dest_addr_len))
|
|
|
|
|
|
class VppMplsRoute(VppObject):
|
|
"""
|
|
MPLS Route/LSP
|
|
"""
|
|
|
|
def __init__(self, test, local_label, eos_bit, paths, table_id=0):
|
|
self._test = test
|
|
self.paths = paths
|
|
self.local_label = local_label
|
|
self.eos_bit = eos_bit
|
|
self.table_id = table_id
|
|
|
|
def add_vpp_config(self):
|
|
for path in self.paths:
|
|
self._test.vapi.mpls_route_add_del(
|
|
self.local_label,
|
|
self.eos_bit,
|
|
1,
|
|
path.nh_addr,
|
|
path.nh_itf,
|
|
table_id=self.table_id,
|
|
next_hop_out_label_stack=path.nh_labels,
|
|
next_hop_n_out_labels=len(
|
|
path.nh_labels),
|
|
next_hop_via_label=path.nh_via_label,
|
|
next_hop_table_id=path.nh_table_id)
|
|
self._test.registry.register(self, self._test.logger)
|
|
|
|
def remove_vpp_config(self):
|
|
for path in self.paths:
|
|
self._test.vapi.mpls_route_add_del(self.local_label,
|
|
self.eos_bit,
|
|
1,
|
|
path.nh_addr,
|
|
path.nh_itf,
|
|
table_id=self.table_id,
|
|
is_add=0)
|
|
|
|
def query_vpp_config(self):
|
|
dump = self._test.vapi.mpls_fib_dump()
|
|
for e in dump:
|
|
if self.local_label == e.label \
|
|
and self.eos_bit == e.eos_bit \
|
|
and self.table_id == e.table_id:
|
|
return True
|
|
return False
|
|
|
|
def __str__(self):
|
|
return self.object_id()
|
|
|
|
def object_id(self):
|
|
return ("%d:%s/%d"
|
|
% (self.table_id,
|
|
self.local_label,
|
|
20+self.eos_bit))
|