Make network API replace IP/WiFi settings

supervisor/api/network.py

@@ -2,7 +2,6 @@
 
 import asyncio
 from collections.abc import Awaitable
-from dataclasses import replace
 from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
 from typing import Any
 
@@ -208,24 +207,26 @@ async def interface_update(self, request: web.Request) -> None:
         # Apply config
         for key, config in body.items():
             if key == ATTR_IPV4:
-                interface.ipv4setting = replace(
-                    interface.ipv4setting
-                    or IpSetting(InterfaceMethod.STATIC, [], None, []),
-                    **config,
+                interface.ipv4setting = IpSetting(
+                    config.get(ATTR_METHOD, InterfaceMethod.STATIC),
+                    config.get(ATTR_ADDRESS, []),
+                    config.get(ATTR_GATEWAY),
+                    config.get(ATTR_NAMESERVERS, []),
                 )
             elif key == ATTR_IPV6:
-                interface.ipv6setting = replace(
-                    interface.ipv6setting
-                    or IpSetting(InterfaceMethod.STATIC, [], None, []),
-                    **config,
+                interface.ipv6setting = IpSetting(
+                    config.get(ATTR_METHOD, InterfaceMethod.STATIC),
+                    config.get(ATTR_ADDRESS, []),
+                    config.get(ATTR_GATEWAY),
+                    config.get(ATTR_NAMESERVERS, []),
                 )
             elif key == ATTR_WIFI:
-                interface.wifi = replace(
-                    interface.wifi
-                    or WifiConfig(
-                        WifiMode.INFRASTRUCTURE, "", AuthMethod.OPEN, None, None
-                    ),
-                    **config,
+                interface.wifi = WifiConfig(
+                    config.get(ATTR_MODE, WifiMode.INFRASTRUCTURE),
+                    config.get(ATTR_SSID, ""),
+                    config.get(ATTR_AUTH, AuthMethod.OPEN),
+                    config.get(ATTR_PSK, None),
+                    None,
                 )
             elif key == ATTR_ENABLED:
                 interface.enabled = config

supervisor/dbus/network/configuration.py

@@ -74,7 +74,7 @@ class IpProperties:
     method: str | None
     address_data: list[IpAddress] | None
     gateway: str | None
-    dns: list[str] | None
+    dns: list[bytes | int] | None
 
 
 @dataclass(slots=True)

supervisor/dbus/network/setting/generate.py

@@ -50,6 +50,89 @@
     from ....host.configuration import Interface
 
 
+def _get_ipv4_connection_settings(ipv4setting) -> dict:
+    ipv4 = {}
+    if not ipv4setting or ipv4setting.method == InterfaceMethod.AUTO:
+        ipv4[CONF_ATTR_IPV4_METHOD] = Variant("s", "auto")
+    elif ipv4setting.method == InterfaceMethod.DISABLED:
+        ipv4[CONF_ATTR_IPV4_METHOD] = Variant("s", "disabled")
+    elif ipv4setting.method == InterfaceMethod.STATIC:
+        ipv4[CONF_ATTR_IPV4_METHOD] = Variant("s", "manual")
+
+        address_data = []
+        for address in ipv4setting.address:
+            address_data.append(
+                {
+                    "address": Variant("s", str(address.ip)),
+                    "prefix": Variant("u", int(address.with_prefixlen.split("/")[-1])),
+                }
+            )
+
+        ipv4[CONF_ATTR_IPV4_ADDRESS_DATA] = Variant("aa{sv}", address_data)
+        if ipv4setting.gateway:
+            ipv4[CONF_ATTR_IPV4_GATEWAY] = Variant("s", str(ipv4setting.gateway))
+    else:
+        raise RuntimeError("Invalid IPv4 InterfaceMethod")
+
+    if (
+        ipv4setting
+        and ipv4setting.nameservers
+        and ipv4setting.method
+        in (
+            InterfaceMethod.AUTO,
+            InterfaceMethod.STATIC,
+        )
+    ):
+        nameservers = ipv4setting.nameservers if ipv4setting else []
+        ipv4[CONF_ATTR_IPV4_DNS] = Variant(
+            "au",
+            [socket.htonl(int(ip_address)) for ip_address in nameservers],
+        )
+
+    return ipv4
+
+
+def _get_ipv6_connection_settings(ipv6setting) -> dict:
+    ipv6 = {}
+    if not ipv6setting or ipv6setting.method == InterfaceMethod.AUTO:
+        ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "auto")
+    elif ipv6setting.method == InterfaceMethod.DISABLED:
+        ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "link-local")
+    elif ipv6setting.method == InterfaceMethod.STATIC:
+        ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "manual")
+
+        address_data = []
+        for address in ipv6setting.address:
+            address_data.append(
+                {
+                    "address": Variant("s", str(address.ip)),
+                    "prefix": Variant("u", int(address.with_prefixlen.split("/")[-1])),
+                }
+            )
+
+        ipv6[CONF_ATTR_IPV6_ADDRESS_DATA] = Variant("aa{sv}", address_data)
+        if ipv6setting.gateway:
+            ipv6[CONF_ATTR_IPV6_GATEWAY] = Variant("s", str(ipv6setting.gateway))
+    else:
+        raise RuntimeError("Invalid IPv6 InterfaceMethod")
+
+    if (
+        ipv6setting
+        and ipv6setting.nameservers
+        and ipv6setting.method
+        in (
+            InterfaceMethod.AUTO,
+            InterfaceMethod.STATIC,
+        )
+    ):
+        nameservers = ipv6setting.nameservers if ipv6setting else []
+        ipv6[CONF_ATTR_IPV6_DNS] = Variant(
+            "aay",
+            [ip_address.packed for ip_address in nameservers],
+        )
+    return ipv6
+
+
 def get_connection_from_interface(
     interface: Interface,
     network_manager: NetworkManager,
@@ -94,66 +177,9 @@ def get_connection_from_interface(
         else:
             conn[CONF_ATTR_CONNECTION]["interface-name"] = Variant("s", interface.name)
 
-    ipv4 = {}
-    if (
-        not interface.ipv4setting
-        or interface.ipv4setting.method == InterfaceMethod.AUTO
-    ):
-        ipv4[CONF_ATTR_IPV4_METHOD] = Variant("s", "auto")
-    elif interface.ipv4setting.method == InterfaceMethod.DISABLED:
-        ipv4[CONF_ATTR_IPV4_METHOD] = Variant("s", "disabled")
-    else:
-        ipv4[CONF_ATTR_IPV4_METHOD] = Variant("s", "manual")
-        ipv4[CONF_ATTR_IPV4_DNS] = Variant(
-            "au",
-            [
-                socket.htonl(int(ip_address))
-                for ip_address in interface.ipv4setting.nameservers
-            ],
-        )
-
-        address_data = []
-        for address in interface.ipv4setting.address:
-            address_data.append(
-                {
-                    "address": Variant("s", str(address.ip)),
-                    "prefix": Variant("u", int(address.with_prefixlen.split("/")[-1])),
-                }
-            )
-
-        ipv4[CONF_ATTR_IPV4_ADDRESS_DATA] = Variant("aa{sv}", address_data)
-        ipv4[CONF_ATTR_IPV4_GATEWAY] = Variant("s", str(interface.ipv4setting.gateway))
-
-    conn[CONF_ATTR_IPV4] = ipv4
-
-    ipv6 = {}
-    if (
-        not interface.ipv6setting
-        or interface.ipv6setting.method == InterfaceMethod.AUTO
-    ):
-        ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "auto")
-    elif interface.ipv6setting.method == InterfaceMethod.DISABLED:
-        ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "link-local")
-    else:
-        ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "manual")
-        ipv6[CONF_ATTR_IPV6_DNS] = Variant(
-            "aay",
-            [ip_address.packed for ip_address in interface.ipv6setting.nameservers],
-        )
-
-        address_data = []
-        for address in interface.ipv6setting.address:
-            address_data.append(
-                {
-                    "address": Variant("s", str(address.ip)),
-                    "prefix": Variant("u", int(address.with_prefixlen.split("/")[-1])),
-                }
-            )
-
-        ipv6[CONF_ATTR_IPV6_ADDRESS_DATA] = Variant("aa{sv}", address_data)
-        ipv6[CONF_ATTR_IPV6_GATEWAY] = Variant("s", str(interface.ipv6setting.gateway))
+    conn[CONF_ATTR_IPV4] = _get_ipv4_connection_settings(interface.ipv4setting)
 
-    conn[CONF_ATTR_IPV6] = ipv6
+    conn[CONF_ATTR_IPV6] = _get_ipv6_connection_settings(interface.ipv6setting)
 
     if interface.type == InterfaceType.ETHERNET:
         conn[CONF_ATTR_802_ETHERNET] = {

supervisor/host/configuration.py

@@ -105,7 +105,9 @@ def from_dbus_interface(inet: NetworkInterface) -> "Interface":
                 ]
                 if inet.settings.ipv4.address_data
                 else [],
-                gateway=inet.settings.ipv4.gateway,
+                gateway=IPv4Address(inet.settings.ipv4.gateway)
+                if inet.settings.ipv4.gateway
+                else None,
                 nameservers=[
                     IPv4Address(socket.ntohl(ip)) for ip in inet.settings.ipv4.dns
                 ]
@@ -124,7 +126,9 @@ def from_dbus_interface(inet: NetworkInterface) -> "Interface":
                 ]
                 if inet.settings.ipv6.address_data
                 else [],
-                gateway=inet.settings.ipv6.gateway,
+                gateway=IPv6Address(inet.settings.ipv6.gateway)
+                if inet.settings.ipv6.gateway
+                else None,
                 nameservers=[IPv6Address(bytes(ip)) for ip in inet.settings.ipv6.dns]
                 if inet.settings.ipv6.dns
                 else [],

tests/api/test_network.py

@@ -184,7 +184,7 @@ async def test_api_network_interface_update_ethernet(
     assert settings["ipv4"]["dns"] == Variant("au", [16843009])
     assert settings["ipv4"]["gateway"] == Variant("s", "192.168.2.1")
 
-    # Partial static configuration, updates only provided settings (e.g. by CLI)
+    # Partial static configuration, clears other settings (e.g. by CLI)
     resp = await api_client.post(
         f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
         json={
@@ -205,15 +205,16 @@ async def test_api_network_interface_update_ethernet(
         "aa{sv}",
         [{"address": Variant("s", "192.168.2.149"), "prefix": Variant("u", 24)}],
     )
-    assert settings["ipv4"]["dns"] == Variant("au", [16843009])
-    assert settings["ipv4"]["gateway"] == Variant("s", "192.168.2.1")
+    assert "dns" not in settings["ipv4"]
+    assert "gateway" not in settings["ipv4"]
 
     # Auto configuration, clears all settings (represents frontend auto config)
     resp = await api_client.post(
         f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
         json={
             "ipv4": {
                 "method": "auto",
+                "nameservers": ["8.8.8.8"],
             }
         },
     )
@@ -228,7 +229,7 @@ async def test_api_network_interface_update_ethernet(
     assert "address-data" not in settings["ipv4"]
     assert "addresses" not in settings["ipv4"]
     assert "dns-data" not in settings["ipv4"]
-    assert "dns" not in settings["ipv4"]
+    assert settings["ipv4"]["dns"] == Variant("au", [134744072])
     assert "gateway" not in settings["ipv4"]
 
 

tests/dbus/network/setting/test_init.py

@@ -77,7 +77,8 @@ async def test_ethernet_update(
     assert "ipv4" in settings
     assert settings["ipv4"]["method"] == Variant("s", "auto")
     assert "gateway" not in settings["ipv4"]
-    assert "dns" not in settings["ipv4"]
+    # Only DNS settings need to be preserved with auto
+    assert settings["ipv4"]["dns"] == Variant("au", [16951488])
     assert "dns-data" not in settings["ipv4"]
     assert "address-data" not in settings["ipv4"]
     assert "addresses" not in settings["ipv4"]
@@ -94,7 +95,10 @@ async def test_ethernet_update(
     assert "ipv6" in settings
     assert settings["ipv6"]["method"] == Variant("s", "auto")
     assert "gateway" not in settings["ipv6"]
-    assert "dns" not in settings["ipv6"]
+    # Only DNS settings need to be preserved with auto
+    assert settings["ipv6"]["dns"] == Variant(
+        "aay", [bytearray(b" \x01H`H`\x00\x00\x00\x00\x00\x00\x00\x00\x88\x88")]
+    )
     assert "dns-data" not in settings["ipv6"]
     assert "address-data" not in settings["ipv6"]
     assert "addresses" not in settings["ipv6"]

tests/host/test_network.py

@@ -72,13 +72,15 @@ async def test_load(
     assert name_dict["eth0"].ipv4setting.method == InterfaceMethod.AUTO
     assert name_dict["eth0"].ipv4setting.address == []
     assert name_dict["eth0"].ipv4setting.gateway is None
-    assert name_dict["eth0"].ipv4setting.nameservers == []
+    assert name_dict["eth0"].ipv4setting.nameservers == [IPv4Address("192.168.2.1")]
     assert name_dict["eth0"].ipv6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69")
     assert name_dict["eth0"].ipv6.ready is True
     assert name_dict["eth0"].ipv6setting.method == InterfaceMethod.AUTO
     assert name_dict["eth0"].ipv6setting.address == []
     assert name_dict["eth0"].ipv6setting.gateway is None
-    assert name_dict["eth0"].ipv6setting.nameservers == []
+    assert name_dict["eth0"].ipv6setting.nameservers == [
+        IPv6Address("2001:4860:4860::8888")
+    ]
     assert "wlan0" in name_dict
     assert name_dict["wlan0"].enabled is False
 
@@ -87,13 +89,17 @@ async def test_load(
         "aa{sv}", []
     )
     assert "gateway" not in connection_settings_service.settings["ipv4"]
-    assert "dns" not in connection_settings_service.settings["ipv4"]
+    assert connection_settings_service.settings["ipv4"]["dns"] == Variant(
+        "au", [16951488]
+    )
     assert connection_settings_service.settings["ipv6"]["method"].value == "auto"
     assert connection_settings_service.settings["ipv6"]["address-data"] == Variant(
         "aa{sv}", []
     )
     assert "gateway" not in connection_settings_service.settings["ipv6"]
-    assert "dns" not in connection_settings_service.settings["ipv6"]
+    assert connection_settings_service.settings["ipv6"]["dns"] == Variant(
+        "aay", [bytearray(b" \x01H`H`\x00\x00\x00\x00\x00\x00\x00\x00\x88\x88")]
+    )
 
     assert network_manager_service.ActivateConnection.calls == [
         (