| """ |
| 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_END_UN_PERF = 10 |
| SR_BEHAVIOR_END_UN = 11 |
| SR_BEHAVIOR_LAST = 12 # 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_addr, |
| end_psp, |
| sw_if_index, |
| vlan_index, |
| fib_table, |
| ): |
| self._test = test |
| self.localsid = localsid |
| self.behavior = behavior |
| self.nh_addr = nh_addr |
| 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( |
| localsid=self.localsid, |
| behavior=self.behavior, |
| nh_addr=self.nh_addr, |
| 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( |
| localsid=self.localsid, |
| behavior=self.behavior, |
| nh_addr=self.nh_addr, |
| 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 |
| self.is_encap = is_encap |
| self.sr_type = sr_type |
| self.weight = weight |
| self.fib_table = fib_table |
| self.segments = segments |
| 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( |
| bsid=self.bsid, |
| weight=self.weight, |
| is_encap=self.is_encap, |
| is_spray=self.sr_type, |
| fib_table=self.fib_table, |
| sids={"num_sids": self.n_segments, "sids": 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 |
| self.prefix = 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( |
| is_del=0, |
| bsid=self.bsid, |
| sr_policy_index=self.sr_policy_index, |
| table_id=self.table_id, |
| prefix={"address": self.prefix, "len": self.mask_width}, |
| sw_if_index=self.sw_if_index, |
| traffic_type=self.traffic_type, |
| ) |
| self._configured = True |
| |
| def remove_vpp_config(self): |
| self._test.vapi.sr_steering_add_del( |
| is_del=1, |
| bsid=self.bsid, |
| sr_policy_index=self.sr_policy_index, |
| table_id=self.table_id, |
| prefix={"address": self.prefix, "len": self.mask_width}, |
| sw_if_index=self.sw_if_index, |
| traffic_type=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, |
| ) |