Use udev path instead of mac or name for nm match (#4476)

This commit is contained in:
Mike Degatano 2023-08-04 17:39:35 -04:00 committed by GitHub
parent a98334ede8
commit 86f004e45a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 33 additions and 83 deletions

View File

@ -277,6 +277,7 @@ class APINetwork(CoreSysAttributes):
)
vlan_interface = Interface(
"",
"",
"",
True,

View File

@ -144,6 +144,7 @@ DBUS_ATTR_OPERATION = "Operation"
DBUS_ATTR_OPTIONS = "Options"
DBUS_ATTR_PARSER_VERSION = "ParserVersion"
DBUS_ATTR_PARTITIONS = "Partitions"
DBUS_ATTR_PATH = "Path"
DBUS_ATTR_POWER_LED = "PowerLED"
DBUS_ATTR_PRIMARY_CONNECTION = "PrimaryConnection"
DBUS_ATTR_READ_ONLY = "ReadOnly"

View File

@ -66,7 +66,7 @@ class IpProperties:
@dataclass(slots=True)
class DeviceProperties:
"""Device properties object for Network Manager."""
class MatchProperties:
"""Match properties object for Network Manager."""
match_device: str | None
path: list[str] | None = None

View File

@ -11,6 +11,7 @@ from ..const import (
DBUS_ATTR_DRIVER,
DBUS_ATTR_HWADDRESS,
DBUS_ATTR_MANAGED,
DBUS_ATTR_PATH,
DBUS_IFACE_DEVICE,
DBUS_NAME_NM,
DBUS_OBJECT_BASE,
@ -74,6 +75,12 @@ class NetworkInterface(DBusInterfaceProxy):
"""Return hardware address (i.e. mac address) of device."""
return self.properties[DBUS_ATTR_HWADDRESS]
@property
@dbus_property
def path(self) -> str:
"""Return The path of the device as exposed by the udev property ID_PATH."""
return self.properties[DBUS_ATTR_PATH]
@property
def connection(self) -> NetworkConnection | None:
"""Return the connection used for this interface."""

View File

@ -11,9 +11,9 @@ from ...interface import DBusInterface
from ...utils import dbus_connected
from ..configuration import (
ConnectionProperties,
DeviceProperties,
EthernetProperties,
IpProperties,
MatchProperties,
VlanProperties,
WirelessProperties,
WirelessSecurityProperties,
@ -26,7 +26,8 @@ CONF_ATTR_802_WIRELESS_SECURITY = "802-11-wireless-security"
CONF_ATTR_VLAN = "vlan"
CONF_ATTR_IPV4 = "ipv4"
CONF_ATTR_IPV6 = "ipv6"
CONF_ATTR_DEVICE = "device"
CONF_ATTR_MATCH = "match"
CONF_ATTR_PATH = "path"
ATTR_ID = "id"
ATTR_UUID = "uuid"
@ -37,7 +38,7 @@ ATTR_POWERSAVE = "powersave"
ATTR_AUTH_ALG = "auth-alg"
ATTR_KEY_MGMT = "key-mgmt"
ATTR_INTERFACE_NAME = "interface-name"
ATTR_MATCH_DEVICE = "match-device"
ATTR_PATH = "path"
IPV4_6_IGNORE_FIELDS = [
"addresses",
@ -88,7 +89,7 @@ class NetworkSetting(DBusInterface):
self._vlan: VlanProperties | None = None
self._ipv4: IpProperties | None = None
self._ipv6: IpProperties | None = None
self._device: DeviceProperties | None = None
self._match: MatchProperties | None = None
@property
def connection(self) -> ConnectionProperties | None:
@ -126,9 +127,9 @@ class NetworkSetting(DBusInterface):
return self._ipv6
@property
def device(self) -> DeviceProperties | None:
"""Return device properties if any."""
return self._device
def match(self) -> MatchProperties | None:
"""Return match properties if any."""
return self._match
@dbus_connected
async def get_settings(self) -> dict[str, Any]:
@ -166,7 +167,7 @@ class NetworkSetting(DBusInterface):
CONF_ATTR_IPV6,
ignore_current_value=IPV4_6_IGNORE_FIELDS,
)
_merge_settings_attribute(new_settings, settings, CONF_ATTR_DEVICE)
_merge_settings_attribute(new_settings, settings, CONF_ATTR_MATCH)
await self.dbus.Settings.Connection.call_update(new_settings)
@ -233,7 +234,5 @@ class NetworkSetting(DBusInterface):
data[CONF_ATTR_IPV6].get(ATTR_METHOD),
)
if CONF_ATTR_DEVICE in data:
self._device = DeviceProperties(
data[CONF_ATTR_DEVICE].get(ATTR_MATCH_DEVICE)
)
if CONF_ATTR_MATCH in data:
self._match = MatchProperties(data[CONF_ATTR_MATCH].get(ATTR_PATH))

View File

@ -9,14 +9,14 @@ from dbus_fast import Variant
from . import (
ATTR_ASSIGNED_MAC,
ATTR_MATCH_DEVICE,
CONF_ATTR_802_ETHERNET,
CONF_ATTR_802_WIRELESS,
CONF_ATTR_802_WIRELESS_SECURITY,
CONF_ATTR_CONNECTION,
CONF_ATTR_DEVICE,
CONF_ATTR_IPV4,
CONF_ATTR_IPV6,
CONF_ATTR_MATCH,
CONF_ATTR_PATH,
CONF_ATTR_VLAN,
)
from ....host.const import InterfaceMethod, InterfaceType
@ -59,11 +59,7 @@ def get_connection_from_interface(
}
if interface.type != InterfaceType.VLAN:
conn[CONF_ATTR_DEVICE] = {
ATTR_MATCH_DEVICE: Variant(
"s", f"mac:{interface.mac},interface-name:{interface.name}"
)
}
conn[CONF_ATTR_MATCH] = {CONF_ATTR_PATH: Variant("as", [interface.path])}
ipv4 = {}
if not interface.ipv4 or interface.ipv4.method == InterfaceMethod.AUTO:

View File

@ -61,6 +61,7 @@ class Interface:
name: str
mac: str
path: str
enabled: bool
connected: bool
primary: bool
@ -75,12 +76,8 @@ class Interface:
if not inet.settings:
return False
if inet.settings.device and inet.settings.device.match_device:
matchers = inet.settings.device.match_device.split(",", 1)
return (
f"mac:{self.mac}" in matchers
or f"interface-name:{self.name}" in matchers
)
if inet.settings.match and inet.settings.match.path:
return inet.settings.match.path == [self.path]
return inet.settings.connection.interface_name == self.name
@ -108,6 +105,7 @@ class Interface:
return Interface(
inet.name,
inet.hw_address,
inet.path,
inet.settings is not None,
Interface._map_nm_connected(inet.connection),
inet.primary,

View File

@ -17,10 +17,7 @@ async def test_get_connection_from_interface(network_manager: NetworkManager):
assert "interface-name" not in connection_payload["connection"]
assert connection_payload["connection"]["type"].value == "802-3-ethernet"
assert (
connection_payload["device"]["match-device"].value
== "mac:AA:BB:CC:DD:EE:FF,interface-name:eth0"
)
assert connection_payload["match"]["path"].value == ["platform-ff3f0000.ethernet"]
assert connection_payload["ipv4"]["method"].value == "auto"
assert "address-data" not in connection_payload["ipv4"]

View File

@ -58,9 +58,7 @@ async def test_update(
)
assert settings["connection"]["autoconnect"] == Variant("b", True)
assert settings["device"] == {
"match-device": Variant("s", "mac:AA:BB:CC:DD:EE:FF,interface-name:eth0")
}
assert settings["match"] == {"path": Variant("as", ["platform-ff3f0000.ethernet"])}
assert "ipv4" in settings
assert settings["ipv4"]["method"] == Variant("s", "auto")

View File

@ -77,9 +77,7 @@ SETINGS_FIXTURES: dict[str, dict[str, dict[str, Variant]]] = {
"proxy": {},
"802-3-ethernet": SETTINGS_FIXTURE["802-3-ethernet"],
"802-11-wireless": SETTINGS_FIXTURE["802-11-wireless"],
"device": {
"match-device": Variant("s", "mac:AA:BB:CC:DD:EE:FF,interface-name:eth0"),
},
"match": {"path": Variant("as", ["platform-ff3f0000.ethernet"])},
},
}

View File

@ -13,7 +13,6 @@ from supervisor.exceptions import HostNotSupportedError
from supervisor.homeassistant.const import WSEvent, WSType
from supervisor.host.const import WifiMode
from tests.common import mock_dbus_services
from tests.dbus_service_mocks.base import DBusServiceMock
from tests.dbus_service_mocks.network_active_connection import (
ActiveConnection as ActiveConnectionService,
@ -253,47 +252,3 @@ async def test_host_connectivity_disabled(
}
)
assert "connectivity_check" not in coresys.resolution.unsupported
@pytest.mark.parametrize(
"interface_obj_path",
[
"/org/freedesktop/NetworkManager/Devices/4",
"/org/freedesktop/NetworkManager/Devices/5",
],
)
async def test_load_with_mac_or_name_change(
coresys: CoreSys,
network_manager_service: NetworkManagerService,
interface_obj_path: str,
):
"""Test load fixes match-device settings if mac address or interface name has changed."""
await mock_dbus_services(
{
"network_active_connection": "/org/freedesktop/NetworkManager/ActiveConnection/2",
"network_connection_settings": "/org/freedesktop/NetworkManager/Settings/2",
"network_device": interface_obj_path,
},
coresys.dbus.bus,
)
await coresys.dbus.network.update({"Devices": [interface_obj_path]})
network_manager_service.ActivateConnection.calls.clear()
assert len(coresys.dbus.network.interfaces) == 1
interface = next(iter(coresys.dbus.network.interfaces))
assert interface.object_path == interface_obj_path
expected_match_device = (
f"mac:{interface.hw_address},interface-name:{interface.name}"
)
assert interface.settings.device.match_device != expected_match_device
await coresys.host.network.load()
assert network_manager_service.ActivateConnection.calls == [
(
"/org/freedesktop/NetworkManager/Settings/2",
interface_obj_path,
"/",
)
]
assert interface.settings.device.match_device == expected_match_device