Skip to content

Commit

Permalink
Use custom string constants for connection settings
Browse files Browse the repository at this point in the history 
Use separate string constants for all connection settings. This makes
it easier to search where a particular NetworkManager connection
setting is used.
  • Loading branch information
@agners
agners committed Aug 21, 2024
1 parent fcc8d89 commit 270604d
Show file tree
Hide file tree
Showing 2 changed files with 143 additions and 69 deletions.
106 changes: 68 additions & 38 deletions supervisor/dbus/network/setting/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus

from ....const import ATTR_DNS, ATTR_METHOD, ATTR_MODE, ATTR_PSK, ATTR_SSID
from ...const import DBUS_NAME_NM
from ...interface import DBusInterface
from ...utils import dbus_connected
Expand All @@ -21,25 +20,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 @@ -149,7 +169,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,53 +214,63 @@ 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:
self._ipv4 = IpProperties(
data[CONF_ATTR_IPV4].get(ATTR_METHOD),
data[CONF_ATTR_IPV4].get("address-data"),
data[CONF_ATTR_IPV4].get("gateway"),
data[CONF_ATTR_IPV4].get(ATTR_DNS),
data[CONF_ATTR_IPV4].get(CONF_ATTR_IPV4_METHOD),
data[CONF_ATTR_IPV4].get(CONF_ATTR_IPV4_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:
self._ipv6 = IpProperties(
data[CONF_ATTR_IPV6].get(ATTR_METHOD),
data[CONF_ATTR_IPV6].get("address-data"),
data[CONF_ATTR_IPV6].get("gateway"),
data[CONF_ATTR_IPV6].get(ATTR_DNS),
data[CONF_ATTR_IPV6].get(CONF_ATTR_IPV6_METHOD),
data[CONF_ATTR_IPV6].get(CONF_ATTR_IPV6_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)
)
106 changes: 75 additions & 31 deletions supervisor/dbus/network/setting/generate.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,16 +11,39 @@
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:
Expand Down Expand Up @@ -54,18 +77,20 @@ 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)

Expand All @@ -74,12 +99,12 @@ def get_connection_from_interface(
not interface.ipv4setting
or interface.ipv4setting.method == InterfaceMethod.AUTO
):
ipv4["method"] = Variant("s", "auto")
ipv4[CONF_ATTR_IPV4_METHOD] = Variant("s", "auto")
elif interface.ipv4setting.method == InterfaceMethod.DISABLED:
ipv4["method"] = Variant("s", "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))
Expand All @@ -96,8 +121,8 @@ def get_connection_from_interface(
}
)

ipv4["address-data"] = Variant("aa{sv}", address_data)
ipv4["gateway"] = Variant("s", str(interface.ipv4setting.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

Expand All @@ -106,12 +131,12 @@ def get_connection_from_interface(
not interface.ipv6setting
or interface.ipv6setting.method == InterfaceMethod.AUTO
):
ipv6["method"] = Variant("s", "auto")
ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "auto")
elif interface.ipv6setting.method == InterfaceMethod.DISABLED:
ipv6["method"] = Variant("s", "link-local")
ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "link-local")
else:
ipv6["method"] = Variant("s", "manual")
ipv6["dns"] = Variant(
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],
)
Expand All @@ -125,18 +150,25 @@ def get_connection_from_interface(
}
)

<<<<<<< HEAD

Check failure on line 153 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:153:1: SyntaxError: Expected a statement

Check failure on line 153 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:153:3: SyntaxError: Expected a statement

Check failure on line 153 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:153:5: SyntaxError: Expected a statement

Check failure on line 153 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:153:7: SyntaxError: Expected a statement
<<<<<<< HEAD

Check failure on line 154 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:154:1: SyntaxError: Expected a statement

Check failure on line 154 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:154:3: SyntaxError: Expected a statement

Check failure on line 154 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:154:5: SyntaxError: Expected a statement

Check failure on line 154 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:154:7: SyntaxError: Expected a statement
ipv6["address-data"] = Variant("aa{sv}", adressdata)

Check failure on line 155 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:155:1: SyntaxError: Unexpected indentation
ipv6["gateway"] = Variant("s", str(interface.ipv6setting.gateway))
=======

Check failure on line 157 in supervisor/dbus/network/setting/generate.py

View workflow job for this annotation

GitHub Actions / Check ruff

Ruff 

supervisor/dbus/network/setting/generate.py:157:1: SyntaxError: Expected a statement
ipv6[CONF_ATTR_IPV6_ADDRESS_DATA] = Variant("aa{sv}", address_data)
ipv6[CONF_ATTR_IPV6_GATEWAY] = Variant("s", str(interface.ipv6setting.gateway))
>>>>>>> 4d84bb9f (Fix typo in address data variable)
=======
ipv6[CONF_ATTR_IPV6_ADDRESS_DATA] = Variant("aa{sv}", adressdata)
ipv6[CONF_ATTR_IPV6_GATEWAY] = Variant("s", str(interface.ipv6setting.gateway))
>>>>>>> 386b7b24 (Use custom string constants for connection settings)

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 (
Expand All @@ -145,30 +177,42 @@ 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

if interface.wifi.auth != "open":
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

0 comments on commit 270604d

Please sign in to comment.