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