diff --git a/octavia/api/common/types.py b/octavia/api/common/types.py index 24392f972d..89dd862f2d 100644 --- a/octavia/api/common/types.py +++ b/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 diff --git a/octavia/common/utils.py b/octavia/common/utils.py index 50c1d72569..5bf29d4ce4 100644 --- a/octavia/common/utils.py +++ b/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): diff --git a/octavia/common/validate.py b/octavia/common/validate.py index b51eb98909..846587292a 100644 --- a/octavia/common/validate.py +++ b/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 import rfc3986 from wsme import types as wtypes @@ -391,7 +390,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 diff --git a/octavia/tests/unit/common/test_utils.py b/octavia/tests/unit/common/test_utils.py index 29aa528ffa..45a9da243e 100644 --- a/octavia/tests/unit/common/test_utils.py +++ b/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')) diff --git a/octavia/tests/unit/common/test_validate.py b/octavia/tests/unit/common/test_validate.py index 1a1261115d..2897e123da 100644 --- a/octavia/tests/unit/common/test_validate.py +++ b/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')) diff --git a/releasenotes/notes/Remove-netaddr-requirement-0ce7f8605a86172a.yaml b/releasenotes/notes/Remove-netaddr-requirement-0ce7f8605a86172a.yaml new file mode 100644 index 0000000000..f3aa45ec5e --- /dev/null +++ b/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. diff --git a/requirements.txt b/requirements.txt index 1e52559570..d9e4e8ccd7 100644 --- a/requirements.txt +++ b/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