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 def get_dpo_proto(addr):
31 if ip_address(addr).version == 6:
32 return DpoProto.DPO_PROTO_IP6
34 return DpoProto.DPO_PROTO_IP4
37 class VppIpAddressUnion():
38 def __init__(self, addr):
40 self.ip_addr = ip_address(text_type(self.addr))
44 return {'ip6': self.ip_addr}
46 return {'ip4': self.ip_addr}
50 return self.ip_addr.version
58 return self.ip_addr.max_prefixlen
62 return self.ip_addr.packed
64 def __eq__(self, other):
65 if isinstance(other, self.__class__):
66 return self.ip_addr == other.ip_addr
67 elif hasattr(other, "ip4") and hasattr(other, "ip6"):
68 # vl_api_address_union_t
70 return self.ip_addr == other.ip4
72 return self.ip_addr == other.ip6
74 raise Exception("Comparing VppIpAddressUnions:%s"
75 " with incomparable type: %s",
78 def __ne__(self, other):
79 return not (self == other)
82 return str(self.ip_addr)
86 def __init__(self, saddr, gaddr, glen):
90 if ip_address(self.saddr).version != \
91 ip_address(self.gaddr).version:
92 raise ValueError('Source and group addresses must be of the '
93 'same address family.')
97 'af': ip_address(self.gaddr).vapi_af,
99 ip_address(self.gaddr).vapi_af_name: self.gaddr
102 ip_address(self.saddr).vapi_af_name: self.saddr
104 'grp_address_length': self.glen,
113 return ip_address(self.gaddr).version
116 return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
118 def __eq__(self, other):
119 if isinstance(other, self.__class__):
120 return (self.glen == other.glen and
121 self.saddr == other.gaddr and
122 self.saddr == other.saddr)
123 elif (hasattr(other, "grp_address_length") and
124 hasattr(other, "grp_address") and
125 hasattr(other, "src_address")):
127 if 4 == self.version:
128 return (self.glen == other.grp_address_length and
129 self.gaddr == str(other.grp_address.ip4) and
130 self.saddr == str(other.src_address.ip4))
132 return (self.glen == other.grp_address_length and
133 self.gaddr == str(other.grp_address.ip6) and
134 self.saddr == str(other.src_address.ip6))
135 return NotImplemented