Network: abstract dbus and supervisor - ipv6/wifi/vlan (#2217)

* Abstract code between dbus - supervisor

* cleanup v2

* fix address vs interface

* fix API calls

* Fix methodnames

* add vlan type

* add vlan support

* Fix tests

* Add wifi support

* more OOO

* fix typing import

* typing part 2

* Fix profile

* fix test payload

* ignore powersafe

* support privancy

* fix property

* Fix tests

* full support all API

* Fix all

* more robust

* Update supervisor/dbus/network/connection.py

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

* Fix gateway

* fix empty gateway

* Allow no ipv6 or ipv4 kernel support

* Exclude device drivers

* Add wifi

* Use loop on api

* refactory p1

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* refactory p2

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* refactory p3

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* refactory p4

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* refactory p5

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* refactory p6

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* refactory p7

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* refactory p8

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* Fix lint

* update sup p1

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* update sup p2

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* fix tests

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* fix logging

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* improve mock handling

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* add fixtures

Signed-off-by: Pascal Vizeli <pvizeli@syshack.ch>

* fix tests

* better testing

* Add more tests

* Fix API test

* Add test for vlan payload

* Support variation

* Fix doc string

* support remove & wifi scan

* make sure we ignore local-link on ipv6

* remove privancy - add vlan

* Fix tests

* fix isort

* Fixture dbus by commands

* Add dnsmanager fixture

* expose commands called by dbus

* Add wifi tests

* Update supervisor/plugins/dns.py

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

* Address comments & fix tests

* change url to be closer on others

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>
This commit is contained in:
Pascal Vizeli 2020-11-09 08:56:42 +01:00 committed by GitHub
parent ffaeb2b96d
commit bd786811a3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
52 changed files with 1995 additions and 510 deletions

View File

@ -117,6 +117,14 @@ class RestAPI(CoreSysAttributes):
"/network/interface/{interface}/update", "/network/interface/{interface}/update",
api_network.interface_update, api_network.interface_update,
), ),
web.get(
"/network/interface/{interface}/accesspoints",
api_network.scan_accesspoints,
),
web.post(
"/network/interface/{interface}/vlan/{vlan}",
api_network.create_vlan,
),
] ]
) )

View File

@ -1,74 +1,158 @@
"""REST API for network.""" """REST API for network."""
import asyncio import asyncio
from ipaddress import ip_address, ip_interface
from typing import Any, Dict from typing import Any, Dict
from aiohttp import web from aiohttp import web
import attr
import voluptuous as vol import voluptuous as vol
from ..const import ( from ..const import (
ATTR_ACCESSPOINTS,
ATTR_ADDRESS, ATTR_ADDRESS,
ATTR_AUTH,
ATTR_CONNECTED,
ATTR_DNS, ATTR_DNS,
ATTR_DOCKER, ATTR_DOCKER,
ATTR_ENABLED,
ATTR_FREQUENCY,
ATTR_GATEWAY, ATTR_GATEWAY,
ATTR_ID,
ATTR_INTERFACE, ATTR_INTERFACE,
ATTR_INTERFACES, ATTR_INTERFACES,
ATTR_IP_ADDRESS, ATTR_IPV4,
ATTR_IPV6,
ATTR_MAC,
ATTR_METHOD, ATTR_METHOD,
ATTR_METHODS, ATTR_MODE,
ATTR_NAMESERVERS, ATTR_NAMESERVERS,
ATTR_PRIMARY, ATTR_PRIMARY,
ATTR_PSK,
ATTR_SIGNAL,
ATTR_SSID,
ATTR_TYPE, ATTR_TYPE,
ATTR_VLAN,
ATTR_WIFI,
DOCKER_NETWORK, DOCKER_NETWORK,
DOCKER_NETWORK_MASK, DOCKER_NETWORK_MASK,
) )
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..dbus.const import InterfaceMethodSimple from ..exceptions import APIError, HostNetworkNotFound
from ..dbus.network.interface import NetworkInterface from ..host.const import AuthMethod, InterfaceType, WifiMode
from ..dbus.network.utils import int2ip from ..host.network import (
from ..exceptions import APIError AccessPoint,
Interface,
InterfaceMethod,
IpConfig,
VlanConfig,
WifiConfig,
)
from .utils import api_process, api_validate from .utils import api_process, api_validate
SCHEMA_UPDATE = vol.Schema( _SCHEMA_IP_CONFIG = vol.Schema(
{ {
vol.Optional(ATTR_ADDRESS): vol.Coerce(str), vol.Optional(ATTR_ADDRESS): [vol.Coerce(ip_interface)],
vol.Optional(ATTR_METHOD): vol.In(ATTR_METHODS), vol.Optional(ATTR_METHOD): vol.Coerce(InterfaceMethod),
vol.Optional(ATTR_GATEWAY): vol.Coerce(str), vol.Optional(ATTR_GATEWAY): vol.Coerce(ip_address),
vol.Optional(ATTR_DNS): [str], vol.Optional(ATTR_NAMESERVERS): [vol.Coerce(ip_address)],
}
)
_SCHEMA_WIFI_CONFIG = vol.Schema(
{
vol.Optional(ATTR_MODE): vol.Coerce(WifiMode),
vol.Optional(ATTR_AUTH): vol.Coerce(AuthMethod),
vol.Optional(ATTR_SSID): str,
vol.Optional(ATTR_PSK): str,
} }
) )
def interface_information(interface: NetworkInterface) -> dict: # pylint: disable=no-value-for-parameter
SCHEMA_UPDATE = vol.Schema(
{
vol.Optional(ATTR_IPV4): _SCHEMA_IP_CONFIG,
vol.Optional(ATTR_IPV6): _SCHEMA_IP_CONFIG,
vol.Optional(ATTR_WIFI): _SCHEMA_WIFI_CONFIG,
vol.Optional(ATTR_ENABLED): vol.Boolean(),
}
)
def ipconfig_struct(config: IpConfig) -> dict:
"""Return a dict with information about ip configuration."""
return {
ATTR_METHOD: config.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),
}
def wifi_struct(config: WifiConfig) -> dict:
"""Return a dict with information about wifi configuration."""
return {
ATTR_MODE: config.mode,
ATTR_AUTH: config.auth,
ATTR_SSID: config.ssid,
ATTR_SIGNAL: config.signal,
}
def interface_struct(interface: Interface) -> dict:
"""Return a dict with information of a interface to be used in th API.""" """Return a dict with information of a interface to be used in th API."""
return { return {
ATTR_INTERFACE: interface.name, ATTR_INTERFACE: interface.name,
ATTR_IP_ADDRESS: f"{interface.ip_address}/{interface.prefix}",
ATTR_GATEWAY: interface.gateway,
ATTR_ID: interface.id,
ATTR_TYPE: interface.type, ATTR_TYPE: interface.type,
ATTR_NAMESERVERS: [int2ip(x) for x in interface.nameservers], ATTR_ENABLED: interface.enabled,
ATTR_METHOD: InterfaceMethodSimple.DHCP ATTR_CONNECTED: interface.connected,
if interface.method == "auto"
else InterfaceMethodSimple.STATIC,
ATTR_PRIMARY: interface.primary, ATTR_PRIMARY: interface.primary,
ATTR_IPV4: ipconfig_struct(interface.ipv4) if interface.ipv4 else None,
ATTR_IPV6: ipconfig_struct(interface.ipv6) if interface.ipv6 else None,
ATTR_WIFI: wifi_struct(interface.wifi) if interface.wifi else None,
ATTR_VLAN: wifi_struct(interface.vlan) if interface.vlan else None,
}
def accesspoint_struct(accesspoint: AccessPoint) -> dict:
"""Return a dict for AccessPoint."""
return {
ATTR_MODE: accesspoint.mode,
ATTR_SSID: accesspoint.ssid,
ATTR_FREQUENCY: accesspoint.frequency,
ATTR_SIGNAL: accesspoint.signal,
ATTR_MAC: accesspoint.mac,
} }
class APINetwork(CoreSysAttributes): class APINetwork(CoreSysAttributes):
"""Handle REST API for network.""" """Handle REST API for network."""
def _get_interface(self, name: str) -> Interface:
"""Get Interface by name or default."""
name = name.lower()
if name == "default":
for interface in self.sys_host.network.interfaces:
if not interface.primary:
continue
return interface
else:
try:
return self.sys_host.network.get(name)
except HostNetworkNotFound:
pass
raise APIError(f"Interface {name} does not exsist") from None
@api_process @api_process
async def info(self, request: web.Request) -> Dict[str, Any]: async def info(self, request: web.Request) -> Dict[str, Any]:
"""Return network information.""" """Return network information."""
interfaces = {}
for interface in self.sys_host.network.interfaces:
interfaces[
self.sys_host.network.interfaces[interface].name
] = interface_information(self.sys_host.network.interfaces[interface])
return { return {
ATTR_INTERFACES: interfaces, ATTR_INTERFACES: [
interface_struct(interface)
for interface in self.sys_host.network.interfaces
],
ATTR_DOCKER: { ATTR_DOCKER: {
ATTR_INTERFACE: DOCKER_NETWORK, ATTR_INTERFACE: DOCKER_NETWORK,
ATTR_ADDRESS: str(DOCKER_NETWORK_MASK), ATTR_ADDRESS: str(DOCKER_NETWORK_MASK),
@ -80,42 +164,88 @@ class APINetwork(CoreSysAttributes):
@api_process @api_process
async def interface_info(self, request: web.Request) -> Dict[str, Any]: async def interface_info(self, request: web.Request) -> Dict[str, Any]:
"""Return network information for a interface.""" """Return network information for a interface."""
req_interface = request.match_info.get(ATTR_INTERFACE) interface = self._get_interface(request.match_info.get(ATTR_INTERFACE))
if req_interface.lower() == "default": return interface_struct(interface)
for interface in self.sys_host.network.interfaces:
if not self.sys_host.network.interfaces[interface].primary:
continue
return interface_information(
self.sys_host.network.interfaces[interface]
)
else:
for interface in self.sys_host.network.interfaces:
if req_interface != self.sys_host.network.interfaces[interface].name:
continue
return interface_information(
self.sys_host.network.interfaces[interface]
)
return {}
@api_process @api_process
async def interface_update(self, request: web.Request) -> Dict[str, Any]: async def interface_update(self, request: web.Request) -> None:
"""Update the configuration of an interface.""" """Update the configuration of an interface."""
req_interface = request.match_info.get(ATTR_INTERFACE) interface = self._get_interface(request.match_info.get(ATTR_INTERFACE))
if not self.sys_host.network.interfaces.get(req_interface): # Validate data
raise APIError(f"Interface {req_interface} does not exsist") body = await api_validate(SCHEMA_UPDATE, request)
if not body:
args = await api_validate(SCHEMA_UPDATE, request)
if not args:
raise APIError("You need to supply at least one option to update") raise APIError("You need to supply at least one option to update")
await asyncio.shield( # Apply config
self.sys_host.network.interfaces[req_interface].update_settings(**args) for key, config in body.items():
if key == ATTR_IPV4:
interface.ipv4 = attr.evolve(interface.ipv4, **config)
elif key == ATTR_IPV6:
interface.ipv6 = attr.evolve(interface.ipv6, **config)
elif key == ATTR_WIFI:
interface.wifi = attr.evolve(interface.wifi, **config)
elif key == ATTR_ENABLED:
interface.enabled = config
await asyncio.shield(self.sys_host.network.apply_changes(interface))
@api_process
async def scan_accesspoints(self, request: web.Request) -> Dict[str, Any]:
"""Scan and return a list of available networks."""
interface = self._get_interface(request.match_info.get(ATTR_INTERFACE))
# Only wlan is supported
if interface.type != InterfaceType.WIRELESS:
raise APIError(f"Interface {interface.name} is not a valid wireless card!")
ap_list = await self.sys_host.network.scan_wifi(interface)
return {ATTR_ACCESSPOINTS: [accesspoint_struct(ap) for ap in ap_list]}
@api_process
async def create_vlan(self, request: web.Request) -> None:
"""Create a new vlan."""
interface = self._get_interface(request.match_info.get(ATTR_INTERFACE))
vlan = int(request.match_info.get(ATTR_VLAN))
# Only ethernet is supported
if interface.type != InterfaceType.ETHERNET:
raise APIError(
f"Interface {interface.name} is not a valid ethernet card for vlan!"
)
body = await api_validate(SCHEMA_UPDATE, request)
vlan_config = VlanConfig(vlan, interface.name)
ipv4_config = None
if ATTR_IPV4 in body:
ipv4_config = IpConfig(
body[ATTR_IPV4].get(ATTR_METHOD, InterfaceMethod.DHCP),
body[ATTR_IPV4].get(ATTR_ADDRESS, []),
body[ATTR_IPV4].get(ATTR_GATEWAY, None),
body[ATTR_IPV4].get(ATTR_NAMESERVERS, []),
)
ipv6_config = None
if ATTR_IPV6 in body:
ipv6_config = IpConfig(
body[ATTR_IPV6].get(ATTR_METHOD, InterfaceMethod.DHCP),
body[ATTR_IPV6].get(ATTR_ADDRESS, []),
body[ATTR_IPV6].get(ATTR_GATEWAY, None),
body[ATTR_IPV6].get(ATTR_NAMESERVERS, []),
)
vlan_interface = Interface(
"",
True,
True,
False,
InterfaceType.VLAN,
ipv4_config,
ipv6_config,
None,
vlan_config,
) )
await asyncio.shield(self.sys_host.network.apply_changes(vlan_interface))
await asyncio.shield(self.sys_host.reload())
return await asyncio.shield(self.interface_info(request))

View File

@ -158,7 +158,6 @@ ATTR_HOST_PID = "host_pid"
ATTR_HOSTNAME = "hostname" ATTR_HOSTNAME = "hostname"
ATTR_ICON = "icon" ATTR_ICON = "icon"
ATTR_ISSUES = "issues" ATTR_ISSUES = "issues"
ATTR_ID = "id"
ATTR_IMAGE = "image" ATTR_IMAGE = "image"
ATTR_IMAGES = "images" ATTR_IMAGES = "images"
ATTR_INDEX = "index" ATTR_INDEX = "index"
@ -176,6 +175,7 @@ ATTR_INTERFACE = "interface"
ATTR_INTERFACES = "interfaces" ATTR_INTERFACES = "interfaces"
ATTR_IP_ADDRESS = "ip_address" ATTR_IP_ADDRESS = "ip_address"
ATTR_IPV4 = "ipv4" ATTR_IPV4 = "ipv4"
ATTR_IPV6 = "ipv6"
ATTR_KERNEL = "kernel" ATTR_KERNEL = "kernel"
ATTR_KERNEL_MODULES = "kernel_modules" ATTR_KERNEL_MODULES = "kernel_modules"
ATTR_LAST_BOOT = "last_boot" ATTR_LAST_BOOT = "last_boot"
@ -193,7 +193,6 @@ ATTR_MEMORY_PERCENT = "memory_percent"
ATTR_MEMORY_USAGE = "memory_usage" ATTR_MEMORY_USAGE = "memory_usage"
ATTR_MESSAGE = "message" ATTR_MESSAGE = "message"
ATTR_METHOD = "method" ATTR_METHOD = "method"
ATTR_METHODS = ["dhcp", "static"]
ATTR_MODE = "mode" ATTR_MODE = "mode"
ATTR_MULTICAST = "multicast" ATTR_MULTICAST = "multicast"
ATTR_NAME = "name" ATTR_NAME = "name"
@ -277,6 +276,17 @@ ATTR_WATCHDOG = "watchdog"
ATTR_WEBUI = "webui" ATTR_WEBUI = "webui"
ATTR_OBSERVER = "observer" ATTR_OBSERVER = "observer"
ATTR_UPDATE_AVAILABLE = "update_available" ATTR_UPDATE_AVAILABLE = "update_available"
ATTR_WIFI = "wifi"
ATTR_VLAN = "vlan"
ATTR_SSD = "ssid"
ATTR_AUTH = "auth"
ATTR_PSK = "psk"
ATTR_CONNECTED = "connected"
ATTR_ENABLED = "enabled"
ATTR_SIGNAL = "signal"
ATTR_MAC = "mac"
ATTR_FREQUENCY = "frequency"
ATTR_ACCESSPOINTS = "accesspoints"
PROVIDE_SERVICE = "provide" PROVIDE_SERVICE = "provide"
NEED_SERVICE = "need" NEED_SERVICE = "need"

View File

@ -3,9 +3,12 @@ from enum import Enum
DBUS_NAME_CONNECTION_ACTIVE = "org.freedesktop.NetworkManager.Connection.Active" DBUS_NAME_CONNECTION_ACTIVE = "org.freedesktop.NetworkManager.Connection.Active"
DBUS_NAME_DEVICE = "org.freedesktop.NetworkManager.Device" DBUS_NAME_DEVICE = "org.freedesktop.NetworkManager.Device"
DBUS_NAME_DEVICE_WIRELESS = "org.freedesktop.NetworkManager.Device.Wireless"
DBUS_NAME_DNS = "org.freedesktop.NetworkManager.DnsManager" DBUS_NAME_DNS = "org.freedesktop.NetworkManager.DnsManager"
DBUS_NAME_ACCESSPOINT = "org.freedesktop.NetworkManager.AccessPoint"
DBUS_NAME_HOSTNAME = "org.freedesktop.hostname1" DBUS_NAME_HOSTNAME = "org.freedesktop.hostname1"
DBUS_NAME_IP4CONFIG = "org.freedesktop.NetworkManager.IP4Config" DBUS_NAME_IP4CONFIG = "org.freedesktop.NetworkManager.IP4Config"
DBUS_NAME_IP6CONFIG = "org.freedesktop.NetworkManager.IP6Config"
DBUS_NAME_NM = "org.freedesktop.NetworkManager" DBUS_NAME_NM = "org.freedesktop.NetworkManager"
DBUS_NAME_RAUC = "de.pengutronix.rauc" DBUS_NAME_RAUC = "de.pengutronix.rauc"
DBUS_NAME_RAUC_INSTALLER = "de.pengutronix.rauc.Installer" DBUS_NAME_RAUC_INSTALLER = "de.pengutronix.rauc.Installer"
@ -15,13 +18,14 @@ DBUS_NAME_SYSTEMD = "org.freedesktop.systemd1"
DBUS_OBJECT_BASE = "/" DBUS_OBJECT_BASE = "/"
DBUS_OBJECT_DNS = "/org/freedesktop/NetworkManager/DnsManager" DBUS_OBJECT_DNS = "/org/freedesktop/NetworkManager/DnsManager"
DBUS_OBJECT_SETTINGS = "/org/freedesktop/NetworkManager/Settings"
DBUS_OBJECT_HOSTNAME = "/org/freedesktop/hostname1" DBUS_OBJECT_HOSTNAME = "/org/freedesktop/hostname1"
DBUS_OBJECT_NM = "/org/freedesktop/NetworkManager" DBUS_OBJECT_NM = "/org/freedesktop/NetworkManager"
DBUS_OBJECT_SYSTEMD = "/org/freedesktop/systemd1" DBUS_OBJECT_SYSTEMD = "/org/freedesktop/systemd1"
DBUS_ATTR_802_WIRELESS = "802-11-wireless"
DBUS_ATTR_802_WIRELESS_SECURITY = "802-11-wireless-security"
DBUS_ATTR_ACTIVE_CONNECTIONS = "ActiveConnections" DBUS_ATTR_ACTIVE_CONNECTIONS = "ActiveConnections"
DBUS_ATTR_ACTIVE_CONNECTION = "ActiveConnection"
DBUS_ATTR_ACTIVE_ACCESSPOINT = "ActiveAccessPoint"
DBUS_ATTR_ADDRESS_DATA = "AddressData" DBUS_ATTR_ADDRESS_DATA = "AddressData"
DBUS_ATTR_BOOT_SLOT = "BootSlot" DBUS_ATTR_BOOT_SLOT = "BootSlot"
DBUS_ATTR_CHASSIS = "Chassis" DBUS_ATTR_CHASSIS = "Chassis"
@ -33,25 +37,32 @@ DBUS_ATTR_DEPLOYMENT = "Deployment"
DBUS_ATTR_DEVICE_INTERFACE = "Interface" DBUS_ATTR_DEVICE_INTERFACE = "Interface"
DBUS_ATTR_DEVICE_TYPE = "DeviceType" DBUS_ATTR_DEVICE_TYPE = "DeviceType"
DBUS_ATTR_DEVICES = "Devices" DBUS_ATTR_DEVICES = "Devices"
DBUS_ATTR_DRIVER = "Driver"
DBUS_ATTR_GATEWAY = "Gateway" DBUS_ATTR_GATEWAY = "Gateway"
DBUS_ATTR_ID = "Id" DBUS_ATTR_ID = "Id"
DBUS_ATTR_IP4ADDRESS = "Ip4Address" DBUS_ATTR_SSID = "Ssid"
DBUS_ATTR_FREQUENCY = "Frequency"
DBUS_ATTR_HWADDRESS = "HwAddress"
DBUS_ATTR_MODE = "Mode"
DBUS_ATTR_STRENGTH = "Strength"
DBUS_ATTR_IP4CONFIG = "Ip4Config" DBUS_ATTR_IP4CONFIG = "Ip4Config"
DBUS_ATTR_IP6CONFIG = "Ip6Config"
DBUS_ATTR_KERNEL_RELEASE = "KernelRelease" DBUS_ATTR_KERNEL_RELEASE = "KernelRelease"
DBUS_ATTR_LAST_ERROR = "LastError" DBUS_ATTR_LAST_ERROR = "LastError"
DBUS_ATTR_MODE = "Mode" DBUS_ATTR_MODE = "Mode"
DBUS_ATTR_NAMESERVERS = "Nameservers" DBUS_ATTR_NAMESERVERS = "Nameservers"
DBUS_ATTR_NAMESERVER_DATA = "NameserverData"
DBUS_ATTR_OPERATING_SYSTEM_PRETTY_NAME = "OperatingSystemPrettyName" DBUS_ATTR_OPERATING_SYSTEM_PRETTY_NAME = "OperatingSystemPrettyName"
DBUS_ATTR_OPERATION = "Operation" DBUS_ATTR_OPERATION = "Operation"
DBUS_ATTR_PRIMARY_CONNECTION = "PrimaryConnection" DBUS_ATTR_PRIMARY_CONNECTION = "PrimaryConnection"
DBUS_ATTR_RCMANAGER = "RcManager" DBUS_ATTR_RCMANAGER = "RcManager"
DBUS_ATTR_REAL = "Real"
DBUS_ATTR_STATE = "State" DBUS_ATTR_STATE = "State"
DBUS_ATTR_STATIC_HOSTNAME = "StaticHostname" DBUS_ATTR_STATIC_HOSTNAME = "StaticHostname"
DBUS_ATTR_STATIC_OPERATING_SYSTEM_CPE_NAME = "OperatingSystemCPEName" DBUS_ATTR_STATIC_OPERATING_SYSTEM_CPE_NAME = "OperatingSystemCPEName"
DBUS_ATTR_TYPE = "Type" DBUS_ATTR_TYPE = "Type"
DBUS_ATTR_UUID = "Uuid" DBUS_ATTR_UUID = "Uuid"
DBUS_ATTR_VARIANT = "Variant" DBUS_ATTR_VARIANT = "Variant"
DBUS_ATTR_MANAGED = "Managed"
class RaucState(str, Enum): class RaucState(str, Enum):
@ -67,13 +78,7 @@ class InterfaceMethod(str, Enum):
AUTO = "auto" AUTO = "auto"
MANUAL = "manual" MANUAL = "manual"
DISABLED = "disabled"
class InterfaceMethodSimple(str, Enum):
"""Interface method."""
DHCP = "dhcp"
STATIC = "static"
class ConnectionType(str, Enum): class ConnectionType(str, Enum):
@ -81,3 +86,41 @@ class ConnectionType(str, Enum):
ETHERNET = "802-3-ethernet" ETHERNET = "802-3-ethernet"
WIRELESS = "802-11-wireless" WIRELESS = "802-11-wireless"
class ConnectionStateType(int, Enum):
"""Connection states.
https://developer.gnome.org/NetworkManager/stable/nm-dbus-types.html#NMActiveConnectionState
"""
UNKNOWN = 0
ACTIVATING = 1
ACTIVATED = 2
DEACTIVATING = 3
DEACTIVATED = 4
class DeviceType(int, Enum):
"""Device types.
https://developer.gnome.org/NetworkManager/stable/nm-dbus-types.html#NMDeviceType
"""
UNKNOWN = 0
ETHERNET = 1
WIRELESS = 2
BLUETOOTH = 5
VLAN = 11
TUN = 16
VETH = 20
class WirelessMethodType(int, Enum):
"""Device Type."""
UNKNOWN = 0
ADHOC = 1
INFRASTRUCTURE = 2
ACCESSPOINT = 3
MESH = 4

View File

@ -1,6 +1,6 @@
"""Interface class for D-Bus wrappers.""" """Interface class for D-Bus wrappers."""
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import Optional from typing import Any, Dict, Optional
from ..utils.gdbus import DBus from ..utils.gdbus import DBus
@ -18,3 +18,15 @@ class DBusInterface(ABC):
@abstractmethod @abstractmethod
async def connect(self): async def connect(self):
"""Connect to D-Bus.""" """Connect to D-Bus."""
class DBusInterfaceProxy(ABC):
"""Handle D-Bus interface proxy."""
dbus: Optional[DBus] = None
object_path: Optional[str] = None
properties: Optional[Dict[str, Any]] = None
@abstractmethod
async def connect(self):
"""Connect to D-Bus."""

View File

@ -1,20 +1,24 @@
"""Network Manager implementation for DBUS.""" """Network Manager implementation for DBUS."""
import logging import logging
from typing import Dict, Optional from typing import Any, Awaitable, Dict
from ...exceptions import DBusError, DBusFatalError, DBusInterfaceError import sentry_sdk
from ...exceptions import DBusError, DBusInterfaceError
from ...utils.gdbus import DBus from ...utils.gdbus import DBus
from ..const import ( from ..const import (
DBUS_ATTR_ACTIVE_CONNECTIONS, DBUS_ATTR_DEVICES,
DBUS_ATTR_PRIMARY_CONNECTION, DBUS_ATTR_PRIMARY_CONNECTION,
DBUS_NAME_NM, DBUS_NAME_NM,
DBUS_OBJECT_BASE,
DBUS_OBJECT_NM, DBUS_OBJECT_NM,
ConnectionType, DeviceType,
) )
from ..interface import DBusInterface from ..interface import DBusInterface
from ..utils import dbus_connected from ..utils import dbus_connected
from .dns import NetworkManagerDNS from .dns import NetworkManagerDNS
from .interface import NetworkInterface from .interface import NetworkInterface
from .settings import NetworkManagerSettings
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -25,23 +29,48 @@ class NetworkManager(DBusInterface):
def __init__(self) -> None: def __init__(self) -> None:
"""Initialize Properties.""" """Initialize Properties."""
self._dns: NetworkManagerDNS = NetworkManagerDNS() self._dns: NetworkManagerDNS = NetworkManagerDNS()
self._interfaces: Optional[Dict[str, NetworkInterface]] = [] self._settings: NetworkManagerSettings = NetworkManagerSettings()
self._interfaces: Dict[str, NetworkInterface] = {}
@property @property
def dns(self) -> NetworkManagerDNS: def dns(self) -> NetworkManagerDNS:
"""Return NetworkManager DNS interface.""" """Return NetworkManager DNS interface."""
return self._dns return self._dns
@property
def settings(self) -> NetworkManagerSettings:
"""Return NetworkManager global settings."""
return self._settings
@property @property
def interfaces(self) -> Dict[str, NetworkInterface]: def interfaces(self) -> Dict[str, NetworkInterface]:
"""Return a dictionary of active interfaces.""" """Return a dictionary of active interfaces."""
return self._interfaces return self._interfaces
@dbus_connected
def activate_connection(
self, connection_object: str, device_object: str
) -> Awaitable[Any]:
"""Activate a connction on a device."""
return self.dbus.ActivateConnection(
connection_object, device_object, DBUS_OBJECT_BASE
)
@dbus_connected
def add_and_activate_connection(
self, settings: str, device_object: str
) -> Awaitable[Any]:
"""Activate a connction on a device."""
return self.dbus.AddAndActivateConnection(
settings, device_object, DBUS_OBJECT_BASE
)
async def connect(self) -> None: async def connect(self) -> None:
"""Connect to system's D-Bus.""" """Connect to system's D-Bus."""
try: try:
self.dbus = await DBus.connect(DBUS_NAME_NM, DBUS_OBJECT_NM) self.dbus = await DBus.connect(DBUS_NAME_NM, DBUS_OBJECT_NM)
await self.dns.connect() await self.dns.connect()
await self.settings.connect()
except DBusError: except DBusError:
_LOGGER.warning("Can't connect to Network Manager") _LOGGER.warning("Can't connect to Network Manager")
except DBusInterfaceError: except DBusInterfaceError:
@ -60,28 +89,33 @@ class NetworkManager(DBusInterface):
_LOGGER.warning("Can't get properties for Network Manager") _LOGGER.warning("Can't get properties for Network Manager")
return return
self._interfaces = {} self._interfaces.clear()
for connection in data.get(DBUS_ATTR_ACTIVE_CONNECTIONS, []): for device in data.get(DBUS_ATTR_DEVICES, []):
interface = NetworkInterface() interface = NetworkInterface(self.dbus, device)
await interface.connect(self.dbus, connection) # Connect to interface
if interface.connection.type not in [
ConnectionType.ETHERNET,
ConnectionType.WIRELESS,
]:
continue
try: try:
await interface.connection.update_information() await interface.connect()
except (IndexError, DBusFatalError, KeyError): except Exception as err: # pylint: disable=broad-except
_LOGGER.exception("Error while processing interface: %s", err)
sentry_sdk.capture_exception(err)
continue continue
if not interface.connection.ip4_config: # Skeep interface
if (
interface.type
not in [
DeviceType.ETHERNET,
DeviceType.WIRELESS,
DeviceType.VLAN,
]
or not interface.managed
):
continue continue
if interface.connection.object_path == data.get( if interface.connection and interface.connection.object_path == data.get(
DBUS_ATTR_PRIMARY_CONNECTION DBUS_ATTR_PRIMARY_CONNECTION
): ):
interface.connection.primary = True interface.primary = True
self._interfaces[interface.name] = interface self._interfaces[interface.name] = interface

View File

@ -0,0 +1,52 @@
"""Connection object for Network Manager."""
from ...utils.gdbus import DBus
from ..const import (
DBUS_ATTR_FREQUENCY,
DBUS_ATTR_HWADDRESS,
DBUS_ATTR_MODE,
DBUS_ATTR_SSID,
DBUS_ATTR_STRENGTH,
DBUS_NAME_ACCESSPOINT,
DBUS_NAME_NM,
)
from ..interface import DBusInterfaceProxy
class NetworkWirelessAP(DBusInterfaceProxy):
"""NetworkWireless AP object for Network Manager."""
def __init__(self, object_path: str) -> None:
"""Initialize NetworkWireless AP object."""
self.object_path = object_path
self.properties = {}
@property
def ssid(self) -> str:
"""Return details about ssid."""
return bytes(self.properties[DBUS_ATTR_SSID]).decode()
@property
def frequency(self) -> int:
"""Return details about frequency."""
return self.properties[DBUS_ATTR_FREQUENCY]
@property
def mac(self) -> str:
"""Return details about mac address."""
return self.properties[DBUS_ATTR_HWADDRESS]
@property
def mode(self) -> int:
"""Return details about mac address."""
return self.properties[DBUS_ATTR_MODE]
@property
def strength(self) -> int:
"""Return details about mac address."""
return int(self.properties[DBUS_ATTR_STRENGTH])
async def connect(self) -> None:
"""Get connection information."""
self.dbus = await DBus.connect(DBUS_NAME_NM, self.object_path)
self.properties = await self.dbus.get_properties(DBUS_NAME_ACCESSPOINT)

View File

@ -1,71 +1,75 @@
"""NetworkConnection object4s for Network Manager.""" """NetworkConnection object4s for Network Manager."""
from typing import List from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
from typing import List, Optional, Union
import attr import attr
from ...utils.gdbus import DBus
@attr.s(slots=True)
class NetworkAttributes:
"""NetworkAttributes object for Network Manager."""
def __init__(self, object_path: str, properties: dict) -> None:
"""Initialize NetworkAttributes object."""
self._properties = properties
self.object_path = object_path
@attr.s
class AddressData:
"""AddressData object for Network Manager."""
address: str = attr.ib()
prefix: int = attr.ib()
@attr.s
class IpConfiguration: class IpConfiguration:
"""NetworkSettingsIPConfig object for Network Manager.""" """NetworkSettingsIPConfig object for Network Manager."""
gateway: str = attr.ib() gateway: Optional[Union[IPv6Address, IPv6Address]] = attr.ib()
method: str = attr.ib() nameservers: List[Union[IPv6Address, IPv6Address]] = attr.ib()
nameservers: List[int] = attr.ib() address: List[Union[IPv4Interface, IPv6Interface]] = attr.ib()
address_data: AddressData = attr.ib()
@attr.s @attr.s(slots=True)
class DNSConfiguration: class DNSConfiguration:
"""DNS configuration Object.""" """DNS configuration Object."""
nameservers: List[str] = attr.ib() nameservers: List[Union[IPv4Address, IPv6Address]] = attr.ib()
domains: List[str] = attr.ib() domains: List[str] = attr.ib()
interface: str = attr.ib() interface: str = attr.ib()
priority: int = attr.ib() priority: int = attr.ib()
vpn: bool = attr.ib() vpn: bool = attr.ib()
@attr.s @attr.s(slots=True)
class NetworkSettings: class ConnectionProperties:
"""NetworkSettings object for Network Manager.""" """Connection Properties object for Network Manager."""
dbus: DBus = attr.ib() id: Optional[str] = attr.ib()
uuid: Optional[str] = attr.ib()
type: Optional[str] = attr.ib()
@attr.s @attr.s(slots=True)
class NetworkDevice:
"""Device properties."""
dbus: DBus = attr.ib()
interface: str = attr.ib()
ip4_address: int = attr.ib()
device_type: int = attr.ib()
real: bool = attr.ib()
@attr.s
class WirelessProperties: class WirelessProperties:
"""WirelessProperties object for Network Manager.""" """Wireless Properties object for Network Manager."""
properties: dict = attr.ib() ssid: Optional[str] = attr.ib()
security: dict = attr.ib() assigned_mac: Optional[str] = attr.ib()
ssid: str = attr.ib() mode: Optional[str] = attr.ib()
powersave: Optional[int] = attr.ib()
@attr.s(slots=True)
class WirelessSecurityProperties:
"""Wireless Security Properties object for Network Manager."""
auth_alg: Optional[str] = attr.ib()
key_mgmt: Optional[str] = attr.ib()
psk: Optional[str] = attr.ib()
@attr.s(slots=True)
class EthernetProperties:
"""Ethernet properties object for Network Manager."""
assigned_mac: Optional[str] = attr.ib()
@attr.s(slots=True)
class VlanProperties:
"""Ethernet properties object for Network Manager."""
id: Optional[int] = attr.ib()
parent: Optional[str] = attr.ib()
@attr.s(slots=True)
class IpProperties:
"""IP properties object for Network Manager."""
method: Optional[str] = attr.ib()

View File

@ -1,146 +1,116 @@
"""Connection object for Network Manager.""" """Connection object for Network Manager."""
from ipaddress import ip_address, ip_interface
from typing import Optional from typing import Optional
from ...const import ATTR_ADDRESS, ATTR_IPV4, ATTR_METHOD, ATTR_PREFIX, ATTR_SSID from ...const import ATTR_ADDRESS, ATTR_PREFIX
from ...utils.gdbus import DBus from ...utils.gdbus import DBus
from ..const import ( from ..const import (
DBUS_ATTR_802_WIRELESS,
DBUS_ATTR_802_WIRELESS_SECURITY,
DBUS_ATTR_ADDRESS_DATA, DBUS_ATTR_ADDRESS_DATA,
DBUS_ATTR_CONNECTION, DBUS_ATTR_CONNECTION,
DBUS_ATTR_DEFAULT,
DBUS_ATTR_DEVICE_INTERFACE,
DBUS_ATTR_DEVICE_TYPE,
DBUS_ATTR_DEVICES,
DBUS_ATTR_GATEWAY, DBUS_ATTR_GATEWAY,
DBUS_ATTR_ID, DBUS_ATTR_ID,
DBUS_ATTR_IP4ADDRESS,
DBUS_ATTR_IP4CONFIG, DBUS_ATTR_IP4CONFIG,
DBUS_ATTR_IP6CONFIG,
DBUS_ATTR_NAMESERVER_DATA,
DBUS_ATTR_NAMESERVERS, DBUS_ATTR_NAMESERVERS,
DBUS_ATTR_REAL,
DBUS_ATTR_STATE, DBUS_ATTR_STATE,
DBUS_ATTR_TYPE, DBUS_ATTR_TYPE,
DBUS_ATTR_UUID, DBUS_ATTR_UUID,
DBUS_NAME_DEVICE, DBUS_NAME_CONNECTION_ACTIVE,
DBUS_NAME_IP4CONFIG, DBUS_NAME_IP4CONFIG,
DBUS_NAME_IP6CONFIG,
DBUS_NAME_NM, DBUS_NAME_NM,
DBUS_OBJECT_BASE, DBUS_OBJECT_BASE,
ConnectionType,
)
from .configuration import (
AddressData,
IpConfiguration,
NetworkAttributes,
NetworkDevice,
NetworkSettings,
WirelessProperties,
) )
from ..interface import DBusInterfaceProxy
from .configuration import IpConfiguration
class NetworkConnection(NetworkAttributes): class NetworkConnection(DBusInterfaceProxy):
"""NetworkConnection object for Network Manager.""" """NetworkConnection object for Network Manager."""
def __init__(self, object_path: str, properties: dict) -> None: def __init__(self, object_path: str) -> None:
"""Initialize NetworkConnection object.""" """Initialize NetworkConnection object."""
super().__init__(object_path, properties) self.object_path = object_path
self._device_dbus: DBus = None self.properties = {}
self._settings_dbus: DBus = None
self._settings: Optional[NetworkSettings] = None
self._ip4_config: Optional[IpConfiguration] = None
self._device: Optional[NetworkDevice]
self._wireless: Optional[WirelessProperties] = None
self.primary: bool = False
@property self._ipv4: Optional[IpConfiguration] = None
def settings(self) -> NetworkSettings: self._ipv6: Optional[IpConfiguration] = None
"""Return a settings object for the connection."""
return self._settings
@property
def device(self) -> NetworkDevice:
"""Return the device used in the connection."""
return self._device
@property
def default(self) -> bool:
"""Return a boolean connection is marked as default."""
return self._properties[DBUS_ATTR_DEFAULT]
@property @property
def id(self) -> str: def id(self) -> str:
"""Return the id of the connection.""" """Return the id of the connection."""
return self._properties[DBUS_ATTR_ID] return self.properties[DBUS_ATTR_ID]
@property @property
def type(self) -> str: def type(self) -> str:
"""Return the type of the connection.""" """Return the type of the connection."""
return self._properties[DBUS_ATTR_TYPE] return self.properties[DBUS_ATTR_TYPE]
@property
def ip4_config(self) -> IpConfiguration:
"""Return a ip configuration object for the connection."""
return self._ip4_config
@property @property
def uuid(self) -> str: def uuid(self) -> str:
"""Return the uuid of the connection.""" """Return the uuid of the connection."""
return self._properties[DBUS_ATTR_UUID] return self.properties[DBUS_ATTR_UUID]
@property
def wireless(self) -> str:
"""Return wireless properties if any."""
if self.type != ConnectionType.WIRELESS:
return None
return self._wireless
@property @property
def state(self) -> int: def state(self) -> int:
""" """Return the state of the connection."""
Return the state of the connection. return self.properties[DBUS_ATTR_STATE]
https://developer.gnome.org/NetworkManager/stable/nm-dbus-types.html#NMActiveConnectionState @property
""" def setting_object(self) -> int:
return self._properties[DBUS_ATTR_STATE] """Return the connection object path."""
return self.properties[DBUS_ATTR_CONNECTION]
async def update_information(self): @property
"""Update the information for childs .""" def ipv4(self) -> Optional[IpConfiguration]:
if self._properties[DBUS_ATTR_IP4CONFIG] == DBUS_OBJECT_BASE: """Return a ip4 configuration object for the connection."""
return return self._ipv4
settings = await DBus.connect( @property
DBUS_NAME_NM, self._properties[DBUS_ATTR_CONNECTION] def ipv6(self) -> Optional[IpConfiguration]:
) """Return a ip6 configuration object for the connection."""
device = await DBus.connect( return self._ipv6
DBUS_NAME_NM, self._properties[DBUS_ATTR_DEVICES][0]
)
ip4 = await DBus.connect(DBUS_NAME_NM, self._properties[DBUS_ATTR_IP4CONFIG])
data = (await settings.Settings.Connection.GetSettings())[0] async def connect(self) -> None:
device_data = await device.get_properties(DBUS_NAME_DEVICE) """Get connection information."""
ip4_data = await ip4.get_properties(DBUS_NAME_IP4CONFIG) self.dbus = await DBus.connect(DBUS_NAME_NM, self.object_path)
self.properties = await self.dbus.get_properties(DBUS_NAME_CONNECTION_ACTIVE)
self._settings = NetworkSettings(settings) # IPv4
if self.properties[DBUS_ATTR_IP4CONFIG] != DBUS_OBJECT_BASE:
ip4 = await DBus.connect(DBUS_NAME_NM, self.properties[DBUS_ATTR_IP4CONFIG])
ip4_data = await ip4.get_properties(DBUS_NAME_IP4CONFIG)
self._ip4_config = IpConfiguration( self._ipv4 = IpConfiguration(
ip4_data.get(DBUS_ATTR_GATEWAY), ip_address(ip4_data[DBUS_ATTR_GATEWAY])
data[ATTR_IPV4].get(ATTR_METHOD), if ip4_data.get(DBUS_ATTR_GATEWAY)
ip4_data.get(DBUS_ATTR_NAMESERVERS), else None,
AddressData( [
ip4_data.get(DBUS_ATTR_ADDRESS_DATA)[0].get(ATTR_ADDRESS), ip_address(nameserver[ATTR_ADDRESS])
ip4_data.get(DBUS_ATTR_ADDRESS_DATA)[0].get(ATTR_PREFIX), for nameserver in ip4_data.get(DBUS_ATTR_NAMESERVER_DATA, [])
), ],
) [
ip_interface(f"{address[ATTR_ADDRESS]}/{address[ATTR_PREFIX]}")
for address in ip4_data.get(DBUS_ATTR_ADDRESS_DATA, [])
],
)
self._wireless = WirelessProperties( # IPv6
data.get(DBUS_ATTR_802_WIRELESS, {}), if self.properties[DBUS_ATTR_IP6CONFIG] != DBUS_OBJECT_BASE:
data.get(DBUS_ATTR_802_WIRELESS_SECURITY, {}), ip6 = await DBus.connect(DBUS_NAME_NM, self.properties[DBUS_ATTR_IP6CONFIG])
bytes(data.get(DBUS_ATTR_802_WIRELESS, {}).get(ATTR_SSID, [])).decode(), ip6_data = await ip6.get_properties(DBUS_NAME_IP6CONFIG)
)
self._device = NetworkDevice( self._ipv6 = IpConfiguration(
device, ip_address(ip6_data[DBUS_ATTR_GATEWAY])
device_data.get(DBUS_ATTR_DEVICE_INTERFACE), if ip6_data.get(DBUS_ATTR_GATEWAY)
device_data.get(DBUS_ATTR_IP4ADDRESS), else None,
device_data.get(DBUS_ATTR_DEVICE_TYPE), [
device_data.get(DBUS_ATTR_REAL), ip_address(bytes(nameserver))
) for nameserver in ip6_data.get(DBUS_ATTR_NAMESERVERS)
],
[
ip_interface(f"{address[ATTR_ADDRESS]}/{address[ATTR_PREFIX]}")
for address in ip6_data.get(DBUS_ATTR_ADDRESS_DATA, [])
],
)

View File

@ -1,4 +1,5 @@
"""D-Bus interface for hostname.""" """D-Bus interface for hostname."""
from ipaddress import ip_address
import logging import logging
from typing import List, Optional from typing import List, Optional
@ -75,7 +76,7 @@ class NetworkManagerDNS(DBusInterface):
# Parse configuraton # Parse configuraton
self._configuration = [ self._configuration = [
DNSConfiguration( DNSConfiguration(
config.get(ATTR_NAMESERVERS), [ip_address(nameserver) for nameserver in config.get(ATTR_NAMESERVERS)],
config.get(ATTR_DOMAINS), config.get(ATTR_DOMAINS),
config.get(ATTR_INTERFACE), config.get(ATTR_INTERFACE),
config.get(ATTR_PRIORITY), config.get(ATTR_PRIORITY),

View File

@ -1,101 +1,96 @@
"""NetworkInterface object for Network Manager.""" """NetworkInterface object for Network Manager."""
from typing import Optional
from ...utils.gdbus import DBus from ...utils.gdbus import DBus
from ..const import ( from ..const import (
DBUS_NAME_CONNECTION_ACTIVE, DBUS_ATTR_ACTIVE_CONNECTION,
DBUS_ATTR_DEVICE_INTERFACE,
DBUS_ATTR_DEVICE_TYPE,
DBUS_ATTR_DRIVER,
DBUS_ATTR_MANAGED,
DBUS_NAME_DEVICE,
DBUS_NAME_NM, DBUS_NAME_NM,
DBUS_OBJECT_BASE, DBUS_OBJECT_BASE,
ConnectionType, DeviceType,
InterfaceMethod,
) )
from ..payloads.generate import interface_update_payload from ..interface import DBusInterfaceProxy
from .connection import NetworkConnection from .connection import NetworkConnection
from .setting import NetworkSetting
from .wireless import NetworkWireless
class NetworkInterface: class NetworkInterface(DBusInterfaceProxy):
"""NetworkInterface object for Network Manager, this serves as a proxy to other objects.""" """NetworkInterface object for Network Manager."""
def __init__(self) -> None: def __init__(self, nm_dbus: DBus, object_path: str) -> None:
"""Initialize NetworkConnection object.""" """Initialize NetworkConnection object."""
self._connection = None self.object_path = object_path
self._nm_dbus = None self.properties = {}
self.primary = True
self._connection: Optional[NetworkConnection] = None
self._settings: Optional[NetworkSetting] = None
self._wireless: Optional[NetworkWireless] = None
self._nm_dbus: DBus = nm_dbus
@property @property
def nm_dbus(self) -> DBus: def name(self) -> str:
"""Return the NM DBus connection.""" """Return interface name."""
return self._nm_dbus return self.properties[DBUS_ATTR_DEVICE_INTERFACE]
@property @property
def connection(self) -> NetworkConnection: def type(self) -> int:
"""Return interface type."""
return self.properties[DBUS_ATTR_DEVICE_TYPE]
@property
def driver(self) -> str:
"""Return interface driver."""
return self.properties[DBUS_ATTR_DRIVER]
@property
def managed(self) -> bool:
"""Return interface driver."""
return self.properties[DBUS_ATTR_MANAGED]
@property
def connection(self) -> Optional[NetworkConnection]:
"""Return the connection used for this interface.""" """Return the connection used for this interface."""
return self._connection return self._connection
@property @property
def name(self) -> str: def settings(self) -> Optional[NetworkSetting]:
"""Return the interface name.""" """Return the connection settings used for this interface."""
return self.connection.device.interface return self._settings
@property @property
def primary(self) -> bool: def wireless(self) -> Optional[NetworkWireless]:
"""Return true if it's the primary interfac.""" """Return the wireless data for this interface."""
return self.connection.primary return self._wireless
@property async def connect(self) -> None:
def ip_address(self) -> str: """Get device information."""
"""Return the ip_address.""" self.dbus = await DBus.connect(DBUS_NAME_NM, self.object_path)
return self.connection.ip4_config.address_data.address self.properties = await self.dbus.get_properties(DBUS_NAME_DEVICE)
@property # Abort if device is not managed
def prefix(self) -> str: if not self.managed:
"""Return the network prefix.""" return
return self.connection.ip4_config.address_data.prefix
@property # If connection exists
def type(self) -> ConnectionType: if self.properties[DBUS_ATTR_ACTIVE_CONNECTION] != DBUS_OBJECT_BASE:
"""Return the interface type.""" self._connection = NetworkConnection(
return self.connection.type self.properties[DBUS_ATTR_ACTIVE_CONNECTION]
)
await self._connection.connect()
@property # Attach settings
def id(self) -> str: if self.connection and self.connection.setting_object != DBUS_OBJECT_BASE:
"""Return the interface id.""" self._settings = NetworkSetting(self.connection.setting_object)
return self.connection.id await self._settings.connect()
@property # Wireless
def uuid(self) -> str: if self.type == DeviceType.WIRELESS:
"""Return the interface uuid.""" self._wireless = NetworkWireless(self.object_path)
return self.connection.uuid await self._wireless.connect()
@property
def method(self) -> InterfaceMethod:
"""Return the interface method."""
return InterfaceMethod(self.connection.ip4_config.method)
@property
def gateway(self) -> str:
"""Return the gateway."""
return self.connection.ip4_config.gateway
@property
def nameservers(self) -> str:
"""Return the nameservers."""
return self.connection.ip4_config.nameservers
async def connect(self, nm_dbus: DBus, connection_object: str) -> None:
"""Get connection information."""
self._nm_dbus = nm_dbus
connection_bus = await DBus.connect(DBUS_NAME_NM, connection_object)
connection_properties = await connection_bus.get_properties(
DBUS_NAME_CONNECTION_ACTIVE
)
self._connection = NetworkConnection(connection_object, connection_properties)
async def update_settings(self, **kwargs) -> None:
"""Update IP configuration used for this interface."""
payload = interface_update_payload(self, **kwargs)
await self.connection.settings.dbus.Settings.Connection.Update(payload)
await self.nm_dbus.ActivateConnection(
self.connection.settings.dbus.object_path,
self.connection.device.dbus.object_path,
DBUS_OBJECT_BASE,
)

View File

@ -0,0 +1,148 @@
"""Connection object for Network Manager."""
from typing import Any, Awaitable, Optional
from ...const import ATTR_METHOD, ATTR_MODE, ATTR_PSK, ATTR_SSID
from ...utils.gdbus import DBus
from ..const import DBUS_NAME_NM
from ..interface import DBusInterfaceProxy
from ..utils import dbus_connected
from .configuration import (
ConnectionProperties,
EthernetProperties,
IpProperties,
VlanProperties,
WirelessProperties,
WirelessSecurityProperties,
)
CONF_ATTR_CONNECTION = "connection"
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"
ATTR_ID = "id"
ATTR_UUID = "uuid"
ATTR_TYPE = "type"
ATTR_PARENT = "parent"
ATTR_ASSIGNED_MAC = "assigned-mac-address"
ATTR_POWERSAVE = "powersave"
ATTR_AUTH_ALGO = "auth-algo"
ATTR_KEY_MGMT = "key-mgmt"
class NetworkSetting(DBusInterfaceProxy):
"""NetworkConnection object for Network Manager."""
def __init__(self, object_path: str) -> None:
"""Initialize NetworkConnection object."""
self.object_path = object_path
self.properties = {}
self._connection: Optional[ConnectionProperties] = None
self._wireless: Optional[WirelessProperties] = None
self._wireless_security: Optional[WirelessSecurityProperties] = None
self._ethernet: Optional[EthernetProperties] = None
self._vlan: Optional[VlanProperties] = None
self._ipv4: Optional[IpProperties] = None
self._ipv6: Optional[IpProperties] = None
@property
def connection(self) -> Optional[ConnectionProperties]:
"""Return connection properties if any."""
return self._connection
@property
def wireless(self) -> Optional[WirelessProperties]:
"""Return wireless properties if any."""
return self._wireless
@property
def wireless_security(self) -> Optional[WirelessSecurityProperties]:
"""Return wireless security properties if any."""
return self._wireless_security
@property
def ethernet(self) -> Optional[EthernetProperties]:
"""Return Ethernet properties if any."""
return self._ethernet
@property
def vlan(self) -> Optional[VlanProperties]:
"""Return Vlan properties if any."""
return self._vlan
@property
def ipv4(self) -> Optional[IpProperties]:
"""Return ipv4 properties if any."""
return self._ipv4
@property
def ipv6(self) -> Optional[IpProperties]:
"""Return ipv6 properties if any."""
return self._ipv6
@dbus_connected
def get_settings(self) -> Awaitable[Any]:
"""Return connection settings."""
return self.dbus.Settings.Connection.GetSettings()
@dbus_connected
def update(self, settings: str) -> Awaitable[None]:
"""Update connection settings."""
return self.dbus.Settings.Connection.Update(settings)
@dbus_connected
def delete(self) -> Awaitable[None]:
"""Delete connection settings."""
return self.dbus.Settings.Connection.Delete()
async def connect(self) -> None:
"""Get connection information."""
self.dbus = await DBus.connect(DBUS_NAME_NM, self.object_path)
data = (await self.get_settings())[0]
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),
)
if CONF_ATTR_802_ETHERNET in data:
self._ethernet = EthernetProperties(
data[CONF_ATTR_802_ETHERNET].get(ATTR_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),
)
if CONF_ATTR_802_WIRELESS_SECURITY in data:
self._wireless_security = WirelessSecurityProperties(
data[CONF_ATTR_802_WIRELESS_SECURITY].get(ATTR_AUTH_ALGO),
data[CONF_ATTR_802_WIRELESS_SECURITY].get(ATTR_KEY_MGMT),
data[CONF_ATTR_802_WIRELESS_SECURITY].get(ATTR_PSK),
)
if CONF_ATTR_VLAN in data:
self._vlan = VlanProperties(
data[CONF_ATTR_VLAN].get(ATTR_ID),
data[CONF_ATTR_VLAN].get(ATTR_PARENT),
)
if CONF_ATTR_IPV4 in data:
self._ipv4 = IpProperties(
data[CONF_ATTR_IPV4].get(ATTR_METHOD),
)
if CONF_ATTR_IPV6 in data:
self._ipv6 = IpProperties(
data[CONF_ATTR_IPV6].get(ATTR_METHOD),
)

View File

@ -0,0 +1,36 @@
"""Network Manager implementation for DBUS."""
import logging
from typing import Any, Awaitable
from ...exceptions import DBusError, DBusInterfaceError
from ...utils.gdbus import DBus
from ..const import DBUS_NAME_NM, DBUS_OBJECT_SETTINGS
from ..interface import DBusInterface
from ..utils import dbus_connected
_LOGGER: logging.Logger = logging.getLogger(__name__)
class NetworkManagerSettings(DBusInterface):
"""Handle D-Bus interface for Network Manager."""
async def connect(self) -> None:
"""Connect to system's D-Bus."""
try:
self.dbus = await DBus.connect(DBUS_NAME_NM, DBUS_OBJECT_SETTINGS)
except DBusError:
_LOGGER.warning("Can't connect to Network Manager Settings")
except DBusInterfaceError:
_LOGGER.warning(
"No Network Manager Settings support on the host. Local network functions have been disabled."
)
@dbus_connected
def add_connection(self, settings: str) -> Awaitable[Any]:
"""Add new connection."""
return self.dbus.Settings.AddConnection(settings)
@dbus_connected
def reload_connections(self) -> Awaitable[Any]:
"""Reload all local connection files."""
return self.dbus.Settings.ReloadConnections()

View File

@ -1,14 +0,0 @@
"""Network utils."""
from ipaddress import ip_address
# Return a 32bit representation of a IP Address
def ip2int(address: str) -> int:
"""Return a 32bit representation for a IP address."""
return int(ip_address(".".join(address.split(".")[::-1])))
def int2ip(bitaddress: int) -> int:
"""Return a IP Address object from a 32bit representation."""
return ".".join([str(bitaddress >> (i << 3) & 0xFF) for i in range(0, 4)])

View File

@ -0,0 +1,51 @@
"""Connection object for Network Manager."""
from typing import Any, Awaitable, Optional
from ...utils.gdbus import DBus
from ..const import (
DBUS_ATTR_ACTIVE_ACCESSPOINT,
DBUS_NAME_DEVICE_WIRELESS,
DBUS_NAME_NM,
DBUS_OBJECT_BASE,
)
from ..interface import DBusInterfaceProxy
from ..utils import dbus_connected
from .accesspoint import NetworkWirelessAP
class NetworkWireless(DBusInterfaceProxy):
"""NetworkWireless object for Network Manager."""
def __init__(self, object_path: str) -> None:
"""Initialize NetworkConnection object."""
self.object_path = object_path
self.properties = {}
self._active: Optional[NetworkWirelessAP] = None
@property
def active(self) -> Optional[NetworkWirelessAP]:
"""Return details about active connection."""
return self._active
@dbus_connected
def request_scan(self) -> Awaitable[None]:
"""Request a new AP scan."""
return self.dbus.Device.Wireless.RequestScan("[]")
@dbus_connected
def get_all_accesspoints(self) -> Awaitable[Any]:
"""Return a list of all access points path."""
return self.dbus.Device.Wireless.GetAllAccessPoints()
async def connect(self) -> None:
"""Get connection information."""
self.dbus = await DBus.connect(DBUS_NAME_NM, self.object_path)
self.properties = await self.dbus.get_properties(DBUS_NAME_DEVICE_WIRELESS)
# Get details from current active
if self.properties[DBUS_ATTR_ACTIVE_ACCESSPOINT] != DBUS_OBJECT_BASE:
self._active = NetworkWirelessAP(
self.properties[DBUS_ATTR_ACTIVE_ACCESSPOINT]
)
await self._active.connect()

View File

@ -1,42 +1,43 @@
"""Payload generators for DBUS communication.""" """Payload generators for DBUS communication."""
from __future__ import annotations
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, Optional
from uuid import uuid4
import jinja2 import jinja2
from ...const import ATTR_ADDRESS, ATTR_DNS, ATTR_METHOD, ATTR_PREFIX, ATTR_SSID from ...host.const import InterfaceType
from ..const import ConnectionType, InterfaceMethod
from ..network.utils import ip2int if TYPE_CHECKING:
from ...host.network import Interface
INTERFACE_UPDATE_TEMPLATE: Path = ( INTERFACE_UPDATE_TEMPLATE: Path = (
Path(__file__).parents[2].joinpath("dbus/payloads/interface_update.tmpl") Path(__file__).parents[2].joinpath("dbus/payloads/interface_update.tmpl")
) )
def interface_update_payload(interface, **kwargs) -> str: def interface_update_payload(
interface: Interface, name: Optional[str] = None, uuid: Optional[str] = None
) -> str:
"""Generate a payload for network interface update.""" """Generate a payload for network interface update."""
template = jinja2.Template(INTERFACE_UPDATE_TEMPLATE.read_text()) template = jinja2.Template(INTERFACE_UPDATE_TEMPLATE.read_text())
if kwargs.get(ATTR_DNS):
kwargs[ATTR_DNS] = [ip2int(x.strip()) for x in kwargs[ATTR_DNS]]
if kwargs.get(ATTR_METHOD): # Generate UUID
kwargs[ATTR_METHOD] = ( if not uuid:
InterfaceMethod.MANUAL uuid = str(uuid4())
if kwargs[ATTR_METHOD] == "static"
else InterfaceMethod.AUTO # Generate ID/name
if not name and interface.type != InterfaceType.VLAN:
name = f"Supervisor {interface.name} - {interface.type!s}"
elif not name:
name = f"Supervisor {interface.name}.{interface.vlan.id}"
# Fix SSID
if interface.wifi:
interface.wifi.ssid = ", ".join(
[f"0x{x}" for x in interface.wifi.ssid.encode().hex(",").split(",")]
) )
if kwargs.get(ATTR_ADDRESS): return template.render(interface=interface, name=name, uuid=uuid)
if "/" in kwargs[ATTR_ADDRESS]:
kwargs[ATTR_PREFIX] = kwargs[ATTR_ADDRESS].split("/")[-1]
kwargs[ATTR_ADDRESS] = kwargs[ATTR_ADDRESS].split("/")[0]
kwargs[ATTR_METHOD] = InterfaceMethod.MANUAL
if interface.type == ConnectionType.WIRELESS:
kwargs[ATTR_SSID] = ", ".join(
[
f"0x{x}"
for x in interface.connection.wireless.ssid.encode().hex(",").split(",")
]
)
return template.render(interface=interface, options=kwargs)

View File

@ -1,41 +1,104 @@
{ {
'connection': 'connection':
{ {
'id': <'{{interface.id}}'>, 'id': <'{{ name }}'>,
'type': <'{{interface.type}}'>, {% if interface.type != "vlan" %}
'uuid': <'{{interface.uuid}}'> 'interface-name': <'{{ interface.name }}'>,
}, {% endif %}
'type': <'{% if interface.type == "ethernet" %}802-3-ethernet{% elif interface.type == "wireless" %}802-11-wireless{% else %}{{ interface.type.value }}{% endif %}'>,
{% if options.get("method") == "auto" %} 'uuid': <'{{ uuid }}'>
'ipv4':
{
'method': <'auto'>
} }
{% else %}
{% if interface.ipv4 %}
,
'ipv4': 'ipv4':
{ {
{% if interface.ipv4.method == "dhcp" %}
'method': <'auto'>
{% elif interface.ipv4.method == "disable" %}
'method': <'disabled'>
{% else %}
'method': <'manual'>, 'method': <'manual'>,
'dns': <[uint32 {{ options.get("dns", interface.nameservers) | list | join(",") }}]>, 'dns': <[uint32 {{ interface.ipv4.nameservers | map("int") | join(",") }}]>,
'address-data': <[ 'address-data': <[
{% for address in interface.ipv4.address %}
{ {
'address': <'{{ options.get("address", interface.ip_address) }}'>, 'address': <'{{ address.ip | string }}'>,
'prefix': <uint32 {{ options.get("prefix", interface.prefix) }}> 'prefix': <uint32 {{ address.with_prefixlen.partition("/") | last | int }}>
}]>, }]>,
'gateway': <'{{ options.get("gateway", interface.gateway) }}'> {% endfor %}
'gateway': <'{{ interface.ipv4.gateway | string }}'>
{% endif %}
} }
{% endif %} {% endif %}
{% if interface.type == "802-11-wireless" %}
{% if interface.ipv6 %}
,
'ipv6':
{
{% if interface.ipv6.method == "dhcp" %}
'method': <'auto'>
{% elif interface.ipv6.method == "disable" %}
'method': <'disabled'>
{% else %}
'method': <'manual'>,
'dns': <[uint32 {{ interface.ipv6.nameservers | map("int") | join(",") }}]>,
'address-data': <[
{% for address in interface.ipv6.address if not address.with_prefixlen.startswith("fe80::") %}
{
'address': <'{{ address.ip | string }}'>,
'prefix': <uint32 {{ address.with_prefixlen.partition("/") | last | int }}>
}]>,
{% endfor %}
'gateway': <'{{ interface.ipv6.gateway | string }}'>
{% endif %}
}
{% endif %}
{% if interface.type == "ethernet" %}
,
'802-3-ethernet':
{
'assigned-mac-address': <'stable'>
}
{% endif %}
{% if interface.type == "vlan" %}
,
'vlan':
{
'id': <uint32 {{ interface.vlan.id }}>,
'parent': <'{{ interface.vlan.interface }}'>
}
{% endif %}
{% if interface.type == "wireless" %}
, ,
'802-11-wireless': '802-11-wireless':
{ {
'security': <'802-11-wireless-security'>, 'security': <'802-11-wireless-security'>,
'ssid': <[byte {{ options.ssid }}]> 'assigned-mac-address': <'stable'>,
'ssid': <[byte {{ interface.wifi.ssid }}]>,
'mode': <'{{ interface.wifi.mode.value }}'>,
'powersave': <uint32 1>
}, },
'802-11-wireless-security': '802-11-wireless-security':
{ {
'auth-alg': <'{{ interface.connection.wireless.security['auth-alg'] }}'>, {% if interface.wifi.auth == "web" %}
'key-mgmt': <'{{ interface.connection.wireless.security['key-mgmt'] }}'> 'auth-alg': <'none'>,
'key-mgmt': <'none'>
{% elif interface.wifi.auth == "wpa-psk" %}
'auth-alg': <'shared'>,
'key-mgmt': <'wpa-psk'>
{% elif interface.wifi.auth == "open" %}
'auth-alg': <'open'>,
'key-mgmt': <'none'>
{% endif %}
{% if interface.wifi.psk %}
,
'psk': <'{{ interface.wifi.psk }}'>
{% endif %}
} }
{% endif %} {% endif %}
} }

View File

@ -172,6 +172,14 @@ class HostAppArmorError(HostError):
"""Host apparmor functions failed.""" """Host apparmor functions failed."""
class HostNetworkError(HostError):
"""Error with host network."""
class HostNetworkNotFound(HostError):
"""Return if host interface is not found."""
# API # API

35
supervisor/host/const.py Normal file
View File

@ -0,0 +1,35 @@
"""Const for host."""
from enum import Enum
class InterfaceMethod(str, Enum):
"""Configuration of an interface."""
DISABLED = "disabled"
STATIC = "static"
DHCP = "dhcp"
class InterfaceType(str, Enum):
"""Configuration of an interface."""
ETHERNET = "ethernet"
WIRELESS = "wireless"
VLAN = "vlan"
class AuthMethod(str, Enum):
"""Authentication method."""
OPEN = "open"
WEB = "web"
WPA_PSK = "wpa-psk"
class WifiMode(str, Enum):
"""Wifi mode."""
INFRASTRUCTURE = "infrastructure"
MESH = "mesh"
ADHOC = "adhoc"
AP = "ap"

View File

@ -1,11 +1,32 @@
"""Info control for host.""" """Info control for host."""
import logging from __future__ import annotations
from typing import Dict, List
from supervisor.dbus.network.interface import NetworkInterface import asyncio
from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
import logging
from typing import List, Optional, Union
import attr
from ..coresys import CoreSys, CoreSysAttributes from ..coresys import CoreSys, CoreSysAttributes
from ..exceptions import DBusError, DBusNotConnectedError, HostNotSupportedError from ..dbus.const import (
ConnectionStateType,
DeviceType,
InterfaceMethod as NMInterfaceMethod,
WirelessMethodType,
)
from ..dbus.network.accesspoint import NetworkWirelessAP
from ..dbus.network.connection import NetworkConnection
from ..dbus.network.interface import NetworkInterface
from ..dbus.payloads.generate import interface_update_payload
from ..exceptions import (
DBusError,
DBusNotConnectedError,
HostNetworkError,
HostNetworkNotFound,
HostNotSupportedError,
)
from .const import AuthMethod, InterfaceMethod, InterfaceType, WifiMode
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -18,9 +39,13 @@ class NetworkManager(CoreSysAttributes):
self.coresys: CoreSys = coresys self.coresys: CoreSys = coresys
@property @property
def interfaces(self) -> Dict[str, NetworkInterface]: def interfaces(self) -> List[Interface]:
"""Return a dictionary of active interfaces.""" """Return a dictionary of active interfaces."""
return self.sys_dbus.network.interfaces interfaces: List[Interface] = []
for inet in self.sys_dbus.network.interfaces.values():
interfaces.append(Interface.from_dbus_interface(inet))
return interfaces
@property @property
def dns_servers(self) -> List[str]: def dns_servers(self) -> List[str]:
@ -32,7 +57,16 @@ class NetworkManager(CoreSysAttributes):
continue continue
servers.extend(config.nameservers) servers.extend(config.nameservers)
return [f"dns://{server}" for server in list(dict.fromkeys(servers))] return list(dict.fromkeys(servers))
def get(self, inet_name: str) -> Interface:
"""Return interface from interface name."""
if inet_name not in self.sys_dbus.network.interfaces:
raise HostNetworkNotFound()
return Interface.from_dbus_interface(
self.sys_dbus.network.interfaces[inet_name]
)
async def update(self): async def update(self):
"""Update properties over dbus.""" """Update properties over dbus."""
@ -44,3 +78,237 @@ class NetworkManager(CoreSysAttributes):
except DBusNotConnectedError as err: except DBusNotConnectedError as err:
_LOGGER.error("No hostname D-Bus connection available") _LOGGER.error("No hostname D-Bus connection available")
raise HostNotSupportedError() from err raise HostNotSupportedError() from err
async def apply_changes(self, interface: Interface) -> None:
"""Apply Interface changes to host."""
inet = self.sys_dbus.network.interfaces.get(interface.name)
# Update exist configuration
if inet and inet.settings and interface.enabled:
settings = interface_update_payload(
interface,
name=inet.settings.connection.id,
uuid=inet.settings.connection.uuid,
)
try:
await inet.settings.update(settings)
except DBusError as err:
_LOGGER.error("Can't update config on %s", interface.name)
raise HostNetworkError() from err
# Create new configuration and activate interface
elif inet and interface.enabled:
settings = interface_update_payload(interface)
try:
await self.sys_dbus.network.add_and_activate_connection(
settings, inet.object_path
)
except DBusError as err:
_LOGGER.error("Can't create config and activate %s", interface.name)
raise HostNetworkError() from err
# Remove config from interface
elif inet and not interface.enabled:
try:
await inet.settings.delete()
except DBusError as err:
_LOGGER.error("Can't remove %s", interface.name)
raise HostNetworkError() from err
# Create new interface (like vlan)
elif not inet:
settings = interface_update_payload(interface)
try:
await self.sys_dbus.network.settings.add_connection(settings)
except DBusError as err:
_LOGGER.error("Can't create new interface")
raise HostNetworkError() from err
await self.update()
async def scan_wifi(self, interface: Interface) -> List[AccessPoint]:
"""Scan on Interface for AccessPoint."""
inet = self.sys_dbus.network.interfaces.get(interface.name)
if inet.type != DeviceType.WIRELESS:
_LOGGER.error("Can only scan with wireless card - %s", interface.name)
raise HostNotSupportedError()
await inet.wireless.request_scan()
await asyncio.sleep(5)
accesspoints: List[AccessPoint] = []
for ap_object in (await inet.wireless.get_all_accesspoints())[0]:
accesspoint = NetworkWirelessAP(ap_object)
try:
await accesspoint.connect()
except DBusError as err:
_LOGGER.waring("Can't process an AP: %s", err)
continue
else:
accesspoints.append(
AccessPoint(
WifiMode[WirelessMethodType(accesspoint.mode).name],
accesspoint.ssid,
accesspoint.mac,
accesspoint.frequency,
accesspoint.strength,
)
)
return accesspoints
@attr.s(slots=True)
class AccessPoint:
"""Represent a wifi configuration."""
mode: WifiMode = attr.ib()
ssid: str = attr.ib()
mac: str = attr.ib()
frequency: int = attr.ib()
signal: int = attr.ib()
@attr.s(slots=True)
class IpConfig:
"""Represent a IP configuration."""
method: InterfaceMethod = attr.ib()
address: List[Union[IPv4Interface, IPv6Interface]] = attr.ib()
gateway: Optional[Union[IPv4Address, IPv6Address]] = attr.ib()
nameservers: List[Union[IPv4Address, IPv6Address]] = attr.ib()
@attr.s(slots=True)
class WifiConfig:
"""Represent a wifi configuration."""
mode: WifiMode = attr.ib()
ssid: str = attr.ib()
auth: AuthMethod = attr.ib()
psk: Optional[str] = attr.ib()
signal: Optional[int] = attr.ib()
@attr.s(slots=True)
class VlanConfig:
"""Represent a vlan configuration."""
id: int = attr.ib()
interface: str = attr.ib()
@attr.s(slots=True)
class Interface:
"""Represent a host network interface."""
name: str = attr.ib()
enabled: bool = attr.ib()
connected: bool = attr.ib()
primary: bool = attr.ib()
type: InterfaceType = attr.ib()
ipv4: Optional[IpConfig] = attr.ib()
ipv6: Optional[IpConfig] = attr.ib()
wifi: Optional[WifiConfig] = attr.ib()
vlan: Optional[VlanConfig] = attr.ib()
@staticmethod
def from_dbus_interface(inet: NetworkInterface) -> Interface:
"""Concert a dbus interface into normal Interface."""
return Interface(
inet.name,
inet.settings is not None,
Interface._map_nm_connected(inet.connection),
inet.primary,
Interface._map_nm_type(inet.type),
IpConfig(
Interface._map_nm_method(inet.settings.ipv4.method),
inet.connection.ipv4.address,
inet.connection.ipv4.gateway,
inet.connection.ipv4.nameservers,
)
if inet.connection and inet.connection.ipv4
else None,
IpConfig(
Interface._map_nm_method(inet.settings.ipv6.method),
inet.connection.ipv6.address,
inet.connection.ipv6.gateway,
inet.connection.ipv6.nameservers,
)
if inet.connection and inet.connection.ipv6
else None,
Interface._map_nm_wifi(inet),
Interface._map_nm_vlan(inet),
)
@staticmethod
def _map_nm_method(method: str) -> InterfaceMethod:
"""Map IP interface method."""
mapping = {
NMInterfaceMethod.AUTO: InterfaceMethod.DHCP,
NMInterfaceMethod.DISABLED: InterfaceMethod.DISABLED,
NMInterfaceMethod.MANUAL: InterfaceMethod.STATIC,
}
return mapping.get(method, InterfaceMethod.DISABLED)
@staticmethod
def _map_nm_connected(connection: Optional[NetworkConnection]) -> bool:
"""Map connectivity state."""
if not connection:
return False
return connection.state in (
ConnectionStateType.ACTIVATED,
ConnectionStateType.ACTIVATING,
)
@staticmethod
def _map_nm_type(device_type: int) -> InterfaceType:
mapping = {
DeviceType.ETHERNET: InterfaceType.ETHERNET,
DeviceType.WIRELESS: InterfaceType.WIRELESS,
DeviceType.VLAN: InterfaceType.VLAN,
}
return mapping[device_type]
@staticmethod
def _map_nm_wifi(inet: NetworkInterface) -> Optional[WifiConfig]:
"""Create mapping to nm wifi property."""
if inet.type != DeviceType.WIRELESS or not inet.settings:
return None
# Authentication
if inet.settings.wireless_security.key_mgmt == "none":
auth = AuthMethod.WEB
elif inet.settings.wireless_security.key_mgmt == "wpa-psk":
auth = AuthMethod.WPA_PSK
else:
auth = AuthMethod.OPEN
# Signal
if inet.wireless:
signal = inet.wireless.active.strength
else:
signal = None
return WifiConfig(
WifiMode[WirelessMethodType(inet.settings.wireless.mode).name],
inet.settings.wireless.ssid,
auth,
inet.settings.wireless_security.psk,
signal,
)
@staticmethod
def _map_nm_vlan(inet: NetworkInterface) -> Optional[WifiConfig]:
"""Create mapping to nm vlan property."""
if inet.type != DeviceType.VLAN or not inet.settings:
return None
return VlanConfig(inet.settings.vlan.id, inet.settings.vlan.parent)

View File

@ -70,12 +70,11 @@ class CoreDNS(JsonConfig, CoreSysAttributes):
def locals(self) -> List[str]: def locals(self) -> List[str]:
"""Return list of local system DNS servers.""" """Return list of local system DNS servers."""
servers: List[str] = [] servers: List[str] = []
for server in self.sys_host.network.dns_servers: for server in [
if server in servers: f"dns://{server!s}" for server in self.sys_host.network.dns_servers
continue ]:
with suppress(vol.Invalid): with suppress(vol.Invalid):
dns_url(server) servers.append(dns_url(server))
servers.append(server)
return servers return servers

View File

@ -1,9 +1,11 @@
"""Test NetworkInterface API.""" """Test NetwrokInterface API."""
from unittest.mock import AsyncMock, patch
import pytest import pytest
from supervisor.const import DOCKER_NETWORK, DOCKER_NETWORK_MASK from supervisor.const import DOCKER_NETWORK, DOCKER_NETWORK_MASK
from tests.const import TEST_INTERFACE from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN
@pytest.mark.asyncio @pytest.mark.asyncio
@ -11,7 +13,9 @@ async def test_api_network_info(api_client, coresys):
"""Test network manager api.""" """Test network manager api."""
resp = await api_client.get("/network/info") resp = await api_client.get("/network/info")
result = await resp.json() result = await resp.json()
assert TEST_INTERFACE in result["data"]["interfaces"] assert TEST_INTERFACE in (
inet["interface"] for inet in result["data"]["interfaces"]
)
assert result["data"]["docker"]["interface"] == DOCKER_NETWORK assert result["data"]["docker"]["interface"] == DOCKER_NETWORK
assert result["data"]["docker"]["address"] == str(DOCKER_NETWORK_MASK) assert result["data"]["docker"]["address"] == str(DOCKER_NETWORK_MASK)
@ -24,7 +28,21 @@ async def test_api_network_interface_info(api_client):
"""Test network manager api.""" """Test network manager api."""
resp = await api_client.get(f"/network/interface/{TEST_INTERFACE}/info") resp = await api_client.get(f"/network/interface/{TEST_INTERFACE}/info")
result = await resp.json() result = await resp.json()
assert result["data"]["ip_address"] == "192.168.2.148/24" assert result["data"]["ipv4"]["address"][-1] == "192.168.2.148/24"
assert result["data"]["ipv4"]["gateway"] == "192.168.2.1"
assert result["data"]["ipv4"]["nameservers"] == ["192.168.2.2"]
assert (
result["data"]["ipv6"]["address"][0] == "2a03:169:3df5:0:6be9:2588:b26a:a679/64"
)
assert (
result["data"]["ipv6"]["address"][1]
== "fd14:949b:c9cc:0:522b:8108:8ff8:cca3/64"
)
assert result["data"]["ipv6"]["gateway"] == "fe80::da58:d7ff:fe00:9c69"
assert result["data"]["ipv6"]["nameservers"] == [
"2001:1620:2777:1::10",
"2001:1620:2777:2::20",
]
assert result["data"]["interface"] == TEST_INTERFACE assert result["data"]["interface"] == TEST_INTERFACE
@ -33,7 +51,21 @@ async def test_api_network_interface_info_default(api_client):
"""Test network manager default api.""" """Test network manager default api."""
resp = await api_client.get("/network/interface/default/info") resp = await api_client.get("/network/interface/default/info")
result = await resp.json() result = await resp.json()
assert result["data"]["ip_address"] == "192.168.2.148/24" assert result["data"]["ipv4"]["address"][-1] == "192.168.2.148/24"
assert result["data"]["ipv4"]["gateway"] == "192.168.2.1"
assert result["data"]["ipv4"]["nameservers"] == ["192.168.2.2"]
assert (
result["data"]["ipv6"]["address"][0] == "2a03:169:3df5:0:6be9:2588:b26a:a679/64"
)
assert (
result["data"]["ipv6"]["address"][1]
== "fd14:949b:c9cc:0:522b:8108:8ff8:cca3/64"
)
assert result["data"]["ipv6"]["gateway"] == "fe80::da58:d7ff:fe00:9c69"
assert result["data"]["ipv6"]["nameservers"] == [
"2001:1620:2777:1::10",
"2001:1620:2777:2::20",
]
assert result["data"]["interface"] == TEST_INTERFACE assert result["data"]["interface"] == TEST_INTERFACE
@ -42,7 +74,14 @@ async def test_api_network_interface_update(api_client):
"""Test network manager api.""" """Test network manager api."""
resp = await api_client.post( resp = await api_client.post(
f"/network/interface/{TEST_INTERFACE}/update", f"/network/interface/{TEST_INTERFACE}/update",
json={"method": "static", "dns": ["1.1.1.1"], "address": "192.168.2.148/24"}, json={
"ipv4": {
"method": "static",
"nameservers": ["1.1.1.1"],
"address": ["192.168.2.148/24"],
"gateway": "192.168.1.1",
}
},
) )
result = await resp.json() result = await resp.json()
assert result["result"] == "ok" assert result["result"] == "ok"
@ -53,7 +92,9 @@ async def test_api_network_interface_info_invalid(api_client):
"""Test network manager api.""" """Test network manager api."""
resp = await api_client.get("/network/interface/invalid/info") resp = await api_client.get("/network/interface/invalid/info")
result = await resp.json() result = await resp.json()
assert not result["data"]
assert result["message"]
assert result["result"] == "error"
@pytest.mark.asyncio @pytest.mark.asyncio
@ -68,10 +109,26 @@ async def test_api_network_interface_update_invalid(api_client):
assert result["message"] == "You need to supply at least one option to update" assert result["message"] == "You need to supply at least one option to update"
resp = await api_client.post( resp = await api_client.post(
f"/network/interface/{TEST_INTERFACE}/update", json={"dns": "1.1.1.1"} f"/network/interface/{TEST_INTERFACE}/update",
json={"ipv4": {"nameservers": "1.1.1.1"}},
) )
result = await resp.json() result = await resp.json()
assert ( assert (
result["message"] result["message"]
== "expected a list for dictionary value @ data['dns']. Got '1.1.1.1'" == "expected a list for dictionary value @ data['ipv4']['nameservers']. Got '1.1.1.1'"
) )
@pytest.mark.asyncio
async def test_api_network_wireless_scan(api_client):
"""Test network manager api."""
with patch("asyncio.sleep", return_value=AsyncMock()):
resp = await api_client.get(
f"/network/interface/{TEST_INTERFACE_WLAN}/accesspoints"
)
result = await resp.json()
assert ["UPC4814466", "VQ@35(55720"] == [
ap["ssid"] for ap in result["data"]["accesspoints"]
]
assert [47, 63] == [ap["signal"] for ap in result["data"]["accesspoints"]]

View File

@ -13,3 +13,9 @@ def load_fixture(filename: str) -> str:
"""Load a fixture.""" """Load a fixture."""
path = Path(Path(__file__).parent.joinpath("fixtures"), filename) path = Path(Path(__file__).parent.joinpath("fixtures"), filename)
return path.read_text() return path.read_text()
def exists_fixture(filename: str) -> bool:
"""Check if a fixture exists."""
path = Path(Path(__file__).parent.joinpath("fixtures"), filename)
return path.exists()

View File

@ -1,5 +1,6 @@
"""Common test functions.""" """Common test functions."""
from pathlib import Path from pathlib import Path
import re
from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch
from uuid import uuid4 from uuid import uuid4
@ -10,13 +11,11 @@ import pytest
from supervisor.api import RestAPI from supervisor.api import RestAPI
from supervisor.bootstrap import initialize_coresys from supervisor.bootstrap import initialize_coresys
from supervisor.coresys import CoreSys from supervisor.coresys import CoreSys
from supervisor.dbus.const import DBUS_NAME_NM, DBUS_OBJECT_BASE
from supervisor.dbus.network import NetworkManager from supervisor.dbus.network import NetworkManager
from supervisor.dbus.network.interface import NetworkInterface
from supervisor.docker import DockerAPI from supervisor.docker import DockerAPI
from supervisor.utils.gdbus import DBus from supervisor.utils.gdbus import DBus
from tests.common import load_fixture, load_json_fixture from tests.common import exists_fixture, load_fixture, load_json_fixture
# pylint: disable=redefined-outer-name, protected-access # pylint: disable=redefined-outer-name, protected-access
@ -53,38 +52,53 @@ def docker() -> DockerAPI:
def dbus() -> DBus: def dbus() -> DBus:
"""Mock DBUS.""" """Mock DBUS."""
async def mock_get_properties(_, interface): dbus_commands = []
return load_json_fixture(f"{interface.replace('.', '_')}.json")
async def mock_get_properties(dbus_obj, interface):
latest = dbus_obj.object_path.split("/")[-1]
fixture = interface.replace(".", "_")
if latest.isnumeric():
fixture = f"{fixture}_{latest}"
return load_json_fixture(f"{fixture}.json")
async def mock_send(_, command, silent=False): async def mock_send(_, command, silent=False):
if silent: if silent:
return "" return ""
filetype = "xml" if "--xml" in command else "fixture" fixture = command[6].replace("/", "_")[1:]
fixture = f"{command[6].replace('/', '_')[1:]}.{filetype}" if command[1] == "introspect":
return load_fixture(fixture) filetype = "xml"
if not exists_fixture(f"{fixture}.{filetype}"):
fixture = re.sub(r"_[0-9]+$", "", fixture)
# special case
if exists_fixture(f"{fixture}_*.{filetype}"):
fixture = f"{fixture}_*"
else:
fixture = f"{fixture}-{command[10].split('.')[-1]}"
filetype = "fixture"
dbus_commands.append(fixture)
return load_fixture(f"{fixture}.{filetype}")
with patch("supervisor.utils.gdbus.DBus._send", new=mock_send), patch( with patch("supervisor.utils.gdbus.DBus._send", new=mock_send), patch(
"supervisor.dbus.interface.DBusInterface.is_connected", "supervisor.dbus.interface.DBusInterface.is_connected",
return_value=True, return_value=True,
), patch("supervisor.utils.gdbus.DBus.get_properties", new=mock_get_properties): ), patch("supervisor.utils.gdbus.DBus.get_properties", new=mock_get_properties):
yield dbus_commands
dbus_obj = DBus(DBUS_NAME_NM, DBUS_OBJECT_BASE)
yield dbus_obj
@pytest.fixture @pytest.fixture
async def network_manager(dbus) -> NetworkManager: async def network_manager(dbus) -> NetworkManager:
"""Mock NetworkManager.""" """Mock NetworkManager."""
nm_obj = NetworkManager()
async def dns_update():
pass
with patch("supervisor.dbus.network.NetworkManager.dns", return_value=MagicMock()):
nm_obj = NetworkManager()
nm_obj.dns.update = dns_update
nm_obj.dbus = dbus nm_obj.dbus = dbus
# Init
await nm_obj.connect() await nm_obj.connect()
await nm_obj.update() await nm_obj.update()
@ -92,7 +106,7 @@ async def network_manager(dbus) -> NetworkManager:
@pytest.fixture @pytest.fixture
async def coresys(loop, docker, dbus, network_manager, aiohttp_client) -> CoreSys: async def coresys(loop, docker, network_manager, aiohttp_client) -> CoreSys:
"""Create a CoreSys Mock.""" """Create a CoreSys Mock."""
with patch("supervisor.bootstrap.initialize_system_data"), patch( with patch("supervisor.bootstrap.initialize_system_data"), patch(
"supervisor.bootstrap.setup_diagnostics" "supervisor.bootstrap.setup_diagnostics"
@ -106,10 +120,10 @@ async def coresys(loop, docker, dbus, network_manager, aiohttp_client) -> CoreSy
coresys_obj = await initialize_coresys() coresys_obj = await initialize_coresys()
# Mock save json # Mock save json
coresys_obj.ingress.save_data = MagicMock() coresys_obj._ingress.save_data = MagicMock()
coresys_obj.auth.save_data = MagicMock() coresys_obj._auth.save_data = MagicMock()
coresys_obj.updater.save_data = MagicMock() coresys_obj._updater.save_data = MagicMock()
coresys_obj.config.save_data = MagicMock() coresys_obj._config.save_data = MagicMock()
# Mock test client # Mock test client
coresys_obj.arch._default_arch = "amd64" coresys_obj.arch._default_arch = "amd64"
@ -117,8 +131,7 @@ async def coresys(loop, docker, dbus, network_manager, aiohttp_client) -> CoreSy
coresys_obj._machine_id = uuid4() coresys_obj._machine_id = uuid4()
# Mock host communication # Mock host communication
coresys_obj._dbus = dbus coresys_obj._dbus._network = network_manager
coresys_obj._dbus.network = network_manager
# Mock docker # Mock docker
coresys_obj._docker = docker coresys_obj._docker = docker
@ -153,15 +166,6 @@ async def api_client(aiohttp_client, coresys: CoreSys):
yield await aiohttp_client(api.webapp) yield await aiohttp_client(api.webapp)
@pytest.fixture
async def network_interface(dbus):
"""Fixture for a network interface."""
interface = NetworkInterface()
await interface.connect(dbus, "/org/freedesktop/NetworkManager/ActiveConnection/1")
await interface.connection.update_information()
yield interface
@pytest.fixture @pytest.fixture
def store_manager(coresys: CoreSys): def store_manager(coresys: CoreSys):
"""Fixture for the store manager.""" """Fixture for the store manager."""

View File

@ -1,3 +1,4 @@
"""Consts for tests.""" """Consts for tests."""
TEST_INTERFACE = "eth0" TEST_INTERFACE = "eth0"
TEST_INTERFACE_WLAN = "wlan0"

View File

@ -1,15 +1,49 @@
"""Test NetwrokInterface.""" """Test NetwrokInterface."""
from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
import pytest import pytest
from supervisor.dbus.const import DeviceType, InterfaceMethod
from supervisor.dbus.network import NetworkManager from supervisor.dbus.network import NetworkManager
from tests.const import TEST_INTERFACE from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_network_interface(network_manager: NetworkManager): async def test_network_interface_ethernet(network_manager: NetworkManager):
"""Test network interface.""" """Test network interface."""
interface = network_manager.interfaces[TEST_INTERFACE] interface = network_manager.interfaces[TEST_INTERFACE]
assert interface.name == TEST_INTERFACE assert interface.name == TEST_INTERFACE
assert interface.type == DeviceType.ETHERNET
assert interface.connection.state == 2 assert interface.connection.state == 2
assert interface.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" assert interface.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6"
assert interface.connection.ipv4.address == [IPv4Interface("192.168.2.148/24")]
assert interface.connection.ipv6.address == [
IPv6Interface("2a03:169:3df5:0:6be9:2588:b26a:a679/64"),
IPv6Interface("fd14:949b:c9cc:0:522b:8108:8ff8:cca3/64"),
IPv6Interface("2a03:169:3df5::2f1/128"),
IPv6Interface("fd14:949b:c9cc::2f1/128"),
IPv6Interface("fe80::ffe3:319e:c630:9f51/64"),
]
assert interface.connection.ipv4.gateway == IPv4Address("192.168.2.1")
assert interface.connection.ipv6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69")
assert interface.connection.ipv4.nameservers == [IPv4Address("192.168.2.2")]
assert interface.connection.ipv6.nameservers == [
IPv6Address("2001:1620:2777:1::10"),
IPv6Address("2001:1620:2777:2::20"),
]
assert interface.settings.ipv4.method == InterfaceMethod.AUTO
assert interface.settings.ipv6.method == InterfaceMethod.AUTO
assert interface.settings.connection.id == "Wired connection 1"
@pytest.mark.asyncio
async def test_network_interface_wlan(network_manager: NetworkManager):
"""Test network interface."""
interface = network_manager.interfaces[TEST_INTERFACE_WLAN]
assert interface.name == TEST_INTERFACE_WLAN
assert interface.type == DeviceType.WIRELESS

View File

@ -1,12 +0,0 @@
"""Test network utils."""
from supervisor.dbus.network.utils import int2ip, ip2int
def test_int2ip():
"""Test int2ip."""
assert int2ip(16885952) == "192.168.1.1"
def test_ip2int():
"""Test ip2int."""
assert ip2int("192.168.1.1") == 16885952

View File

@ -1,39 +1,180 @@
"""Test interface update payload.""" """Test interface update payload."""
from ipaddress import ip_address, ip_interface
import pytest import pytest
from supervisor.dbus.const import ConnectionType
from supervisor.dbus.payloads.generate import interface_update_payload from supervisor.dbus.payloads.generate import interface_update_payload
from supervisor.host.const import AuthMethod, InterfaceMethod, InterfaceType, WifiMode
from supervisor.host.network import VlanConfig, WifiConfig
from supervisor.utils.gdbus import DBus from supervisor.utils.gdbus import DBus
from tests.const import TEST_INTERFACE
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_interface_update_payload_ethernet(network_interface): async def test_interface_update_payload_ethernet(coresys):
"""Test interface update payload.""" """Test interface update payload."""
data = interface_update_payload(network_interface, **{"method": "auto"}) interface = coresys.host.network.get(TEST_INTERFACE)
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "auto"
data = interface_update_payload(interface)
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "auto"
assert DBus.parse_gvariant(data)["ipv6"]["method"] == "auto"
data = interface_update_payload(
network_interface, **{"address": "1.1.1.1", "dns": ["1.1.1.1", "1.0.0.1"]}
)
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "manual"
assert DBus.parse_gvariant(data)["ipv4"]["address-data"][0]["address"] == "1.1.1.1"
assert DBus.parse_gvariant(data)["ipv4"]["dns"] == [16843009, 16777217]
assert ( assert (
DBus.parse_gvariant(data)["connection"]["uuid"] DBus.parse_gvariant(data)["802-3-ethernet"]["assigned-mac-address"] == "stable"
== "0c23631e-2118-355c-bbb0-8943229cb0d6"
) )
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_interface_update_payload_wireless(network_interface): async def test_interface_update_payload_ethernet_ipv4(coresys):
"""Test interface update payload.""" """Test interface update payload."""
network_interface.connection._properties["Type"] = ConnectionType.WIRELESS interface = coresys.host.network.get(TEST_INTERFACE)
data = interface_update_payload(network_interface, **{"method": "auto"}) inet = coresys.dbus.network.interfaces[TEST_INTERFACE]
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "auto"
interface.ipv4.method = InterfaceMethod.STATIC
interface.ipv4.address = [ip_interface("192.168.1.1/24")]
interface.ipv4.nameservers = [ip_address("1.1.1.1"), ip_address("1.0.1.1")]
interface.ipv4.gateway = ip_address("192.168.1.1")
data = interface_update_payload( data = interface_update_payload(
network_interface, **{"address": "1.1.1.1", "dns": ["1.1.1.1", "1.0.0.1"]} interface,
name=inet.settings.connection.id,
uuid=inet.settings.connection.uuid,
) )
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "manual" assert DBus.parse_gvariant(data)["ipv4"]["method"] == "manual"
assert DBus.parse_gvariant(data)["ipv4"]["address-data"][0]["address"] == "1.1.1.1" assert (
assert DBus.parse_gvariant(data)["802-11-wireless"]["ssid"] == [78, 69, 84, 84] DBus.parse_gvariant(data)["ipv4"]["address-data"][0]["address"] == "192.168.1.1"
)
assert DBus.parse_gvariant(data)["ipv4"]["address-data"][0]["prefix"] == 24
assert DBus.parse_gvariant(data)["ipv4"]["dns"] == [16843009, 16777473]
assert (
DBus.parse_gvariant(data)["connection"]["uuid"] == inet.settings.connection.uuid
)
assert DBus.parse_gvariant(data)["connection"]["id"] == inet.settings.connection.id
assert DBus.parse_gvariant(data)["connection"]["type"] == "802-3-ethernet"
assert DBus.parse_gvariant(data)["connection"]["interface-name"] == interface.name
assert DBus.parse_gvariant(data)["ipv4"]["gateway"] == "192.168.1.1"
@pytest.mark.asyncio
async def test_interface_update_payload_ethernet_ipv6(coresys):
"""Test interface update payload."""
interface = coresys.host.network.get(TEST_INTERFACE)
inet = coresys.dbus.network.interfaces[TEST_INTERFACE]
interface.ipv6.method = InterfaceMethod.STATIC
interface.ipv6.address = [ip_interface("2a03:169:3df5:0:6be9:2588:b26a:a679/64")]
interface.ipv6.nameservers = [
ip_address("2606:4700:4700::64"),
ip_address("2606:4700:4700::6400"),
]
interface.ipv6.gateway = ip_address("fe80::da58:d7ff:fe00:9c69")
data = interface_update_payload(
interface,
name=inet.settings.connection.id,
uuid=inet.settings.connection.uuid,
)
assert DBus.parse_gvariant(data)["ipv6"]["method"] == "manual"
assert (
DBus.parse_gvariant(data)["ipv6"]["address-data"][0]["address"]
== "2a03:169:3df5:0:6be9:2588:b26a:a679"
)
assert DBus.parse_gvariant(data)["ipv6"]["address-data"][0]["prefix"] == 64
assert DBus.parse_gvariant(data)["ipv6"]["dns"] == [
50543257694033307102031451402929176676,
50543257694033307102031451402929202176,
]
assert (
DBus.parse_gvariant(data)["connection"]["uuid"] == inet.settings.connection.uuid
)
assert DBus.parse_gvariant(data)["connection"]["id"] == inet.settings.connection.id
assert DBus.parse_gvariant(data)["connection"]["type"] == "802-3-ethernet"
assert DBus.parse_gvariant(data)["connection"]["interface-name"] == interface.name
assert DBus.parse_gvariant(data)["ipv6"]["gateway"] == "fe80::da58:d7ff:fe00:9c69"
@pytest.mark.asyncio
async def test_interface_update_payload_wireless_wpa_psk(coresys):
"""Test interface update payload."""
interface = coresys.host.network.get(TEST_INTERFACE)
interface.type = InterfaceType.WIRELESS
interface.wifi = WifiConfig(
WifiMode.INFRASTRUCTURE, "Test", AuthMethod.WPA_PSK, "password", 0
)
data = interface_update_payload(interface)
assert DBus.parse_gvariant(data)["connection"]["type"] == "802-11-wireless"
assert DBus.parse_gvariant(data)["802-11-wireless"]["ssid"] == [84, 101, 115, 116]
assert DBus.parse_gvariant(data)["802-11-wireless"]["mode"] == "infrastructure"
assert DBus.parse_gvariant(data)["802-11-wireless-security"]["auth-alg"] == "shared"
assert (
DBus.parse_gvariant(data)["802-11-wireless-security"]["key-mgmt"] == "wpa-psk"
)
assert DBus.parse_gvariant(data)["802-11-wireless-security"]["psk"] == "password"
@pytest.mark.asyncio
async def test_interface_update_payload_wireless_web(coresys):
"""Test interface update payload."""
interface = coresys.host.network.get(TEST_INTERFACE)
interface.type = InterfaceType.WIRELESS
interface.wifi = WifiConfig(
WifiMode.INFRASTRUCTURE, "Test", AuthMethod.WEB, "password", 0
)
data = interface_update_payload(interface)
assert DBus.parse_gvariant(data)["connection"]["type"] == "802-11-wireless"
assert DBus.parse_gvariant(data)["802-11-wireless"]["ssid"] == [84, 101, 115, 116]
assert DBus.parse_gvariant(data)["802-11-wireless"]["mode"] == "infrastructure"
assert DBus.parse_gvariant(data)["802-11-wireless-security"]["auth-alg"] == "none"
assert DBus.parse_gvariant(data)["802-11-wireless-security"]["key-mgmt"] == "none"
assert DBus.parse_gvariant(data)["802-11-wireless-security"]["psk"] == "password"
@pytest.mark.asyncio
async def test_interface_update_payload_wireless_open(coresys):
"""Test interface update payload."""
interface = coresys.host.network.get(TEST_INTERFACE)
interface.type = InterfaceType.WIRELESS
interface.wifi = WifiConfig(
WifiMode.INFRASTRUCTURE, "Test", AuthMethod.OPEN, None, 0
)
data = interface_update_payload(interface)
assert DBus.parse_gvariant(data)["connection"]["type"] == "802-11-wireless"
assert DBus.parse_gvariant(data)["802-11-wireless"]["ssid"] == [84, 101, 115, 116]
assert DBus.parse_gvariant(data)["802-11-wireless"]["mode"] == "infrastructure"
assert (
DBus.parse_gvariant(data)["802-11-wireless"]["assigned-mac-address"] == "stable"
)
assert DBus.parse_gvariant(data)["802-11-wireless-security"]["auth-alg"] == "open"
assert DBus.parse_gvariant(data)["802-11-wireless-security"]["key-mgmt"] == "none"
assert "psk" not in DBus.parse_gvariant(data)["802-11-wireless-security"]
@pytest.mark.asyncio
async def test_interface_update_payload_vlan(coresys):
"""Test interface update payload."""
interface = coresys.host.network.get(TEST_INTERFACE)
interface.type = InterfaceType.VLAN
interface.vlan = VlanConfig(10, interface.name)
data = interface_update_payload(interface)
assert DBus.parse_gvariant(data)["ipv4"]["method"] == "auto"
assert DBus.parse_gvariant(data)["ipv6"]["method"] == "auto"
assert DBus.parse_gvariant(data)["vlan"]["id"] == 10
assert DBus.parse_gvariant(data)["vlan"]["parent"] == interface.name
assert DBus.parse_gvariant(data)["connection"]["type"] == "vlan"
assert "interface-name" not in DBus.parse_gvariant(data)["connection"]

View File

@ -1 +1 @@
({'connection': {'id': <'Wired connection 1'>, 'permissions': <@as []>, 'timestamp': <uint64 1598125548>, 'type': <'802-3-ethernet'>, 'uuid': <'0c23631e-2118-355c-bbb0-8943229cb0d6'>}, 'ipv4': {'address-data': <[{'address': <'192.168.2.148'>, 'prefix': <uint32 24>}]>, 'addresses': <[[uint32 2483202240, 24, 16951488]]>, 'dns': <[uint32 16951488]>, 'dns-search': <@as []>, 'gateway': <'192.168.2.1'>, 'method': <'auto'>, 'route-data': <@aa{sv} []>, 'routes': <@aau []>}, 'ipv6': {'address-data': <@aa{sv} []>, 'addresses': <@a(ayuay) []>, 'dns': <@aay []>, 'dns-search': <@as []>, 'method': <'auto'>, 'route-data': <@aa{sv} []>, 'routes': <@a(ayuayu) []>}, 'proxy': {}, '802-3-ethernet': {'auto-negotiate': <false>, 'mac-address-blacklist': <@as []>, 's390-options': <@a{ss} {}>}, '802-11-wireless': {'ssid': <[byte 0x4e, 0x45, 0x54, 0x54]>}},) ()

View File

@ -1,8 +1,12 @@
{ {
"Devices": ["/org/freedesktop/NetworkManager/Devices/1"], "Devices": [
"/org/freedesktop/NetworkManager/Devices/1",
"/org/freedesktop/NetworkManager/Devices/3"
],
"AllDevices": [ "AllDevices": [
"/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/1",
"/org/freedesktop/NetworkManager/Devices/2" "/org/freedesktop/NetworkManager/Devices/2",
"/org/freedesktop/NetworkManager/Devices/3"
], ],
"Checkpoints": [], "Checkpoints": [],
"NetworkingEnabled": true, "NetworkingEnabled": true,

View File

@ -0,0 +1,52 @@
<!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN"
"http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
<!-- GDBus 2.62.5 -->
<node>
<interface name="org.freedesktop.DBus.Properties">
<method name="Get">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="out"/>
</method>
<method name="GetAll">
<arg type="s" name="interface_name" direction="in"/>
<arg type="a{sv}" name="properties" direction="out"/>
</method>
<method name="Set">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="in"/>
</method>
<signal name="PropertiesChanged">
<arg type="s" name="interface_name"/>
<arg type="a{sv}" name="changed_properties"/>
<arg type="as" name="invalidated_properties"/>
</signal>
</interface>
<interface name="org.freedesktop.DBus.Introspectable">
<method name="Introspect">
<arg type="s" name="xml_data" direction="out"/>
</method>
</interface>
<interface name="org.freedesktop.DBus.Peer">
<method name="Ping"/>
<method name="GetMachineId">
<arg type="s" name="machine_uuid" direction="out"/>
</method>
</interface>
<interface name="org.freedesktop.NetworkManager.AccessPoint">
<signal name="PropertiesChanged">
<arg type="a{sv}" name="properties"/>
</signal>
<property type="u" name="Flags" access="read"/>
<property type="u" name="WpaFlags" access="read"/>
<property type="u" name="RsnFlags" access="read"/>
<property type="ay" name="Ssid" access="read"/>
<property type="u" name="Frequency" access="read"/>
<property type="s" name="HwAddress" access="read"/>
<property type="u" name="Mode" access="read"/>
<property type="u" name="MaxBitrate" access="read"/>
<property type="y" name="Strength" access="read"/>
<property type="i" name="LastSeen" access="read"/>
</interface>
</node>

View File

@ -0,0 +1,12 @@
{
"Flags": 3,
"WpaFlags": 0,
"RsnFlags": 392,
"Ssid": [85, 80, 67, 52, 56, 49, 52, 52, 54, 54],
"Frequency": 2462,
"HwAddress": "E4:57:40:A9:D7:DE",
"Mode": 2,
"MaxBitrate": 195000,
"Strength": 47,
"LastSeen": 1398776
}

View File

@ -0,0 +1,12 @@
{
"Flags": 1,
"WpaFlags": 0,
"RsnFlags": 392,
"Ssid": [86, 81, 64, 51, 53, 40, 53, 53, 55, 50, 48],
"Frequency": 5660,
"HwAddress": "18:4B:0D:23:A1:9C",
"Mode": 2,
"MaxBitrate": 540000,
"Strength": 63,
"LastSeen": 1398839
}

View File

@ -0,0 +1,30 @@
{
"Udi": "/sys/devices/platform/soc/fe300000.mmcnr/mmc_host/mmc1/mmc1:0001/mmc1:0001:1/net/wlan0",
"Interface": "wlan0",
"IpInterface": "",
"Driver": "brcmfmac",
"DriverVersion": "7.45.154",
"FirmwareVersion": "01-4fbe0b04",
"Capabilities": 1,
"Ip4Address": 0,
"State": 30,
"StateReason": [30, 42],
"ActiveConnection": "/",
"Ip4Config": "/",
"Dhcp4Config": "/",
"Ip6Config": "/",
"Dhcp6Config": "/",
"Managed": true,
"Autoconnect": true,
"FirmwareMissing": false,
"NmPluginMissing": false,
"DeviceType": 2,
"AvailableConnections": [],
"PhysicalPortId": "",
"Mtu": 1500,
"Metered": 0,
"LldpNeighbors": [],
"Real": true,
"Ip4Connectivity": 1,
"Ip6Connectivity": 1
}

View File

@ -0,0 +1,20 @@
{
"HwAddress": "EA:3C:50:4C:B8:82",
"PermHwAddress": "DC:A6:32:02:BA:21",
"Mode": 2,
"Bitrate": 0,
"AccessPoints": [
"/org/freedesktop/NetworkManager/AccessPoint/41533",
"/org/freedesktop/NetworkManager/AccessPoint/41534",
"/org/freedesktop/NetworkManager/AccessPoint/41535",
"/org/freedesktop/NetworkManager/AccessPoint/41536",
"/org/freedesktop/NetworkManager/AccessPoint/41537",
"/org/freedesktop/NetworkManager/AccessPoint/41538",
"/org/freedesktop/NetworkManager/AccessPoint/41539",
"/org/freedesktop/NetworkManager/AccessPoint/41540",
"/org/freedesktop/NetworkManager/AccessPoint/41541"
],
"ActiveAccessPoint": "/",
"WirelessCapabilities": 2047,
"LastScan": 1343924585
}

View File

@ -1,6 +1,6 @@
<!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN" <!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN"
"http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd"> "http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
<!-- GDBus 2.64.3 --> <!-- GDBus 2.62.5 -->
<node> <node>
<interface name="org.freedesktop.DBus.Properties"> <interface name="org.freedesktop.DBus.Properties">
<method name="Get"> <method name="Get">
@ -42,6 +42,34 @@
<property type="t" name="TxBytes" access="read"/> <property type="t" name="TxBytes" access="read"/>
<property type="t" name="RxBytes" access="read"/> <property type="t" name="RxBytes" access="read"/>
</interface> </interface>
<interface name="org.freedesktop.NetworkManager.Device.Wireless">
<method name="GetAccessPoints">
<arg type="ao" name="access_points" direction="out"/>
</method>
<method name="GetAllAccessPoints">
<arg type="ao" name="access_points" direction="out"/>
</method>
<method name="RequestScan">
<arg type="a{sv}" name="options" direction="in"/>
</method>
<signal name="PropertiesChanged">
<arg type="a{sv}" name="properties"/>
</signal>
<signal name="AccessPointAdded">
<arg type="o" name="access_point"/>
</signal>
<signal name="AccessPointRemoved">
<arg type="o" name="access_point"/>
</signal>
<property type="s" name="HwAddress" access="read"/>
<property type="s" name="PermHwAddress" access="read"/>
<property type="u" name="Mode" access="read"/>
<property type="u" name="Bitrate" access="read"/>
<property type="ao" name="AccessPoints" access="read"/>
<property type="o" name="ActiveAccessPoint" access="read"/>
<property type="u" name="WirelessCapabilities" access="read"/>
<property type="x" name="LastScan" access="read"/>
</interface>
<interface name="org.freedesktop.NetworkManager.Device"> <interface name="org.freedesktop.NetworkManager.Device">
<method name="Reapply"> <method name="Reapply">
<arg type="a{sa{sv}}" name="connection" direction="in"/> <arg type="a{sa{sv}}" name="connection" direction="in"/>
@ -88,16 +116,5 @@
<property type="b" name="Real" access="read"/> <property type="b" name="Real" access="read"/>
<property type="u" name="Ip4Connectivity" access="read"/> <property type="u" name="Ip4Connectivity" access="read"/>
<property type="u" name="Ip6Connectivity" access="read"/> <property type="u" name="Ip6Connectivity" access="read"/>
<property type="u" name="InterfaceFlags" access="read"/>
</interface>
<interface name="org.freedesktop.NetworkManager.Device.Wired">
<signal name="PropertiesChanged">
<arg type="a{sv}" name="properties"/>
</signal>
<property type="s" name="HwAddress" access="read"/>
<property type="s" name="PermHwAddress" access="read"/>
<property type="u" name="Speed" access="read"/>
<property type="as" name="S390Subchannels" access="read"/>
<property type="b" name="Carrier" access="read"/>
</interface> </interface>
</node> </node>

View File

@ -0,0 +1 @@
([objectpath '/org/freedesktop/NetworkManager/AccessPoint/43099', '/org/freedesktop/NetworkManager/AccessPoint/43100'],)

View File

@ -0,0 +1 @@
()

View File

@ -0,0 +1,13 @@
{
"Mode": "default",
"RcManager": "file",
"Configuration": [
{
"nameservers": ["192.168.30.1"],
"domains": ["syshack.ch"],
"interface": "eth0",
"priority": 100,
"vpn": false
}
]
}

View File

@ -11,7 +11,7 @@
{ "dest": "169.254.0.0", "prefix": 16, "metric": 1000 }, { "dest": "169.254.0.0", "prefix": 16, "metric": 1000 },
{ "dest": "0.0.0.0", "prefix": 0, "next-hop": "192.168.2.1", "metric": 100 } { "dest": "0.0.0.0", "prefix": 0, "next-hop": "192.168.2.1", "metric": 100 }
], ],
"NameserverData": [{ "address": "192.168.2.1" }], "NameserverData": [{ "address": "192.168.2.2" }],
"Nameservers": [16951488], "Nameservers": [16951488],
"Domains": [], "Domains": [],
"Searches": [], "Searches": [],

View File

@ -0,0 +1,52 @@
<!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN"
"http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
<!-- GDBus 2.62.5 -->
<node>
<interface name="org.freedesktop.DBus.Properties">
<method name="Get">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="out"/>
</method>
<method name="GetAll">
<arg type="s" name="interface_name" direction="in"/>
<arg type="a{sv}" name="properties" direction="out"/>
</method>
<method name="Set">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="in"/>
</method>
<signal name="PropertiesChanged">
<arg type="s" name="interface_name"/>
<arg type="a{sv}" name="changed_properties"/>
<arg type="as" name="invalidated_properties"/>
</signal>
</interface>
<interface name="org.freedesktop.DBus.Introspectable">
<method name="Introspect">
<arg type="s" name="xml_data" direction="out"/>
</method>
</interface>
<interface name="org.freedesktop.DBus.Peer">
<method name="Ping"/>
<method name="GetMachineId">
<arg type="s" name="machine_uuid" direction="out"/>
</method>
</interface>
<interface name="org.freedesktop.NetworkManager.IP6Config">
<signal name="PropertiesChanged">
<arg type="a{sv}" name="properties"/>
</signal>
<property type="a(ayuay)" name="Addresses" access="read"/>
<property type="aa{sv}" name="AddressData" access="read"/>
<property type="s" name="Gateway" access="read"/>
<property type="a(ayuayu)" name="Routes" access="read"/>
<property type="aa{sv}" name="RouteData" access="read"/>
<property type="aay" name="Nameservers" access="read"/>
<property type="as" name="Domains" access="read"/>
<property type="as" name="Searches" access="read"/>
<property type="as" name="DnsOptions" access="read"/>
<property type="i" name="DnsPriority" access="read"/>
</interface>
</node>

View File

@ -0,0 +1,115 @@
{
"Addresses": [
[
[42, 3, 1, 105, 61, 245, 0, 0, 107, 233, 37, 136, 178, 106, 166, 121],
64,
[254, 128, 0, 0, 0, 0, 0, 0, 218, 88, 215, 255, 254, 0, 256, 105]
],
[
[253, 20, 148, 255, 201, 204, 0, 0, 82, 43, 129, 8, 143, 248, 204, 163],
64,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
],
[
[42, 3, 1, 105, 61, 245, 0, 0, 0, 0, 0, 0, 0, 0, 2, 241],
128,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
],
[
[253, 20, 148, 255, 201, 204, 0, 0, 0, 0, 0, 0, 0, 0, 2, 241],
128,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
],
[
[254, 128, 0, 0, 0, 0, 0, 0, 255, 227, 49, 158, 198, 48, 159, 81],
64,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
]
],
"AddressData": [
{ "address": "2a03:169:3df5:0:6be9:2588:b26a:a679", "prefix": 64 },
{ "address": "fd14:949b:c9cc:0:522b:8108:8ff8:cca3", "prefix": 64 },
{ "address": "2a03:169:3df5::2f1", "prefix": 128 },
{ "address": "fd14:949b:c9cc::2f1", "prefix": 128 },
{ "address": "fe80::ffe3:319e:c630:9f51", "prefix": 64 }
],
"Gateway": "fe80::da58:d7ff:fe00:9c69",
"Routes": [
[
[253, 20, 148, 255, 201, 204, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
48,
[254, 128, 0, 0, 0, 0, 0, 0, 218, 88, 215, 255, 254, 0, 256, 105],
100
],
[
[42, 3, 1, 105, 61, 245, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
48,
[254, 128, 0, 0, 0, 0, 0, 0, 218, 88, 215, 255, 254, 0, 256, 105],
100
],
[
[253, 20, 148, 255, 201, 204, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
64,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
100
],
[
[42, 3, 1, 105, 61, 245, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
64,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
100
],
[
[42, 3, 1, 105, 61, 245, 0, 0, 0, 0, 0, 0, 0, 0, 2, 241],
128,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
100
],
[
[253, 20, 148, 255, 201, 204, 0, 0, 0, 0, 0, 0, 0, 0, 2, 241],
128,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
100
],
[
[254, 128, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
64,
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
100
]
],
"RouteData": [
{
"dest": "fd14:949b:c9cc::",
"prefix": 48,
"next-hop": "fe80::da58:d7ff:fe00:9c69",
"metric": 100
},
{
"dest": "2a03:169:3df5::",
"prefix": 48,
"next-hop": "fe80::da58:d7ff:fe00:9c69",
"metric": 100
},
{ "dest": "fd14:949b:c9cc::", "prefix": 64, "metric": 100 },
{ "dest": "2a03:169:3df5::", "prefix": 64, "metric": 100 },
{
"dest": "::",
"prefix": 0,
"next-hop": "fe80::da58:d7ff:fe00:9c69",
"metric": 100
},
{ "dest": "2a03:169:3df5::2f1", "prefix": 128, "metric": 100 },
{ "dest": "fd14:949b:c9cc::2f1", "prefix": 128, "metric": 100 },
{ "dest": "fe80::", "prefix": 64, "metric": 100 },
{ "dest": "ff00::", "prefix": 8, "metric": 256, "table": 255 }
],
"Nameservers": [
[32, 1, 22, 32, 39, 119, 0, 1, 0, 0, 0, 0, 0, 0, 0, 16],
[32, 1, 22, 32, 39, 119, 0, 2, 0, 0, 0, 0, 0, 0, 0, 32]
],
"Domains": [],
"Searches": [],
"DnsOptions": [],
"DnsPriority": 100
}

View File

@ -0,0 +1,85 @@
<!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN"
"http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
<!-- GDBus 2.62.5 -->
<node>
<interface name="org.freedesktop.DBus.Properties">
<method name="Get">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="out"/>
</method>
<method name="GetAll">
<arg type="s" name="interface_name" direction="in"/>
<arg type="a{sv}" name="properties" direction="out"/>
</method>
<method name="Set">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="in"/>
</method>
<signal name="PropertiesChanged">
<arg type="s" name="interface_name"/>
<arg type="a{sv}" name="changed_properties"/>
<arg type="as" name="invalidated_properties"/>
</signal>
</interface>
<interface name="org.freedesktop.DBus.Introspectable">
<method name="Introspect">
<arg type="s" name="xml_data" direction="out"/>
</method>
</interface>
<interface name="org.freedesktop.DBus.Peer">
<method name="Ping"/>
<method name="GetMachineId">
<arg type="s" name="machine_uuid" direction="out"/>
</method>
</interface>
<interface name="org.freedesktop.NetworkManager.Settings">
<method name="ListConnections">
<arg type="ao" name="connections" direction="out"/>
</method>
<method name="GetConnectionByUuid">
<arg type="s" name="uuid" direction="in"/>
<arg type="o" name="connection" direction="out"/>
</method>
<method name="AddConnection">
<arg type="a{sa{sv}}" name="connection" direction="in"/>
<arg type="o" name="path" direction="out"/>
</method>
<method name="AddConnectionUnsaved">
<arg type="a{sa{sv}}" name="connection" direction="in"/>
<arg type="o" name="path" direction="out"/>
</method>
<method name="AddConnection2">
<arg type="a{sa{sv}}" name="settings" direction="in"/>
<arg type="u" name="flags" direction="in"/>
<arg type="a{sv}" name="args" direction="in"/>
<arg type="o" name="path" direction="out"/>
<arg type="a{sv}" name="result" direction="out"/>
</method>
<method name="LoadConnections">
<arg type="as" name="filenames" direction="in"/>
<arg type="b" name="status" direction="out"/>
<arg type="as" name="failures" direction="out"/>
</method>
<method name="ReloadConnections">
<arg type="b" name="status" direction="out"/>
</method>
<method name="SaveHostname">
<arg type="s" name="hostname" direction="in"/>
</method>
<signal name="PropertiesChanged">
<arg type="a{sv}" name="properties"/>
</signal>
<signal name="NewConnection">
<arg type="o" name="connection"/>
</signal>
<signal name="ConnectionRemoved">
<arg type="o" name="connection"/>
</signal>
<property type="ao" name="Connections" access="read"/>
<property type="s" name="Hostname" access="read"/>
<property type="b" name="CanModify" access="read"/>
</interface>
<node name="1"/>
</node>

View File

@ -0,0 +1 @@
()

View File

@ -17,24 +17,18 @@ async def test_evaluation(coresys: CoreSys):
assert operating_system.reason not in coresys.resolution.unsupported assert operating_system.reason not in coresys.resolution.unsupported
with patch( coresys.host._info = MagicMock(operating_system="unsupported")
"supervisor.utils.gdbus.DBusCallWrapper", await operating_system()
return_value=MagicMock(hostname=MagicMock(operating_system="unsupported")), assert operating_system.reason in coresys.resolution.unsupported
):
await operating_system()
assert operating_system.reason in coresys.resolution.unsupported
coresys.hassos._available = True coresys.hassos._available = True
await operating_system() await operating_system()
assert operating_system.reason not in coresys.resolution.unsupported assert operating_system.reason not in coresys.resolution.unsupported
coresys.hassos._available = False coresys.hassos._available = False
with patch( coresys.host._info = MagicMock(operating_system=SUPPORTED_OS[0])
"supervisor.utils.gdbus.DBusCallWrapper", await operating_system()
return_value=MagicMock(hostname=MagicMock(operating_system=SUPPORTED_OS[0])), assert operating_system.reason not in coresys.resolution.unsupported
):
await operating_system()
assert operating_system.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys): async def test_did_run(coresys: CoreSys):

View File

@ -2,7 +2,7 @@
# pylint: disable=import-error,protected-access # pylint: disable=import-error,protected-access
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from supervisor.const import CoreState from supervisor.const import CoreState, HostFeature
from supervisor.coresys import CoreSys from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.systemd import EvaluateSystemd from supervisor.resolution.evaluations.systemd import EvaluateSystemd
@ -14,28 +14,19 @@ async def test_evaluation(coresys: CoreSys):
assert systemd.reason not in coresys.resolution.unsupported assert systemd.reason not in coresys.resolution.unsupported
with patch( coresys._host = MagicMock()
"supervisor.utils.gdbus.DBusCallWrapper",
return_value=MagicMock(systemd=MagicMock(is_connected=False)),
):
await systemd()
assert systemd.reason in coresys.resolution.unsupported
with patch( coresys.host.supported_features = [HostFeature.HOSTNAME]
"supervisor.utils.gdbus.DBusCallWrapper", await systemd()
return_value=MagicMock(hostname=MagicMock(is_connected=False)), assert systemd.reason in coresys.resolution.unsupported
):
await systemd()
assert systemd.reason in coresys.resolution.unsupported
with patch( coresys.host.supported_features = [
"supervisor.utils.gdbus.DBusCallWrapper", HostFeature.SERVICES,
return_value=MagicMock( HostFeature.SHUTDOWN,
hostname=MagicMock(is_connected=True), systemd=MagicMock(is_connected=True) HostFeature.REBOOT,
), ]
): await systemd()
await systemd() assert systemd.reason in coresys.resolution.unsupported
assert systemd.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys): async def test_did_run(coresys: CoreSys):