Fix local RPC IPv6 detection to use socket binding instead of file checks

The local RPC code previously checked for IPv6 availability by examining
the existence of /proc/sys/net/ipv6/conf/lo/disable_ipv6 and reading its
contents. This approach was unreliable as it depended on filesystem
checks rather than testing actual IPv6 functionality.

The fix replaces the file-based check with an actual socket binding test
to ::1 using a context manager for proper resource management. Socket
reuse is enabled to prevent port conflicts. Debug logging is added when
IPv6 is unavailable. Unit tests are updated to provide comprehensive
coverage of the new implementation.

Change-Id: I1e3afabc78f1382ff5248707ff2ca8114d10dd90
Generated-By: Claude Code Sonnet 4
Signed-off-by: Dmitry Tantsur <dtantsur@protonmail.com>
This commit is contained in:
Dmitry Tantsur
2025-08-06 16:53:42 +02:00
parent ef05073952
commit aa0348678b
2 changed files with 82 additions and 5 deletions

View File

@@ -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):

View File

@@ -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)