Fix IPv6 netmask format for sysconfig

cloudinit/net/__init__.py

@@ -14,7 +14,7 @@
 from typing import Any, Dict
 
 from cloudinit import subp, util
-from cloudinit.net.network_state import mask_to_net_prefix
+from cloudinit.net.network_state import ipv4_mask_to_net_prefix
 from cloudinit.url_helper import UrlError, readurl
 
 LOG = logging.getLogger(__name__)
@@ -1125,9 +1125,12 @@ def __init__(
                 )
             )
         try:
-            self.prefix = mask_to_net_prefix(prefix_or_mask)
+            self.prefix = ipv4_mask_to_net_prefix(prefix_or_mask)
         except ValueError as e:
-            raise ValueError("Cannot setup network: {0}".format(e)) from e
+            raise ValueError(
+                "Cannot setup network, invalid prefix or "
+                "netmask: {0}".format(e)
+            ) from e
 
         self.connectivity_url_data = connectivity_url_data
         self.interface = interface

cloudinit/net/network_state.py

@@ -6,6 +6,7 @@
 
 import copy
 import functools
+import ipaddress
 import logging
 import socket
 import struct
@@ -928,10 +929,16 @@ def _normalize_net_keys(network, address_keys=()):
         try:
             prefix = int(maybe_prefix)
         except ValueError:
-            # this supports input of <address>/255.255.255.0
-            prefix = mask_to_net_prefix(maybe_prefix)
-    elif netmask:
-        prefix = mask_to_net_prefix(netmask)
+            if ipv6:
+                # this supports input of ffff:ffff:ffff::
+                prefix = ipv6_mask_to_net_prefix(maybe_prefix)
+            else:
+                # this supports input of 255.255.255.0
+                prefix = ipv4_mask_to_net_prefix(maybe_prefix)
+    elif netmask and not ipv6:
+        prefix = ipv4_mask_to_net_prefix(netmask)
+    elif netmask and ipv6:
+        prefix = ipv6_mask_to_net_prefix(netmask)
     elif "prefix" in net:
         prefix = int(net["prefix"])
     else:
@@ -1035,88 +1042,42 @@ def ipv4_mask_to_net_prefix(mask):
        str(24)         => 24
        "24"            => 24
     """
-    if isinstance(mask, int):
-        return mask
-    if isinstance(mask, str):
-        try:
-            return int(mask)
-        except ValueError:
-            pass
-    else:
-        raise TypeError("mask '%s' is not a string or int")
-
-    if "." not in mask:
-        raise ValueError("netmask '%s' does not contain a '.'" % mask)
-
-    toks = mask.split(".")
-    if len(toks) != 4:
-        raise ValueError("netmask '%s' had only %d parts" % (mask, len(toks)))
-
-    return sum([bin(int(x)).count("1") for x in toks])
+    return ipaddress.ip_network(f"0.0.0.0/{mask}").prefixlen
 
 
 def ipv6_mask_to_net_prefix(mask):
     """Convert an ipv6 netmask (very uncommon) or prefix (64) to prefix.
 
-    If 'mask' is an integer or string representation of one then
-    int(mask) will be returned.
+    If the input is already an integer or a string representation of
+    an integer, then int(mask) will be returned.
+       "ffff:ffff:ffff::"  => 48
+       "48"                => 48
     """
-
-    if isinstance(mask, int):
-        return mask
-    if isinstance(mask, str):
-        try:
-            return int(mask)
-        except ValueError:
-            pass
-    else:
-        raise TypeError("mask '%s' is not a string or int")
-
-    if ":" not in mask:
-        raise ValueError("mask '%s' does not have a ':'")
-
-    bitCount = [
-        0,
-        0x8000,
-        0xC000,
-        0xE000,
-        0xF000,
-        0xF800,
-        0xFC00,
-        0xFE00,
-        0xFF00,
-        0xFF80,
-        0xFFC0,
-        0xFFE0,
-        0xFFF0,
-        0xFFF8,
-        0xFFFC,
-        0xFFFE,
-        0xFFFF,
-    ]
-    prefix = 0
-    for word in mask.split(":"):
-        if not word or int(word, 16) == 0:
-            break
-        prefix += bitCount.index(int(word, 16))
-
-    return prefix
-
-
-def mask_to_net_prefix(mask):
-    """Return the network prefix for the netmask provided.
-
-    Supports ipv4 or ipv6 netmasks."""
     try:
-        # if 'mask' is a prefix that is an integer.
-        # then just return it.
-        return int(mask)
+        # In the case the mask is already a prefix
+        prefixlen = ipaddress.ip_network(f"::/{mask}").prefixlen
+        return prefixlen
     except ValueError:
+        # ValueError means mask is an IPv6 address representation and need
+        # conversion.
         pass
-    if is_ipv6_addr(mask):
-        return ipv6_mask_to_net_prefix(mask)
-    else:
-        return ipv4_mask_to_net_prefix(mask)
+
+    netmask = ipaddress.ip_address(mask)
+    mask_int = int(netmask)
+    # If the mask is all zeroes, just return it
+    if mask_int == 0:
+        return mask_int
+
+    trailing_zeroes = min(
+        ipaddress.IPV6LENGTH, (~mask_int & (mask_int - 1)).bit_length()
+    )
+    leading_ones = mask_int >> trailing_zeroes
+    prefixlen = ipaddress.IPV6LENGTH - trailing_zeroes
+    all_ones = (1 << prefixlen) - 1
+    if leading_ones != all_ones:
+        raise ValueError("Invalid network mask '%s'" % mask)
+
+    return prefixlen
 
 
 def mask_and_ipv4_to_bcast_addr(mask, ip):

cloudinit/net/sysconfig.py

@@ -10,6 +10,7 @@
 from cloudinit import log as logging
 from cloudinit import subp, util
 from cloudinit.distros.parsers import networkmanager_conf, resolv_conf
+from cloudinit.net import network_state
 
 from . import renderer
 from .network_state import (
@@ -190,69 +191,61 @@ def to_string(self, proto="ipv4"):
         # (because Route can contain a mix of IPv4 and IPv6)
         reindex = -1
         for key in sorted(self._conf.keys()):
-            if "ADDRESS" in key:
-                index = key.replace("ADDRESS", "")
-                address_value = str(self._conf[key])
-                # only accept combinations:
-                # if proto ipv6 only display ipv6 routes
-                # if proto ipv4 only display ipv4 routes
-                # do not add ipv6 routes if proto is ipv4
-                # do not add ipv4 routes if proto is ipv6
-                # (this array will contain a mix of ipv4 and ipv6)
-                if proto == "ipv4" and not self.is_ipv6_route(address_value):
-                    netmask_value = str(self._conf["NETMASK" + index])
-                    gateway_value = str(self._conf["GATEWAY" + index])
-                    # increase IPv4 index
-                    reindex = reindex + 1
-                    buf.write(
-                        "%s=%s\n"
-                        % (
-                            "ADDRESS" + str(reindex),
-                            _quote_value(address_value),
-                        )
-                    )
-                    buf.write(
-                        "%s=%s\n"
-                        % (
-                            "GATEWAY" + str(reindex),
-                            _quote_value(gateway_value),
-                        )
-                    )
+            if "ADDRESS" not in key:
+                continue
+
+            index = key.replace("ADDRESS", "")
+            address_value = str(self._conf[key])
+            netmask_value = str(self._conf["NETMASK" + index])
+            gateway_value = str(self._conf["GATEWAY" + index])
+
+            # only accept combinations:
+            # if proto ipv6 only display ipv6 routes
+            # if proto ipv4 only display ipv4 routes
+            # do not add ipv6 routes if proto is ipv4
+            # do not add ipv4 routes if proto is ipv6
+            # (this array will contain a mix of ipv4 and ipv6)
+            if proto == "ipv4" and not self.is_ipv6_route(address_value):
+                # increase IPv4 index
+                reindex = reindex + 1
+                buf.write(
+                    "%s=%s\n"
+                    % ("ADDRESS" + str(reindex), _quote_value(address_value))
+                )
+                buf.write(
+                    "%s=%s\n"
+                    % ("GATEWAY" + str(reindex), _quote_value(gateway_value))
+                )
+                buf.write(
+                    "%s=%s\n"
+                    % ("NETMASK" + str(reindex), _quote_value(netmask_value))
+                )
+                metric_key = "METRIC" + index
+                if metric_key in self._conf:
+                    metric_value = str(self._conf["METRIC" + index])
                     buf.write(
                         "%s=%s\n"
-                        % (
-                            "NETMASK" + str(reindex),
-                            _quote_value(netmask_value),
-                        )
-                    )
-                    metric_key = "METRIC" + index
-                    if metric_key in self._conf:
-                        metric_value = str(self._conf["METRIC" + index])
-                        buf.write(
-                            "%s=%s\n"
-                            % (
-                                "METRIC" + str(reindex),
-                                _quote_value(metric_value),
-                            )
-                        )
-                elif proto == "ipv6" and self.is_ipv6_route(address_value):
-                    netmask_value = str(self._conf["NETMASK" + index])
-                    gateway_value = str(self._conf["GATEWAY" + index])
-                    metric_value = (
-                        "metric " + str(self._conf["METRIC" + index])
-                        if "METRIC" + index in self._conf
-                        else ""
+                        % ("METRIC" + str(reindex), _quote_value(metric_value))
                     )
-                    buf.write(
-                        "%s/%s via %s %s dev %s\n"
-                        % (
-                            address_value,
-                            netmask_value,
-                            gateway_value,
-                            metric_value,
-                            self._route_name,
-                        )
+            elif proto == "ipv6" and self.is_ipv6_route(address_value):
+                prefix_value = network_state.ipv6_mask_to_net_prefix(
+                    netmask_value
+                )
+                metric_value = (
+                    "metric " + str(self._conf["METRIC" + index])
+                    if "METRIC" + index in self._conf
+                    else ""
+                )
+                buf.write(
+                    "%s/%s via %s %s dev %s\n"
+                    % (
+                        address_value,
+                        prefix_value,
+                        gateway_value,
+                        metric_value,
+                        self._route_name,
                     )
+                )
 
         return buf.getvalue()
 

cloudinit/sources/DataSourceOpenNebula.py

@@ -248,7 +248,7 @@ def gen_conf(self):
             # Set IPv4 address
             devconf["addresses"] = []
             mask = self.get_mask(c_dev)
-            prefix = str(net.mask_to_net_prefix(mask))
+            prefix = str(net.ipv4_mask_to_net_prefix(mask))
             devconf["addresses"].append(self.get_ip(c_dev, mac) + "/" + prefix)
 
             # Set IPv6 Global and ULA address

cloudinit/sources/helpers/vmware/imc/config_nic.py

@@ -10,7 +10,7 @@
 import re
 
 from cloudinit import subp, util
-from cloudinit.net.network_state import mask_to_net_prefix
+from cloudinit.net.network_state import ipv4_mask_to_net_prefix
 
 logger = logging.getLogger(__name__)
 
@@ -182,7 +182,7 @@ def gen_ipv4_route(self, nic, gateways, netmask):
         """
         route_list = []
 
-        cidr = mask_to_net_prefix(netmask)
+        cidr = ipv4_mask_to_net_prefix(netmask)
 
         for gateway in gateways:
             destination = "%s/%d" % (gen_subnet(gateway, netmask), cidr)

tests/unittests/net/test_init.py

@@ -629,7 +629,9 @@ def test_ephemeral_ipv4_network_errors_invalid_mask_prefix(self, m_subp):
                 with net.EphemeralIPv4Network(**params):
                     pass
             error = context_manager.exception
-            self.assertIn("Cannot setup network: netmask", str(error))
+            self.assertIn(
+                "Cannot setup network, invalid prefix or netmask: ", str(error)
+            )
             self.assertEqual(0, m_subp.call_count)
 
     def test_ephemeral_ipv4_network_performs_teardown(self, m_subp):