--- /dev/null
+"""
+ SRv6 LocalSIDs
+
+ object abstractions for representing SRv6 localSIDs in VPP
+"""
+
+from vpp_object import VppObject
+from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
+
+
+class SRv6LocalSIDBehaviors():
+ # from src/vnet/srv6/sr.h
+ SR_BEHAVIOR_END = 1
+ SR_BEHAVIOR_X = 2
+ SR_BEHAVIOR_T = 3
+ SR_BEHAVIOR_D_FIRST = 4 # Unused. Separator in between regular and D
+ SR_BEHAVIOR_DX2 = 5
+ SR_BEHAVIOR_DX6 = 6
+ SR_BEHAVIOR_DX4 = 7
+ SR_BEHAVIOR_DT6 = 8
+ SR_BEHAVIOR_DT4 = 9
+ SR_BEHAVIOR_LAST = 10 # Must always be the last one
+
+
+class SRv6PolicyType():
+ # from src/vnet/srv6/sr.h
+ SR_POLICY_TYPE_DEFAULT = 0
+ SR_POLICY_TYPE_SPRAY = 1
+
+
+class SRv6PolicySteeringTypes():
+ # from src/vnet/srv6/sr.h
+ SR_STEER_L2 = 2
+ SR_STEER_IPV4 = 4
+ SR_STEER_IPV6 = 6
+
+
+class VppSRv6LocalSID(VppObject):
+ """
+ SRv6 LocalSID
+ """
+
+ def __init__(self, test, localsid, behavior, nh_addr4, nh_addr6,
+ end_psp, sw_if_index, vlan_index, fib_table):
+ self._test = test
+ self.localsid = localsid
+ # keep binary format in _localsid
+ self.localsid["addr"] = inet_pton(AF_INET6, self.localsid["addr"])
+ self.behavior = behavior
+ self.nh_addr4 = inet_pton(AF_INET, nh_addr4)
+ self.nh_addr6 = inet_pton(AF_INET6, nh_addr6)
+ self.end_psp = end_psp
+ self.sw_if_index = sw_if_index
+ self.vlan_index = vlan_index
+ self.fib_table = fib_table
+ self._configured = False
+
+ def add_vpp_config(self):
+ self._test.vapi.sr_localsid_add_del(
+ self.localsid,
+ self.behavior,
+ self.nh_addr4,
+ self.nh_addr6,
+ is_del=0,
+ end_psp=self.end_psp,
+ sw_if_index=self.sw_if_index,
+ vlan_index=self.vlan_index,
+ fib_table=self.fib_table)
+ self._configured = True
+
+ def remove_vpp_config(self):
+ self._test.vapi.sr_localsid_add_del(
+ self.localsid,
+ self.behavior,
+ self.nh_addr4,
+ self.nh_addr6,
+ is_del=1,
+ end_psp=self.end_psp,
+ sw_if_index=self.sw_if_index,
+ vlan_index=self.vlan_index,
+ fib_table=self.fib_table)
+ self._configured = False
+
+ def query_vpp_config(self):
+ # sr_localsids_dump API is disabled
+ # use _configured flag for now
+ return self._configured
+
+ def object_id(self):
+ return ("%d;%s,%d"
+ % (self.fib_table,
+ self.localsid,
+ self.behavior))
+
+
+class VppSRv6Policy(VppObject):
+ """
+ SRv6 Policy
+ """
+
+ def __init__(self, test, bsid,
+ is_encap, sr_type, weight, fib_table,
+ segments, source):
+ self._test = test
+ self.bsid = bsid
+ # keep binary format in _bsid
+ self._bsid = inet_pton(AF_INET6, bsid)
+ self.is_encap = is_encap
+ self.sr_type = sr_type
+ self.weight = weight
+ self.fib_table = fib_table
+ self.segments = segments
+ # keep binary format in _segments
+ self._segments = []
+ for seg in segments:
+ self._segments.extend(inet_pton(AF_INET6, seg))
+ self.n_segments = len(segments)
+ # source not passed to API
+ # self.source = inet_pton(AF_INET6, source)
+ self.source = source
+ self._configured = False
+
+ def add_vpp_config(self):
+ self._test.vapi.sr_policy_add(
+ self._bsid,
+ self.weight,
+ self.is_encap,
+ self.sr_type,
+ self.fib_table,
+ self.n_segments,
+ self._segments)
+ self._configured = True
+
+ def remove_vpp_config(self):
+ self._test.vapi.sr_policy_del(
+ self._bsid)
+ self._configured = False
+
+ def query_vpp_config(self):
+ # no API to query SR Policies
+ # use _configured flag for now
+ return self._configured
+
+ def object_id(self):
+ return ("%d;%s-><%s>;%d"
+ % (self.sr_type,
+ self.bsid,
+ ','.join(self.segments),
+ self.is_encap))
+
+
+class VppSRv6Steering(VppObject):
+ """
+ SRv6 Steering
+ """
+
+ def __init__(self, test,
+ bsid,
+ prefix,
+ mask_width,
+ traffic_type,
+ sr_policy_index,
+ table_id,
+ sw_if_index):
+ self._test = test
+ self.bsid = bsid
+ # keep binary format in _bsid
+ self._bsid = inet_pton(AF_INET6, bsid)
+ self.prefix = prefix
+ # keep binary format in _prefix
+ if ':' in prefix:
+ # IPv6
+ self._prefix = inet_pton(AF_INET6, prefix)
+ else:
+ # IPv4
+ # API expects 16 octets (128 bits)
+ # last 4 octets are used for IPv4
+ # --> prepend 12 octets
+ self._prefix = ('\x00' * 12) + inet_pton(AF_INET, prefix)
+ self.mask_width = mask_width
+ self.traffic_type = traffic_type
+ self.sr_policy_index = sr_policy_index
+ self.sw_if_index = sw_if_index
+ self.table_id = table_id
+ self._configured = False
+
+ def add_vpp_config(self):
+ self._test.vapi.sr_steering_add_del(
+ 0,
+ self._bsid,
+ self.sr_policy_index,
+ self.table_id,
+ self._prefix,
+ self.mask_width,
+ self.sw_if_index,
+ self.traffic_type)
+ self._configured = True
+
+ def remove_vpp_config(self):
+ self._test.vapi.sr_steering_add_del(
+ 1,
+ self._bsid,
+ self.sr_policy_index,
+ self.table_id,
+ self._prefix,
+ self.mask_width,
+ self.sw_if_index,
+ self.traffic_type)
+ self._configured = False
+
+ def query_vpp_config(self):
+ # no API to query steering entries
+ # use _configured flag for now
+ return self._configured
+
+ def object_id(self):
+ return ("%d;%d;%s/%d->%s"
+ % (self.table_id,
+ self.traffic_type,
+ self.prefix,
+ self.mask_width,
+ self.bsid))
Code Review / vpp.git / blobdiff