From 7bd016008429f0a18393a070d88e669f3ed89caa Mon Sep 17 00:00:00 2001

From: Emanuele Giuseppe Esposito <eesposit@redhat.com>

Date: Fri, 11 Feb 2022 14:37:46 +0100

Subject: [PATCH] Fix IPv6 netmask format for sysconfig (#1215)

RH-Author: Emanuele Giuseppe Esposito <eesposit@redhat.com>

RH-MergeRequest: 48: Fix IPv6 netmask format for sysconfig (#1215)

RH-Commit: [1/1] 4c940bbcf85dba1fce9f4acb9fc7820c0d7777f6

RH-Bugzilla: 2046540

RH-Acked-by: Eduardo Otubo <otubo@redhat.com>

RH-Acked-by: Vitaly Kuznetsov <vkuznets@redhat.com>

commit b97a30f0a05c1dea918c46ca9c05c869d15fe2d5

Author: Harald <hjensas@redhat.com>

Date:   Tue Feb 8 15:49:00 2022 +0100

    Fix IPv6 netmask format for sysconfig (#1215)

    This change converts the IPv6 netmask from the network_data.json[1]

    format to the CIDR style, <IPv6_addr>/<prefix>.

    Using an IPv6 address like ffff:ffff:ffff:ffff:: does not work with

    NetworkManager, nor networkscripts.

    NetworkManager will ignore the route, logging:

      ifcfg-rh: ignoring invalid route at \

        "::/:: via fd00:fd00:fd00:2::fffe dev $DEV" \

        (/etc/sysconfig/network-scripts/route6-$DEV:3): \

        Argument for "::/::" is not ADDR/PREFIX format

    Similarly if using networkscripts, ip route fail with error:

      Error: inet6 prefix is expected rather than \

        "fd00:fd00:fd00::/ffff:ffff:ffff:ffff::".

    Also a bit of refactoring ...

    cloudinit.net.sysconfig.Route.to_string:

    * Move a couple of lines around to reduce repeated code.

    * if "ADDRESS" not in key -> continute, so that the

      code block following it can be de-indented.

    cloudinit.net.network_state:

    * Refactors the ipv4_mask_to_net_prefix, ipv6_mask_to_net_prefix

      removes mask_to_net_prefix methods. Utilize ipaddress library to

      do some of the heavy lifting.

    LP: #1959148

Signed-off-by: Emanuele Giuseppe Esposito <eesposit@redhat.com>

---

 cloudinit/net/__init__.py                     |   7 +-

 cloudinit/net/network_state.py                | 103 +++++++-----------

 cloudinit/net/sysconfig.py                    |  91 ++++++++++------

 cloudinit/sources/DataSourceOpenNebula.py     |   2 +-

 .../sources/helpers/vmware/imc/config_nic.py  |   4 +-

 tests/unittests/test_net.py                   |  78 ++++++++++++-

 6 files changed, 176 insertions(+), 109 deletions(-)

diff --git a/cloudinit/net/__init__.py b/cloudinit/net/__init__.py

index 003efa2a..12bf64de 100644

--- a/cloudinit/net/__init__.py

+++ b/cloudinit/net/__init__.py

@@ -14,7 +14,7 @@ import re

 from cloudinit import subp

 from cloudinit import 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__)

@@ -1048,10 +1048,11 @@ class EphemeralIPv4Network(object):

                 'Cannot init network on {0} with {1}/{2} and bcast {3}'.format(

                     interface, ip, prefix_or_mask, broadcast))

         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)

+                "Cannot setup network, invalid prefix or "

+                "netmask: {0}".format(e)

             ) from e

         self.connectivity_url = connectivity_url

diff --git a/cloudinit/net/network_state.py b/cloudinit/net/network_state.py

index e8bf9e39..2768ef94 100644

--- a/cloudinit/net/network_state.py

+++ b/cloudinit/net/network_state.py

@@ -6,6 +6,7 @@

 import copy

 import functools

+import ipaddress

 import logging

 import socket

 import struct

@@ -872,12 +873,18 @@ 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)

-    elif 'prefix' in net:

-        prefix = int(net['prefix'])

+            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:

         prefix = 64 if ipv6 else 24

@@ -972,72 +979,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):

diff --git a/cloudinit/net/sysconfig.py b/cloudinit/net/sysconfig.py

index d5440998..7ecbe1c3 100644

--- a/cloudinit/net/sysconfig.py

+++ b/cloudinit/net/sysconfig.py

@@ -12,6 +12,7 @@ from cloudinit import util

 from cloudinit import subp

 from cloudinit.distros.parsers import networkmanager_conf

 from cloudinit.distros.parsers import resolv_conf

+from cloudinit.net import network_state

 from . import renderer

 from .network_state import (

@@ -171,43 +172,61 @@ class Route(ConfigMap):

         # (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)))

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

+            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 via %s %s dev %s\n" % (address_value,

-                                                      netmask_value,

-                                                      gateway_value,

-                                                      metric_value,

-                                                      self._route_name))

+                        "%s=%s\n"

+                        % ("METRIC" + str(reindex), _quote_value(metric_value))

+                    )

+            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()

diff --git a/cloudinit/sources/DataSourceOpenNebula.py b/cloudinit/sources/DataSourceOpenNebula.py

index 730ec586..e7980ab1 100644

--- a/cloudinit/sources/DataSourceOpenNebula.py

+++ b/cloudinit/sources/DataSourceOpenNebula.py

@@ -233,7 +233,7 @@ class OpenNebulaNetwork(object):

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

diff --git a/cloudinit/sources/helpers/vmware/imc/config_nic.py b/cloudinit/sources/helpers/vmware/imc/config_nic.py

index 9cd2c0c0..3a45c67e 100644

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

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

@@ -9,7 +9,7 @@ import logging

 import os

 import re

-from cloudinit.net.network_state import mask_to_net_prefix

+from cloudinit.net.network_state import ipv4_mask_to_net_prefix

 from cloudinit import subp

 from cloudinit import util

@@ -180,7 +180,7 @@ class NicConfigurator(object):

         """

         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)

diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py

index 14d3462f..a7f6a1f7 100644

--- a/tests/unittests/test_net.py

+++ b/tests/unittests/test_net.py

@@ -2025,10 +2025,10 @@ pre-down route del -net 10.0.0.0/8 gw 11.0.0.1 metric 3 || true

                     routes:

                         - gateway: 2001:67c:1562:1

                           network: 2001:67c:1

-                          netmask: ffff:ffff:0

+                          netmask: "ffff:ffff::"

                         - gateway: 3001:67c:1562:1

                           network: 3001:67c:1

-                          netmask: ffff:ffff:0

+                          netmask: "ffff:ffff::"

                           metric: 10000

             """),

         'expected_netplan': textwrap.dedent("""

@@ -2295,8 +2295,8 @@ iface bond0 inet6 static

             'route6-bond0': textwrap.dedent("""\

         # Created by cloud-init on instance boot automatically, do not edit.

         #

-        2001:67c:1/ffff:ffff:0 via 2001:67c:1562:1  dev bond0

-        3001:67c:1/ffff:ffff:0 via 3001:67c:1562:1 metric 10000 dev bond0

+        2001:67c:1/32 via 2001:67c:1562:1  dev bond0

+        3001:67c:1/32 via 3001:67c:1562:1 metric 10000 dev bond0

             """),

             'route-bond0': textwrap.dedent("""\

         ADDRESS0=10.1.3.0

@@ -3088,6 +3088,76 @@ USERCTL=no

             renderer.render_network_state(ns, target=render_dir)

         self.assertEqual([], os.listdir(render_dir))

+    def test_invalid_network_mask_ipv6(self):

+        net_json = {

+            "services": [{"type": "dns", "address": "172.19.0.12"}],

+            "networks": [

+                {

+                    "network_id": "public-ipv6",

+                    "type": "ipv6",

+                    "netmask": "",

+                    "link": "tap1a81968a-79",

+                    "routes": [

+                        {

+                            "gateway": "2001:DB8::1",

+                            "netmask": "ff:ff:ff:ff::",

+                            "network": "2001:DB8:1::1",

+                        },

+                    ],

+                    "ip_address": "2001:DB8::10",

+                    "id": "network1",

+                }

+            ],

+            "links": [

+                {

+                    "ethernet_mac_address": "fa:16:3e:ed:9a:59",

+                    "mtu": None,

+                    "type": "bridge",

+                    "id": "tap1a81968a-79",

+                    "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f",

+                },

+            ],

+        }

+        macs = {"fa:16:3e:ed:9a:59": "eth0"}

+        network_cfg = openstack.convert_net_json(net_json, known_macs=macs)

+        with self.assertRaises(ValueError):

+            network_state.parse_net_config_data(network_cfg, skip_broken=False)

+

+    def test_invalid_network_mask_ipv4(self):

+        net_json = {

+            "services": [{"type": "dns", "address": "172.19.0.12"}],

+            "networks": [

+                {

+                    "network_id": "public-ipv4",

+                    "type": "ipv4",

+                    "netmask": "",

+                    "link": "tap1a81968a-79",

+                    "routes": [

+                        {

+                            "gateway": "172.20.0.1",

+                            "netmask": "255.234.255.0",

+                            "network": "172.19.0.0",

+                        },

+                    ],

+                    "ip_address": "172.20.0.10",

+                    "id": "network1",

+                }

+            ],

+            "links": [

+                {

+                    "ethernet_mac_address": "fa:16:3e:ed:9a:59",

+                    "mtu": None,

+                    "type": "bridge",

+                    "id": "tap1a81968a-79",

+                    "vif_id": "1a81968a-797a-400f-8a80-567f997eb93f",

+                },

+            ],

+        }

+        macs = {"fa:16:3e:ed:9a:59": "eth0"}

+        network_cfg = openstack.convert_net_json(net_json, known_macs=macs)

+        with self.assertRaises(ValueError):

+            network_state.parse_net_config_data(network_cfg, skip_broken=False)

+

     def test_openstack_rendering_samples(self):

         for os_sample in OS_SAMPLES:

             render_dir = self.tmp_dir()

-- 

2.27.0