Merge pull request #36676 from matthewoliver/cephadm_orch_ipv6

cephadm: auto wrap and unwrap ipv6 addresses

Reviewed-by: Laura Paduano <lpaduano@suse.com>
Reviewed-by: Sebastian Wagner <sebastian.wagner@suse.com>

src/cephadm/cephadm

@@ -534,10 +534,9 @@ def check_ip_port(ip, port):
     # type: (str, int) -> None
     if not args.skip_ping_check:
         logger.info('Verifying IP %s port %d ...' % (ip, port))
-        if ip.startswith('[') or '::' in ip:
+        if is_ipv6(ip):
             s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
-            if ip.startswith('[') and ip.endswith(']'):
-                ip = ip[1:-1]
+            ip = unwrap_ipv6(ip)
         else:
             s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         try:
@@ -2488,6 +2487,7 @@ def command_inspect_image():
 
 ##################################
 
+
 def unwrap_ipv6(address):
     # type: (str) -> str
     if address.startswith('[') and address.endswith(']'):
@@ -2495,6 +2495,21 @@ def unwrap_ipv6(address):
     return address
 
 
+def wrap_ipv6(address):
+    # type: (str) -> str
+
+    # We cannot assume it's already wrapped or even an IPv6 address if
+    # it's already wrapped it'll not pass (like if it's a hostname) and trigger
+    # the ValueError
+    try:
+        if ipaddress.ip_address(unicode(address)).version == 6:
+            return f"[{address}]"
+    except ValueError:
+        pass
+
+    return address
+
+
 def is_ipv6(address):
     # type: (str) -> bool
     address = unwrap_ipv6(address)
@@ -2550,6 +2565,8 @@ def command_bootstrap():
     base_ip = ''
     if args.mon_ip:
         ipv6 = is_ipv6(args.mon_ip)
+        if ipv6:
+            args.mon_ip = wrap_ipv6(args.mon_ip)
         hasport = r.findall(args.mon_ip)
         if hasport:
             port = int(hasport[0])

src/cephadm/tests/test_cephadm.py

@@ -177,6 +177,20 @@ default via fe80::2480:28ec:5097:3fe2 dev wlp2s0 proto ra metric 20600 pref medi
         for address, expected in tests:
             unwrap_test(address, expected)
 
+    def test_wrap_ipv6(self):
+        def wrap_test(address, expected):
+            assert cd.wrap_ipv6(address) == expected
+
+        tests = [
+            ('::1', '[::1]'), ('[::1]', '[::1]'),
+            ('fde4:8dba:82e1:0:5054:ff:fe6a:357',
+             '[fde4:8dba:82e1:0:5054:ff:fe6a:357]'),
+            ('myhost.example.com', 'myhost.example.com'),
+            ('192.168.0.1', '192.168.0.1'),
+            ('', ''), ('fd00::1::1', 'fd00::1::1')]
+        for address, expected in tests:
+            wrap_test(address, expected)
+
     @mock.patch('cephadm.call_throws')
     @mock.patch('cephadm.get_parm')
     def test_registry_login(self, get_parm, call_throws):

src/pybind/mgr/cephadm/services/cephadmservice.py

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, List, Callable, Any, TypeVar, Generic,  Option
 from mgr_module import HandleCommandResult, MonCommandFailed
 
 from ceph.deployment.service_spec import ServiceSpec, RGWSpec
+from ceph.deployment.utils import is_ipv6, unwrap_ipv6
 from orchestrator import OrchestratorError, DaemonDescription
 from cephadm import utils
 
@@ -250,6 +251,8 @@ class MonService(CephadmService):
                 extra_config += 'public network = %s\n' % network
             elif network.startswith('[v') and network.endswith(']'):
                 extra_config += 'public addrv = %s\n' % network
+            elif is_ipv6(network):
+                extra_config += 'public addr = %s\n' % unwrap_ipv6(network)
             elif ':' not in network:
                 extra_config += 'public addr = %s\n' % network
             else:

src/pybind/mgr/dashboard/tools.py

@@ -3,7 +3,6 @@ from __future__ import absolute_import
 
 import inspect
 import json
-import ipaddress
 import logging
 
 import collections
@@ -16,6 +15,8 @@ import urllib
 
 import cherrypy
 
+from ceph.deployment.utils import wrap_ipv6
+
 from . import mgr
 from .exceptions import ViewCacheNoDataException
 from .settings import Settings
@@ -686,11 +687,7 @@ def build_url(host, scheme=None, port=None):
     :type port: int
     :rtype: str
     """
-    try:
-        ipaddress.IPv6Address(host)
-        netloc = '[{}]'.format(host)
-    except ValueError:
-        netloc = host
+    netloc = wrap_ipv6(host)
     if port:
         netloc += ':{}'.format(port)
     pr = urllib.parse.ParseResult(

src/python-common/ceph/deployment/service_spec.py

@@ -8,6 +8,7 @@ from typing import Optional, Dict, Any, List, Union, Callable, Iterator
 import yaml
 
 from ceph.deployment.hostspec import HostSpec
+from ceph.deployment.utils import unwrap_ipv6
 
 
 class ServiceSpecValidationError(Exception):
@@ -121,13 +122,16 @@ class HostPlacementSpec(namedtuple('HostPlacementSpec', ['hostname', 'network',
         for network in networks:
             # only if we have versioned network configs
             if network.startswith('v') or network.startswith('[v'):
-                network = network.split(':')[1]
+                # if this is ipv6 we can't just simply split on ':' so do
+                # a split once and rsplit once to leave us with just ipv6 addr
+                network = network.split(':', 1)[1]
+                network = network.rsplit(':', 1)[0]
             try:
                 # if subnets are defined, also verify the validity
                 if '/' in network:
                     ip_network(network)
                 else:
-                    ip_address(network)
+                    ip_address(unwrap_ipv6(network))
             except ValueError as e:
                 # logging?
                 raise e

src/python-common/ceph/deployment/utils.py

@@ -0,0 +1,36 @@
+import ipaddress
+import sys
+
+if sys.version_info > (3, 0):
+    unicode = str
+
+
+def unwrap_ipv6(address):
+    # type: (str) -> str
+    if address.startswith('[') and address.endswith(']'):
+        return address[1:-1]
+    return address
+
+
+def wrap_ipv6(address):
+    # type: (str) -> str
+
+    # We cannot assume it's already wrapped or even an IPv6 address if
+    # it's already wrapped it'll not pass (like if it's a hostname) and trigger
+    # the ValueError
+    try:
+        if ipaddress.ip_address(unicode(address)).version == 6:
+            return f"[{address}]"
+    except ValueError:
+        pass
+
+    return address
+
+
+def is_ipv6(address):
+    # type: (str) -> bool
+    address = unwrap_ipv6(address)
+    try:
+        return ipaddress.ip_address(unicode(address)).version == 6
+    except ValueError:
+        return False

src/python-common/ceph/tests/test_utils.py

@@ -0,0 +1,37 @@
+from ceph.deployment.utils import is_ipv6, unwrap_ipv6, wrap_ipv6
+
+
+def test_is_ipv6():
+    for good in ("[::1]", "::1",
+                 "fff:ffff:ffff:ffff:ffff:ffff:ffff:ffff"):
+        assert is_ipv6(good)
+    for bad in ("127.0.0.1",
+                "ffff:ffff:ffff:ffff:ffff:ffff:ffff:fffg",
+                "1:2:3:4:5:6:7:8:9", "fd00::1::1", "[fg::1]"):
+        assert not is_ipv6(bad)
+
+
+def test_unwrap_ipv6():
+    def unwrap_test(address, expected):
+        assert unwrap_ipv6(address) == expected
+
+    tests = [
+        ('::1', '::1'), ('[::1]', '::1'),
+        ('[fde4:8dba:82e1:0:5054:ff:fe6a:357]', 'fde4:8dba:82e1:0:5054:ff:fe6a:357'),
+        ('can actually be any string', 'can actually be any string'),
+        ('[but needs to be stripped] ', '[but needs to be stripped] ')]
+    for address, expected in tests:
+        unwrap_test(address, expected)
+
+
+def test_wrap_ipv6():
+    def wrap_test(address, expected):
+        assert wrap_ipv6(address) == expected
+
+    tests = [
+        ('::1', '[::1]'), ('[::1]', '[::1]'),
+        ('fde4:8dba:82e1:0:5054:ff:fe6a:357', '[fde4:8dba:82e1:0:5054:ff:fe6a:357]'),
+        ('myhost.example.com', 'myhost.example.com'), ('192.168.0.1', '192.168.0.1'),
+        ('', ''), ('fd00::1::1', 'fd00::1::1')]
+    for address, expected in tests:
+        wrap_test(address, expected)