7 from ipaddress import ip_address
8 from socket import AF_INET, AF_INET6
9 from vpp_papi import VppEnum
15 _log = logging.getLogger(__name__)
22 DPO_PROTO_ETHERNET = 3
27 INVALID_INDEX = 0xffffffff
30 class VppIpAddressUnion():
31 def __init__(self, addr):
33 self.ip_addr = ip_address(text_type(self.addr))
37 return {'ip6': self.ip_addr}
39 return {'ip4': self.ip_addr}
43 return self.ip_addr.version
51 return self.ip_addr.max_prefixlen
55 return self.ip_addr.packed
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
63 return self.ip_addr.packed == other.ip4
65 return self.ip_addr.packed == other.ip6
67 _log.error("Comparing VppIpAddressUnions:%s"
68 " with incomparable type: %s",
73 return str(self.ip_addr)
77 def __init__(self, addr):
78 self.addr = VppIpAddressUnion(addr)
81 if self.addr.version == 6:
83 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
84 'un': self.addr.encode()
88 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
89 'un': self.addr.encode()
92 def __eq__(self, other):
93 if isinstance(other, self.__class__):
94 return self.addr == other.addr
95 elif hasattr(other, "af") and hasattr(other, "un"):
99 VppEnum.vl_api_address_family_t.ADDRESS_IP4 and \
100 other.un == self.addr
103 VppEnum.vl_api_address_family_t.ADDRESS_IP6 and \
104 other.un == self.addr
107 "Comparing VppIpAddress:<%s> %s with incomparable "
109 self.__class__.__name__, self,
110 other.__class__.__name__, other)
111 return NotImplemented
113 def __ne__(self, other):
114 return not (self == other)
121 return self.addr.bytes
125 return self.addr.address
129 return self.addr.length
133 return self.addr.version
137 return (self.version == 6)
141 if self.version == 6:
148 if self.version == 6:
149 return DpoProto.DPO_PROTO_IP6
151 return DpoProto.DPO_PROTO_IP4
155 def __init__(self, addr, len):
156 self.addr = VppIpAddress(addr)
159 def __eq__(self, other):
160 if self.address == other.address and self.len == other.len:
165 return {'address': self.addr.encode(),
170 return self.addr.version
174 return self.addr.address
178 return self.addr.bytes
186 return self.addr.is_ip6
189 return "%s/%d" % (self.address, self.length)
191 def __eq__(self, other):
192 if isinstance(other, self.__class__):
193 return (self.len == other.len and self.addr == other.addr)
194 elif hasattr(other, "address") and hasattr(other, "len"):
196 return self.len == other.len and \
197 self.addr == other.address
200 "Comparing VppIpPrefix:%s with incomparable type: %s" %
202 return NotImplemented
205 class VppIpMPrefix():
206 def __init__(self, saddr, gaddr, glen):
210 self.ip_saddr = VppIpAddressUnion(text_type(self.saddr))
211 self.ip_gaddr = VppIpAddressUnion(text_type(self.gaddr))
212 if self.ip_saddr.version != self.ip_gaddr.version:
213 raise ValueError('Source and group addresses must be of the '
214 'same address family.')
217 if 6 == self.ip_saddr.version:
219 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
226 'grp_address_length': self.glen,
230 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
237 'grp_address_length': self.glen,
247 return self.ip_gaddr.version
250 return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
252 def __eq__(self, other):
253 if isinstance(other, self.__class__):
254 return (self.glen == other.glen and
255 self.ip_saddr == other.ip_gaddr and
256 self.ip_saddr == other.ip_saddr)
257 elif (hasattr(other, "grp_address_length") and
258 hasattr(other, "grp_address") and
259 hasattr(other, "src_address")):
261 if 4 == self.ip_saddr.version:
262 if self.glen == other.grp_address_length and \
263 self.gaddr == str(other.grp_address.ip4) and \
264 self.saddr == str(other.src_address.ip4):
268 return (self.glen == other.grp_address_length and
269 self.gaddr == other.grp_address.ip6 and
270 self.saddr == other.src_address.ip6)
272 raise Exception("Comparing VppIpPrefix:%s with unknown type: %s" %