z_mirror/dotfiles
git clone https://git.raiguard.me/z_mirror/dotfiles
9b06fc2
trunk
1#!/usr/bin/env python3 2# encoding:utf8 3"""NetworkManager command line dmenu script. 4 5To add new connections or enable/disable networking requires policykit 6permissions setup per: 7https://wiki.archlinux.org/index.php/NetworkManager#Set_up_PolicyKit_permissions 8 9OR running the script as root 10 11Add dmenu options and default terminal if desired to 12~/.config/networkmanager-dmenu/config.ini 13 14""" 15import pathlib 16import struct 17import configparser 18import locale 19import os 20from os.path import basename, expanduser 21import shlex 22from shutil import which 23import sys 24from time import sleep 25import uuid 26import subprocess 27 28# pylint: disable=import-error 29import gi 30gi.require_version('NM', '1.0') 31from gi.repository import GLib, NM # noqa pylint: disable=wrong-import-position 32# pylint: enable=import-error 33 34ENV = os.environ.copy() 35ENC = locale.getpreferredencoding() 36 37CONF = configparser.ConfigParser() 38CONF.read(expanduser("~/.config/networkmanager-dmenu/config.ini")) 39 40 41def cli_args(): 42 """ Don't override dmenu_cmd function arguments with CLI args. Removes -l 43 and -p if those are passed on the command line. 44 45 Exception: if -l is passed and dmenu_command is not defined, assume that the 46 user wants to switch dmenu to the vertical layout and include -l. 47 48 Returns: List of additional CLI arguments 49 50 """ 51 args = sys.argv[1:] 52 cmd = CONF.get('dmenu', 'dmenu_command', fallback=False) 53 if "-l" in args or "-p" in args: 54 for nope in ['-l', '-p'] if cmd is not False else ['-p']: 55 try: 56 nope_idx = args.index(nope) 57 del args[nope_idx] 58 del args[nope_idx] 59 except ValueError: 60 pass 61 return args 62 63 64def dmenu_pass(command, color): 65 """Check if dmenu passphrase patch is applied and return the correct command 66 line arg list 67 68 Args: command - string 69 color - obscure color string 70 Returns: list or None 71 72 """ 73 if command != 'dmenu': 74 return None 75 try: 76 # Check for dmenu password patch 77 dm_patch = b'P' in subprocess.run(["dmenu", "-h"], 78 capture_output=True, 79 check=False).stderr 80 except FileNotFoundError: 81 dm_patch = False 82 return ["-P"] if dm_patch else ["-nb", color, "-nf", color] 83 84 85def dmenu_cmd(num_lines, prompt="Networks", active_lines=None): 86 """Parse config.ini for menu options 87 88 Args: args - num_lines: number of lines to display 89 prompt: prompt to show 90 active_lines: list of line numbers to tag as active 91 Returns: command invocation (as a list of strings) for example 92 ["dmenu", "-l", "<num_lines>", "-p", "<prompt>", "-i"] 93 94 """ 95 # Create command string 96 commands = {"dmenu": ["-p", str(prompt)], 97 "rofi": ["-dmenu", "-p", str(prompt)], 98 "bemenu": ["-p", str(prompt)], 99 "wofi": ["-p", str(prompt)], 100 "fuzzel": ["-p", str(prompt), "--log-level", "none"]} 101 command = shlex.split(CONF.get('dmenu', 'dmenu_command', fallback="dmenu")) 102 cmd_base = basename(command[0]) 103 command.extend(cli_args()) 104 command.extend(commands.get(cmd_base, [])) 105 # Highlighting 106 highlight = CONF.getboolean('dmenu', 'highlight', fallback=False) 107 if highlight is True: 108 # Rofi 109 if cmd_base == "rofi" and active_lines: 110 command.extend(["-a", ",".join([str(num) for num in active_lines])]) 111 # Wofi 112 if cmd_base == "wofi" and active_lines: 113 # add '-q' to prevent tag name and properties of pango markup from searchable 114 command.extend(["-m", "-q"]) 115 # Passphrase prompts 116 obscure = CONF.getboolean('dmenu_passphrase', 'obscure', fallback=False) 117 if prompt == "Passphrase" and obscure is True: 118 obscure_color = CONF.get('dmenu_passphrase', 'obscure_color', fallback='#222222') 119 pass_prompts = {"dmenu": dmenu_pass(cmd_base, obscure_color), 120 "rofi": ['-password'], 121 "bemenu": ['-x'], 122 "wofi": ['-P'], 123 "fuzzel": ['--password']} 124 command.extend(pass_prompts.get(cmd_base, [])) 125 return command 126 127 128def choose_adapter(client): 129 """If there is more than one wifi adapter installed, ask which one to use 130 131 """ 132 devices = client.get_devices() 133 devices = [i for i in devices if i.get_device_type() == NM.DeviceType.WIFI] 134 if not devices: 135 return None 136 if len(devices) == 1: 137 return devices[0] 138 device_names = "\n".join([d.get_iface() for d in devices]) 139 sel = subprocess.run(dmenu_cmd(len(devices), "CHOOSE ADAPTER:"), 140 capture_output=True, 141 check=False, 142 env=ENV, 143 input=device_names, 144 encoding=ENC).stdout 145 if not sel.strip(): 146 sys.exit() 147 devices = [i for i in devices if i.get_iface() == sel.strip()] 148 if len(devices) != 1: 149 raise ValueError(f"Selection was ambiguous: '{str(sel.strip())}'") 150 return devices[0] 151 152 153def is_installed(cmd): 154 """Check if a utility is installed""" 155 return which(cmd) is not None 156 157 158def is_running(cmd): 159 try: 160 subprocess.check_output(["pidof", cmd]) 161 except subprocess.CalledProcessError: 162 return False 163 else: 164 return True 165 166 167def bluetooth_get_enabled(): 168 """Check if bluetooth is enabled. Try bluetoothctl first, then rfkill. 169 170 Returns None if no bluetooth device was found. 171 """ 172 if is_installed('bluetoothctl') and is_running('bluetoothd'): 173 # Times out in 2 seconds, otherwise bluetoothctl will hang if bluetooth 174 # service isn't running. 175 try: 176 res = subprocess.run(['bluetoothctl', 'show'], 177 timeout=2, 178 capture_output=True, 179 text=True) 180 return "Powered: yes" in res.stdout 181 except subprocess.TimeoutExpired: 182 pass 183 # See https://www.kernel.org/doc/Documentation/ABI/stable/sysfs-class-rfkill 184 for path in pathlib.Path('/sys/class/rfkill/').glob('rfkill*'): 185 if (path / 'type').read_text().strip() == 'bluetooth': 186 return (path / 'soft').read_text().strip() == '0' 187 return None 188 189 190def create_other_actions(client): 191 """Return list of other actions that can be taken 192 193 """ 194 networking_enabled = client.networking_get_enabled() 195 networking_action = "Disable" if networking_enabled else "Enable" 196 197 wifi_enabled = client.wireless_get_enabled() 198 wifi_action = "Disable" if wifi_enabled else "Enable" 199 200 bluetooth_enabled = bluetooth_get_enabled() 201 bluetooth_action = "Disable" if bluetooth_enabled else "Enable" 202 203 actions = [Action(f"{wifi_action} Wifi", toggle_wifi, 204 not wifi_enabled), 205 Action(f"{networking_action} Networking", 206 toggle_networking, not networking_enabled)] 207 if bluetooth_enabled is not None: 208 actions.append(Action(f"{bluetooth_action} Bluetooth", 209 toggle_bluetooth, not bluetooth_enabled)) 210 actions += [Action("Launch Connection Manager", launch_connection_editor), 211 Action("Delete a Connection", delete_connection)] 212 if wifi_enabled: 213 actions.append(Action("Rescan Wifi Networks", rescan_wifi)) 214 return actions 215 216 217def rescan_wifi(): 218 """ 219 Rescan Wifi Access Points 220 """ 221 delay = CONF.getint('nmdm', 'rescan_delay', fallback=5) 222 for dev in CLIENT.get_devices(): 223 if gi.repository.NM.DeviceWifi == type(dev): 224 try: 225 dev.request_scan_async(None, rescan_cb, None) 226 LOOP.run() 227 sleep(delay) 228 notify("Wifi scan complete") 229 main() 230 except gi.repository.GLib.Error as err: 231 # Too frequent rescan error 232 notify("Wifi rescan failed", urgency="critical") 233 if not err.code == 6: # pylint: disable=no-member 234 raise err 235 236 237def rescan_cb(dev, res, data): 238 """Callback for rescan_wifi. Just for notifications 239 240 """ 241 if dev.request_scan_finish(res) is True: 242 notify("Wifi scan running...") 243 else: 244 notify("Wifi scan failed", urgency="critical") 245 LOOP.quit() 246 247 248def ssid_to_utf8(nm_ap): 249 """ Convert binary ssid to utf-8 """ 250 ssid = nm_ap.get_ssid() 251 if not ssid: 252 return "" 253 ret = NM.utils_ssid_to_utf8(ssid.get_data()) 254 return ret 255 256 257def prompt_saved(saved_cons): 258 """Prompt for a saved connection.""" 259 actions = create_saved_actions(saved_cons) 260 sel = get_selection(actions) 261 sel() 262 263 264def ap_security(nm_ap): 265 """Parse the security flags to return a string with 'WPA2', etc. """ 266 flags = nm_ap.get_flags() 267 wpa_flags = nm_ap.get_wpa_flags() 268 rsn_flags = nm_ap.get_rsn_flags() 269 sec_str = "" 270 if ((flags & getattr(NM, '80211ApFlags').PRIVACY) and 271 (wpa_flags == 0) and (rsn_flags == 0)): 272 sec_str = " WEP" 273 if wpa_flags: 274 sec_str = " WPA1" 275 if rsn_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_PSK: 276 sec_str += " WPA2" 277 if rsn_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_SAE: 278 sec_str += " WPA3" 279 if ((wpa_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_802_1X) or 280 (rsn_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_802_1X)): 281 sec_str += " 802.1X" 282 if ((wpa_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_OWE) or 283 (rsn_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_OWE)): 284 sec_str += " OWE" 285 286 # If there is no security use "--" 287 if sec_str == "": 288 sec_str = "--" 289 return sec_str.lstrip() 290 291 292class Action(): # pylint: disable=too-few-public-methods 293 """Helper class to execute functions from a string variable""" 294 def __init__(self, 295 name, 296 func, 297 args=None, 298 active=False): 299 self.name = name 300 self.func = func 301 self.is_active = active 302 if args is None: 303 self.args = None 304 elif isinstance(args, list): 305 self.args = args 306 else: 307 self.args = [args] 308 309 def __str__(self): 310 return self.name 311 312 def __call__(self): 313 if self.args is None: 314 self.func() 315 else: 316 self.func(*self.args) 317 318 319def conn_matches_adapter(conn, adapter): 320 """Return True if the connection is applicable for the given adapter. 321 322 There seem to be two ways for a connection specify what interface it belongs 323 to: 324 325 - By setting 'mac-address' in [wifi] to the adapter's MAC 326 - By setting 'interface-name` in [connection] to the adapter's name. 327 328 Depending on how the connection was added, it seems like either 329 'mac-address', 'interface-name' or neither of both is set. 330 """ 331 # [wifi] mac-address 332 setting_wireless = conn.get_setting_wireless() 333 mac = setting_wireless.get_mac_address() 334 if mac is not None: 335 return mac == adapter.get_permanent_hw_address() 336 337 # [connection] interface-name 338 setting_connection = conn.get_setting_connection() 339 interface = setting_connection.get_interface_name() 340 if interface is not None: 341 return interface == adapter.get_iface() 342 343 # Neither is set, let's assume this connection is for multiple/all adapters. 344 return True 345 346 347def process_ap(nm_ap, is_active, adapter): 348 """Activate/Deactivate a connection and get password if required""" 349 if is_active: 350 CLIENT.deactivate_connection_async(nm_ap, None, deactivate_cb, nm_ap) 351 LOOP.run() 352 else: 353 conns_cur = [i for i in CONNS if 354 i.get_setting_wireless() is not None and 355 conn_matches_adapter(i, adapter)] 356 con = nm_ap.filter_connections(conns_cur) 357 if len(con) > 1: 358 raise ValueError("There are multiple connections possible") 359 360 if len(con) == 1: 361 CLIENT.activate_connection_async(con[0], adapter, nm_ap.get_path(), 362 None, activate_cb, nm_ap) 363 LOOP.run() 364 else: 365 if ap_security(nm_ap) != "--": 366 password = get_passphrase() 367 else: 368 password = "" 369 set_new_connection(nm_ap, password, adapter) 370 371 372def activate_cb(dev, res, data): 373 """Notification if activate connection completed successfully 374 375 """ 376 try: 377 conn = dev.activate_connection_finish(res) 378 except GLib.Error: 379 conn = None 380 if conn is not None: 381 notify(f"Activated {conn.get_id()}") 382 else: 383 notify(f"Problem activating {data.get_id()}", urgency="critical") 384 LOOP.quit() 385 386 387def deactivate_cb(dev, res, data): 388 """Notification if deactivate connection completed successfully 389 390 """ 391 if dev.deactivate_connection_finish(res) is True: 392 notify(f"Deactivated {data.get_id()}") 393 else: 394 notify(f"Problem deactivating {data.get_id()}", urgency="critical") 395 LOOP.quit() 396 397 398def process_vpngsm(con, activate): 399 """Activate/deactive VPN or GSM connections""" 400 if activate: 401 CLIENT.activate_connection_async(con, None, None, 402 None, activate_cb, con) 403 else: 404 CLIENT.deactivate_connection_async(con, None, deactivate_cb, con) 405 LOOP.run() 406 407def strength_bars(signal_strength): 408 bars = NM.utils_wifi_strength_bars(signal_strength) 409 wifi_chars = CONF.get("dmenu", "wifi_chars", fallback=False) 410 if wifi_chars: 411 bars = "".join([wifi_chars[i] for i, j in enumerate(bars) if j == '*']) 412 return bars 413 414 415def strength_icon(signal_strength): 416 wifi_icons = CONF.get("dmenu", "wifi_icons", fallback=False) 417 if wifi_icons: 418 return wifi_icons[round(signal_strength / 100 * (len(wifi_icons) - 1))] 419 return "" 420 421 422def create_ap_actions(aps, active_ap, active_connection, adapter): # noqa pylint: disable=too-many-locals,line-too-long 423 """For each AP in a list, create the string and its attached function 424 (activate/deactivate) 425 426 """ 427 active_ap_bssid = active_ap.get_bssid() if active_ap is not None else "" 428 429 names = [ssid_to_utf8(ap) for ap in aps] 430 max_len_name = max([len(name) for name in names]) if names else 0 431 secs = [ap_security(ap) for ap in aps] 432 max_len_sec = max([len(sec) for sec in secs]) if secs else 0 433 434 ap_actions = [] 435 436 if CONF.getboolean("dmenu", "compact", fallback=False): 437 format = CONF.get("dmenu", "format", fallback="{name} {sec} {bars}") 438 else: 439 format = CONF.get("dmenu", "format", fallback="{name:<{max_len_name}s} {sec:<{max_len_sec}s} {bars:>4}") 440 441 for nm_ap, name, sec in zip(aps, names, secs): 442 is_active = nm_ap.get_bssid() == active_ap_bssid 443 signal_strength = nm_ap.get_strength() 444 bars = strength_bars(signal_strength) 445 icon = strength_icon(signal_strength) 446 action_name = format.format(name=name, sec=sec, signal=signal_strength, bars=bars, icon=icon, 447 max_len_name=max_len_name, max_len_sec=max_len_sec) 448 if is_active: 449 ap_actions.append(Action(action_name, process_ap, 450 [active_connection, True, adapter], 451 active=True)) 452 else: 453 ap_actions.append(Action(action_name, process_ap, 454 [nm_ap, False, adapter])) 455 return ap_actions 456 457 458def create_vpn_actions(vpns, active): 459 """Create the list of strings to display with associated function 460 (activate/deactivate) for VPN connections. 461 462 """ 463 active_vpns = [i for i in active if i.get_vpn()] 464 return _create_vpngsm_actions(vpns, active_vpns, "VPN") 465 466 467def create_vlan_actions(vlans, active): 468 """Create the list of strings to display with associated function 469 (activate/deactivate) for VLAN connections. 470 471 """ 472 active_vlans = [i for i in active if "vlan" == i.get_connection_type()] 473 return _create_vpngsm_actions(vlans, active_vlans, "VLAN") 474 475 476def create_wireguard_actions(wgs, active): 477 """Create the list of strings to display with associated function 478 (activate/deactivate) for Wireguard connections. 479 480 """ 481 active_wgs = [i for i in active if i.get_connection_type() == "wireguard"] 482 return _create_vpngsm_actions(wgs, active_wgs, "Wireguard") 483 484 485def create_eth_actions(eths, active): 486 """Create the list of strings to display with associated function 487 (activate/deactivate) for Ethernet connections. 488 489 """ 490 active_eths = [i for i in active if 'ethernet' in i.get_connection_type()] 491 return _create_vpngsm_actions(eths, active_eths, "Eth") 492 493 494def create_gsm_actions(gsms, active): 495 """Create the list of strings to display with associated function 496 (activate/deactivate) GSM connections.""" 497 active_gsms = [i for i in active if 498 i.get_connection() is not None and 499 i.get_connection().is_type(NM.SETTING_GSM_SETTING_NAME)] 500 return _create_vpngsm_actions(gsms, active_gsms, "GSM") 501 502 503def create_blue_actions(blues, active): 504 """Create the list of strings to display with associated function 505 (activate/deactivate) Bluetooth connections.""" 506 active_blues = [i for i in active if 507 i.get_connection() is not None and 508 i.get_connection().is_type(NM.SETTING_BLUETOOTH_SETTING_NAME)] 509 return _create_vpngsm_actions(blues, active_blues, "Bluetooth") 510 511 512def create_saved_actions(saved): 513 """Create the list of strings to display with associated function 514 (activate/deactivate) for VPN connections. 515 516 """ 517 return _create_vpngsm_actions(saved, [], "SAVED") 518 519 520def _create_vpngsm_actions(cons, active_cons, label): 521 active_con_ids = [a.get_id() for a in active_cons] 522 actions = [] 523 for con in cons: 524 is_active = con.get_id() in active_con_ids 525 action_name = f"{con.get_id()}:{label}" 526 if is_active: 527 active_connection = [a for a in active_cons 528 if a.get_id() == con.get_id()] 529 if len(active_connection) != 1: 530 raise ValueError(f"Multiple active connections match {con.get_id()}") 531 active_connection = active_connection[0] 532 533 actions.append(Action(action_name, process_vpngsm, 534 [active_connection, False], active=True)) 535 else: 536 actions.append(Action(action_name, process_vpngsm, 537 [con, True])) 538 return actions 539 540 541def create_wwan_actions(client): 542 """Create WWWAN actions 543 544 """ 545 wwan_enabled = client.wwan_get_enabled() 546 wwan_action = "Disable" if wwan_enabled else "Enable" 547 return [Action(f"{wwan_action} WWAN", toggle_wwan, not wwan_enabled)] 548 549 550def combine_actions(eths, aps, vlans, vpns, wgs, gsms, blues, wwan, others, saved): 551 # pylint: disable=too-many-arguments 552 """Combine all given actions into a list of actions. 553 554 Args: args - eths: list of Actions 555 aps: list of Actions 556 vpns: list of Actions 557 gsms: list of Actions 558 blues: list of Actions 559 wwan: list of Actions 560 others: list of Actions 561 """ 562 compact = CONF.getboolean("dmenu", "compact", fallback=False) 563 empty_action = [Action('', None)] if not compact else [] 564 all_actions = [] 565 all_actions += eths + empty_action if eths else [] 566 all_actions += aps + empty_action if aps else [] 567 all_actions += vlans + empty_action if vlans else [] 568 all_actions += vpns + empty_action if vpns else [] 569 all_actions += wgs + empty_action if wgs else [] 570 all_actions += gsms + empty_action if (gsms and wwan) else [] 571 all_actions += blues + empty_action if blues else [] 572 all_actions += wwan + empty_action if wwan else [] 573 all_actions += others + empty_action if others else [] 574 all_actions += saved + empty_action if saved else [] 575 return all_actions 576 577 578def get_wofi_highlight_markup(action): 579 highlight_fg = CONF.get('dmenu', 'highlight_fg', fallback=None) 580 highlight_bg = CONF.get('dmenu', 'highlight_bg', fallback=None) 581 highlight_bold = CONF.getboolean('dmenu', 'highlight_bold', fallback=True) 582 583 style = "" 584 if highlight_fg: 585 style += f'foreground="{highlight_fg}" ' 586 if highlight_bg: 587 style += f'background="{highlight_bg}" ' 588 if highlight_bold: 589 style += 'weight="bold" ' 590 591 return f"<span {style}>" + str(action) + "</span>" 592 593 594def get_selection(all_actions): 595 """Spawn dmenu for selection and execute the associated action.""" 596 command = shlex.split(CONF.get("dmenu", "dmenu_command", fallback="dmenu")) 597 cmd_base = basename(command[0]) 598 active_chars = CONF.get("dmenu", "active_chars", fallback="==") 599 highlight = CONF.getboolean("dmenu", "highlight", fallback=False) 600 inp = [] 601 602 if highlight is True and cmd_base == "rofi": 603 inp = [str(action) for action in all_actions] 604 elif highlight is True and cmd_base == "wofi": 605 inp = [get_wofi_highlight_markup(action) if action.is_active else str(action) 606 for action in all_actions] 607 else: 608 inp = [(active_chars if action.is_active else " " * len(active_chars)) + " " + str(action) 609 for action in all_actions] 610 active_lines = [index for index, action in enumerate(all_actions) 611 if action.is_active] 612 613 command = dmenu_cmd(len(inp), active_lines=active_lines) 614 sel = subprocess.run(command, 615 capture_output=True, 616 check=False, 617 input="\n".join(inp), 618 encoding=ENC, 619 env=ENV).stdout 620 621 if not sel.rstrip(): 622 sys.exit() 623 624 if highlight is True and cmd_base == "rofi": 625 action = [i for i in all_actions if str(i).strip() == sel.strip()] 626 elif highlight is True and cmd_base == "wofi": 627 action = [i for i in all_actions 628 if str(i).strip() == sel.strip() or 629 get_wofi_highlight_markup(i) == sel.strip()] 630 else: 631 action = [i for i in all_actions 632 if ((str(i).strip() == str(sel.strip()) 633 and not i.is_active) or 634 (active_chars + " " + str(i) == str(sel.rstrip('\n')) 635 and i.is_active))] 636 if len(action) != 1: 637 raise ValueError(f"Selection was ambiguous: '{str(sel.strip())}'") 638 return action[0] 639 640 641def toggle_networking(enable): 642 """Enable/disable networking 643 644 Args: enable - boolean 645 646 """ 647 toggle = GLib.Variant.new_tuple(GLib.Variant.new_boolean(enable)) 648 try: 649 CLIENT.dbus_call(NM.DBUS_PATH, NM.DBUS_INTERFACE, "Enable", toggle, 650 None, -1, None, None, None) 651 except AttributeError: 652 # Workaround for older versions of python-gobject 653 CLIENT.networking_set_enabled(enable) 654 notify(f"Networking {'enabled' if enable is True else 'disabled'}") 655 656 657def toggle_wifi(enable): 658 """Enable/disable Wifi 659 660 Args: enable - boolean 661 662 """ 663 toggle = GLib.Variant.new_boolean(enable) 664 try: 665 CLIENT.dbus_set_property(NM.DBUS_PATH, NM.DBUS_INTERFACE, "WirelessEnabled", toggle, 666 -1, None, None, None) 667 except AttributeError: 668 # Workaround for older versions of python-gobject 669 CLIENT.wireless_set_enabled(enable) 670 notify(f"Wifi {'enabled' if enable is True else 'disabled'}") 671 672 673def toggle_wwan(enable): 674 """Enable/disable WWAN 675 676 Args: enable - boolean 677 678 """ 679 toggle = GLib.Variant.new_boolean(enable) 680 try: 681 CLIENT.dbus_set_property(NM.DBUS_PATH, NM.DBUS_INTERFACE, "WwanEnabled", toggle, 682 -1, None, None, None) 683 except AttributeError: 684 # Workaround for older versions of python-gobject 685 CLIENT.wwan_set_enabled(enable) 686 notify(f"Wwan {'enabled' if enable is True else 'disabled'}") 687 688 689def toggle_bluetooth(enable): 690 """Enable/disable Bluetooth 691 692 Try bluetoothctl first, then drop to rfkill if it's not installed or 693 bluetooth service isn't running. 694 695 Args: enable - boolean 696 697 References: 698 https://github.com/blueman-project/blueman/blob/master/blueman/plugins/mechanism/RfKill.py 699 https://www.kernel.org/doc/html/latest/driver-api/rfkill.html 700 https://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git/tree/include/uapi/linux/rfkill.h?h=v5.8.9 701 702 """ 703 if is_installed('bluetoothctl') and is_running('bluetoothd'): 704 # Times out in 2 seconds, otherwise bluetoothctl will hang if bluetooth 705 # service isn't running. 706 try: 707 res = subprocess.run(['bluetoothctl', 'power', 'on' if enable is True else 'off'], 708 timeout=2, 709 capture_output=True) 710 except subprocess.TimeoutExpired: 711 pass 712 try: 713 res = subprocess.run(['bluetoothctl', 'show'], 714 timeout=2, 715 capture_output=True, 716 text=True) 717 if "Powered: yes" in res.stdout: 718 notify("Bluetooth enabled") 719 return 720 except subprocess.TimeoutExpired: 721 pass 722 # Now try using rfkill 723 type_bluetooth = 2 724 op_change_all = 3 725 idx = 0 726 soft_state = 0 if enable else 1 727 hard_state = 0 728 729 data = struct.pack("IBBBB", idx, type_bluetooth, op_change_all, 730 soft_state, hard_state) 731 732 try: 733 with open('/dev/rfkill', 'r+b', buffering=0) as rff: 734 rff.write(data) 735 except PermissionError: 736 notify("Lacking permission to write to /dev/rfkill.", 737 "Check README for configuration options.", 738 urgency="critical") 739 else: 740 notify(f"Bluetooth {'enabled' if enable else 'disabled'}") 741 742 743def launch_connection_editor(): 744 """Launch nmtui or the gui nm-connection-editor 745 746 """ 747 terminal = CONF.get("editor", "terminal", fallback="xterm") 748 gui_if_available = CONF.getboolean("editor", "gui_if_available", fallback=True) 749 guis = ["gnome-control-center", "nm-connection-editor"] 750 if gui_if_available is True: 751 for gui in guis: 752 if is_installed(gui): 753 subprocess.run(gui, check=False) 754 return 755 if is_installed("nmtui"): 756 subprocess.run([terminal, "-e", "nmtui"], check=False) 757 return 758 notify("No network connection editor installed", urgency="critical") 759 760 761def get_passphrase(): 762 """Get a password 763 764 Returns: string 765 766 """ 767 pinentry = CONF.get("dmenu", "pinentry", fallback=None) 768 if pinentry: 769 description = CONF.get("pinentry", "description", fallback="Get network password") 770 prompt = CONF.get("pinentry", "prompt", fallback="Password: ") 771 pin = "" 772 out = subprocess.run(pinentry, 773 capture_output=True, 774 check=False, 775 encoding=ENC, 776 input=f"setdesc {description}\nsetprompt {prompt}\ngetpin\n").stdout 777 if out: 778 res = [i for i in out.split("\n") if i.startswith("D ")] 779 if res and res[0].startswith("D "): 780 pin = res[0].split("D ")[1] 781 return pin 782 return subprocess.run(dmenu_cmd(0, "Passphrase"), 783 stdin=subprocess.DEVNULL, 784 capture_output=True, 785 check=False, 786 encoding=ENC).stdout 787 788 789def delete_connection(): 790 """Display list of NM connections and delete the selected one 791 792 """ 793 conn_acts = [Action(i.get_id(), i.delete_async, args=[None, delete_cb, None]) for i in CONNS] 794 conn_names = "\n".join([str(i) for i in conn_acts]) 795 sel = subprocess.run(dmenu_cmd(len(conn_acts), "CHOOSE CONNECTION TO DELETE:"), 796 capture_output=True, 797 check=False, 798 input=conn_names, 799 encoding=ENC, 800 env=ENV).stdout 801 if not sel.strip(): 802 sys.exit() 803 action = [i for i in conn_acts if str(i) == sel.rstrip("\n")] 804 if len(action) != 1: 805 raise ValueError(f"Selection was ambiguous: {str(sel)}") 806 action[0]() 807 LOOP.run() 808 809 810def delete_cb(dev, res, data): 811 """Notification if delete completed successfully 812 813 """ 814 if dev.delete_finish(res) is True: 815 notify(f"Deleted {dev.get_id()}") 816 else: 817 notify(f"Problem deleting {dev.get_id()}", urgency="critical") 818 LOOP.quit() 819 820 821def set_new_connection(nm_ap, nm_pw, adapter): 822 """Setup a new NetworkManager connection 823 824 Args: ap - NM.AccessPoint 825 pw - string 826 827 """ 828 nm_pw = str(nm_pw).strip() 829 profile = create_wifi_profile(nm_ap, nm_pw, adapter) 830 CLIENT.add_and_activate_connection_async(profile, adapter, nm_ap.get_path(), 831 None, verify_conn, profile) 832 LOOP.run() 833 834 835def create_wifi_profile(nm_ap, password, adapter): 836 # pylint: disable=line-too-long 837 # noqa From https://cgit.freedesktop.org/NetworkManager/NetworkManager/tree/examples/python/gi/add_connection.py 838 # noqa and https://cgit.freedesktop.org/NetworkManager/NetworkManager/tree/examples/python/dbus/add-wifi-psk-connection.py 839 # pylint: enable=line-too-long 840 """Create the NM profile given the AP and passphrase""" 841 ap_sec = ap_security(nm_ap) 842 profile = NM.SimpleConnection.new() 843 844 s_con = NM.SettingConnection.new() 845 s_con.set_property(NM.SETTING_CONNECTION_ID, ssid_to_utf8(nm_ap)) 846 s_con.set_property(NM.SETTING_CONNECTION_UUID, str(uuid.uuid4())) 847 s_con.set_property(NM.SETTING_CONNECTION_TYPE, "802-11-wireless") 848 profile.add_setting(s_con) 849 850 s_wifi = NM.SettingWireless.new() 851 s_wifi.set_property(NM.SETTING_WIRELESS_SSID, nm_ap.get_ssid()) 852 s_wifi.set_property(NM.SETTING_WIRELESS_MODE, 'infrastructure') 853 s_wifi.set_property(NM.SETTING_WIRELESS_MAC_ADDRESS, adapter.get_permanent_hw_address()) 854 profile.add_setting(s_wifi) 855 856 s_ip4 = NM.SettingIP4Config.new() 857 s_ip4.set_property(NM.SETTING_IP_CONFIG_METHOD, "auto") 858 profile.add_setting(s_ip4) 859 860 s_ip6 = NM.SettingIP6Config.new() 861 s_ip6.set_property(NM.SETTING_IP_CONFIG_METHOD, "auto") 862 profile.add_setting(s_ip6) 863 864 if ap_sec != "--": 865 s_wifi_sec = NM.SettingWirelessSecurity.new() 866 if "WPA" in ap_sec: 867 if "WPA3" in ap_sec: 868 s_wifi_sec.set_property(NM.SETTING_WIRELESS_SECURITY_KEY_MGMT, 869 "sae") 870 else: 871 s_wifi_sec.set_property(NM.SETTING_WIRELESS_SECURITY_KEY_MGMT, 872 "wpa-psk") 873 s_wifi_sec.set_property(NM.SETTING_WIRELESS_SECURITY_AUTH_ALG, 874 "open") 875 s_wifi_sec.set_property(NM.SETTING_WIRELESS_SECURITY_PSK, password) 876 elif "WEP" in ap_sec: 877 s_wifi_sec.set_property(NM.SETTING_WIRELESS_SECURITY_KEY_MGMT, 878 "None") 879 s_wifi_sec.set_property(NM.SETTING_WIRELESS_SECURITY_WEP_KEY_TYPE, 880 NM.WepKeyType.PASSPHRASE) 881 s_wifi_sec.set_wep_key(0, password) 882 profile.add_setting(s_wifi_sec) 883 884 return profile 885 886 887def verify_conn(client, result, data): 888 """Callback function for add_and_activate_connection_async 889 890 Check if connection completes successfully. Delete the connection if there 891 is an error. 892 893 """ 894 try: 895 act_conn = client.add_and_activate_connection_finish(result) 896 conn = act_conn.get_connection() 897 if not all([conn.verify(), 898 conn.verify_secrets(), 899 data.verify(), 900 data.verify_secrets()]): 901 raise GLib.Error 902 notify(f"Added {conn.get_id()}") 903 except GLib.Error: 904 try: 905 notify(f"Connection to {conn.get_id()} failed", 906 urgency="critical") 907 conn.delete_async(None, None, None) 908 except UnboundLocalError: 909 pass 910 finally: 911 LOOP.quit() 912 913 914def create_ap_list(adapter, active_connections): 915 """Generate list of access points. Remove duplicate APs , keeping strongest 916 ones and the active AP 917 918 Args: adapter 919 active_connections - list of all active connections 920 Returns: aps - list of access points 921 active_ap - active AP 922 active_ap_con - active Connection 923 adapter 924 925 """ 926 aps = [] 927 ap_names = [] 928 active_ap = adapter.get_active_access_point() 929 aps_all = sorted(adapter.get_access_points(), 930 key=lambda a: a.get_strength(), reverse=True) 931 conns_cur = [i for i in CONNS if 932 i.get_setting_wireless() is not None and 933 conn_matches_adapter(i, adapter)] 934 try: 935 ap_conns = active_ap.filter_connections(conns_cur) 936 active_ap_name = ssid_to_utf8(active_ap) 937 active_ap_con = [active_conn for active_conn in active_connections 938 if active_conn.get_connection() in ap_conns] 939 except AttributeError: 940 active_ap_name = None 941 active_ap_con = [] 942 if len(active_ap_con) > 1: 943 raise ValueError("Multiple connection profiles match" 944 " the wireless AP") 945 active_ap_con = active_ap_con[0] if active_ap_con else None 946 for nm_ap in aps_all: 947 ap_name = ssid_to_utf8(nm_ap) 948 if nm_ap != active_ap and ap_name == active_ap_name: 949 # Skip adding AP if it's not active but same name as active AP 950 continue 951 if ap_name not in ap_names: 952 ap_names.append(ap_name) 953 aps.append(nm_ap) 954 return aps, active_ap, active_ap_con, adapter 955 956 957def notify(message, details=None, urgency="low"): 958 """Use notify-send if available for notifications 959 960 """ 961 delay = CONF.getint('nmdm', 'rescan_delay', fallback=5) 962 args = ["-u", urgency, "-a", "networkmanager-dmenu", 963 "-t", str(delay * 1000), message] 964 if details is not None: 965 args.append(details) 966 if is_installed("notify-send"): 967 subprocess.run(["notify-send"] + args, check=False) 968 969 970def run(): # pylint: disable=too-many-locals 971 """Main script entrypoint""" 972 try: 973 subprocess.check_output(["pidof", "NetworkManager"]) 974 except subprocess.CalledProcessError: 975 notify("WARNING: NetworkManager don't seems to be running") 976 print("WARNING: NetworkManager don't seems to be running") 977 active = CLIENT.get_active_connections() 978 adapter = choose_adapter(CLIENT) 979 if adapter: 980 ap_actions = create_ap_actions(*create_ap_list(adapter, active)) 981 else: 982 ap_actions = [] 983 984 vpns = [i for i in CONNS if i.is_type(NM.SETTING_VPN_SETTING_NAME)] 985 try: 986 wgs = [i for i in CONNS if i.is_type(NM.SETTING_WIREGUARD_SETTING_NAME)] 987 except AttributeError: 988 # Workaround for older versions of python-gobject with no wireguard support 989 wgs = [] 990 eths = [i for i in CONNS if i.is_type(NM.SETTING_WIRED_SETTING_NAME)] 991 vlans = [i for i in CONNS if i.is_type(NM.SETTING_VLAN_SETTING_NAME)] 992 blues = [i for i in CONNS if i.is_type(NM.SETTING_BLUETOOTH_SETTING_NAME)] 993 994 vpn_actions = create_vpn_actions(vpns, active) 995 wg_actions = create_wireguard_actions(wgs, active) 996 eth_actions = create_eth_actions(eths, active) 997 vlan_actions = create_vlan_actions(vlans, active) 998 blue_actions = create_blue_actions(blues, active) 999 other_actions = create_other_actions(CLIENT) 1000 wwan_installed = is_installed("ModemManager") 1001 if wwan_installed: 1002 gsms = [i for i in CONNS if i.is_type(NM.SETTING_GSM_SETTING_NAME)] 1003 gsm_actions = create_gsm_actions(gsms, active) 1004 wwan_actions = create_wwan_actions(CLIENT) 1005 else: 1006 gsm_actions = [] 1007 wwan_actions = [] 1008 1009 list_saved = CONF.getboolean('dmenu', 'list_saved', fallback=False) 1010 saved_cons = [i for i in CONNS if i not in vpns + wgs + eths + blues] 1011 if list_saved: 1012 saved_actions = create_saved_actions(saved_cons) 1013 else: 1014 saved_actions = [Action("Saved connections", prompt_saved, [saved_cons])] 1015 1016 1017 actions = combine_actions(eth_actions, ap_actions, vlan_actions, vpn_actions, 1018 wg_actions, gsm_actions, blue_actions, wwan_actions, 1019 other_actions, saved_actions) 1020 sel = get_selection(actions) 1021 sel() 1022 1023 1024def main(): 1025 """Main. Enables script to be re-run after a wifi rescan 1026 1027 """ 1028 global CLIENT, CONNS, LOOP # noqa pylint: disable=global-variable-undefined 1029 CLIENT = NM.Client.new(None) 1030 LOOP = GLib.MainLoop() 1031 CONNS = CLIENT.get_connections() 1032 1033 run() 1034 1035 1036if __name__ == '__main__': 1037 main() 1038 1039# vim: set et ts=4 sw=4 :