import os import subprocess from typing import Any, List, Literal import gi from gi.repository import Gio from loguru import logger from fabric.core.service import Property, Service, Signal from fabric.utils import bulk_connect # FIX: HAHA Could be better if loc is less try: gi.require_version("NM", "1.0") from gi.repository import NM except ValueError: raise logger.error("Failed to start network manager") class Wifi(Service): """A service to manage wifi devices""" @Signal def changed(self) -> None: ... @Signal def enabled(self) -> bool: ... @Signal def scanning(self, is_scanning: bool) -> None: ... def __init__(self, client: NM.Client, device: NM.DeviceWifi, **kwargs): self._client: NM.Client = client self._device: NM.DeviceWifi = device self._ap: NM.AccessPoint | None = None self._ap_signal: int | None = None self._scan_cancellable: Gio.Cancellable | None = None self._is_scanning: bool = False super().__init__(**kwargs) self._client.connect( "notify::wireless-enabled", lambda *args: self.notifier("enabled", args), ) if self._device: bulk_connect( self._device, { "notify::active-access-point": self._activate_ap, "access-point-added": lambda *_: self.emit("changed"), "access-point-removed": lambda *_: self.emit("changed"), "state-changed": self.ap_update, }, ) self._activate_ap() def ap_update(self, *_): self.emit("changed") for sn in [ "enabled", "internet", "strength", "frequency", "access-points", "ssid", "state", "icon-name", ]: self.notify(sn) def _activate_ap(self, *_): if self._ap: self._ap.disconnect(self._ap_signal) self._ap = self._device.get_active_access_point() if not self._ap: return self._ap_signal = self._ap.connect( "notify::strength", lambda *args: self.ap_update() ) # type: ignore def toggle_wifi(self): self._client.wireless_set_enabled(not self._client.wireless_get_enabled()) def get_ap_security( self, nm_ap: NM.AccessPoint ) -> Literal["WEP", "WPA1", "WPA2", "802.1X", "unsecured"]: """Parse the security flags to return a string with 'WPA2', etc.""" flags = nm_ap.get_flags() wpa_flags = nm_ap.get_wpa_flags() rsn_flags = nm_ap.get_rsn_flags() sec_str = "" if ( (flags & getattr(NM, "80211ApFlags").PRIVACY) and (wpa_flags == 0) and (rsn_flags == 0) ): sec_str += " WEP" if wpa_flags != 0: sec_str += " WPA1" if rsn_flags != 0: sec_str += " WPA2" if (wpa_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X) or ( rsn_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X ): sec_str += " 802.1X" # If there is no security use "--" if sec_str == "": sec_str = "unsecured" return sec_str.lstrip() def scan(self): """Start scanning for WiFi networks and emit scanning signal""" if self._device and not self._is_scanning: self._is_scanning = True self._scan_cancellable = Gio.Cancellable() self.notify("scanning") # Notify property change self._device.request_scan_async( self._scan_cancellable, lambda device, result: self._scan_finished(device, result), ) def _scan_finished(self, device, result): """Handle scan completion""" try: device.request_scan_finish(result) except Exception as e: logger.error(f"WiFi scan error: {e}") finally: self._is_scanning = False self._scan_cancellable = None self.notify("scanning") # Notify property change def toggle_scan(self): """Toggle WiFi scanning on/off""" if self._is_scanning: # Stop current scan if self._scan_cancellable: self._scan_cancellable.cancel() self._scan_cancellable = None self._is_scanning = False self.notify("scanning") # Notify property change else: # Start new scan self.scan() @Property(bool, "readable", default_value=False) def scanning(self) -> bool: """Check if currently scanning""" return self._is_scanning def is_active_ap(self, name) -> bool: return self._ap.get_bssid() == name if self._ap else False def notifier(self, name: str, *_): self.notify(name) self.emit("changed") return def forget_access_point(self, ssid): try: # List all saved connections result = subprocess.check_output( "nmcli connection show", shell=True, text=True ) # Find connection ID that matches SSID for line in result.splitlines(): if ssid in line: connection_id = line.split()[0] subprocess.check_call( f"nmcli connection delete id '{connection_id}'", shell=True ) logger.info( f"[NetworkService] Deleted saved connection: {connection_id}" ) return True logger.warning( f"[NetworkService] No saved connection found for SSID: {ssid}" ) return False except subprocess.CalledProcessError as e: logger.error(f"[NetworkService] Error forgetting connection: {e}") return False def connect_network( self, ssid: str, password: str = "", remember: bool = True ) -> bool: """Connect to a WiFi network""" if not ssid: logger.error("[NetworkService] SSID cannot be empty") return False # Check if nm-applet is running before we kill it nm_applet_was_running = self._is_nm_applet_running() # Kill any running NetworkManager authentication agents to prevent GUI dialogs try: subprocess.run(["pkill", "-f", "nm-applet"], capture_output=True) subprocess.run(["pkill", "-f", "nm-connection-editor"], capture_output=True) subprocess.run( ["pkill", "-f", "polkit-gnome-authentication-agent"], capture_output=True, ) except: pass # Ignore errors if processes don't exist try: # First try to connect using saved connection (suppress GUI) try: env = os.environ.copy() env.update( { "DISPLAY": "", "WAYLAND_DISPLAY": "", "XDG_SESSION_TYPE": "tty", } ) subprocess.run( ["nmcli", "--terse", "con", "up", ssid], check=True, capture_output=True, env=env, ) return True except subprocess.CalledProcessError: # If saved connection fails, try with password if provided if password: # Create environment without GUI components to prevent dialogs env = os.environ.copy() env.update( { "DISPLAY": "", # No X11 display "WAYLAND_DISPLAY": "", # No Wayland display "XDG_SESSION_TYPE": "tty", # Force TTY session "DESKTOP_SESSION": "", # No desktop session "NM_EDITOR": "/bin/false", # Disable editor "NM_POLKIT_AGENT": "/bin/false", # Disable polkit agent } ) # First, try to delete any existing connection with the same name to avoid conflicts try: subprocess.run( ["nmcli", "connection", "delete", ssid], capture_output=True, env=env, ) except: pass # Ignore if connection doesn't exist # Create a temporary connection profile with the password # Use a unique temporary name to avoid conflicts temp_connection_name = f"temp_{ssid}_{os.getpid()}" create_cmd = [ "nmcli", "connection", "add", "type", "wifi", "con-name", temp_connection_name, "ssid", ssid, "wifi-sec.key-mgmt", "wpa-psk", "wifi-sec.psk", password, "connection.autoconnect", "no", # Don't auto-connect ] create_result = subprocess.run( create_cmd, capture_output=True, text=True, env=env ) if create_result.returncode == 0: # Try to activate the temporary connection with shorter timeout activate_cmd = [ "nmcli", "--wait", "3", "connection", "up", temp_connection_name, ] try: activate_result = subprocess.run( activate_cmd, capture_output=True, text=True, env=env, timeout=5, # 5 second timeout for faster failure detection ) except subprocess.TimeoutExpired: # Connection timed out - treat as failure activate_result = subprocess.CompletedProcess( activate_cmd, 1, "", "Connection timeout" ) if activate_result.returncode == 0: # Connection successful! if remember: # Rename the temporary connection to the final name rename_cmd = [ "nmcli", "connection", "modify", temp_connection_name, "connection.id", ssid, "connection.autoconnect", "yes", ] subprocess.run(rename_cmd, capture_output=True, env=env) else: # Don't remember - delete the temporary connection subprocess.run( [ "nmcli", "connection", "delete", temp_connection_name, ], capture_output=True, env=env, ) return True else: # Connection failed - delete the temporary connection immediately # This ensures incorrect passwords are never saved subprocess.run( ["nmcli", "connection", "delete", temp_connection_name], capture_output=True, env=env, ) return False else: # Failed to create connection profile return False # Fallback: try the old method but always use temporary connections first # This ensures incorrect passwords are never permanently saved fallback_cmd = [ "nmcli", "--terse", "--wait", "3", # Reduced wait time for faster response "device", "wifi", "connect", ssid, "password", password, "--temporary", # Always use temporary for initial attempt ] # Set additional environment variables to suppress dialogs env.update( { "SSH_ASKPASS": "/bin/false", "GIT_ASKPASS": "/bin/false", "SUDO_ASKPASS": "/bin/false", "NM_POLKIT_AGENT": "", } ) result = subprocess.run( fallback_cmd, capture_output=True, text=True, env=env, stdin=subprocess.DEVNULL, ) # If connection succeeded and we want to remember it if result.returncode == 0 and remember: # Create a permanent connection profile save_cmd = [ "nmcli", "connection", "add", "type", "wifi", "con-name", ssid, "ssid", ssid, "wifi-sec.key-mgmt", "wpa-psk", "wifi-sec.psk", password, "connection.autoconnect", "yes", ] subprocess.run(save_cmd, capture_output=True, env=env) return result.returncode == 0 return False except subprocess.CalledProcessError as e: logger.error(f"[NetworkService] Failed connecting to network: {e}") return False finally: # Cleanup: Remove any leftover temporary connections for this process self._cleanup_temp_connections() # Restart nm-applet if it was running before we killed it if nm_applet_was_running: self._restart_nm_applet() def _is_nm_applet_running(self): """Check if nm-applet is currently running""" try: result = subprocess.run( ["pgrep", "-f", "nm-applet"], capture_output=True, text=True ) return result.returncode == 0 except Exception: return False def _restart_nm_applet(self): """Restart nm-applet with indicator support""" try: # Small delay to ensure connection process is complete import time time.sleep(0.5) # Start nm-applet in the background with indicator support subprocess.Popen( ["nm-applet", "--indicator"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, # Detach from parent process ) logger.debug("[NetworkService] Restarted nm-applet with indicator support") except Exception as e: logger.debug(f"[NetworkService] Failed to restart nm-applet: {e}") def _cleanup_temp_connections(self): """Clean up any temporary connections created by this process""" try: # List all connections and find temporary ones created by this process result = subprocess.run( ["nmcli", "-t", "-f", "NAME", "connection", "show"], capture_output=True, text=True, ) if result.returncode == 0: connections = result.stdout.strip().split("\n") temp_prefix = "temp_" process_suffix = f"_{os.getpid()}" for connection in connections: if connection.startswith(temp_prefix) and connection.endswith( process_suffix ): # Delete this temporary connection subprocess.run( ["nmcli", "connection", "delete", connection], capture_output=True, ) logger.debug( f"[NetworkService] Cleaned up temporary connection: { connection }" ) except Exception as e: logger.debug(f"[NetworkService] Error during temp connection cleanup: {e}") def disconnect_network(self, ssid: str) -> bool: """Disconnect from a WiFi network""" if not ssid: logger.error("[NetworkService] SSID cannot be empty") return False try: # Method 1: Try to disconnect using device-based approach (more reliable) if self._device and self._device.get_active_connection(): active_connection = self._device.get_active_connection() # Verify this is the connection we want to disconnect if ( self._ap and NM.utils_ssid_to_utf8(self._ap.get_ssid().get_data()) == ssid ): try: # Use NetworkManager API to deactivate the connection self._client.deactivate_connection_async( active_connection, None, lambda client, result: self._disconnect_finished( client, result, ssid ), ) logger.info( f"[NetworkService] Initiated disconnect from {ssid}" ) # Force immediate state update to reflect disconnection from gi.repository import GLib GLib.timeout_add(100, lambda: self._force_state_update()) return True except Exception as e: logger.warning( f"[NetworkService] API disconnect failed: {e}, trying nmcli" ) # Method 2: Fallback to nmcli with proper connection ID resolution try: # First, get the connection ID that matches the SSID result = subprocess.check_output( [ "nmcli", "-t", "-f", "NAME,TYPE", "connection", "show", "--active", ], text=True, ) connection_id = None for line in result.strip().split("\n"): if line: name, conn_type = line.split(":", 1) if conn_type == "802-11-wireless" and name == ssid: connection_id = name break if connection_id: subprocess.run(["nmcli", "con", "down", connection_id], check=True) logger.info( f"[NetworkService] Disconnected from {ssid} using connection ID" ) # Force immediate state update from gi.repository import GLib GLib.timeout_add(100, lambda: self._force_state_update()) return True else: # Method 3: Try disconnecting by device device_name = self._device.get_iface() if self._device else None if device_name: subprocess.run( ["nmcli", "device", "disconnect", device_name], check=True ) logger.info( f"[NetworkService] Disconnected device {device_name}" ) # Force immediate state update from gi.repository import GLib GLib.timeout_add(100, lambda: self._force_state_update()) return True else: logger.error( f"[NetworkService] No active connection found for SSID: { ssid }" ) return False except subprocess.CalledProcessError as e: logger.error(f"[NetworkService] nmcli disconnect failed: {e}") return False except Exception as e: logger.error(f"[NetworkService] Failed disconnecting from network: {e}") return False def _disconnect_finished(self, client, result, ssid): """Handle async disconnect completion""" try: client.deactivate_connection_finish(result) logger.info(f"[NetworkService] Successfully disconnected from {ssid}") # Force state update after async completion self._force_state_update() except Exception as e: logger.error(f"[NetworkService] Async disconnect failed: {e}") def _force_state_update(self): """Force an immediate state update to reflect changes""" # Trigger all relevant property notifications self.ap_update() return False # Remove timeout @Property(bool, "read-write", default_value=False) def enabled(self) -> bool: # noqa: F811 return bool(self._client.wireless_get_enabled()) @enabled.setter def enabled(self, value: bool): self._client.wireless_set_enabled(value) @Property(int, "readable") def strength(self): return self._ap.get_strength() if self._ap else -1 @Property(str, "readable") def icon_name(self): if not self._ap: return "network-wireless-disabled-symbolic" if self.internet == "activated": return { 80: "network-wireless-signal-excellent-symbolic", 60: "network-wireless-signal-good-symbolic", 40: "network-wireless-signal-ok-symbolic", 20: "network-wireless-signal-weak-symbolic", 00: "network-wireless-signal-none-symbolic", }.get( min(80, 20 * round(self._ap.get_strength() / 20)), "network-wireless-no-route-symbolic", ) if self.internet == "activating": return "network-wireless-acquiring-symbolic" return "network-wireless-offline-symbolic" @Property(int, "readable") def frequency(self): return self._ap.get_frequency() if self._ap else -1 @Property(int, "readable") def internet(self): active_connection = self._device.get_active_connection() if not active_connection: return "disconnected" return { NM.ActiveConnectionState.ACTIVATED: "activated", NM.ActiveConnectionState.ACTIVATING: "activating", NM.ActiveConnectionState.DEACTIVATING: "deactivating", NM.ActiveConnectionState.DEACTIVATED: "deactivated", }.get( active_connection.get_state(), "unknown", ) @Property(object, "readable") def access_points(self) -> List[object]: points: list[NM.AccessPoint] = self._device.get_access_points() def make_ap_dict(ap: NM.AccessPoint): return { "bssid": ap.get_bssid(), "last_seen": ap.get_last_seen(), "ssid": ( NM.utils_ssid_to_utf8(ap.get_ssid().get_data()) if ap.get_ssid() else "Unknown" ), "active-ap": self._ap, "strength": ap.get_strength(), "frequency": ap.get_frequency(), "icon-name": { 80: "network-wireless-signal-excellent-symbolic", 60: "network-wireless-signal-good-symbolic", 40: "network-wireless-signal-ok-symbolic", 20: "network-wireless-signal-weak-symbolic", 00: "network-wireless-signal-none-symbolic", }.get( min(80, 20 * round(ap.get_strength() / 20)), "network-wireless-no-route-symbolic", ), } return list(map(make_ap_dict, points)) @Property(str, "readable") def ssid(self): # Check if we have an active connection first active_connection = self._device.get_active_connection() if not active_connection or active_connection.get_state() in [ NM.ActiveConnectionState.DEACTIVATED, NM.ActiveConnectionState.DEACTIVATING, ]: return "Disconnected" if not self._ap: return "Disconnected" ssid = self._ap.get_ssid().get_data() return NM.utils_ssid_to_utf8(ssid) if ssid else "Unknown" @Property(int, "readable") def state(self): return { NM.DeviceState.UNMANAGED: "unmanaged", NM.DeviceState.UNAVAILABLE: "unavailable", NM.DeviceState.DISCONNECTED: "disconnected", NM.DeviceState.PREPARE: "prepare", NM.DeviceState.CONFIG: "config", NM.DeviceState.NEED_AUTH: "need_auth", NM.DeviceState.IP_CONFIG: "ip_config", NM.DeviceState.IP_CHECK: "ip_check", NM.DeviceState.SECONDARIES: "secondaries", NM.DeviceState.ACTIVATED: "activated", NM.DeviceState.DEACTIVATING: "deactivating", NM.DeviceState.FAILED: "failed", }.get(self._device.get_state(), "unknown") class Ethernet(Service): """A service to manage ethernet devices""" @Signal def changed(self) -> None: ... @Signal def enabled(self) -> bool: ... @Property(int, "readable") def speed(self) -> int: return self._device.get_speed() @Property(str, "readable") def internet(self) -> str: return { NM.ActiveConnectionState.ACTIVATED: "activated", NM.ActiveConnectionState.ACTIVATING: "activating", NM.ActiveConnectionState.DEACTIVATING: "deactivating", NM.ActiveConnectionState.DEACTIVATED: "deactivated", }.get( self._device.get_active_connection().get_state(), "disconnected", ) @Property(str, "readable") def icon_name(self) -> str: network = self.internet if network == "activated": return "network-wired-symbolic" elif network == "activating": return "network-wired-acquiring-symbolic" elif self._device.get_connectivity != NM.ConnectivityState.FULL: return "network-wired-no-route-symbolic" return "network-wired-disconnected-symbolic" def __init__(self, client: NM.Client, device: NM.DeviceEthernet, **kwargs) -> None: super().__init__(**kwargs) self._client: NM.Client = client self._device: NM.DeviceEthernet = device for names in ( "active-connection", "icon-name", "internet", "speed", "state", ): self._device.connect(f"notify::{names}", lambda *_: self.notifier(names)) self._device.connect("notify::speed", lambda *_: print(_)) def notifier(self, names): self.notify(names) self.emit("changed") class NetworkService(Service): """A service to manage network devices""" _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance @Signal def device_ready(self) -> None: ... def __init__(self, **kwargs): self._client: NM.Client | None = None self.wifi_device: Wifi | None = None self.ethernet_device: Ethernet | None = None super().__init__(**kwargs) NM.Client.new_async( cancellable=None, callback=self._init_network_client, **kwargs, ) def _init_network_client(self, client: NM.Client, task: Gio.Task, **kwargs): self._client = client wifi_device: NM.DeviceWifi | None = self._get_device(NM.DeviceType.WIFI) # type: ignore ethernet_device: NM.DeviceEthernet | None = self._get_device( NM.DeviceType.ETHERNET ) if wifi_device: self.wifi_device = Wifi(self._client, wifi_device) # Clean up any leftover temporary connections from previous runs self._cleanup_temp_connections() self.emit("device-ready") if ethernet_device: self.ethernet_device = Ethernet(client=self._client, device=ethernet_device) self.emit("device-ready") self.notify("primary-device") def _get_device(self, device_type) -> Any: devices: List[NM.Device] = self._client.get_devices() # type: ignore return next( ( x for x in devices if x.get_device_type() == device_type and x.get_active_connection() is not None ), None, ) def _get_primary_device(self) -> Literal["wifi", "wired"] | None: if not self._client: return None if self._client.get_primary_connection() is None: return "wifi" return ( "wifi" if "wireless" in str(self._client.get_primary_connection().get_connection_type()) else ( "wired" if "ethernet" in str(self._client.get_primary_connection().get_connection_type()) else None ) ) @Property(str, "readable") def primary_device(self) -> Literal["wifi", "wired"] | None: return self._get_primary_device() def _is_nm_applet_running(self): """Check if nm-applet is currently running""" try: result = subprocess.run( ["pgrep", "-f", "nm-applet"], capture_output=True, text=True ) return result.returncode == 0 except Exception: return False def _restart_nm_applet(self): """Restart nm-applet with indicator support""" try: # Small delay to ensure connection process is complete import time time.sleep(0.5) # Start nm-applet in the background with indicator support subprocess.Popen( ["nm-applet", "--indicator"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, # Detach from parent process ) logger.debug("[NetworkService] Restarted nm-applet with indicator support") except Exception as e: logger.debug(f"[NetworkService] Failed to restart nm-applet: {e}") def _cleanup_temp_connections(self): """Clean up any temporary connections created by this process""" try: # List all connections and find temporary ones created by this process result = subprocess.run( ["nmcli", "-t", "-f", "NAME", "connection", "show"], capture_output=True, text=True, ) if result.returncode == 0: connections = result.stdout.strip().split("\n") temp_prefix = "temp_" process_suffix = f"_{os.getpid()}" for connection in connections: if connection.startswith(temp_prefix) and connection.endswith( process_suffix ): # Delete this temporary connection subprocess.run( ["nmcli", "connection", "delete", connection], capture_output=True, ) logger.debug( f"[NetworkService] Cleaned up temporary connection: { connection }" ) except Exception as e: logger.debug(f"[NetworkService] Error during temp connection cleanup: {e}")