fix(net/networkd): add GatewayOnLink flag when necessary

cloudinit/net/__init__.py

@@ -1283,6 +1283,48 @@ def is_ipv6_network(address: str) -> bool:
     )
 
 
+def is_ip_in_subnet(address: str, subnet: str) -> bool:
+    """Returns a bool indicating if ``s`` is in subnet.
+
+    :param address:
+        The string of IP address.
+
+    :param subnet:
+        The string of subnet.
+
+    :return:
+        A bool indicating if the string is in subnet.
+    """
+    ip_address = ipaddress.ip_address(address)
+    subnet_network = ipaddress.ip_network(subnet, strict=False)
+    return ip_address in subnet_network
+
+
+def should_add_gateway_onlink_flag(gateway: str, subnet: str) -> bool:
+    """Returns a bool indicating if should add gateway onlink flag.
+
+    :param gateway:
+        The string of gateway address.
+
+    :param subnet:
+        The string of subnet.
+
+    :return:
+        A bool indicating if the string is in subnet.
+    """
+    try:
+        return not is_ip_in_subnet(gateway, subnet)
+    except ValueError as e:
+        LOG.warning(
+            "Failed to check whether gateway %s"
+            " is contained within subnet %s: %s",
+            gateway,
+            subnet,
+            e,
+        )
+        return False
+
+
 def subnet_is_ipv6(subnet) -> bool:
     """Common helper for checking network_state subnets for ipv6."""
     # 'static6', 'dhcp6', 'ipv6_dhcpv6-stateful', 'ipv6_dhcpv6-stateless' or

cloudinit/net/netplan.py

@@ -2,7 +2,6 @@
 
 import copy
 import io
-import ipaddress
 import logging
 import os
 import textwrap
@@ -14,6 +13,7 @@
     SYS_CLASS_NET,
     get_devicelist,
     renderer,
+    should_add_gateway_onlink_flag,
     subnet_is_ipv6,
 )
 from cloudinit.net.network_state import NET_CONFIG_TO_V2, NetworkState
@@ -123,28 +123,17 @@ def _listify(obj, token=" "):
                     "via": subnet.get("gateway"),
                     "to": "default",
                 }
-                try:
-                    subnet_gateway = ipaddress.ip_address(subnet["gateway"])
-                    subnet_network = ipaddress.ip_network(addr, strict=False)
-                    # If the gateway is not contained within the subnet's
-                    # network, mark it as on-link so that it can still be
-                    # reached.
-                    if subnet_gateway not in subnet_network:
-                        LOG.debug(
-                            "Gateway %s is not contained within subnet %s,"
-                            " adding on-link flag",
-                            subnet["gateway"],
-                            addr,
-                        )
-                        new_route["on-link"] = True
-                except ValueError as e:
-                    LOG.warning(
-                        "Failed to check whether gateway %s"
-                        " is contained within subnet %s: %s",
+                # If the gateway is not contained within the subnet's
+                # network, mark it as on-link so that it can still be
+                # reached.
+                if should_add_gateway_onlink_flag(subnet["gateway"], addr):
+                    LOG.debug(
+                        "Gateway %s is not contained within subnet %s,"
+                        " adding on-link flag",
                         subnet["gateway"],
                         addr,
-                        e,
                     )
+                    new_route["on-link"] = True
                 routes.append(new_route)
             if "dns_nameservers" in subnet:
                 nameservers += _listify(subnet.get("dns_nameservers", []))

cloudinit/net/networkd.py

@@ -9,7 +9,7 @@
 from typing import Optional
 
 from cloudinit import subp, util
-from cloudinit.net import renderer
+from cloudinit.net import renderer, should_add_gateway_onlink_flag
 from cloudinit.net.network_state import NetworkState
 
 LOG = logging.getLogger(__name__)
@@ -68,7 +68,7 @@ def get_final_conf(self):
                     contents += "[" + k + "]\n"
                     for e in sorted(v[n]):
                         contents += e + "\n"
-                        contents += "\n"
+                    contents += "\n"
             else:
                 contents += "[" + k + "]\n"
                 for e in sorted(v):
@@ -169,6 +169,9 @@ def parse_subnets(self, iface, cfg: CfgParser):
                     self.parse_routes(f"r{rid}", i, cfg)
                     rid = rid + 1
             if "address" in e:
+                addr = e["address"]
+                if "prefix" in e:
+                    addr += "/" + str(e["prefix"])
                 subnet_cfg_map = {
                     "address": "Address",
                     "gateway": "Gateway",
@@ -177,15 +180,23 @@ def parse_subnets(self, iface, cfg: CfgParser):
                 }
                 for k, v in e.items():
                     if k == "address":
-                        if "prefix" in e:
-                            v += "/" + str(e["prefix"])
-                        cfg.update_section("Address", subnet_cfg_map[k], v)
+                        cfg.update_section("Address", subnet_cfg_map[k], addr)
                     elif k == "gateway":
                         # Use "a" as a dict key prefix for this route to
                         # isolate it from other sources of routes
                         cfg.update_route_section(
                             "Route", f"a{rid}", subnet_cfg_map[k], v
                         )
+                        if should_add_gateway_onlink_flag(v, addr):
+                            LOG.debug(
+                                "Gateway %s is not contained within subnet %s,"
+                                " adding GatewayOnLink flag",
+                                v,
+                                addr,
+                            )
+                            cfg.update_route_section(
+                                "Route", f"a{rid}", "GatewayOnLink", "yes"
+                            )
                         rid = rid + 1
                     elif k == "dns_nameservers" or k == "dns_search":
                         cfg.update_section(sec, subnet_cfg_map[k], " ".join(v))

tests/unittests/net/test_init.py

@@ -1892,3 +1892,29 @@ class TestIsIpNetwork:
     )
     def test_is_ip_network(self, func, arg, expected_return):
         assert func(arg) == expected_return
+
+
+class TestIsIpInSubnet:
+    """Tests for net.is_ip_in_subnet()."""
+
+    @pytest.mark.parametrize(
+        "func,ip,subnet,expected_return",
+        (
+            (net.is_ip_in_subnet, "192.168.1.1", "2001:67c::1/64", False),
+            (net.is_ip_in_subnet, "2001:67c::1", "192.168.1.1/24", False),
+            (net.is_ip_in_subnet, "192.168.1.1", "192.168.1.1/24", True),
+            (net.is_ip_in_subnet, "192.168.1.1", "192.168.1.1/32", True),
+            (net.is_ip_in_subnet, "192.168.1.2", "192.168.1.1/24", True),
+            (net.is_ip_in_subnet, "192.168.1.2", "192.168.1.1/32", False),
+            (net.is_ip_in_subnet, "192.168.2.2", "192.168.1.1/24", False),
+            (net.is_ip_in_subnet, "192.168.2.2", "192.168.1.1/32", False),
+            (net.is_ip_in_subnet, "2001:67c1::1", "2001:67c1::1/64", True),
+            (net.is_ip_in_subnet, "2001:67c1::1", "2001:67c1::1/128", True),
+            (net.is_ip_in_subnet, "2001:67c1::2", "2001:67c1::1/64", True),
+            (net.is_ip_in_subnet, "2001:67c1::2", "2001:67c1::1/128", False),
+            (net.is_ip_in_subnet, "2002:67c1::1", "2001:67c1::1/8", True),
+            (net.is_ip_in_subnet, "2002:67c1::1", "2001:67c1::1/16", False),
+        ),
+    )
+    def test_is_ip_in_subnet(self, func, ip, subnet, expected_return):
+        assert func(ip, subnet) == expected_return