Something went wrong. Try again.
A Material 3 inspired desktop shell for Niri and Hyprland created with Ignis.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278from __future__ import annotations
import asyncioimport weakref
import gifrom 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")