From 1ba621be60e65105533a1cceeecccfebad0197d7 Mon Sep 17 00:00:00 2001 From: Stefan Agner Date: Thu, 22 Aug 2024 08:08:55 +0200 Subject: [PATCH] Use separate data structure for IP configuration (#5262) * Use separate data structure for IP configuration So far we use the same IpConfig data structure to represent the users IP setting and the currently applied IP configuration. This commit separates the two in IpConfig (for the currently applied IP configuration) and IpSetting (representing the user provided IP setting). * Use custom string constants for connection settings Use separate string constants for all connection settings. This makes it easier to search where a particular NetworkManager connection setting is used. * Use Python typing for IpAddress in IpProperties * Address pytest issue --- supervisor/api/network.py | 37 +++--- supervisor/dbus/network/configuration.py | 11 ++ supervisor/dbus/network/setting/__init__.py | 107 ++++++++++----- supervisor/dbus/network/setting/generate.py | 136 +++++++++++++------- supervisor/host/configuration.py | 91 +++++++++---- supervisor/host/network.py | 4 +- tests/dbus/network/setting/test_generate.py | 6 +- tests/dbus/network/setting/test_init.py | 4 +- tests/host/test_network.py | 38 +++++- 9 files changed, 304 insertions(+), 130 deletions(-) diff --git a/supervisor/api/network.py b/supervisor/api/network.py index b3c903f4a..204cf58f0 100644 --- a/supervisor/api/network.py +++ b/supervisor/api/network.py @@ -49,6 +49,7 @@ from ..host.configuration import ( Interface, InterfaceMethod, IpConfig, + IpSetting, VlanConfig, WifiConfig, ) @@ -85,10 +86,10 @@ SCHEMA_UPDATE = vol.Schema( ) -def ipconfig_struct(config: IpConfig) -> dict[str, Any]: +def ipconfig_struct(config: IpConfig, setting: IpSetting) -> dict[str, Any]: """Return a dict with information about ip configuration.""" return { - ATTR_METHOD: config.method, + ATTR_METHOD: setting.method, ATTR_ADDRESS: [address.with_prefixlen for address in config.address], ATTR_NAMESERVERS: [str(address) for address in config.nameservers], ATTR_GATEWAY: str(config.gateway) if config.gateway else None, @@ -123,8 +124,8 @@ def interface_struct(interface: Interface) -> dict[str, Any]: ATTR_CONNECTED: interface.connected, ATTR_PRIMARY: interface.primary, ATTR_MAC: interface.mac, - ATTR_IPV4: ipconfig_struct(interface.ipv4) if interface.ipv4 else None, - ATTR_IPV6: ipconfig_struct(interface.ipv6) if interface.ipv6 else None, + ATTR_IPV4: ipconfig_struct(interface.ipv4, interface.ipv4setting), + ATTR_IPV6: ipconfig_struct(interface.ipv6, interface.ipv6setting), ATTR_WIFI: wifi_struct(interface.wifi) if interface.wifi else None, ATTR_VLAN: vlan_struct(interface.vlan) if interface.vlan else None, } @@ -198,15 +199,15 @@ class APINetwork(CoreSysAttributes): # Apply config for key, config in body.items(): if key == ATTR_IPV4: - interface.ipv4 = replace( - interface.ipv4 - or IpConfig(InterfaceMethod.STATIC, [], None, [], None), + interface.ipv4setting = replace( + interface.ipv4setting + or IpSetting(InterfaceMethod.STATIC, [], None, []), **config, ) elif key == ATTR_IPV6: - interface.ipv6 = replace( - interface.ipv6 - or IpConfig(InterfaceMethod.STATIC, [], None, [], None), + interface.ipv6setting = replace( + interface.ipv6setting + or IpSetting(InterfaceMethod.STATIC, [], None, []), **config, ) elif key == ATTR_WIFI: @@ -257,24 +258,22 @@ class APINetwork(CoreSysAttributes): vlan_config = VlanConfig(vlan, interface.name) - ipv4_config = None + ipv4_setting = None if ATTR_IPV4 in body: - ipv4_config = IpConfig( + ipv4_setting = IpSetting( body[ATTR_IPV4].get(ATTR_METHOD, InterfaceMethod.AUTO), body[ATTR_IPV4].get(ATTR_ADDRESS, []), body[ATTR_IPV4].get(ATTR_GATEWAY, None), body[ATTR_IPV4].get(ATTR_NAMESERVERS, []), - None, ) - ipv6_config = None + ipv6_setting = None if ATTR_IPV6 in body: - ipv6_config = IpConfig( + ipv6_setting = IpSetting( body[ATTR_IPV6].get(ATTR_METHOD, InterfaceMethod.AUTO), body[ATTR_IPV6].get(ATTR_ADDRESS, []), body[ATTR_IPV6].get(ATTR_GATEWAY, None), body[ATTR_IPV6].get(ATTR_NAMESERVERS, []), - None, ) vlan_interface = Interface( @@ -285,8 +284,10 @@ class APINetwork(CoreSysAttributes): True, False, InterfaceType.VLAN, - ipv4_config, - ipv6_config, + None, + ipv4_setting, + None, + ipv6_setting, None, vlan_config, ) diff --git a/supervisor/dbus/network/configuration.py b/supervisor/dbus/network/configuration.py index c598a726d..36a81e34a 100644 --- a/supervisor/dbus/network/configuration.py +++ b/supervisor/dbus/network/configuration.py @@ -59,11 +59,22 @@ class VlanProperties: parent: str | None +@dataclass(slots=True) +class IpAddress: + """IP address object for Network Manager.""" + + address: str + prefix: int + + @dataclass(slots=True) class IpProperties: """IP properties object for Network Manager.""" method: str | None + address_data: list[IpAddress] | None + gateway: str | None + dns: list[str] | None @dataclass(slots=True) diff --git a/supervisor/dbus/network/setting/__init__.py b/supervisor/dbus/network/setting/__init__.py index 80758988d..5bf1dbad2 100644 --- a/supervisor/dbus/network/setting/__init__.py +++ b/supervisor/dbus/network/setting/__init__.py @@ -6,13 +6,13 @@ from typing import Any from dbus_fast import Variant from dbus_fast.aio.message_bus import MessageBus -from ....const import ATTR_METHOD, ATTR_MODE, ATTR_PSK, ATTR_SSID from ...const import DBUS_NAME_NM from ...interface import DBusInterface from ...utils import dbus_connected from ..configuration import ( ConnectionProperties, EthernetProperties, + IpAddress, IpProperties, MatchProperties, VlanProperties, @@ -21,25 +21,46 @@ from ..configuration import ( ) CONF_ATTR_CONNECTION = "connection" +CONF_ATTR_MATCH = "match" CONF_ATTR_802_ETHERNET = "802-3-ethernet" CONF_ATTR_802_WIRELESS = "802-11-wireless" CONF_ATTR_802_WIRELESS_SECURITY = "802-11-wireless-security" CONF_ATTR_VLAN = "vlan" CONF_ATTR_IPV4 = "ipv4" CONF_ATTR_IPV6 = "ipv6" -CONF_ATTR_MATCH = "match" -CONF_ATTR_PATH = "path" -ATTR_ID = "id" -ATTR_UUID = "uuid" -ATTR_TYPE = "type" -ATTR_PARENT = "parent" -ATTR_ASSIGNED_MAC = "assigned-mac-address" -ATTR_POWERSAVE = "powersave" -ATTR_AUTH_ALG = "auth-alg" -ATTR_KEY_MGMT = "key-mgmt" -ATTR_INTERFACE_NAME = "interface-name" -ATTR_PATH = "path" +CONF_ATTR_CONNECTION_ID = "id" +CONF_ATTR_CONNECTION_UUID = "uuid" +CONF_ATTR_CONNECTION_TYPE = "type" +CONF_ATTR_CONNECTION_LLMNR = "llmnr" +CONF_ATTR_CONNECTION_MDNS = "mdns" +CONF_ATTR_CONNECTION_AUTOCONNECT = "autoconnect" +CONF_ATTR_CONNECTION_INTERFACE_NAME = "interface-name" + +CONF_ATTR_MATCH_PATH = "path" + +CONF_ATTR_VLAN_ID = "id" +CONF_ATTR_VLAN_PARENT = "parent" + +CONF_ATTR_802_ETHERNET_ASSIGNED_MAC = "assigned-mac-address" + +CONF_ATTR_802_WIRELESS_MODE = "mode" +CONF_ATTR_802_WIRELESS_ASSIGNED_MAC = "assigned-mac-address" +CONF_ATTR_802_WIRELESS_SSID = "ssid" +CONF_ATTR_802_WIRELESS_POWERSAVE = "powersave" +CONF_ATTR_802_WIRELESS_SECURITY_AUTH_ALG = "auth-alg" +CONF_ATTR_802_WIRELESS_SECURITY_KEY_MGMT = "key-mgmt" +CONF_ATTR_802_WIRELESS_SECURITY_PSK = "psk" + +CONF_ATTR_IPV4_METHOD = "method" +CONF_ATTR_IPV4_ADDRESS_DATA = "address-data" +CONF_ATTR_IPV4_GATEWAY = "gateway" +CONF_ATTR_IPV4_DNS = "dns" + +CONF_ATTR_IPV6_METHOD = "method" +CONF_ATTR_IPV6_ADDRESS_DATA = "address-data" +CONF_ATTR_IPV6_GATEWAY = "gateway" +CONF_ATTR_IPV6_DNS = "dns" IPV4_6_IGNORE_FIELDS = [ "addresses", @@ -75,7 +96,7 @@ def _merge_settings_attribute( class NetworkSetting(DBusInterface): """Network connection setting object for Network Manager. - https://developer.gnome.org/NetworkManager/stable/gdbus-org.freedesktop.NetworkManager.Settings.Connection.html + https://networkmanager.dev/docs/api/1.48.0/gdbus-org.freedesktop.NetworkManager.Settings.Connection.html """ bus_name: str = DBUS_NAME_NM @@ -149,7 +170,7 @@ class NetworkSetting(DBusInterface): new_settings, settings, CONF_ATTR_CONNECTION, - ignore_current_value=[ATTR_INTERFACE_NAME], + ignore_current_value=[CONF_ATTR_CONNECTION_INTERFACE_NAME], ) _merge_settings_attribute(new_settings, settings, CONF_ATTR_802_ETHERNET) _merge_settings_attribute(new_settings, settings, CONF_ATTR_802_WIRELESS) @@ -194,47 +215,69 @@ class NetworkSetting(DBusInterface): # See: https://developer-old.gnome.org/NetworkManager/stable/ch01.html if CONF_ATTR_CONNECTION in data: self._connection = ConnectionProperties( - data[CONF_ATTR_CONNECTION].get(ATTR_ID), - data[CONF_ATTR_CONNECTION].get(ATTR_UUID), - data[CONF_ATTR_CONNECTION].get(ATTR_TYPE), - data[CONF_ATTR_CONNECTION].get(ATTR_INTERFACE_NAME), + data[CONF_ATTR_CONNECTION].get(CONF_ATTR_CONNECTION_ID), + data[CONF_ATTR_CONNECTION].get(CONF_ATTR_CONNECTION_UUID), + data[CONF_ATTR_CONNECTION].get(CONF_ATTR_CONNECTION_TYPE), + data[CONF_ATTR_CONNECTION].get(CONF_ATTR_CONNECTION_INTERFACE_NAME), ) if CONF_ATTR_802_ETHERNET in data: self._ethernet = EthernetProperties( - data[CONF_ATTR_802_ETHERNET].get(ATTR_ASSIGNED_MAC), + data[CONF_ATTR_802_ETHERNET].get(CONF_ATTR_802_ETHERNET_ASSIGNED_MAC), ) if CONF_ATTR_802_WIRELESS in data: self._wireless = WirelessProperties( - bytes(data[CONF_ATTR_802_WIRELESS].get(ATTR_SSID, [])).decode(), - data[CONF_ATTR_802_WIRELESS].get(ATTR_ASSIGNED_MAC), - data[CONF_ATTR_802_WIRELESS].get(ATTR_MODE), - data[CONF_ATTR_802_WIRELESS].get(ATTR_POWERSAVE), + bytes( + data[CONF_ATTR_802_WIRELESS].get(CONF_ATTR_802_WIRELESS_SSID, []) + ).decode(), + data[CONF_ATTR_802_WIRELESS].get(CONF_ATTR_802_WIRELESS_ASSIGNED_MAC), + data[CONF_ATTR_802_WIRELESS].get(CONF_ATTR_802_WIRELESS_MODE), + data[CONF_ATTR_802_WIRELESS].get(CONF_ATTR_802_WIRELESS_POWERSAVE), ) if CONF_ATTR_802_WIRELESS_SECURITY in data: self._wireless_security = WirelessSecurityProperties( - data[CONF_ATTR_802_WIRELESS_SECURITY].get(ATTR_AUTH_ALG), - data[CONF_ATTR_802_WIRELESS_SECURITY].get(ATTR_KEY_MGMT), - data[CONF_ATTR_802_WIRELESS_SECURITY].get(ATTR_PSK), + data[CONF_ATTR_802_WIRELESS_SECURITY].get( + CONF_ATTR_802_WIRELESS_SECURITY_AUTH_ALG + ), + data[CONF_ATTR_802_WIRELESS_SECURITY].get( + CONF_ATTR_802_WIRELESS_SECURITY_KEY_MGMT + ), + data[CONF_ATTR_802_WIRELESS_SECURITY].get( + CONF_ATTR_802_WIRELESS_SECURITY_PSK + ), ) if CONF_ATTR_VLAN in data: self._vlan = VlanProperties( - data[CONF_ATTR_VLAN].get(ATTR_ID), - data[CONF_ATTR_VLAN].get(ATTR_PARENT), + data[CONF_ATTR_VLAN].get(CONF_ATTR_VLAN_ID), + data[CONF_ATTR_VLAN].get(CONF_ATTR_VLAN_PARENT), ) if CONF_ATTR_IPV4 in data: + address_data = None + if ips := data[CONF_ATTR_IPV4].get(CONF_ATTR_IPV4_ADDRESS_DATA): + address_data = [IpAddress(ip["address"], ip["prefix"]) for ip in ips] self._ipv4 = IpProperties( - data[CONF_ATTR_IPV4].get(ATTR_METHOD), + data[CONF_ATTR_IPV4].get(CONF_ATTR_IPV4_METHOD), + address_data, + data[CONF_ATTR_IPV4].get(CONF_ATTR_IPV4_GATEWAY), + data[CONF_ATTR_IPV4].get(CONF_ATTR_IPV4_DNS), ) if CONF_ATTR_IPV6 in data: + address_data = None + if ips := data[CONF_ATTR_IPV6].get(CONF_ATTR_IPV6_ADDRESS_DATA): + address_data = [IpAddress(ip["address"], ip["prefix"]) for ip in ips] self._ipv6 = IpProperties( - data[CONF_ATTR_IPV6].get(ATTR_METHOD), + data[CONF_ATTR_IPV6].get(CONF_ATTR_IPV6_METHOD), + address_data, + data[CONF_ATTR_IPV6].get(CONF_ATTR_IPV6_GATEWAY), + data[CONF_ATTR_IPV6].get(CONF_ATTR_IPV6_DNS), ) if CONF_ATTR_MATCH in data: - self._match = MatchProperties(data[CONF_ATTR_MATCH].get(ATTR_PATH)) + self._match = MatchProperties( + data[CONF_ATTR_MATCH].get(CONF_ATTR_MATCH_PATH) + ) diff --git a/supervisor/dbus/network/setting/generate.py b/supervisor/dbus/network/setting/generate.py index c25fea29f..d2825a71a 100644 --- a/supervisor/dbus/network/setting/generate.py +++ b/supervisor/dbus/network/setting/generate.py @@ -11,16 +11,39 @@ from dbus_fast import Variant from ....host.const import InterfaceMethod, InterfaceType from .. import NetworkManager from . import ( - ATTR_ASSIGNED_MAC, CONF_ATTR_802_ETHERNET, + CONF_ATTR_802_ETHERNET_ASSIGNED_MAC, CONF_ATTR_802_WIRELESS, + CONF_ATTR_802_WIRELESS_ASSIGNED_MAC, + CONF_ATTR_802_WIRELESS_MODE, + CONF_ATTR_802_WIRELESS_POWERSAVE, CONF_ATTR_802_WIRELESS_SECURITY, + CONF_ATTR_802_WIRELESS_SECURITY_AUTH_ALG, + CONF_ATTR_802_WIRELESS_SECURITY_KEY_MGMT, + CONF_ATTR_802_WIRELESS_SECURITY_PSK, + CONF_ATTR_802_WIRELESS_SSID, CONF_ATTR_CONNECTION, + CONF_ATTR_CONNECTION_AUTOCONNECT, + CONF_ATTR_CONNECTION_ID, + CONF_ATTR_CONNECTION_LLMNR, + CONF_ATTR_CONNECTION_MDNS, + CONF_ATTR_CONNECTION_TYPE, + CONF_ATTR_CONNECTION_UUID, CONF_ATTR_IPV4, + CONF_ATTR_IPV4_ADDRESS_DATA, + CONF_ATTR_IPV4_DNS, + CONF_ATTR_IPV4_GATEWAY, + CONF_ATTR_IPV4_METHOD, CONF_ATTR_IPV6, + CONF_ATTR_IPV6_ADDRESS_DATA, + CONF_ATTR_IPV6_DNS, + CONF_ATTR_IPV6_GATEWAY, + CONF_ATTR_IPV6_METHOD, CONF_ATTR_MATCH, - CONF_ATTR_PATH, + CONF_ATTR_MATCH_PATH, CONF_ATTR_VLAN, + CONF_ATTR_VLAN_ID, + CONF_ATTR_VLAN_PARENT, ) if TYPE_CHECKING: @@ -54,77 +77,88 @@ def get_connection_from_interface( conn: dict[str, dict[str, Variant]] = { CONF_ATTR_CONNECTION: { - "id": Variant("s", name), - "type": Variant("s", iftype), - "uuid": Variant("s", uuid), - "llmnr": Variant("i", 2), - "mdns": Variant("i", 2), - "autoconnect": Variant("b", True), + CONF_ATTR_CONNECTION_ID: Variant("s", name), + CONF_ATTR_CONNECTION_UUID: Variant("s", uuid), + CONF_ATTR_CONNECTION_TYPE: Variant("s", iftype), + CONF_ATTR_CONNECTION_LLMNR: Variant("i", 2), + CONF_ATTR_CONNECTION_MDNS: Variant("i", 2), + CONF_ATTR_CONNECTION_AUTOCONNECT: Variant("b", True), }, } if interface.type != InterfaceType.VLAN: if interface.path: - conn[CONF_ATTR_MATCH] = {CONF_ATTR_PATH: Variant("as", [interface.path])} + conn[CONF_ATTR_MATCH] = { + CONF_ATTR_MATCH_PATH: Variant("as", [interface.path]) + } else: conn[CONF_ATTR_CONNECTION]["interface-name"] = Variant("s", interface.name) ipv4 = {} - if not interface.ipv4 or interface.ipv4.method == InterfaceMethod.AUTO: - ipv4["method"] = Variant("s", "auto") - elif interface.ipv4.method == InterfaceMethod.DISABLED: - ipv4["method"] = Variant("s", "disabled") + 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["method"] = Variant("s", "manual") - ipv4["dns"] = Variant( + 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.ipv4.nameservers + for ip_address in interface.ipv4setting.nameservers ], ) - adressdata = [] - for address in interface.ipv4.address: - adressdata.append( + 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["address-data"] = Variant("aa{sv}", adressdata) - ipv4["gateway"] = Variant("s", str(interface.ipv4.gateway)) + 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.ipv6 or interface.ipv6.method == InterfaceMethod.AUTO: - ipv6["method"] = Variant("s", "auto") - elif interface.ipv6.method == InterfaceMethod.DISABLED: - ipv6["method"] = Variant("s", "link-local") + 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["method"] = Variant("s", "manual") - ipv6["dns"] = Variant( - "aay", [ip_address.packed for ip_address in interface.ipv6.nameservers] + 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], ) - adressdata = [] - for address in interface.ipv6.address: - adressdata.append( + 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["address-data"] = Variant("aa{sv}", adressdata) - ipv6["gateway"] = Variant("s", str(interface.ipv6.gateway)) + 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_IPV6] = ipv6 if interface.type == InterfaceType.ETHERNET: - conn[CONF_ATTR_802_ETHERNET] = {ATTR_ASSIGNED_MAC: Variant("s", "preserve")} + conn[CONF_ATTR_802_ETHERNET] = { + CONF_ATTR_802_ETHERNET_ASSIGNED_MAC: Variant("s", "preserve") + } elif interface.type == "vlan": parent = interface.vlan.interface if parent in network_manager and ( @@ -133,15 +167,17 @@ def get_connection_from_interface( parent = parent_connection.uuid conn[CONF_ATTR_VLAN] = { - "id": Variant("u", interface.vlan.id), - "parent": Variant("s", parent), + CONF_ATTR_VLAN_ID: Variant("u", interface.vlan.id), + CONF_ATTR_VLAN_PARENT: Variant("s", parent), } elif interface.type == InterfaceType.WIRELESS: wireless = { - ATTR_ASSIGNED_MAC: Variant("s", "preserve"), - "ssid": Variant("ay", interface.wifi.ssid.encode("UTF-8")), - "mode": Variant("s", "infrastructure"), - "powersave": Variant("i", 1), + CONF_ATTR_802_WIRELESS_ASSIGNED_MAC: Variant("s", "preserve"), + CONF_ATTR_802_WIRELESS_SSID: Variant( + "ay", interface.wifi.ssid.encode("UTF-8") + ), + CONF_ATTR_802_WIRELESS_MODE: Variant("s", "infrastructure"), + CONF_ATTR_802_WIRELESS_POWERSAVE: Variant("i", 1), } conn[CONF_ATTR_802_WIRELESS] = wireless @@ -149,14 +185,24 @@ def get_connection_from_interface( wireless["security"] = Variant("s", CONF_ATTR_802_WIRELESS_SECURITY) wireless_security = {} if interface.wifi.auth == "wep": - wireless_security["auth-alg"] = Variant("s", "open") - wireless_security["key-mgmt"] = Variant("s", "none") + wireless_security[CONF_ATTR_802_WIRELESS_SECURITY_AUTH_ALG] = Variant( + "s", "open" + ) + wireless_security[CONF_ATTR_802_WIRELESS_SECURITY_KEY_MGMT] = Variant( + "s", "none" + ) elif interface.wifi.auth == "wpa-psk": - wireless_security["auth-alg"] = Variant("s", "open") - wireless_security["key-mgmt"] = Variant("s", "wpa-psk") + wireless_security[CONF_ATTR_802_WIRELESS_SECURITY_AUTH_ALG] = Variant( + "s", "open" + ) + wireless_security[CONF_ATTR_802_WIRELESS_SECURITY_KEY_MGMT] = Variant( + "s", "wpa-psk" + ) if interface.wifi.psk: - wireless_security["psk"] = Variant("s", interface.wifi.psk) + wireless_security[CONF_ATTR_802_WIRELESS_SECURITY_PSK] = Variant( + "s", interface.wifi.psk + ) conn[CONF_ATTR_802_WIRELESS_SECURITY] = wireless_security return conn diff --git a/supervisor/host/configuration.py b/supervisor/host/configuration.py index 8abf5c2fc..5c6799093 100644 --- a/supervisor/host/configuration.py +++ b/supervisor/host/configuration.py @@ -2,6 +2,7 @@ from dataclasses import dataclass from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface +import socket from ..dbus.const import ( ConnectionStateFlags, @@ -27,13 +28,22 @@ class AccessPoint: @dataclass(slots=True) class IpConfig: - """Represent a IP configuration.""" + """Represent a (current) IP configuration.""" + + address: list[IPv4Interface | IPv6Interface] + gateway: IPv4Address | IPv6Address | None + nameservers: list[IPv4Address | IPv6Address] + ready: bool | None + + +@dataclass(slots=True) +class IpSetting: + """Represent a user IP setting.""" method: InterfaceMethod address: list[IPv4Interface | IPv6Interface] gateway: IPv4Address | IPv6Address | None nameservers: list[IPv4Address | IPv6Address] - ready: bool | None @dataclass(slots=True) @@ -67,7 +77,9 @@ class Interface: primary: bool type: InterfaceType ipv4: IpConfig | None + ipv4setting: IpSetting | None ipv6: IpConfig | None + ipv6setting: IpSetting | None wifi: WifiConfig | None vlan: VlanConfig | None @@ -84,16 +96,42 @@ class Interface: @staticmethod def from_dbus_interface(inet: NetworkInterface) -> "Interface": """Coerce a dbus interface into normal Interface.""" - ipv4_method = ( - Interface._map_nm_method(inet.settings.ipv4.method) - if inet.settings and inet.settings.ipv4 - else InterfaceMethod.DISABLED - ) - ipv6_method = ( - Interface._map_nm_method(inet.settings.ipv6.method) - if inet.settings and inet.settings.ipv6 - else InterfaceMethod.DISABLED - ) + if inet.settings and inet.settings.ipv4: + ipv4_setting = IpSetting( + method=Interface._map_nm_method(inet.settings.ipv4.method), + address=[ + IPv4Interface(f"{ip.address}/{ip.prefix}") + for ip in inet.settings.ipv4.address_data + ] + if inet.settings.ipv4.address_data + else [], + gateway=inet.settings.ipv4.gateway, + nameservers=[ + IPv4Address(socket.ntohl(ip)) for ip in inet.settings.ipv4.dns + ] + if inet.settings.ipv4.dns + else [], + ) + else: + ipv4_setting = IpSetting(InterfaceMethod.DISABLED, [], None, []) + + if inet.settings and inet.settings.ipv6: + ipv6_setting = IpSetting( + method=Interface._map_nm_method(inet.settings.ipv6.method), + address=[ + IPv6Interface(f"{ip.address}/{ip.prefix}") + for ip in inet.settings.ipv6.address_data + ] + if inet.settings.ipv6.address_data + else [], + gateway=inet.settings.ipv6.gateway, + nameservers=[IPv6Address(bytes(ip)) for ip in inet.settings.ipv6.dns] + if inet.settings.ipv6.dns + else [], + ) + else: + ipv6_setting = IpSetting(InterfaceMethod.DISABLED, [], None, []) + ipv4_ready = ( bool(inet.connection) and ConnectionStateFlags.IP4_READY in inet.connection.state_flags @@ -102,6 +140,7 @@ class Interface: bool(inet.connection) and ConnectionStateFlags.IP6_READY in inet.connection.state_flags ) + return Interface( inet.name, inet.hw_address, @@ -111,27 +150,31 @@ class Interface: inet.primary, Interface._map_nm_type(inet.type), IpConfig( - ipv4_method, - inet.connection.ipv4.address if inet.connection.ipv4.address else [], - inet.connection.ipv4.gateway, - inet.connection.ipv4.nameservers + address=inet.connection.ipv4.address + if inet.connection.ipv4.address + else [], + gateway=inet.connection.ipv4.gateway, + nameservers=inet.connection.ipv4.nameservers if inet.connection.ipv4.nameservers else [], - ipv4_ready, + ready=ipv4_ready, ) if inet.connection and inet.connection.ipv4 - else IpConfig(ipv4_method, [], None, [], ipv4_ready), + else IpConfig([], None, [], ipv4_ready), + ipv4_setting, IpConfig( - ipv6_method, - inet.connection.ipv6.address if inet.connection.ipv6.address else [], - inet.connection.ipv6.gateway, - inet.connection.ipv6.nameservers + address=inet.connection.ipv6.address + if inet.connection.ipv6.address + else [], + gateway=inet.connection.ipv6.gateway, + nameservers=inet.connection.ipv6.nameservers if inet.connection.ipv6.nameservers else [], - ipv6_ready, + ready=ipv6_ready, ) if inet.connection and inet.connection.ipv6 - else IpConfig(ipv6_method, [], None, [], ipv6_ready), + else IpConfig([], None, [], ipv6_ready), + ipv6_setting, Interface._map_nm_wifi(inet), Interface._map_nm_vlan(inet), ) diff --git a/supervisor/host/network.py b/supervisor/host/network.py index f9fd00f95..57e288b12 100644 --- a/supervisor/host/network.py +++ b/supervisor/host/network.py @@ -128,8 +128,8 @@ class NetworkManager(CoreSysAttributes): for interface in interfaces if interface.enabled and ( - interface.ipv4.method != InterfaceMethod.DISABLED - or interface.ipv6.method != InterfaceMethod.DISABLED + interface.ipv4setting.method != InterfaceMethod.DISABLED + or interface.ipv6setting.method != InterfaceMethod.DISABLED ) ] ) diff --git a/tests/dbus/network/setting/test_generate.py b/tests/dbus/network/setting/test_generate.py index ac6aa5ed8..58c7b32ad 100644 --- a/tests/dbus/network/setting/test_generate.py +++ b/tests/dbus/network/setting/test_generate.py @@ -5,7 +5,7 @@ from unittest.mock import PropertyMock, patch from supervisor.dbus.network import NetworkManager from supervisor.dbus.network.interface import NetworkInterface from supervisor.dbus.network.setting.generate import get_connection_from_interface -from supervisor.host.configuration import IpConfig, VlanConfig +from supervisor.host.configuration import IpConfig, IpSetting, VlanConfig from supervisor.host.const import InterfaceMethod, InterfaceType from supervisor.host.network import Interface @@ -55,8 +55,10 @@ async def test_generate_from_vlan(network_manager: NetworkManager): connected=True, primary=False, type=InterfaceType.VLAN, - ipv4=IpConfig(InterfaceMethod.AUTO, [], None, [], None), + ipv4=IpConfig([], None, [], None), + ipv4setting=IpSetting(InterfaceMethod.AUTO, [], None, []), ipv6=None, + ipv6setting=None, wifi=None, vlan=VlanConfig(1, "eth0"), ) diff --git a/tests/dbus/network/setting/test_init.py b/tests/dbus/network/setting/test_init.py index 40a8e4524..7066c7195 100644 --- a/tests/dbus/network/setting/test_init.py +++ b/tests/dbus/network/setting/test_init.py @@ -106,8 +106,8 @@ async def test_update( async def test_ipv6_disabled_is_link_local(dbus_interface: NetworkInterface): """Test disabled equals link local for ipv6.""" interface = Interface.from_dbus_interface(dbus_interface) - interface.ipv4.method = InterfaceMethod.DISABLED - interface.ipv6.method = InterfaceMethod.DISABLED + interface.ipv4setting.method = InterfaceMethod.DISABLED + interface.ipv6setting.method = InterfaceMethod.DISABLED conn = get_connection_from_interface( interface, MagicMock(), diff --git a/tests/host/test_network.py b/tests/host/test_network.py index a4fc4a27a..4bb1a33b5 100644 --- a/tests/host/test_network.py +++ b/tests/host/test_network.py @@ -46,7 +46,11 @@ async def fixture_wireless_service( yield network_manager_services["network_device_wireless"] -async def test_load(coresys: CoreSys, network_manager_service: NetworkManagerService): +async def test_load( + coresys: CoreSys, + network_manager_service: NetworkManagerService, + connection_settings_service: ConnectionSettingsService, +): """Test network manager load.""" network_manager_service.ActivateConnection.calls.clear() network_manager_service.CheckConnectivity.calls.clear() @@ -63,15 +67,30 @@ async def test_load(coresys: CoreSys, network_manager_service: NetworkManagerSer assert "eth0" in name_dict assert name_dict["eth0"].mac == "AA:BB:CC:DD:EE:FF" assert name_dict["eth0"].enabled is True - assert name_dict["eth0"].ipv4.method == InterfaceMethod.AUTO assert name_dict["eth0"].ipv4.gateway == IPv4Address("192.168.2.1") assert name_dict["eth0"].ipv4.ready is True - assert name_dict["eth0"].ipv6.method == InterfaceMethod.AUTO + 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"].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 "wlan0" in name_dict assert name_dict["wlan0"].enabled is False + assert connection_settings_service.settings["ipv4"]["method"].value == "auto" + assert "address-data" not in connection_settings_service.settings["ipv4"] + assert "gateway" not in connection_settings_service.settings["ipv4"] + assert "dns" not in connection_settings_service.settings["ipv4"] + assert connection_settings_service.settings["ipv6"]["method"].value == "auto" + assert "address-data" not in connection_settings_service.settings["ipv6"] + assert "gateway" not in connection_settings_service.settings["ipv6"] + assert "dns" not in connection_settings_service.settings["ipv6"] + assert network_manager_service.ActivateConnection.calls == [ ( "/org/freedesktop/NetworkManager/Settings/1", @@ -100,6 +119,15 @@ async def test_load_with_disabled_methods( await coresys.host.network.load() assert network_manager_service.ActivateConnection.calls == [] + assert connection_settings_service.settings["ipv4"]["method"].value == "disabled" + assert "address-data" not in connection_settings_service.settings["ipv4"] + assert "gateway" not in connection_settings_service.settings["ipv4"] + assert "dns" not in connection_settings_service.settings["ipv4"] + assert connection_settings_service.settings["ipv6"]["method"].value == "disabled" + assert "address-data" not in connection_settings_service.settings["ipv6"] + assert "gateway" not in connection_settings_service.settings["ipv6"] + assert "dns" not in connection_settings_service.settings["ipv6"] + async def test_load_with_network_connection_issues( coresys: CoreSys, @@ -121,9 +149,9 @@ async def test_load_with_network_connection_issues( name_dict = {intr.name: intr for intr in coresys.host.network.interfaces} assert "eth0" in name_dict assert name_dict["eth0"].enabled is True - assert name_dict["eth0"].ipv4.method == InterfaceMethod.AUTO + assert name_dict["eth0"].ipv4setting.method == InterfaceMethod.AUTO assert name_dict["eth0"].ipv4.gateway is None - assert name_dict["eth0"].ipv6.method == InterfaceMethod.AUTO + assert name_dict["eth0"].ipv6setting.method == InterfaceMethod.AUTO assert name_dict["eth0"].ipv6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69")