diff --git a/ironic/conductor/local_rpc.py b/ironic/conductor/local_rpc.py index 79f55b7230..7725e599eb 100644 --- a/ironic/conductor/local_rpc.py +++ b/ironic/conductor/local_rpc.py @@ -11,8 +11,8 @@ # limitations under the License. import atexit -import os import secrets +import socket import tempfile from keystoneauth1 import loading as ks_loading @@ -34,10 +34,15 @@ _USERNAME = 'ironic' def _lo_has_ipv6(): - return ( - os.path.exists("/proc/sys/net/ipv6/conf/lo/disable_ipv6") - and open("/proc/sys/net/ipv6/conf/lo/disable_ipv6").read() != "1" - ) + """Check if IPv6 is available by attempting to bind to ::1.""" + try: + with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as sock: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(('::1', 0)) + return True + except (OSError, socket.error) as e: + LOG.debug('IPv6 is not available on localhost: %s', e) + return False def _create_tls_files(ip): diff --git a/ironic/tests/unit/conductor/test_local_rpc.py b/ironic/tests/unit/conductor/test_local_rpc.py index e3b2ba1a91..942eaf4aa8 100644 --- a/ironic/tests/unit/conductor/test_local_rpc.py +++ b/ironic/tests/unit/conductor/test_local_rpc.py @@ -12,6 +12,7 @@ import ipaddress import os +import socket from unittest import mock import bcrypt @@ -95,3 +96,74 @@ class ConfigureTestCase(tests_base.TestCase): self.assertEqual('127.0.0.1', CONF.json_rpc.host_ip) self._verify_password() self._verify_tls(ipv6=False) + + +@mock.patch('socket.socket', autospec=True) +class LoHasIpv6TestCase(tests_base.TestCase): + + def test_ipv6_available(self, mock_socket): + # Mock successful IPv6 socket creation and bind + mock_sock = mock.Mock() + mock_sock.__enter__ = mock.Mock(return_value=mock_sock) + mock_sock.__exit__ = mock.Mock(return_value=False) + mock_socket.return_value = mock_sock + + result = local_rpc._lo_has_ipv6() + + # Verify socket operations + mock_socket.assert_called_once_with(socket.AF_INET6, + socket.SOCK_STREAM) + mock_sock.setsockopt.assert_called_once_with(socket.SOL_SOCKET, + socket.SO_REUSEADDR, + 1) + mock_sock.bind.assert_called_once_with(('::1', 0)) + self.assertTrue(result) + + def test_ipv6_not_available_os_error(self, mock_socket): + # Mock failed IPv6 socket bind (IPv6 not available) + mock_sock = mock.Mock() + mock_sock.__enter__ = mock.Mock(return_value=mock_sock) + mock_sock.__exit__ = mock.Mock(return_value=False) + mock_socket.return_value = mock_sock + mock_sock.bind.side_effect = OSError("Cannot assign requested address") + + result = local_rpc._lo_has_ipv6() + + # Verify socket operations attempted + mock_socket.assert_called_once_with(socket.AF_INET6, + socket.SOCK_STREAM) + mock_sock.setsockopt.assert_called_once_with(socket.SOL_SOCKET, + socket.SO_REUSEADDR, + 1) + mock_sock.bind.assert_called_once_with(('::1', 0)) + self.assertFalse(result) + + def test_ipv6_not_available_socket_error(self, mock_socket): + # Mock socket.error during bind + mock_sock = mock.Mock() + mock_sock.__enter__ = mock.Mock(return_value=mock_sock) + mock_sock.__exit__ = mock.Mock(return_value=False) + mock_socket.return_value = mock_sock + mock_sock.bind.side_effect = socket.error("Network unreachable") + + result = local_rpc._lo_has_ipv6() + + # Verify socket operations attempted + mock_socket.assert_called_once_with(socket.AF_INET6, + socket.SOCK_STREAM) + mock_sock.setsockopt.assert_called_once_with(socket.SOL_SOCKET, + socket.SO_REUSEADDR, + 1) + mock_sock.bind.assert_called_once_with(('::1', 0)) + self.assertFalse(result) + + def test_ipv6_not_available_socket_creation_fails(self, mock_socket): + # Mock socket creation failure + mock_socket.side_effect = OSError("Address family not supported") + + result = local_rpc._lo_has_ipv6() + + # Verify socket creation attempted + mock_socket.assert_called_once_with(socket.AF_INET6, + socket.SOCK_STREAM) + self.assertFalse(result)