Replace oslo_utils.netutils type compares with ipaddress

We used netutils earlier on to have a backportable change
however the longer term goal was to replace the change with
using the python native ipaddress module directly.

For the cases where we can change IP version type compares,
we change them with this change.

Note: other uses of netutils still exist, and we should
eventually see if we can phase them out, however the remaining
uses are around MAC address validations.

Change-Id: I44336423194eed99f026c44b6390030a94ed0522
This commit is contained in:
Julia Kreger 2020-04-06 13:45:16 -07:00
parent c871622ff5
commit b1dd9147d2
5 changed files with 20 additions and 14 deletions

View File

@ -17,7 +17,6 @@ from keystoneauth1 import loading as ks_loading
from neutronclient.common import exceptions as neutron_exceptions
from neutronclient.v2_0 import client as clientv20
from oslo_log import log
from oslo_utils import netutils
from oslo_utils import uuidutils
import retrying
@ -248,7 +247,8 @@ def _add_ip_addresses_for_ipv6_stateful(context, port, client):
"""
fixed_ips = port['port']['fixed_ips']
if (not fixed_ips
or not netutils.is_valid_ipv6(fixed_ips[0]['ip_address'])):
or ipaddress.ip_address(
fixed_ips[0]['ip_address']).version != 6):
return
subnet = client.show_subnet(

View File

@ -22,6 +22,7 @@ import contextlib
import datetime
import errno
import hashlib
import ipaddress
import os
import re
import shutil
@ -576,6 +577,6 @@ def pop_node_nested_field(node, collection, field, default=None):
def wrap_ipv6(ip):
"""Wrap the address in square brackets if it's an IPv6 address."""
if netutils.is_valid_ipv6(ip):
if ipaddress.ip_address(ip).version == 6:
return "[%s]" % ip
return ip

View File

@ -19,7 +19,6 @@ import time
from neutronclient.common import exceptions as neutron_client_exc
from oslo_log import log as logging
from oslo_utils import netutils
from ironic.common import exception
from ironic.common.i18n import _
@ -187,12 +186,18 @@ class NeutronDHCPApi(base.BaseDHCP):
ip_address = fixed_ips[0].get('ip_address', None)
if ip_address:
if netutils.is_valid_ipv4(ip_address):
return ip_address
else:
LOG.error("Neutron returned invalid IPv4 "
"address %(ip_address)s on port %(port_uuid)s.",
{'ip_address': ip_address, 'port_uuid': port_uuid})
try:
if ipaddress.ip_address(ip_address).version == 4:
return ip_address
else:
LOG.error("Neutron returned invalid IPv4 "
"address %(ip_address)s on port %(port_uuid)s.",
{'ip_address': ip_address,
'port_uuid': port_uuid})
raise exception.InvalidIPv4Address(ip_address=ip_address)
except ValueError as exc:
LOG.error("An Invalid IP address was supplied and failed "
"basic validation: %s", exc)
raise exception.InvalidIPv4Address(ip_address=ip_address)
else:
LOG.error("No IP address assigned to Neutron port %s.",

View File

@ -21,6 +21,7 @@ Ironic console utilities.
import errno
import fcntl
import ipaddress
import os
import signal
import socket
@ -32,7 +33,6 @@ from oslo_concurrency import lockutils
from oslo_log import log as logging
from oslo_service import loopingcall
from oslo_utils import fileutils
from oslo_utils import netutils
import psutil
from ironic.common import exception
@ -402,7 +402,7 @@ def start_socat_console(node_uuid, port, console_cmd):
args.append('-L%s' % pid_file)
console_host = CONF.console.socat_address
if netutils.is_valid_ipv6(console_host):
if ipaddress.ip_address(console_host).version == 6:
arg = ('TCP6-LISTEN:%(port)s,bind=[%(host)s],reuseaddr,fork,'
'max-children=1')
else:

View File

@ -19,6 +19,7 @@
import errno
import fcntl
import ipaddress
import os
import random
import signal
@ -31,7 +32,6 @@ from unittest import mock
from ironic_lib import utils as ironic_utils
from oslo_config import cfg
from oslo_service import loopingcall
from oslo_utils import netutils
import psutil
from ironic.common import exception
@ -223,7 +223,7 @@ class ConsoleUtilsTestCase(db_base.DbTestCase):
generated_url = (
console_utils.get_shellinabox_console_url(self.info['port']))
console_host = CONF.my_ip
if netutils.is_valid_ipv6(console_host):
if ipaddress.ip_address(console_host).version == 6:
console_host = '[%s]' % console_host
http_url = "%s://%s:%s" % (scheme, console_host, self.info['port'])
self.assertEqual(http_url, generated_url)