Skip to content

Commit

Permalink
Use separate data structure for IP configuration (#5262)
Browse files Browse the repository at this point in the history
* 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
  • Loading branch information
agners authored Aug 22, 2024
1 parent 5117364 commit 1ba621b
Show file tree
Hide file tree
Showing 9 changed files with 305 additions and 131 deletions.
37 changes: 19 additions & 18 deletions supervisor/api/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
Interface,
InterfaceMethod,
IpConfig,
IpSetting,
VlanConfig,
WifiConfig,
)
Expand Down Expand Up @@ -85,10 +86,10 @@
)


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,
Expand Down Expand Up @@ -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,
}
Expand Down Expand Up @@ -198,15 +199,15 @@ async def interface_update(self, request: web.Request) -> None:
# 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:
Expand Down Expand Up @@ -257,24 +258,22 @@ async def create_vlan(self, request: web.Request) -> None:

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(
Expand All @@ -285,8 +284,10 @@ async def create_vlan(self, request: web.Request) -> None:
True,
False,
InterfaceType.VLAN,
ipv4_config,
ipv6_config,
None,
ipv4_setting,
None,
ipv6_setting,
None,
vlan_config,
)
Expand Down
11 changes: 11 additions & 0 deletions supervisor/dbus/network/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
109 changes: 76 additions & 33 deletions supervisor/dbus/network/setting/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,13 @@
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,
Expand All @@ -21,25 +21,46 @@
)

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",
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -149,7 +170,7 @@ async def update(self, settings: dict[str, dict[str, Variant]]) -> None:
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)
Expand Down Expand Up @@ -194,47 +215,69 @@ async def reload(self):
# 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)
)
Loading

0 comments on commit 1ba621b

Please sign in to comment.