Merge "Remove netaddr module requirement"

octavia/api/common/types.py

@@ -13,9 +13,9 @@
 #    under the License.
 
 import copy
+import ipaddress
 
 from dateutil import parser
-import netaddr
 from wsme import types as wtypes
 
 from octavia.common import constants
@@ -50,8 +50,8 @@ class CidrType(wtypes.UserType):
     def validate(value):
         """Validates whether value is an IPv4 or IPv6 CIDR."""
         try:
-            return str(netaddr.IPNetwork(value).cidr)
-        except (ValueError, netaddr.core.AddrFormatError) as e:
+            return ipaddress.ip_network(value, strict=False).with_prefixlen
+        except Exception as e:
             error = 'Value should be IPv4 or IPv6 CIDR format'
             raise ValueError(error) from e
 

octavia/common/utils.py

@@ -20,10 +20,10 @@
 
 import base64
 import hashlib
+import ipaddress
 import re
 import socket
 
-import netaddr
 from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import excutils
@@ -73,47 +73,42 @@ def get_network_driver():
 
 def is_ipv4(ip_address):
     """Check if ip address is IPv4 address."""
-    ip = netaddr.IPAddress(ip_address)
+    ip = ipaddress.ip_address(ip_address)
     return ip.version == 4
 
 
 def is_ipv6(ip_address):
     """Check if ip address is IPv6 address."""
-    ip = netaddr.IPAddress(ip_address)
+    ip = ipaddress.ip_address(ip_address)
     return ip.version == 6
 
 
 def is_cidr_ipv6(cidr):
     """Check if CIDR is IPv6 address with subnet prefix."""
-    ip = netaddr.IPNetwork(cidr)
+    ip = ipaddress.ip_network(cidr, strict=False)
     return ip.version == 6
 
 
 def is_ipv6_lla(ip_address):
     """Check if ip address is IPv6 link local address."""
-    ip = netaddr.IPAddress(ip_address)
-    return ip.version == 6 and ip.is_link_local()
+    ip = ipaddress.ip_address(ip_address)
+    return ip.version == 6 and ip.is_link_local
 
 
 def ip_port_str(ip_address, port):
     """Return IP port as string representation depending on address family."""
-    ip = netaddr.IPAddress(ip_address)
+    ip = ipaddress.ip_address(ip_address)
     if ip.version == 4:
         return "{ip}:{port}".format(ip=ip, port=port)
     return "[{ip}]:{port}".format(ip=ip, port=port)
 
 
 def netmask_to_prefix(netmask):
-    return netaddr.IPAddress(netmask).netmask_bits()
+    return ipaddress.ip_network(f'0.0.0.0/{netmask}', strict=False).prefixlen
 
 
 def ip_netmask_to_cidr(ip, netmask):
-    net = netaddr.IPNetwork("0.0.0.0/0")
-    if ip and netmask:
-        net = netaddr.IPNetwork(
-            "{ip}/{netmask}".format(ip=ip, netmask=netmask)
-        )
-    return "{ip}/{netmask}".format(ip=net.network, netmask=net.prefixlen)
+    return ipaddress.ip_network(f'{ip}/{netmask}', strict=False).with_prefixlen
 
 
 def get_vip_security_group_name(loadbalancer_id):

octavia/common/validate.py

@@ -22,7 +22,6 @@ Defined here so these can also be used at deeper levels than the API.
 import ipaddress
 import re
 
-import netaddr
 from oslo_config import cfg
 from rfc3986 import uri_reference
 from rfc3986 import validators
@@ -397,7 +396,7 @@ def network_allowed_by_config(network_id, valid_networks=None):
 
 
 def is_ip_member_of_cidr(address, cidr):
-    if netaddr.IPAddress(address) in netaddr.IPNetwork(cidr):
+    if ipaddress.ip_address(address) in ipaddress.ip_network(cidr):
         return True
     return False
 

octavia/tests/unit/common/test_utils.py

@@ -41,6 +41,10 @@ class TestConfig(base.TestCase):
         self.assertTrue(utils.is_ipv6('2001:db8::1'))
         self.assertTrue(utils.is_ipv6('fe80::225:90ff:fefb:53ad'))
 
+    def test_is_cidr_ipv6(self):
+        self.assertTrue(utils.is_cidr_ipv6('2001:db8::/32'))
+        self.assertFalse(utils.is_cidr_ipv6('192.0.2.0/32'))
+
     def test_is_ipv6_lla(self):
         self.assertFalse(utils.is_ipv6_lla('192.0.2.10'))
         self.assertFalse(utils.is_ipv6_lla('169.254.0.10'))

octavia/tests/unit/common/test_validate.py

@@ -526,3 +526,13 @@ class TestValidations(base.TestCase):
             exceptions.ValidationException,
             validate.check_alpn_protocols,
             [])
+
+    def test_is_ip_member_of_cidr(self):
+        self.assertTrue(validate.is_ip_member_of_cidr('192.0.0.5',
+                                                      '192.0.0.0/24'))
+        self.assertFalse(validate.is_ip_member_of_cidr('198.51.100.5',
+                                                       '192.0.0.0/24'))
+        self.assertTrue(validate.is_ip_member_of_cidr('2001:db8::5',
+                                                      '2001:db8::/32'))
+        self.assertFalse(validate.is_ip_member_of_cidr('::ffff:0:203.0.113.5',
+                                                       '2001:db8::/32'))

releasenotes/notes/Remove-netaddr-requirement-0ce7f8605a86172a.yaml

@@ -0,0 +1,5 @@
+---
+other:
+  - |
+    The netaddr python module has been removed as an Octavia requirement. It
+    has been replaced with the python standard library 'ipaddress' module.

requirements.txt

@@ -47,7 +47,6 @@ tenacity>=5.0.4  # Apache-2.0
 distro>=1.2.0 # Apache-2.0
 jsonschema>=3.2.0 # MIT
 octavia-lib>=2.5.0 # Apache-2.0
-netaddr>=0.7.19 # BSD
 simplejson>=3.13.2 # MIT
 setproctitle>=1.1.10 # BSD
 python-dateutil>=2.7.0 # BSD