Check NetworkManager Version if it is supported (#2340)

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

supervisor/dbus/const.py

@@ -65,6 +65,7 @@ DBUS_ATTR_STATIC_OPERATING_SYSTEM_CPE_NAME = "OperatingSystemCPEName"
 DBUS_ATTR_TYPE = "Type"
 DBUS_ATTR_UUID = "Uuid"
 DBUS_ATTR_VARIANT = "Variant"
+DBUS_ATTR_VERSION = "Version"
 DBUS_ATTR_MANAGED = "Managed"
 DBUS_ATTR_CONNECTION_ENABLED = "ConnectivityCheckEnabled"
 

supervisor/dbus/interface.py

@@ -20,6 +20,10 @@ class DBusInterface(ABC):
     async def connect(self):
         """Connect to D-Bus."""
 
+    def disconnect(self):
+        """Disconnect from D-Bus."""
+        self.dbus = None
+
 
 class DBusInterfaceProxy(ABC):
     """Handle D-Bus interface proxy."""

supervisor/dbus/network/__init__.py

@@ -2,14 +2,21 @@
 import logging
 from typing import Any, Awaitable, Dict
 
+from packaging.version import parse as pkg_parse
 import sentry_sdk
 
-from ...exceptions import DBusError, DBusInterfaceError, DBusProgramError
+from ...exceptions import (
+    DBusError,
+    DBusInterfaceError,
+    DBusProgramError,
+    HostNotSupportedError,
+)
 from ...utils.gdbus import DBus
 from ..const import (
     DBUS_ATTR_CONNECTION_ENABLED,
     DBUS_ATTR_DEVICES,
     DBUS_ATTR_PRIMARY_CONNECTION,
+    DBUS_ATTR_VERSION,
     DBUS_NAME_NM,
     DBUS_OBJECT_BASE,
     DBUS_OBJECT_NM,
@@ -23,6 +30,8 @@ from .settings import NetworkManagerSettings
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
+MINIMAL_VERSION = "1.14.6"
+
 
 class NetworkManager(DBusInterface):
     """Handle D-Bus interface for Network Manager."""
@@ -55,7 +64,12 @@ class NetworkManager(DBusInterface):
     @property
     def connectivity_enabled(self) -> bool:
         """Return if connectivity check is enabled."""
-        return self.properties.get(DBUS_ATTR_CONNECTION_ENABLED, False)
+        return self.properties[DBUS_ATTR_CONNECTION_ENABLED]
+
+    @property
+    def version(self) -> bool:
+        """Return if connectivity check is enabled."""
+        return self.properties[DBUS_ATTR_VERSION]
 
     @dbus_connected
     def activate_connection(
@@ -93,6 +107,28 @@ class NetworkManager(DBusInterface):
                 "No Network Manager support on the host. Local network functions have been disabled."
             )
 
+        # Make Sure we only connect to supported version
+        if self.is_connected:
+            try:
+                await self._validate_version()
+            except (HostNotSupportedError, DBusError):
+                self.disconnect()
+                self.dns.disconnect()
+                self.settings.disconnect()
+
+    async def _validate_version(self) -> None:
+        """Validate Version of NetworkManager."""
+        self.properties = await self.dbus.get_properties(DBUS_NAME_NM)
+
+        try:
+            if pkg_parse(self.version) >= pkg_parse(MINIMAL_VERSION):
+                return
+        except (TypeError, ValueError, KeyError):
+            pass
+
+        _LOGGER.error("Version '%s' of NetworkManager is not supported!", self.version)
+        raise HostNotSupportedError()
+
     @dbus_connected
     async def update(self):
         """Update Properties."""

tests/dbus/network/test_network_manager.py

@@ -1,12 +1,29 @@
 """Test NetwrokInterface."""
+from unittest.mock import AsyncMock
+
 import pytest
 
 from supervisor.dbus.network import NetworkManager
+from supervisor.exceptions import HostNotSupportedError
 
 from tests.const import TEST_INTERFACE
 
+# pylint: disable=protected-access
+
 
 @pytest.mark.asyncio
 async def test_network_manager(network_manager: NetworkManager):
     """Test network manager update."""
     assert TEST_INTERFACE in network_manager.interfaces
+
+
+@pytest.mark.asyncio
+async def test_network_manager_version(network_manager: NetworkManager):
+    """Test if version validate work."""
+    await network_manager._validate_version()
+    assert network_manager.version == "1.22.10"
+
+    network_manager.dbus.get_properties = AsyncMock(return_value={"Version": "1.13.9"})
+    with pytest.raises(HostNotSupportedError):
+        await network_manager._validate_version()
+    assert network_manager.version == "1.13.9"