from __future__ import annotations import asyncio import weakref import gi from gi.repository import GLib, GObject gi.require_version("NM", "1.0") from gi.repository import NM from ..model import ( Capabilities, ErrorCode, LinkDetails, LinkState, NetworkError, Security, WifiNetwork, WifiState, connect_blocker, ) from .base import Backend ACTIVATION_TIMEOUT_S = 60 ACTIVATING_STATES = {"prepare", "config", "need_auth", "ip_config", "ip_check", "secondaries"} AP_SECURITY = getattr(NM, "80211ApSecurityFlags") AP_FLAGS = getattr(NM, "80211ApFlags") PASSWORD_REASONS = { NM.ActiveConnectionStateReason.NO_SECRETS, NM.ActiveConnectionStateReason.LOGIN_FAILED, } def security_of(point: NM.AccessPoint) -> Security: wpa, rsn = int(point.get_wpa_flags()), int(point.get_rsn_flags()) if (wpa | rsn) & int(AP_SECURITY.KEY_MGMT_802_1X): return Security.ENTERPRISE if wpa or rsn: return Security.PSK if int(point.get_flags()) & int(AP_FLAGS.PRIVACY): return Security.WEP return Security.OPEN def failure_for(reason, had_passphrase: bool) -> NetworkError: if reason in PASSWORD_REASONS: code = ErrorCode.WRONG_PASSWORD if had_passphrase else ErrorCode.PASSWORD_REQUIRED return NetworkError(code, reason.value_nick) if reason == NM.ActiveConnectionStateReason.CONNECT_TIMEOUT: return NetworkError(ErrorCode.TIMEOUT, reason.value_nick) if reason in ( NM.ActiveConnectionStateReason.DEVICE_DISCONNECTED, NM.ActiveConnectionStateReason.USER_DISCONNECTED, ): return NetworkError(ErrorCode.CANCELLED, reason.value_nick) return NetworkError(ErrorCode.FAILED, reason.value_nick) class NetworkManagerBackend(Backend): name = "NetworkManager" capabilities = Capabilities(forget=True, auto_connect=False, link_details=True) def __init__(self, service): super().__init__() self._wifi = service.wifi self._busy: tuple[str, LinkState] | None = None self._scanning = False self._hooked = weakref.WeakSet() self._wifi.connect("notify::enabled", self._on_change) self._wifi.connect("new-device", lambda _wifi, device: self._hook_device(device)) for device in self._wifi.devices: self._hook_device(device) def _on_change(self, *_): self.notify_changed() def _hook_device(self, device): if device in self._hooked: return self._hooked.add(device) device.connect("new-access-point", lambda _device, point: self._hook_point(point)) for signal in ("notify::access-points", "notify::state", "notify::is-connected", "removed"): device.connect(signal, self._on_change) device.ap.connect("notify::ssid", self._on_change) for point in device.access_points: self._hook_point(point) self.notify_changed() def _hook_point(self, point): if point not in self._hooked: self._hooked.add(point) for signal in ("notify::strength", "notify::is-connected", "notify::psk", "removed"): point.connect(signal, self._on_change) self.notify_changed() def _groups(self) -> dict[str, list]: groups: dict[str, list] = {} for device in self._wifi.devices: for point in device.access_points: if not point.ssid: continue groups.setdefault(self._network_id(point), []).append((device, point)) return groups @staticmethod def _network_id(point) -> str: return f"{point.ssid}\0{security_of(point.point).value}" @staticmethod def _is_saved(point) -> bool: return bool(getattr(point, "_connections", ())) def state(self) -> WifiState: devices = list(self._wifi.devices) if not devices: return WifiState() if not self._wifi.enabled: return WifiState(available=True, enabled=False) networks = [] for network_id, members in self._groups().items(): _device, strongest = max(members, key=lambda member: member[1].strength) networks.append( WifiNetwork( network_id=network_id, ssid=strongest.ssid, security=security_of(strongest.point), strength=strongest.strength, state=self._state_of(network_id, members), saved=any(self._is_saved(point) for _, point in members), ) ) return WifiState( available=True, enabled=True, scanning=self._scanning, station=self._station_state(devices), networks=tuple(networks), link=self._link_details(), ) def _state_of(self, network_id: str, members: list) -> LinkState: if any(point.is_connected for _, point in members): return LinkState.CONNECTED if self._busy is not None and self._busy[0] == network_id: return self._busy[1] for device, point in members: if device.ap.ssid != point.ssid: continue if device.state in ACTIVATING_STATES: return LinkState.CONNECTING if device.state == "deactivating": return LinkState.DISCONNECTING return LinkState.IDLE @staticmethod def _station_state(devices) -> LinkState: states = {device.state for device in devices} if states & ACTIVATING_STATES: return LinkState.CONNECTING if "deactivating" in states: return LinkState.DISCONNECTING if "activated" in states: return LinkState.CONNECTED return LinkState.IDLE def _link_details(self) -> LinkDetails | None: for device in self._wifi.devices: for point in device.access_points: if point.is_connected: return LinkDetails( bssid=point.bssid, frequency_mhz=point.frequency or None, max_mbps=point.max_bitrate / 1000 if point.max_bitrate else None, ) return None async def set_wifi_enabled(self, enabled: bool) -> None: self._wifi.enabled = enabled async def scan(self) -> None: self._scanning = True self.notify_changed() try: for device in self._wifi.devices: try: await device.scan() except GLib.Error: continue finally: self._scanning = False self.notify_changed() async def connect_to(self, network_id: str, passphrase: str | None = None) -> None: members = self._groups().get(network_id) if not members: raise NetworkError(ErrorCode.NOT_FOUND, network_id) _, point = max(members, key=lambda member: member[1].strength) saved = any(self._is_saved(p) for _, p in members) security = security_of(point.point) blocker = connect_blocker(security, saved, passphrase) if blocker is not None: raise NetworkError(blocker, point.ssid) self._busy = (network_id, LinkState.CONNECTING) self.notify_changed() try: try: active = await point.connect_to(passphrase) except GLib.Error as error: raise NetworkError(ErrorCode.FAILED, error.message) from error try: await self._await_activation(active, bool(passphrase)) except NetworkError as failure: if failure.code is ErrorCode.WRONG_PASSWORD and not saved: await point.forget() raise finally: self._busy = None self.notify_changed() async def _await_activation(self, active, had_passphrase: bool) -> None: outcome: asyncio.Future = asyncio.get_running_loop().create_future() def settle(state, reason): if outcome.done(): return if state == NM.ActiveConnectionState.ACTIVATED: outcome.set_result(None) elif state in ( NM.ActiveConnectionState.DEACTIVATING, NM.ActiveConnectionState.DEACTIVATED, ): outcome.set_exception(failure_for(reason, had_passphrase)) handler = active.connect("state-changed", lambda _active, state, reason: settle(state, reason)) try: settle(active.get_state(), NM.ActiveConnectionStateReason.UNKNOWN) try: await asyncio.wait_for(outcome, ACTIVATION_TIMEOUT_S) except TimeoutError as error: raise NetworkError(ErrorCode.TIMEOUT) from error finally: GObject.Object.disconnect(active, handler) async def disconnect_from(self, network_id: str) -> None: members = self._groups().get(network_id, []) self._busy = (network_id, LinkState.DISCONNECTING) self.notify_changed() try: for _, point in members: if point.is_connected: try: await point.disconnect_from() except GLib.Error as error: raise NetworkError(ErrorCode.FAILED, error.message) from error finally: self._busy = None self.notify_changed() async def forget(self, network_id: str) -> None: for _, point in self._groups().get(network_id, []): if self._is_saved(point): try: await point.forget() except GLib.Error as error: raise NetworkError(ErrorCode.FAILED, error.message) from error async def set_auto_connect(self, network_id: str, enabled: bool) -> None: raise NetworkError(ErrorCode.UNAVAILABLE, "auto-connect is not exposed for NetworkManager")