test/vpp_ip.py

   1 """
   2   IP Types
   3
   4 """
   5 import logging
   6
   7 from ipaddress import ip_address
   8 from socket import AF_INET, AF_INET6
   9 from vpp_papi import VppEnum
  10 try:
  11     text_type = unicode
  12 except NameError:
  13     text_type = str
  14
  15 _log = logging.getLogger(__name__)
  16
  17
  18 class DpoProto:
  19     DPO_PROTO_IP4 = 0
  20     DPO_PROTO_IP6 = 1
  21     DPO_PROTO_MPLS = 2
  22     DPO_PROTO_ETHERNET = 3
  23     DPO_PROTO_BIER = 4
  24     DPO_PROTO_NSH = 5
  25
  26
  27 INVALID_INDEX = 0xffffffff
  28
  29
  30 class VppIpAddressUnion():
  31     def __init__(self, addr):
  32         self.addr = addr
  33         self.ip_addr = ip_address(text_type(self.addr))
  34
  35     def encode(self):
  36         if self.version == 6:
  37             return {'ip6': self.ip_addr.packed}
  38         else:
  39             return {'ip4': self.ip_addr.packed}
  40
  41     @property
  42     def version(self):
  43         return self.ip_addr.version
  44
  45     @property
  46     def address(self):
  47         return self.addr
  48
  49     @property
  50     def length(self):
  51         return self.ip_addr.max_prefixlen
  52
  53     @property
  54     def bytes(self):
  55         return self.ip_addr.packed
  56
  57     def __eq__(self, other):
  58         if isinstance(other, self.__class__):
  59             return self.ip_addr == other.ip_addr
  60         elif hasattr(other, "ip4") and hasattr(other, "ip6"):
  61             # vl_api_address_union_t
  62             if 4 == self.version:
  63                 return self.ip_addr.packed == other.ip4
  64             else:
  65                 return self.ip_addr.packed == other.ip6
  66         else:
  67             _log.error("Comparing VppIpAddressUnions:%s"
  68                        " with incomparable type: %s",
  69                        self, other)
  70             return NotImplemented
  71
  72
  73 class VppIpAddress():
  74     def __init__(self, addr):
  75         self.addr = VppIpAddressUnion(addr)
  76
  77     def encode(self):
  78         if self.addr.version == 6:
  79             return {
  80                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
  81                 'un': self.addr.encode()
  82             }
  83         else:
  84             return {
  85                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
  86                 'un': self.addr.encode()
  87             }
  88
  89     def __eq__(self, other):
  90         if isinstance(other, self.__class__):
  91             return self.addr == other.addr
  92         elif hasattr(other, "af") and hasattr(other, "un"):
  93             # a vp_api_address_t
  94             if 4 == self.version:
  95                 return other.af == \
  96                     VppEnum.vl_api_address_family_t.ADDRESS_IP4 and \
  97                     other.un == self.addr
  98             else:
  99                 return other.af == \
 100                     VppEnum.vl_api_address_family_t.ADDRESS_IP6 and \
 101                     other.un == self.addr
 102         else:
 103             _log.error(
 104                 "Comparing VppIpAddress:<%s> %s with incomparable "
 105                 "type: <%s> %s",
 106                 self.__class__.__name__, self,
 107                 other.__class__.__name__, other)
 108             return NotImplemented
 109
 110     def __ne__(self, other):
 111         return not (self == other)
 112
 113     def __str__(self):
 114         return self.address
 115
 116     @property
 117     def bytes(self):
 118         return self.addr.bytes
 119
 120     @property
 121     def address(self):
 122         return self.addr.address
 123
 124     @property
 125     def length(self):
 126         return self.addr.length
 127
 128     @property
 129     def version(self):
 130         return self.addr.version
 131
 132     @property
 133     def is_ip6(self):
 134         return (self.version == 6)
 135
 136     @property
 137     def af(self):
 138         if self.version == 6:
 139             return AF_INET6
 140         else:
 141             return AF_INET
 142
 143     @property
 144     def dpo_proto(self):
 145         if self.version == 6:
 146             return DpoProto.DPO_PROTO_IP6
 147         else:
 148             return DpoProto.DPO_PROTO_IP4
 149
 150
 151 class VppIpPrefix():
 152     def __init__(self, addr, len):
 153         self.addr = VppIpAddress(addr)
 154         self.len = len
 155
 156     def encode(self):
 157         return {'address': self.addr.encode(),
 158                 'address_length': self.len}
 159
 160     @property
 161     def address(self):
 162         return self.addr.address
 163
 164     @property
 165     def bytes(self):
 166         return self.addr.bytes
 167
 168     @property
 169     def length(self):
 170         return self.len
 171
 172     @property
 173     def is_ip6(self):
 174         return self.addr.is_ip6
 175
 176     def __str__(self):
 177         return "%s/%d" % (self.address, self.length)
 178
 179     def __eq__(self, other):
 180         if isinstance(other, self.__class__):
 181             return (self.len == other.len and self.addr == other.addr)
 182         elif hasattr(other, "address") and hasattr(other, "address_length"):
 183             # vl_api_prefix_t
 184             return self.len == other.address_length and \
 185                    self.addr == other.address
 186         else:
 187             _log.error(
 188                 "Comparing VppIpPrefix:%s with incomparable type: %s" %
 189                 (self, other))
 190             return NotImplemented
 191
 192
 193 class VppIpMPrefix():
 194     def __init__(self, saddr, gaddr, len):
 195         self.saddr = saddr
 196         self.gaddr = gaddr
 197         self.len = len
 198         self.ip_saddr = ip_address(text_type(self.saddr))
 199         self.ip_gaddr = ip_address(text_type(self.gaddr))
 200         if self.ip_saddr.version != self.ip_gaddr.version:
 201             raise ValueError('Source and group addresses must be of the '
 202                              'same address family.')
 203
 204     def encode(self):
 205         if 6 == self.ip_saddr.version:
 206             prefix = {
 207                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
 208                 'grp_address': {'ip6': self.ip_gaddr.packed},
 209                 'src_address': {'ip6': self.ip_saddr.packed},
 210                 'grp_address_length': self.len,
 211             }
 212         else:
 213             prefix = {
 214                 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
 215                 'grp_address': {'ip4': self.ip_gaddr.packed},
 216                 'src_address': {'ip4': self.ip_saddr.packed},
 217                 'grp_address_length': self.len,
 218             }
 219         return prefix