src/plugins/srv6-ad/test/vpp_srv6.py

   1 """
   2   SRv6 LocalSIDs
   3
   4   object abstractions for representing SRv6 localSIDs in VPP
   5 """
   6
   7 from vpp_object import VppObject
   8 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
   9
  10
  11 class SRv6LocalSIDBehaviors():
  12     # from src/vnet/srv6/sr.h
  13     SR_BEHAVIOR_END = 1
  14     SR_BEHAVIOR_X = 2
  15     SR_BEHAVIOR_T = 3
  16     SR_BEHAVIOR_D_FIRST = 4   # Unused. Separator in between regular and D
  17     SR_BEHAVIOR_DX2 = 5
  18     SR_BEHAVIOR_DX6 = 6
  19     SR_BEHAVIOR_DX4 = 7
  20     SR_BEHAVIOR_DT6 = 8
  21     SR_BEHAVIOR_DT4 = 9
  22     SR_BEHAVIOR_LAST = 10      # Must always be the last one
  23
  24
  25 class SRv6PolicyType():
  26     # from src/vnet/srv6/sr.h
  27     SR_POLICY_TYPE_DEFAULT = 0
  28     SR_POLICY_TYPE_SPRAY = 1
  29
  30
  31 class SRv6PolicySteeringTypes():
  32     # from src/vnet/srv6/sr.h
  33     SR_STEER_L2 = 2
  34     SR_STEER_IPV4 = 4
  35     SR_STEER_IPV6 = 6
  36
  37
  38 class VppSRv6LocalSID(VppObject):
  39     """
  40     SRv6 LocalSID
  41     """
  42
  43     def __init__(self, test, localsid, behavior, nh_addr4, nh_addr6,
  44                  end_psp, sw_if_index, vlan_index, fib_table):
  45         self._test = test
  46         self.localsid = localsid
  47         # keep binary format in _localsid
  48         self.localsid["addr"] = inet_pton(AF_INET6, self.localsid["addr"])
  49         self.behavior = behavior
  50         self.nh_addr4 = inet_pton(AF_INET, nh_addr4)
  51         self.nh_addr6 = inet_pton(AF_INET6, nh_addr6)
  52         self.end_psp = end_psp
  53         self.sw_if_index = sw_if_index
  54         self.vlan_index = vlan_index
  55         self.fib_table = fib_table
  56         self._configured = False
  57
  58     def add_vpp_config(self):
  59         self._test.vapi.sr_localsid_add_del(
  60             self.localsid,
  61             self.behavior,
  62             self.nh_addr4,
  63             self.nh_addr6,
  64             is_del=0,
  65             end_psp=self.end_psp,
  66             sw_if_index=self.sw_if_index,
  67             vlan_index=self.vlan_index,
  68             fib_table=self.fib_table)
  69         self._configured = True
  70
  71     def remove_vpp_config(self):
  72         self._test.vapi.sr_localsid_add_del(
  73             self.localsid,
  74             self.behavior,
  75             self.nh_addr4,
  76             self.nh_addr6,
  77             is_del=1,
  78             end_psp=self.end_psp,
  79             sw_if_index=self.sw_if_index,
  80             vlan_index=self.vlan_index,
  81             fib_table=self.fib_table)
  82         self._configured = False
  83
  84     def query_vpp_config(self):
  85         # sr_localsids_dump API is disabled
  86         # use _configured flag for now
  87         return self._configured
  88
  89     def object_id(self):
  90         return ("%d;%s,%d"
  91                 % (self.fib_table,
  92                    self.localsid,
  93                    self.behavior))
  94
  95
  96 class VppSRv6Policy(VppObject):
  97     """
  98     SRv6 Policy
  99     """
 100
 101     def __init__(self, test, bsid,
 102                  is_encap, sr_type, weight, fib_table,
 103                  segments, source):
 104         self._test = test
 105         self.bsid = bsid
 106         # keep binary format in _bsid
 107         self._bsid = inet_pton(AF_INET6, bsid)
 108         self.is_encap = is_encap
 109         self.sr_type = sr_type
 110         self.weight = weight
 111         self.fib_table = fib_table
 112         self.segments = segments
 113         # keep binary format in _segments
 114         self._segments = []
 115         for seg in segments:
 116             self._segments.extend(inet_pton(AF_INET6, seg))
 117         self.n_segments = len(segments)
 118         # source not passed to API
 119         # self.source = inet_pton(AF_INET6, source)
 120         self.source = source
 121         self._configured = False
 122
 123     def add_vpp_config(self):
 124         self._test.vapi.sr_policy_add(
 125                      self._bsid,
 126                      self.weight,
 127                      self.is_encap,
 128                      self.sr_type,
 129                      self.fib_table,
 130                      self.n_segments,
 131                      self._segments)
 132         self._configured = True
 133
 134     def remove_vpp_config(self):
 135         self._test.vapi.sr_policy_del(
 136                      self._bsid)
 137         self._configured = False
 138
 139     def query_vpp_config(self):
 140         # no API to query SR Policies
 141         # use _configured flag for now
 142         return self._configured
 143
 144     def object_id(self):
 145         return ("%d;%s-><%s>;%d"
 146                 % (self.sr_type,
 147                    self.bsid,
 148                    ','.join(self.segments),
 149                    self.is_encap))
 150
 151
 152 class VppSRv6Steering(VppObject):
 153     """
 154     SRv6 Steering
 155     """
 156
 157     def __init__(self, test,
 158                  bsid,
 159                  prefix,
 160                  mask_width,
 161                  traffic_type,
 162                  sr_policy_index,
 163                  table_id,
 164                  sw_if_index):
 165         self._test = test
 166         self.bsid = bsid
 167         # keep binary format in _bsid
 168         self._bsid = inet_pton(AF_INET6, bsid)
 169         self.prefix = prefix
 170         # keep binary format in _prefix
 171         if ':' in prefix:
 172             # IPv6
 173             self._prefix = inet_pton(AF_INET6, prefix)
 174         else:
 175             # IPv4
 176             # API expects 16 octets (128 bits)
 177             # last 4 octets are used for IPv4
 178             # --> prepend 12 octets
 179             self._prefix = ('\x00' * 12) + inet_pton(AF_INET, prefix)
 180         self.mask_width = mask_width
 181         self.traffic_type = traffic_type
 182         self.sr_policy_index = sr_policy_index
 183         self.sw_if_index = sw_if_index
 184         self.table_id = table_id
 185         self._configured = False
 186
 187     def add_vpp_config(self):
 188         self._test.vapi.sr_steering_add_del(
 189                      0,
 190                      self._bsid,
 191                      self.sr_policy_index,
 192                      self.table_id,
 193                      self._prefix,
 194                      self.mask_width,
 195                      self.sw_if_index,
 196                      self.traffic_type)
 197         self._configured = True
 198
 199     def remove_vpp_config(self):
 200         self._test.vapi.sr_steering_add_del(
 201                      1,
 202                      self._bsid,
 203                      self.sr_policy_index,
 204                      self.table_id,
 205                      self._prefix,
 206                      self.mask_width,
 207                      self.sw_if_index,
 208                      self.traffic_type)
 209         self._configured = False
 210
 211     def query_vpp_config(self):
 212         # no API to query steering entries
 213         # use _configured flag for now
 214         return self._configured
 215
 216     def object_id(self):
 217         return ("%d;%d;%s/%d->%s"
 218                 % (self.table_id,
 219                    self.traffic_type,
 220                    self.prefix,
 221                    self.mask_width,
 222                    self.bsid))