Migrate from netaddr to ipaddress (#1926)

* Most netaddr calls have been replaced with ipaddress calls, the
  major exception is MAC address handling which is not available
  in ipaddress
* The 'get ip from interface' and 'get prefix from interface'
  functions have been moved to 'utils.routing' and most other
  modules use those functions
* Direct calls to ipaddress functions are use only for more advanced
  use cases.
* Error messages in augment.addressing and augment.links have been
  simplified and include parsing error information from ipaddress
* 'get_addr_mask' and 'get_nth_ip_from_prefix' functions did almost
  the same thing and have been merged into a single function
* Extra properties available in ipaddress make it easier to check
  if an address is a network address, multicast address, or a
  broadcast address

Note:
* Sometimes we still cheat and merge IP address string with a
  prefix length string to get a full prefix. There's no good
  way to get that done with ipaddress functions (you cannot change
  the prefix length in IPv4Network/IPv6Network)
* Some error checking code has been removed from old modules; that
  functionality is now available as generic data validation

Author: ipspace

Diff for: netsim/augment/addressing.py

@@ -52,6 +52,7 @@
 import typing
 import types
 
+import ipaddress
 import netaddr
 from box import Box
 
@@ -65,7 +66,7 @@ def normalize_prefix(pfx: typing.Union[str,Box]) -> Box:
   # Normalize IP addr strings, e.g. 2001:001::/48 becomes 2001:1::/48
   def normalize_ip(ip:typing.Union[str,bool]) -> typing.Union[str,bool]:
     try:
-      return str(netaddr.IPNetwork(ip)) if isinstance(ip,str) else ip
+      return str(ipaddress.ip_network(ip)) if isinstance(ip,str) else ip
     except Exception as ex:
       log.error(
         f'Cannot parse address prefix: {ex}',
@@ -152,14 +153,12 @@ def validate_pools(addrs: Box, topology: Box) -> None:
       if k in pfx:
         if not isinstance(pfx[k],bool):
           try:
-            network = netaddr.IPNetwork(pfx[k])
-            if str(network.cidr) != pfx[k]:
-              log.error( f"pool '{pool}' is using an invalid prefix {pfx[k]} with host bits set ({str(network.cidr)})",
-                            category=log.IncorrectValue, module='addressing')
+            network = ipaddress.ip_network(pfx[k])
             addrs[pool][k+'_pfx'] = network
-          except:
+          except Exception as Ex:
             log.error(
-              "%s prefix %s in addressing pool '%s' is invalid (%s)" % (k,pfx[k],pool,sys.exc_info()[1]),
+              f"{k} prefix {pfx[k]} in addressing pool '{pool}' is invalid",
+              more_data=str(Ex),
               category=log.IncorrectValue,
               module='addressing')
             continue
@@ -221,7 +220,7 @@ def create_pool_generators(addrs: Box, no_copy_list: list) -> Box:
       if "_pfx" in key:
         af   = key.replace('_pfx','')
         plen = pfx['prefix'] if af == 'ipv4' else pfx.get('prefix6',64)
-        gen[pool][af] = data.subnet(plen)
+        gen[pool][af] = data.subnets(new_prefix=plen)
         if (af == 'ipv4' and plen == 32) or (af == 'ipv6' and plen >= 127) or (pool == 'loopback'):
           next(gen[pool][af])
       elif not key in no_copy_list:
@@ -239,7 +238,7 @@ def get_pool(pools: Box, pool_list: typing.List[str]) -> typing.Optional[str]:
     module='addressing')                       # pragma: no cover (impossible to get here due to built-in default pools)
   return None                                  # pragma: no cover
 
-def get_nth_subnet(n: int, subnet: netaddr.IPNetwork.subnet, cache_list: list) -> netaddr.IPNetwork:
+def get_nth_subnet(n: int, subnet: types.GeneratorType, cache_list: list) -> ipaddress._BaseNetwork:
   while len(cache_list) < n:
     cache_list.append(next(subnet))
   return cache_list[n-1]
@@ -253,7 +252,7 @@ def get_pool_prefix(pools: Box, p: str, n: typing.Optional[int] = None) -> Box:
       prefixes[af] = pools[p][af]
       continue
 
-    subnet_cache = 'cache_%s' % af
+    subnet_cache = f'cache_{af}'
     if n:                                                             # Allocating a specific prefix or IP address from a subnet
       if not subnet_cache in pools[p]:                                # Set up a cache to speed up things
         pools[p][subnet_cache] = []
@@ -365,9 +364,13 @@ def parse_prefix(prefix: typing.Union[str,Box],path: str = 'links') -> Box:
       evaluate_named_prefix(topology,prefix)
       return topology.prefix[prefix]
     try:
-      return get_box({ 'ipv4' : netaddr.IPNetwork(prefix) })
+      return get_box({ 'ipv4' : ipaddress.IPv4Network(prefix) })
     except Exception as ex:
-      log.error(f'Cannot parse {prefix} used in {path} as an IPv4 prefix',log.IncorrectValue,'addressing')
+      log.error(
+        f'Cannot parse {prefix} used in {path} as an IPv4 prefix',
+        more_data=str(ex),
+        category=log.IncorrectValue,
+        module='addressing')
       return empty_box
 
   if not isinstance(prefix,Box):                      # pragma: no cover
@@ -394,23 +397,17 @@ def parse_prefix(prefix: typing.Union[str,Box],path: str = 'links') -> Box:
       continue
 
     try:
-      prefix_list[af] = netaddr.IPNetwork(pfx)
+      prefix_list[af] = ipaddress.ip_network(pfx)
     except Exception as ex:
       log.error(
         f'Cannot parse {af} prefix: {prefix} used in {path}',
         more_data= [ str(ex) ],
         category=log.IncorrectValue,
         module='addressing')
       return empty_box
-    if str(prefix_list[af]) != str(prefix_list[af].cidr):
-      log.error(
-        f'{af} prefix used in {path} contains host bits: {prefix}',
-        category=log.IncorrectValue,
-        module='addressing')    
 
   return prefix_list
 
-def get_addr_mask(pfx: netaddr.IPNetwork, host: int) -> str:
-  host_ip = netaddr.IPNetwork(pfx[host])
-  host_ip.prefixlen = pfx.prefixlen
-  return str(host_ip)
+def get_nth_ip_from_prefix(pfx: typing.Union[str,ipaddress._BaseNetwork], host: int) -> str:
+  net = ipaddress.ip_network(pfx) if isinstance(pfx,str) else pfx
+  return f'{str(net[host])}/{net.prefixlen}'

Diff for: netsim/augment/links.py

@@ -4,7 +4,7 @@
 #
 import typing
 
-import netaddr
+import ipaddress
 from box import Box
 
 # Related modules
@@ -373,7 +373,7 @@ def set_fhrp_gateway(link: Box, pfx_list: Box, nodes: Box, link_path: str) -> No
       continue
 
     try:                                                              # Now try to get N-th IP address on that link
-      link.gateway[af] = get_nth_ip_from_prefix(netaddr.IPNetwork(link.prefix[af]),link.gateway.id)
+      link.gateway[af] = addressing.get_nth_ip_from_prefix(link.prefix[af],link.gateway.id)
       fhrp_assigned = True
     except Exception as ex:
       log.error(
@@ -468,15 +468,15 @@ def assign_link_prefix(
 
 Return 'error' if the prefix size is too small
 """
-def get_prefix_IPAM_policy(link: Box, pfx: typing.Union[netaddr.IPNetwork,bool], ndict: Box) -> str:
+def get_prefix_IPAM_policy(link: Box, pfx: typing.Union[ipaddress._BaseNetwork,bool], ndict: Box) -> str:
   if isinstance(pfx,bool):
     return 'unnumbered'
 
   gwid = get_gateway_id(link) or 0                                    # Get link gateway ID (if set) --- must be int for min to work
   if link.type == 'p2p' and not gwid:                                 # P2P allocation policy cannot be used with default gateway
-    return 'p2p' if pfx.first != pfx.last else 'error'
+    return 'p2p' if pfx.num_addresses > 1 else 'error'
 
-  pfx_size = pfx.last - pfx.first + 1
+  pfx_size = pfx.num_addresses
   add_extra_ip = 0
   subtract_reserved_ip = -2
 
@@ -501,19 +501,6 @@ def get_prefix_IPAM_policy(link: Box, pfx: typing.Union[netaddr.IPNetwork,bool],
 
   return 'error'
 
-"""
-Get Nth IP address in a prefix returned as a nice string with a subnet mask
-
-*** WARNING *** WARNING *** WARNING ***
-
-Parent must catch the exception as we don't know what error text to display
-"""
-
-def get_nth_ip_from_prefix(pfx: netaddr.IPNetwork, n_id: int) -> str:
-  node_addr = netaddr.IPNetwork(pfx[n_id])
-  node_addr.prefixlen = pfx.prefixlen
-  return str(node_addr)
-
 """
 check_interface_host_bits: Check whether the IP addresses on an interface have host bits. The check is disabled
 for special prefixes (IPv4: /31, /32, IPv6: /127, /128)
@@ -533,27 +520,27 @@ def check_interface_host_bits(intf: Box, node: Box, link: typing.Optional[Box] =
       continue
     if not isinstance(intf[af],str):            # Skip unnumbered interfaces
       continue
-    pfx = netaddr.IPNetwork(intf[af])
-    if pfx[0].is_multicast():
+    pfx = ipaddress.ip_interface(intf[af])
+    if pfx.ip.is_multicast:
       log.error(
         f'Interfaces cannot have multicast {af} addresses ({intf[af]} on {node.name}/{intf_name})',
         log.IncorrectValue,
         'links')
       OK = False
       continue
 
-    if pfx.last <= pfx.first + 1:               # Are we dealing with special prefix (loopback or /31)
+    if pfx.network.num_addresses <= 2:          # Are we dealing with special prefix (loopback or /31)
       continue                                  # ... then it's OK not to have host bits
 
-    if str(pfx) == str(pfx.cidr):               # Does the IP address have host bits -- is it different from its CIDR subnet?
+    if pfx.ip == pfx.network.network_address:   # Does the IP address have host bits -- is it different from its CIDR subnet?
       log.error(
         f'Address {intf[af]} on node {node.name}/{intf_name} does not contain host bits',
         log.IncorrectValue,
         'links')
       OK = False
       continue
 
-    if str(pfx[-1]) == str(pfx.ip) and af == 'ipv4':
+    if pfx.ip == pfx.network.broadcast_address:
       log.error(
         f'Address {intf[af]} on node {node.name}/{intf_name} is a subnet broadcast address',
         log.IncorrectValue,
@@ -566,7 +553,7 @@ def check_interface_host_bits(intf: Box, node: Box, link: typing.Optional[Box] =
 """
 Set an interface address based on the link prefix and interface sequential number (could be node.id or counter)
 """
-def set_interface_address(intf: Box, af: str, pfx: netaddr.IPNetwork, node_id: int) -> bool:
+def set_interface_address(intf: Box, af: str, pfx: ipaddress._BaseNetwork, node_id: int) -> bool:
   if af in intf:                                # Check static interface addresses
     if isinstance(intf[af],bool):               # unnumbered or unaddressed node, leave it alone
       return True
@@ -575,22 +562,23 @@ def set_interface_address(intf: Box, af: str, pfx: netaddr.IPNetwork, node_id: i
       node_id = intf[af]
     elif isinstance(intf[af],str):              # static address specified on the interface
       try:
-        intf_pfx = netaddr.IPNetwork(intf[af])  # Try to parse the interface IP address
-        if not '/' in intf[af]:
-          intf_pfx.prefixlen = pfx.prefixlen    # If it lacks a prefix, add link prefix
+        if '/' not in intf[af]:                 # Add link prefix if needed
+          intf[af] += f'/{pfx.prefixlen}'
+        intf_pfx = ipaddress.ip_interface(intf[af])
         intf[af] = str(intf_pfx)                # ... and save modified/validated interface IP address
       except Exception as ex:
         log.error(
-          f'Cannot parse {af} address {intf.af} for node {intf.node}\n'+strings.extra_data_printout(str(ex)),
-          log.IncorrectValue,
-          'links')
+          f'Cannot parse {af} address {intf.af} for node {intf.node}',
+          more_data=str(ex),
+          category=log.IncorrectValue,
+          module='links')
         return False
 
       return True                               # Further checks done in check_interface_host_bits
 
   # No static interface address, or static address specified as relative node_id
   try:
-    intf[af] = get_nth_ip_from_prefix(pfx,node_id)
+    intf[af] = addressing.get_nth_ip_from_prefix(pfx,node_id)
     return True
   except Exception as ex:
     log.error(
@@ -625,8 +613,8 @@ def IPAM_unnumbered(link: Box, af: str, pfx: typing.Optional[bool], ndict: Box)
 
   return OK
 
-def IPAM_sequential(link: Box, af: str, pfx: netaddr.IPNetwork, ndict: Box) -> bool:
-  start = 1 if pfx.last != pfx.first + 1 else 0
+def IPAM_sequential(link: Box, af: str, pfx: ipaddress._BaseNetwork, ndict: Box) -> bool:
+  start = 1 if pfx.num_addresses > 2 else 0
   gwid = get_gateway_id(link)
   OK = True
   for count,intf in enumerate(link.interfaces):
@@ -636,25 +624,25 @@ def IPAM_sequential(link: Box, af: str, pfx: netaddr.IPNetwork, ndict: Box) -> b
 
   return OK
 
-def IPAM_p2p(link: Box, af: str, pfx: netaddr.IPNetwork, ndict: Box) -> bool:
-  start = 1 if pfx.last != pfx.first + 1 else 0
+def IPAM_p2p(link: Box, af: str, pfx: ipaddress._BaseNetwork, ndict: Box) -> bool:
+  start = 1 if pfx.num_addresses > 2 else 0
   OK = True
   for count,intf in enumerate(sorted(link.interfaces, key=lambda intf: intf.node)):
     OK = set_interface_address(intf,af,pfx,count+start) and OK
 
   return OK
 
-def IPAM_id_based(link: Box, af: str, pfx: netaddr.IPNetwork, ndict: Box) -> bool:
+def IPAM_id_based(link: Box, af: str, pfx: ipaddress._BaseNetwork, ndict: Box) -> bool:
   OK = True
   for intf in link.interfaces:
     OK = set_interface_address(intf,af,pfx,ndict[intf.node].id) and OK
 
   return OK
 
-def IPAM_loopback(link: Box, af: str, pfx: netaddr.IPNetwork, ndict: Box) -> bool:
+def IPAM_loopback(link: Box, af: str, pfx: ipaddress._BaseNetwork, ndict: Box) -> bool:
   for intf in link.interfaces:
-    pfx.prefixlen = 128 if ':' in str(pfx) else 32
-    intf[af] = str(pfx)
+    prefixlen = 128 if af == 'ipv6' else 32
+    intf[af] = f'{str(pfx.network_address)}/{prefixlen}'
 
   return True
 
@@ -699,14 +687,13 @@ def assign_interface_addresses(link: Box, addr_pools: Box, ndict: Box, defaults:
       pfx_net = pfx_list[af]
     else:
       try:                                            # Parse the AF prefix
-        pfx_net = netaddr.IPNetwork(pfx_list[af])
+        pfx_net = ipaddress.ip_network(pfx_list[af])
       except Exception as ex:                         # Report an error and move on if it cannot be parsed
         log.error(
-          f'Cannot parse {af} prefix {pfx_list[af]} on {link._linkname}\n' + \
-            strings.extra_data_printout(f'{ex}') + '\n' + \
-            strings.extra_data_printout(f'{link}'),
-          log.IncorrectValue,
-          'links')
+          f'Cannot parse {af} prefix {pfx_list[af]} on {link._linkname}',
+          more_data=[ str(ex), str(link) ],
+          category=log.IncorrectValue,
+          module='links')
         error = True
         continue
 

Diff for: netsim/augment/nodes.py

@@ -360,7 +360,7 @@ def loopback_interface(n: Box, pools: Box, topology: Box) -> None:
         if prefix_list[af].prefixlen == 128:
           n.loopback[af] = str(prefix_list[af])
         else:
-          n.loopback[af] = addressing.get_addr_mask(prefix_list[af],1)
+          n.loopback[af] = addressing.get_nth_ip_from_prefix(prefix_list[af],1)
       else:
         n.loopback[af] = str(prefix_list[af])
       n.af[af] = True

Diff for: netsim/data/types.py

@@ -2,7 +2,7 @@
 # Data validation routines
 #
 
-import typing,typing_extensions,types
+import typing,typing_extensions
 import functools
 import ipaddress
 import netaddr
@@ -743,21 +743,14 @@ def transform_to_ipv6(value: typing.Any) -> dict:
     return { '_type': 'IPv4 or IPv6 prefix' }
 
   try:
-    parse = netaddr.IPNetwork(value)                                  # now let's check if we have a valid address
+    parse = ipaddress.ip_network(value)                              # now let's check if we have a valid prefix
   except Exception as ex:
-    return check_named_prefix(value) or { '_value': "IPv4, IPv6, or named prefix" }
+    return check_named_prefix(value) or { '_value': f"IPv4, IPv6, or named prefix" }
 
-  if parse.network != parse.ip:
-    return { '_value': "IPv4 or IPv6 prefix without the host bits" }
-
-  try:                                                                # ... and finally we have to check it's a true IPv4 address
-    parse.ipv4()
-    if not parse.is_ipv4_mapped():
-      return { '_valid': True, '_transform': transform_to_ipv4 }
-  except Exception as ex:
-    pass
-
-  return { '_valid': True, '_transform': transform_to_ipv6 }
+  if isinstance(parse,ipaddress.IPv4Network):
+    return { '_valid': True, '_transform': transform_to_ipv4 }
+  else:
+    return { '_valid': True, '_transform': transform_to_ipv6 }
 
 @type_test()
 def must_be_named_pfx(value: typing.Any) -> dict:
@@ -832,9 +825,7 @@ def must_be_rd(value: typing.Any) -> dict:
     rdt_parsed = int(rdt)
   except Exception as ex:
     try:
-      if '/' in rdt:
-        return { '_value': "route distinguisher in asn:id or ip:id format" }
-      netaddr.IPNetwork(rdt)
+      ipaddress.IPv4Address(rdt)
     except Exception as ex:
       return { '_value': "an RD in asn:id or ip:id format where asn is an integer or ip is an IPv4 address" }
 

Diff for: netsim/extra/ebgp.multihop/plugin.py

@@ -1,5 +1,5 @@
 from box import Box
-import netaddr
+import ipaddress
 
 from netsim.utils import log,routing as _bgp
 from netsim.modules import vrf
@@ -177,7 +177,7 @@ def fix_vrf_loopbacks(ndata: Box, topology: Box) -> None:
       for af in ('ipv4','ipv6'):                                # Now copy remote VRF loopback IPv4/IPv6 address
         ngb.pop(af)                                             # ... into neighbor data
         if af in lb:                                            # ... removing whatever might have come from the
-          ngb[af] = str(netaddr.IPNetwork(lb[af]).ip)           # ... global loopback
+          ngb[af] = str(ipaddress.ip_interface(lb[af]).ip)      # ... global loopback
 
     if '_src_vrf' in ngb:                                       # Is out endpoint in a VRF?
       if not isinstance(features.bgp.multihop,Box) or features.bgp.multihop.vrf is not True:

Diff for: netsim/extra/mlag.vtep/plugin.py

@@ -37,7 +37,7 @@ def pre_link_transform(topology: Box) -> None:
       if 'vxlan' in node.module:
           vtep_loopback = data.get_empty_box()
           vtep_loopback.type = 'loopback'           # Assign same static IP to both nodes
-          vtep_loopback.interfaces = [ { 'node': node_name, 'ipv4': str(vtep_a.network)+"/32" } ]
+          vtep_loopback.interfaces = [ { 'node': node_name, 'ipv4': str(vtep_a.network_address)+"/32" } ]
           vtep_loopback._linkname = f"MLAG VTEP VXLAN interface shared between {' - '.join(peers)}"
           vtep_loopback.vxlan.vtep = True
           topology.links.append(vtep_loopback)

Diff for: netsim/extra/multilab/__init__.py

@@ -1,4 +1,4 @@
-import typing, netaddr
+import typing
 from box import Box
 from netsim import utils,data
 

Diff for: netsim/extra/proxy-arp/plugin.py

@@ -1,4 +1,4 @@
-import typing, netaddr
+import typing
 from box import Box
 
 from netsim import api