| """ |
| SRv6 LocalSIDs |
| |
| object abstractions for representing SRv6 localSIDs in VPP |
| """ |
| |
| from vpp_object import * |
| 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 __str__(self): |
| return self.object_id() |
| |
| 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 __str__(self): |
| return self.object_id() |
| |
| 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 __str__(self): |
| return self.object_id() |
| |
| def object_id(self): |
| return ("%d;%d;%s/%d->%s" |
| % (self.table_id, |
| self.traffic_type, |
| self.prefix, |
| self.mask_width, |
| self.bsid)) |