kexec-installer: generate networkd units instead of iproute

This should be more robust when interfaces appear late.
It also makes ipv6 support easier to implement.

nix/kexec-installer.nix

@@ -36,8 +36,11 @@ in {
 
     # save the networking config for later use
     if type -p ip &>/dev/null; then
-      ip --json addr > addrs.json
+      ip -4 --json addr > addrs-v4.json
-      ip --json route > routes.json
+      ip -4 --json addr > addrs-v6.json
+
+      ip -6 --json route > routes-v4.json
+      ip -6 --json route > routes-v6.json
     else
       echo "Skip saving static network addresses because no iproute2 binary is available." 2>&1
       echo "The image can depends only on DHCP to get network after reboot!" 2>&1
@@ -87,16 +90,20 @@ in {
   # Not really needed. Saves a few bytes and the only service we are running is sshd, which we want to be reachable.
   networking.firewall.enable = false;
 
+  systemd.network.enable = true;
+  networking.dhcpcd.enable = false;
+
   # for detection if we are on kexec
   environment.etc.is_kexec.text = "true";
 
   systemd.services.restoreNetwork = {
-    path = [
+    before = [ "network-pre.target" ];
-      pkgs.iproute2
+    wants = [ "network-pre.target" ];
-    ];
     wantedBy = [ "multi-user.target" ];
-    after = [ "network-online.target" ];
+
-    serviceConfig.ExecStart = "${restoreNetwork} /root/network/addrs.json /root/network/routes.json";
+    serviceConfig.ExecStart = [
+      "${restoreNetwork} /root/network/addrs-v4.json /root/network/addrs-v6.json /root/network/routes-v4.json /root/network/routes-v6.json"
+    ];
 
     unitConfig.ConditionPathExists = [
       "/root/network/addrs.json"

nix/restore_routes.py

@@ -1,94 +1,116 @@
 import json
 import sys
-import subprocess
+from pathlib import Path
+from typing import Any
 
 
-def filter_interfaces(network):
+def filter_interfaces(network: list[dict[str, Any]]) -> list[dict[str, Any]]:
     output = []
     for net in network:
-        if net["ifname"] == "lo":
+        if net.get("link_type") == "loopback":
+            continue
+        if not net.get("address"):
+            # We need a mac address to match devices reliable
             continue
         addr_info = []
+        has_dynamic_address = False
         for addr in net["addr_info"]:
+            # no link-local ipv4/ipv6
+            if addr.get("scope") == "link":
+                continue
+            # do not explicitly configure addresses from dhcp or router advertisment
             if addr.get("dynamic", False):
-                pass
+                has_dynamic_address = True
-            elif addr["local"].startswith("fe80"):
+                continue
-                pass
             else:
                 addr_info.append(addr)
-        if addr_info != []:
+        if addr_info != [] or has_dynamic_address:
             net["addr_info"] = addr_info
             output.append(net)
 
     return output
 
 
+def filter_routes(routes: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    filtered = []
+    for route in routes:
+        # Filter out routes set by addresses with subnets, dhcp and router advertisment
+        if route.get("protocol") in ["dhcp", "kernel", "ra"]:
+            continue
+        filtered.append(route)
+
+    return filtered
+
+
+def generate_networkd_units(
+    interfaces: list[dict[str, Any]], routes: list[dict[str, Any]], directory: Path
+) -> None:
+    directory.mkdir(exist_ok=True)
+    for interface in interfaces:
+        name = f"{interface['ifname']}.network"
+        addresses = [
+            f"Address = {addr['local']}/{addr['prefixlen']}"
+            for addr in interface["addr_info"]
+        ]
+
+        route_sections = []
+        for route in routes:
+            if route["dev"] != interface["ifname"]:
+                continue
+
+            route_section = f"[Route]"
+            if route["dst"] != "default":
+                # can be skipped for default routes
+                route_section += f"Destination = {route['dst']}\n"
+            gateway = route.get("gateway")
+            if gateway:
+                route_section += f"Gateway = {gateway}\n"
+
+            # we may ignore on-link default routes here, but I don't see how
+            # they would be useful for internet connectivity anyway
+            route_sections.append(route_section)
+
+        # FIXME in some networks we might not want to trust dhcp or router advertisments
+        unit = f"""
+[Match]
+MACAddress = {interface["address"]}
+
+[Network]
+DHCP = yes
+IPv6AcceptRA = yes
+"""
+        unit += "\n".join(addresses)
+        unit += "\n" + "\n".join(route_sections)
+        (directory / name).write_text(unit)
+
+
 def main() -> None:
-    if len(sys.argv) < 3:
+    if len(sys.argv) < 5:
-        print(f"USAGE: {sys.argv[0]} addresses routes", file=sys.stderr)
+        print(
+            f"USAGE: {sys.argv[0]} addresses-v4 addresses-v6 routes-v4 routes-v6 [networkd-directory]",
+            file=sys.stderr,
+        )
         sys.exit(1)
 
     with open(sys.argv[1]) as f:
-        addresses = json.load(f)
+        v4_addresses = json.load(f)
     with open(sys.argv[2]) as f:
-        routes = json.load(f)
+        v6_addresses = json.load(f)
-    relevant_interfaces = filter_interfaces(addresses)
+    with open(sys.argv[3]) as f:
-    current_interfaces = json.loads(
+        v4_routes = json.load(f)
-        subprocess.run(
+    with open(sys.argv[4]) as f:
-            ["ip", "--json", "addr"],
+        v6_routes = json.load(f)
-            capture_output=True,
-            check=True
-        ).stdout
-    )
 
-    for interface in relevant_interfaces:
+    if len(sys.argv) >= 5:
-        for current_interface in current_interfaces:
+        networkd_directory = Path(sys.argv[5])
-            if "address" in interface and "address" in current_interface:
-                if interface["address"] == current_interface["address"]:
-                    for addr in interface["addr_info"]:
-                        subprocess.run(
-                            [
-                                "ip",
-                                "addr",
-                                "add",
-                                "dev",
-                                current_interface["ifname"],
-                                f'{addr["local"]}/{addr["prefixlen"]}',
-                            ],
-                            check=True
-                        )
-                    for route in routes:
-                        if route["dev"] == interface["ifname"]:
-                            if route.get("gateway", False):
-                                subprocess.run(
-                                    [
-                                        "ip",
-                                        "route",
-                                        "add",
-                                        route["dst"],
-                                        "via",
-                                        route["gateway"],
-                                        "dev",
-                                        current_interface["ifname"],
-                                        "preference",
-                                        "1",
-                                    ],
-                                    check=True
-                                )
     else:
-                                subprocess.run(
+        networkd_directory = Path("/etc/systemd/network")
-                                    [
+
-                                        "ip",
+    addresses = v4_addresses + v6_addresses
-                                        "route",
+    relevant_interfaces = filter_interfaces(addresses)
-                                        "add",
+    relevant_routes = filter_routes(v4_routes) + filter_routes(v6_routes)
-                                        route["dst"],
+
-                                        "dev",
+    generate_networkd_units(relevant_interfaces, relevant_routes, networkd_directory)
-                                        current_interface["ifname"],
-                                        "preference",
-                                        "1",
-                                    ],
-                                    check=True
-                                )
 
 
 if __name__ == "__main__":