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
This commit is contained in:
Stefan Agner 2024-08-22 08:08:55 +02:00 committed by GitHub
parent 5117364625
commit 1ba621be60
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 304 additions and 130 deletions

View File

@ -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,
)

View File

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

View File

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

View File

@ -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

View File

@ -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),
)

View File

@ -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
)
]
)

View File

@ -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"),
)

View File

@ -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(),

View File

@ -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")