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.packed == other.ip4
72 return self.ip_addr.packed == other.ip6
74 _log.error("Comparing VppIpAddressUnions:%s"
75 " with incomparable type: %s",
80 return str(self.ip_addr)
84 def __init__(self, saddr, gaddr, glen):
88 if ip_address(self.saddr).version != \
89 ip_address(self.gaddr).version:
90 raise ValueError('Source and group addresses must be of the '
91 'same address family.')
96 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP6,
103 'grp_address_length': self.glen,
107 'af': VppEnum.vl_api_address_family_t.ADDRESS_IP4,
114 'grp_address_length': self.glen,
124 return ip_address(self.gaddr).version
127 return "(%s,%s)/%d" % (self.saddr, self.gaddr, self.glen)
129 def __eq__(self, other):
130 if isinstance(other, self.__class__):
131 return (self.glen == other.glen and
132 self.saddr == other.gaddr and
133 self.saddr == other.saddr)
134 elif (hasattr(other, "grp_address_length") and
135 hasattr(other, "grp_address") and
136 hasattr(other, "src_address")):
138 if 4 == self.version:
139 return (self.glen == other.grp_address_length and
140 self.gaddr == str(other.grp_address.ip4) and
141 self.saddr == str(other.src_address.ip4))
143 return (self.glen == other.grp_address_length and
144 self.gaddr == str(other.grp_address.ip6) and
145 self.saddr == str(other.src_address.ip6))
147 raise Exception("Comparing VppIpPrefix:%s with unknown type: %s" %