Experimental network management GUI added. This may be broken.
This commit is contained in:
Storm Dragon committed
1 parent
f18b6955a9
commit
f3a082219a
5 files changed
+1110
No files matched your search
@@ -161,6 +161,7 @@ Panel Mode automatically returns you to Default Mode after a selection is made o
|
||||
`s`: Show system information.<br>
|
||||
`r`: Open the reminder panel, if remind is installed.<br>
|
||||
`n`: Launch the notes application.<br>
|
||||
`Shift` + `n`: Manage wired and Wi-Fi connections.<br>
|
||||
`b`: Open Bluetooth, if blueman was installed when the i3 config was generated.<br>
|
||||
`@MODKEY@` + `b`: Speak a quick battery status.<br>
|
||||
`Shift` + `b`: Show detailed battery information.<br>
|
||||
@@ -173,6 +174,47 @@ Panel Mode automatically returns you to Default Mode after a selection is made o
|
||||
|
||||
The system information panel, `s` in Panel Mode, displays system statistics such as CPU usage, memory usage, disk space, and network status.
|
||||
|
||||
### Network Connections
|
||||
|
||||
Press `Shift` + `n` in Panel Mode to open I38 Network. Use Up and Down to select
|
||||
a connection, then Tab to Connect, Disconnect, Refresh, Details, or Close.
|
||||
Selecting a row does not change connections. Rows include the connection name,
|
||||
adapter, state, and Wi-Fi security when available. Type a name prefix
|
||||
to narrow the list; Escape clears the search, then closes the window. Scans
|
||||
preserve the selected connection and do not reorder existing rows.
|
||||
|
||||
The panel shows nearby Wi-Fi networks and saved wired and Wi-Fi profiles.
|
||||
Connect supports existing profiles and new open, WPA Personal, WPA3 Personal,
|
||||
and enhanced-open Wi-Fi networks. New enterprise, WEP, hidden-network, VPN, and
|
||||
advanced IP/DNS configuration are outside this panel's scope. Details provides
|
||||
reviewable signal strength, connection, and address information. Signal changes
|
||||
do not rewrite list rows, avoiding repeated announcements. Internet connectivity is shown
|
||||
separately from the connection state; a connection alone does not prove Internet
|
||||
access, and captive portals may require signing in through a browser.
|
||||
|
||||
Wi-Fi passwords use a normal, labeled GTK password field with Show password and
|
||||
Remember password controls. NetworkManager owns saved profiles and passwords;
|
||||
I38 has no password store. Remembered passwords are saved with the system-wide
|
||||
connection, subject to NetworkManager's access controls. Unchecking Remember
|
||||
password on a new network saves its profile without the password and disables
|
||||
automatic connection for that profile. On an existing profile, unchecking it
|
||||
uses the supplied password for that attempt without changing saved credentials.
|
||||
Saving a replacement password requires permission to modify that profile.
|
||||
|
||||
Connection changes require the current login session to belong to Stormux's
|
||||
`network` group and NetworkManager to authorize the requested operation. After
|
||||
an administrator changes group membership, log out and back in. The GUI runs
|
||||
as your normal user. A polkit handler scoped to its own process declines
|
||||
administrator password challenges, including late challenges after a policy
|
||||
change. If this protection cannot be registered, changes remain disabled.
|
||||
Other applications' authentication agents are unaffected. Missing permissions
|
||||
leave connection information available and display an explanation.
|
||||
|
||||
Closing the panel leaves established connections running. Disconnect may allow
|
||||
NetworkManager to activate another saved connection; disable Wi-Fi to turn off
|
||||
the wireless radio. This does not disable wired networking. A hardware wireless
|
||||
block must be released using the device's hardware control.
|
||||
|
||||
### Battery Information
|
||||
|
||||
The battery report has two modes. Press `@MODKEY@` + `b` in Panel Mode for a concise spoken announcement of charge level and charging state. Press `Shift` + `b` in Panel Mode for the detailed report. The detailed report is shown in a reviewable text dialog instead of spoken aloud, so the full details can be navigated with the arrow keys.
|
||||
|
||||
@@ -37,6 +37,7 @@ Optional features use these packages when installed:
|
||||
- libnotify: For sending notifications
|
||||
- magic-wormhole: [optional] for file sharing with magic-wormhole GUI
|
||||
- notification-daemon: To handle notifications
|
||||
- networkmanager: [optional] for the Network panel, including libnm's GObject introspection data.
|
||||
- pamixer: for the mute-unmute script
|
||||
- pcmanfm: [optional] Graphical file manager.
|
||||
- playerctl: music controls
|
||||
@@ -110,6 +111,82 @@ Personal (if configured), and Bypass modes; check existing mode sounds and
|
||||
Cthulhu's native mode speech. Also check battery status, OCR, and USB alerts if
|
||||
enabled. Automated checks do not prove audible output or physical key behavior.
|
||||
|
||||
## Network Panel (Optional)
|
||||
|
||||
Press Control+Alt+Tab, then Shift+N to open the accessible wired/Wi-Fi panel.
|
||||
It requires NetworkManager/libnm 1.30 or newer and its `NM-1.0` introspection
|
||||
data, plus I38's existing Python/PyGObject/GTK 3 dependencies. On Stormux,
|
||||
connection management requires membership in `network` in the current session
|
||||
and an appropriate polkit policy. The GUI never elevates or requests an
|
||||
administrator password. If permissions or its process-specific protection
|
||||
against authentication prompts are unavailable, it remains usable for viewing
|
||||
connection information. No group membership or policy is changed by the panel.
|
||||
|
||||
Try the source directly before updating your installation:
|
||||
|
||||
```bash
|
||||
python3 -B scripts/network.py
|
||||
```
|
||||
|
||||
Use the arrows to select a connection; Tab to Connect, Disconnect, Refresh,
|
||||
Details, or Close. Selection alone does not connect or disconnect. Type a
|
||||
case-insensitive name prefix to filter. Escape clears the filter, then closes
|
||||
the window. Nearby access points with the same SSID and security are combined
|
||||
per adapter, while distinct saved profiles remain selectable. Existing rows
|
||||
retain their order and selection when scans change. Signal strength is available
|
||||
in Details; fluctuations do not rewrite the focused row.
|
||||
|
||||
New personal Wi-Fi networks use an accessible GTK password dialog. Remember
|
||||
password saves through NetworkManager with the system-wide profile, under its
|
||||
access controls. Without remembering, a new profile is saved without its
|
||||
password and with autoconnect disabled. Existing profiles can request another
|
||||
password, including after a wrong-password failure. Saving a replacement
|
||||
requires modification permission; otherwise it is used only for that attempt.
|
||||
I38 stores no passwords itself. Enterprise/hidden-network setup, VPN setup, and
|
||||
advanced IP/DNS editing are not included. Closing the window leaves established
|
||||
connections running. See the Network Connections section of the I38 guide.
|
||||
|
||||
For the installed binding, review the diff and regenerate your I38 configuration,
|
||||
then reload i3. `./i38.sh -u` copies scripts but does not add Shift+N to existing
|
||||
bindings. Regeneration uses your saved preferences; review them first.
|
||||
|
||||
Automated checks (no host connections are changed):
|
||||
|
||||
```bash
|
||||
dbus-run-session -- env I38_PRIVATE_NETWORK_TEST=1 python3 -B -m unittest discover -s tests -p test_network.py -v
|
||||
xvfb-run -a dbus-run-session -- env -u AT_SPI_BUS_ADDRESS I38_PRIVATE_NETWORK_GUI_TEST=1 python3 -B tests/native_network_gui.py -v
|
||||
```
|
||||
|
||||
For a native check against the running NetworkManager and polkit services, add
|
||||
`I38_NETWORK_LIVE_READONLY_TEST=1` to the second command's environment. It reads
|
||||
connection information, registers the panel's temporary agents, and checks that
|
||||
a polkit challenge for this process is declined when the local policy offers
|
||||
one. It does not activate, disconnect, scan, or modify connections or policy.
|
||||
|
||||
Before submission, review and test the changes as required by this project's
|
||||
AI-assisted contributions policy. For live testing with Cthulhu and Orca:
|
||||
|
||||
1. Open the source panel. Check initial selection, complete row announcements,
|
||||
prefix search, Tab/Shift+Tab, status review, Details, and Escape. Check Braille
|
||||
if available. Refresh must preserve focus and selection without repeated
|
||||
signal-strength announcements.
|
||||
2. Join a test Wi-Fi network. Check password masking, Show password, Remember
|
||||
password, cancellation, an incorrect password, a correct retry, and reopening
|
||||
the panel. Verify remembered credentials reconnect and an unremembered new
|
||||
profile requires its password again. No administrator dialog should appear.
|
||||
3. Check an existing wired connection and Wi-Fi, including both active together,
|
||||
disconnect/reconnect, Wi-Fi disabled/enabled, and hardware wireless blocking
|
||||
if available. Confirm normal networking still works after the panel closes.
|
||||
4. If a restricted test account is available, check that missing `network`
|
||||
membership or denied authorization leaves information readable and changes
|
||||
disabled. A policy change while a password dialog is open must be rechecked.
|
||||
5. After configuration regeneration, check Panel Mode Shift+N, existing lowercase
|
||||
`n` for Notes, F1 help, and the other panel bindings. Opening the network panel
|
||||
twice should focus its existing window.
|
||||
|
||||
Private GTK and D-Bus checks do not establish audible speech, Braille, physical
|
||||
key behavior, or successful live connection/password persistence.
|
||||
|
||||
## USB Connection Alerts (Optional)
|
||||
|
||||
Setup offers USB connection sounds and spoken device names separately from window
|
||||
|
||||
@@ -1443,6 +1443,9 @@ fi)
|
||||
# Simple notes system bound to n
|
||||
bindsym n exec --no-startup-id ${i3Path}/scripts/notes.py, mode "default"
|
||||
|
||||
# Network connections bound to Shift+n
|
||||
bindsym Shift+n exec --no-startup-id python3 ${i3Path}/scripts/network.py, mode "default"
|
||||
|
||||
# Password manager bound to m
|
||||
bindsym m exec --no-startup-id ${i3Path}/scripts/passmanager.py, mode "default"
|
||||
|
||||
|
||||
@@ -0,0 +1,417 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Accessible, ordinary-user NetworkManager panel for I38.
|
||||
|
||||
Copyright (C) 2026 I38 contributors
|
||||
SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
import gi
|
||||
|
||||
gi.require_version("Atk", "1.0")
|
||||
gi.require_version("Gdk", "3.0")
|
||||
gi.require_version("Gtk", "3.0")
|
||||
from gi.repository import Atk, Gdk, Gio, Gtk
|
||||
|
||||
try:
|
||||
from network_backend import (MODIFY_SYSTEM, NETWORK_CONTROL, NM, NetworkBackend,
|
||||
security_type, valid_password)
|
||||
except (ImportError, ValueError):
|
||||
NetworkBackend = None
|
||||
|
||||
|
||||
def labeled_entry(labelText, entry):
|
||||
label = Gtk.Label.new_with_mnemonic(labelText)
|
||||
label.set_xalign(0)
|
||||
label.set_mnemonic_widget(entry)
|
||||
return label
|
||||
|
||||
|
||||
class PasswordDialog(Gtk.Dialog):
|
||||
def __init__(self, parent, name, keyMgmt, canRemember):
|
||||
super().__init__(title="Wi-Fi password", transient_for=parent, modal=True)
|
||||
self.keyMgmt = keyMgmt
|
||||
self.add_button("_Cancel", Gtk.ResponseType.CANCEL)
|
||||
self.connectButton = self.add_button("_Connect", Gtk.ResponseType.OK)
|
||||
self.set_default_response(Gtk.ResponseType.OK)
|
||||
box = self.get_content_area()
|
||||
box.set_spacing(10)
|
||||
box.set_border_width(12)
|
||||
heading = Gtk.Label(label="Enter the Wi-Fi password for {}.".format(name))
|
||||
heading.set_line_wrap(True)
|
||||
box.add(heading)
|
||||
self.entry = Gtk.Entry()
|
||||
self.entry.set_visibility(False)
|
||||
self.entry.set_input_purpose(Gtk.InputPurpose.PASSWORD)
|
||||
self.entry.set_activates_default(True)
|
||||
box.add(labeled_entry("_Password:", self.entry))
|
||||
box.add(self.entry)
|
||||
show = Gtk.CheckButton.new_with_mnemonic("_Show password")
|
||||
show.connect("toggled", lambda button: self.entry.set_visibility(button.get_active()))
|
||||
box.add(show)
|
||||
self.remember = Gtk.CheckButton.new_with_mnemonic("_Remember password")
|
||||
self.remember.set_active(canRemember)
|
||||
self.remember.set_sensitive(canRemember)
|
||||
self.remember.get_accessible().set_description(
|
||||
"NetworkManager saves the password with this connection for system users."
|
||||
if canRemember else "Saving this password is not authorized. It will only be used for this attempt.")
|
||||
box.add(self.remember)
|
||||
self.validation = Gtk.Label(label="")
|
||||
self.validation.set_xalign(0)
|
||||
box.add(self.validation)
|
||||
self.entry.connect("changed", self.on_changed)
|
||||
self.on_changed(self.entry)
|
||||
self.connect("key-press-event", self.on_key)
|
||||
self.show_all()
|
||||
self.entry.grab_focus()
|
||||
|
||||
def on_changed(self, entry):
|
||||
valid = valid_password(entry.get_text(), self.keyMgmt)
|
||||
self.connectButton.set_sensitive(valid)
|
||||
self.validation.set_text("" if valid else (
|
||||
"Use 8–63 characters or 64 hexadecimal digits." if self.keyMgmt == "wpa-psk"
|
||||
else "Enter a password of 1–63 UTF-8 bytes."))
|
||||
|
||||
def on_key(self, widget, event):
|
||||
if event.keyval == Gdk.KEY_Escape:
|
||||
self.response(Gtk.ResponseType.CANCEL)
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class NetworkWindow(Gtk.ApplicationWindow):
|
||||
def __init__(self, application, backendFactory=None):
|
||||
super().__init__(application=application, title="I38 Network")
|
||||
self.set_default_size(850, 500)
|
||||
self.backend = None
|
||||
self.passwordDialog = None
|
||||
self.resultDialog = None
|
||||
self.lastBusy = False
|
||||
self.updating = False
|
||||
self.rowsByKey = {}
|
||||
self.connect("destroy", self.on_destroy)
|
||||
self.connect("key-press-event", self.on_key)
|
||||
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=10)
|
||||
box.set_border_width(12)
|
||||
self.add(box)
|
||||
|
||||
self.summary = Gtk.Label(label="Loading NetworkManager…")
|
||||
self.summary.set_xalign(0)
|
||||
self.summary.set_line_wrap(True)
|
||||
box.pack_start(self.summary, False, False, 0)
|
||||
self.wireless = Gtk.CheckButton.new_with_mnemonic("_Enable Wi-Fi")
|
||||
self.wireless.connect("toggled", self.on_wireless)
|
||||
box.pack_start(self.wireless, False, False, 0)
|
||||
|
||||
self.search = Gtk.SearchEntry()
|
||||
self.search.get_accessible().set_name("Find connection by name prefix")
|
||||
box.pack_start(labeled_entry("_Find connection:", self.search), False, False, 0)
|
||||
box.pack_start(self.search, False, False, 0)
|
||||
# A single text column keeps a row's complete state in one accessible cell.
|
||||
self.store = Gtk.ListStore(object, str, str)
|
||||
self.filtered = self.store.filter_new()
|
||||
self.filtered.set_visible_func(self.row_visible)
|
||||
self.tree = Gtk.TreeView(model=self.filtered)
|
||||
self.tree.set_headers_visible(False)
|
||||
self.tree.set_enable_search(False)
|
||||
self.tree.get_accessible().set_name("Network connections")
|
||||
self.tree.get_accessible().set_description(
|
||||
"Use Up and Down to select a connection, then Tab to Connect, Disconnect, or Details. Selection does not change the network.")
|
||||
self.tree.append_column(Gtk.TreeViewColumn("Connection", Gtk.CellRendererText(), text=1))
|
||||
self.tree.get_selection().connect("changed", self.on_selection)
|
||||
self.tree.connect("key-press-event", self.on_list_key)
|
||||
scroll = Gtk.ScrolledWindow()
|
||||
scroll.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
|
||||
scroll.add(self.tree)
|
||||
box.pack_start(scroll, True, True, 0)
|
||||
self.search.connect("search-changed", self.on_search)
|
||||
|
||||
self.permissionText = Gtk.Label(label="Checking network permissions…")
|
||||
self.permissionText.set_xalign(0)
|
||||
self.permissionText.set_line_wrap(True)
|
||||
self.permissionText.set_selectable(True)
|
||||
box.pack_start(self.permissionText, False, False, 0)
|
||||
self.statusEntry = Gtk.Entry()
|
||||
self.statusEntry.set_editable(False)
|
||||
self.statusEntry.set_text("Loading connections…")
|
||||
self.statusEntry.get_accessible().set_role(Atk.Role.STATUSBAR)
|
||||
box.pack_start(labeled_entry("_Status:", self.statusEntry), False, False, 0)
|
||||
box.pack_start(self.statusEntry, False, False, 0)
|
||||
buttons = Gtk.Box(spacing=8)
|
||||
box.pack_start(buttons, False, False, 0)
|
||||
self.connectButton = self.add_button(buttons, "_Connect", self.on_connect)
|
||||
self.disconnectButton = self.add_button(buttons, "_Disconnect", self.on_disconnect)
|
||||
self.refreshButton = self.add_button(buttons, "_Refresh", lambda button: self.backend.scan())
|
||||
self.detailsButton = self.add_button(buttons, "De_tails", self.on_details)
|
||||
self.add_button(buttons, "Cl_ose", lambda button: self.close())
|
||||
self.show_all()
|
||||
if backendFactory is None:
|
||||
backendFactory = NetworkBackend
|
||||
if backendFactory:
|
||||
self.backend = backendFactory(self.refresh, self.on_status, self.request_password)
|
||||
self.refresh()
|
||||
else:
|
||||
self.on_status("NetworkManager's Python bindings are unavailable. Install NetworkManager/libnm 1.30 or newer and its GObject introspection data.")
|
||||
self.permissionText.set_text("Connection management is unavailable.")
|
||||
for button in (self.connectButton, self.disconnectButton, self.refreshButton, self.wireless):
|
||||
button.set_sensitive(False)
|
||||
self.tree.grab_focus()
|
||||
|
||||
def add_button(self, box, text, handler):
|
||||
button = Gtk.Button.new_with_mnemonic(text)
|
||||
button.connect("clicked", handler)
|
||||
box.add(button)
|
||||
return button
|
||||
|
||||
def selected_row(self):
|
||||
model, treeIter = self.tree.get_selection().get_selected()
|
||||
return self.rowsByKey.get(model[treeIter][0]) if treeIter is not None else None
|
||||
|
||||
def row_visible(self, model, treeIter, data):
|
||||
return model[treeIter][2].casefold().startswith(self.search.get_text().casefold())
|
||||
|
||||
def on_search(self, entry):
|
||||
self.filtered.refilter()
|
||||
if self.selected_row() is None and len(self.filtered):
|
||||
self.tree.set_cursor(Gtk.TreePath.new_first())
|
||||
self.on_selection()
|
||||
|
||||
def on_list_key(self, widget, event):
|
||||
if event.state & (Gdk.ModifierType.CONTROL_MASK | Gdk.ModifierType.MOD1_MASK):
|
||||
return False
|
||||
character = Gdk.keyval_to_unicode(event.keyval)
|
||||
if character and chr(character).isprintable():
|
||||
self.search.set_text(self.search.get_text() + chr(character))
|
||||
self.search.set_position(-1)
|
||||
return True
|
||||
if event.keyval == Gdk.KEY_BackSpace:
|
||||
self.search.set_text(self.search.get_text()[:-1])
|
||||
return True
|
||||
return False
|
||||
|
||||
def on_key(self, widget, event):
|
||||
if event.keyval == Gdk.KEY_Escape:
|
||||
if self.search.get_text():
|
||||
self.search.set_text("")
|
||||
self.tree.grab_focus()
|
||||
else:
|
||||
self.close()
|
||||
return True
|
||||
return False
|
||||
|
||||
def refresh(self):
|
||||
if self.backend is None:
|
||||
return
|
||||
oldRow = self.selected_row()
|
||||
oldKey = oldRow.key if oldRow else None
|
||||
rows = self.backend.rows()
|
||||
self.rowsByKey = {row.key: row for row in rows}
|
||||
desiredKeys = set(self.rowsByKey)
|
||||
# Update in place: scans and signal changes never rebuild the focused list.
|
||||
treeIter = self.store.get_iter_first()
|
||||
existing = set()
|
||||
while treeIter is not None:
|
||||
key = self.store[treeIter][0]
|
||||
if key not in desiredKeys:
|
||||
if not self.store.remove(treeIter):
|
||||
break
|
||||
continue
|
||||
existing.add(key)
|
||||
row = self.rowsByKey[key]
|
||||
text = self.row_text(row)
|
||||
if self.store[treeIter][1] != text:
|
||||
self.store[treeIter][1] = text
|
||||
if self.store[treeIter][2] != row.name:
|
||||
self.store[treeIter][2] = row.name
|
||||
treeIter = self.store.iter_next(treeIter)
|
||||
for row in rows:
|
||||
if row.key not in existing:
|
||||
self.store.append((row.key, self.row_text(row), row.name))
|
||||
selected = self.selected_row()
|
||||
if selected is None:
|
||||
firstPath = None
|
||||
for modelRow in self.filtered:
|
||||
row = self.rowsByKey[modelRow[0]]
|
||||
if firstPath is None:
|
||||
firstPath = modelRow.path
|
||||
if row.key == oldKey or (oldKey is None and row.active):
|
||||
firstPath = modelRow.path
|
||||
break
|
||||
if firstPath is not None:
|
||||
self.tree.set_cursor(firstPath)
|
||||
client = self.backend.client
|
||||
self.updating = True
|
||||
self.wireless.set_active(bool(client and client.wireless_get_enabled()))
|
||||
self.updating = False
|
||||
if not client or not client.get_nm_running():
|
||||
self.summary.set_text("NetworkManager is unavailable.")
|
||||
else:
|
||||
if self.statusEntry.get_text() == "Loading connections…":
|
||||
self.statusEntry.set_text("Select a connection, then choose Connect or Disconnect.")
|
||||
connectivity = client.get_connectivity()
|
||||
descriptions = {NM.ConnectivityState.FULL: "Internet connectivity confirmed",
|
||||
NM.ConnectivityState.PORTAL: "Network sign-in may be required",
|
||||
NM.ConnectivityState.LIMITED: "Limited connectivity",
|
||||
NM.ConnectivityState.NONE: "No Internet connectivity",
|
||||
NM.ConnectivityState.UNKNOWN: "Internet connectivity not checked"}
|
||||
activeNames = [row.name for row in rows if row.state == "Connected"]
|
||||
self.summary.set_text("{}. {}. {} connections listed.".format(
|
||||
"Connected: " + ", ".join(activeNames) if activeNames else "No active wired or Wi-Fi connection",
|
||||
descriptions.get(connectivity, "Internet connectivity unknown"), len(rows)))
|
||||
self.on_selection()
|
||||
self.refreshButton.set_sensitive(bool(client and not self.backend.busy and
|
||||
not self.backend.permission_problem([NM.ClientPermission.WIFI_SCAN])))
|
||||
self.wireless.set_sensitive(bool(client and not self.backend.busy and client.wireless_hardware_get_enabled() and
|
||||
not self.backend.permission_problem([NM.ClientPermission.ENABLE_DISABLE_WIFI])))
|
||||
self.lastBusy = self.backend.busy
|
||||
if not self.backend.secretRequest and self.passwordDialog is not None and self.backend.busy:
|
||||
self.passwordDialog.response(Gtk.ResponseType.CANCEL)
|
||||
|
||||
def row_text(self, row, includeSignal=False):
|
||||
parts = [row.name, row.kind, row.device.get_iface(), row.state]
|
||||
if includeSignal and row.signal:
|
||||
parts.append("signal " + row.signal)
|
||||
if row.security:
|
||||
parts.append(row.security)
|
||||
if row.profile:
|
||||
parts.append("saved")
|
||||
return ", ".join(parts)
|
||||
|
||||
def on_selection(self, *args):
|
||||
if not self.backend:
|
||||
return
|
||||
row = self.selected_row()
|
||||
permissions = [NETWORK_CONTROL] + ([MODIFY_SYSTEM] if row and not row.profile else [])
|
||||
problem = self.backend.permission_problem(permissions)
|
||||
self.permissionText.set_text(problem or "Connection management is authorized. Administrator passwords are never requested.")
|
||||
available = bool(row and row.state not in ("Unavailable", "Unmanaged", "Cable unplugged", "Hardware blocked", "Wi-Fi disabled"))
|
||||
supported = bool(row and (row.profile or security_type(row.ap)[1] is not None))
|
||||
self.connectButton.set_sensitive(bool(available and supported and not row.active and
|
||||
not problem and self.backend.agentReady and not self.backend.busy))
|
||||
self.disconnectButton.set_sensitive(bool(row and row.active and not self.backend.busy and
|
||||
not self.backend.permission_problem([NETWORK_CONTROL])))
|
||||
self.detailsButton.set_sensitive(row is not None)
|
||||
|
||||
def on_status(self, message):
|
||||
self.statusEntry.set_text(message)
|
||||
self.statusEntry.set_tooltip_text(message)
|
||||
if self.backend and self.lastBusy and not self.backend.busy:
|
||||
self.lastBusy = False
|
||||
if self.passwordDialog:
|
||||
self.passwordDialog.response(Gtk.ResponseType.CANCEL)
|
||||
self.show_message("Network operation", message)
|
||||
|
||||
def show_message(self, title, message):
|
||||
if self.resultDialog:
|
||||
self.resultDialog.destroy()
|
||||
dialog = Gtk.MessageDialog(transient_for=self, modal=True, message_type=Gtk.MessageType.INFO,
|
||||
buttons=Gtk.ButtonsType.CLOSE, text=title)
|
||||
dialog.format_secondary_text(message)
|
||||
dialog.connect("response", lambda widget, response: widget.destroy())
|
||||
dialog.connect("destroy", self.on_result_closed)
|
||||
self.resultDialog = dialog
|
||||
dialog.show_all()
|
||||
|
||||
def on_result_closed(self, dialog):
|
||||
if self.resultDialog == dialog:
|
||||
self.resultDialog = None
|
||||
|
||||
def request_password(self, name, keyMgmt, canRemember, callback):
|
||||
if self.passwordDialog:
|
||||
self.passwordDialog.response(Gtk.ResponseType.CANCEL)
|
||||
dialog = PasswordDialog(self, name, keyMgmt, canRemember)
|
||||
self.passwordDialog = dialog
|
||||
|
||||
def responded(widget, response):
|
||||
password = widget.entry.get_text() if response == Gtk.ResponseType.OK else None
|
||||
remember = widget.remember.get_active()
|
||||
widget.entry.set_text("")
|
||||
self.passwordDialog = None
|
||||
widget.destroy()
|
||||
callback(password, remember)
|
||||
|
||||
dialog.connect("response", responded)
|
||||
|
||||
def on_connect(self, button):
|
||||
row = self.selected_row()
|
||||
if not row:
|
||||
return
|
||||
permissions = [NETWORK_CONTROL] + ([] if row.profile else [MODIFY_SYSTEM])
|
||||
if not self.backend.authorize(permissions):
|
||||
return
|
||||
if row.profile or security_type(row.ap)[1] not in ("wpa-psk", "sae"):
|
||||
self.backend.connect_row(row)
|
||||
return
|
||||
|
||||
def answered(password, remember):
|
||||
if password is not None:
|
||||
self.backend.connect_row(row, password, remember)
|
||||
|
||||
self.request_password(row.name, security_type(row.ap)[1], True, answered)
|
||||
|
||||
def on_disconnect(self, button):
|
||||
row = self.selected_row()
|
||||
if row:
|
||||
self.backend.disconnect_row(row)
|
||||
|
||||
def on_wireless(self, button):
|
||||
if not self.updating and self.backend:
|
||||
self.backend.set_wireless(button.get_active())
|
||||
self.refresh()
|
||||
|
||||
def on_details(self, button):
|
||||
row = self.selected_row()
|
||||
if not row:
|
||||
return
|
||||
lines = [self.row_text(row, includeSignal=True)]
|
||||
if row.profile:
|
||||
lines.append("Profile UUID: " + row.profile.get_uuid())
|
||||
lines.append("Connect automatically: " + ("Yes" if row.profile.get_setting_connection().get_autoconnect() else "No"))
|
||||
if row.ap and security_type(row.ap)[1] is None:
|
||||
lines.append("Creating profiles for this security type is not supported. Existing profiles can be activated.")
|
||||
if row.active:
|
||||
for version, config in (("IPv4", row.device.get_ip4_config()), ("IPv6", row.device.get_ip6_config())):
|
||||
if config:
|
||||
lines.extend("{} address: {}/{}".format(version, address.get_address(), address.get_prefix()) for address in config.get_addresses())
|
||||
if config.get_gateway():
|
||||
lines.append("{} gateway: {}".format(version, config.get_gateway()))
|
||||
lines.append("{} DNS: {}".format(version, ", ".join(config.get_nameservers())))
|
||||
dialog = Gtk.Dialog(title="Connection details", transient_for=self, modal=True)
|
||||
dialog.add_button("_Close", Gtk.ResponseType.CLOSE)
|
||||
view = Gtk.TextView()
|
||||
view.set_editable(False)
|
||||
view.set_accepts_tab(False)
|
||||
view.set_wrap_mode(Gtk.WrapMode.WORD_CHAR)
|
||||
view.get_accessible().set_name("Connection details")
|
||||
view.get_buffer().set_text("\n".join(lines))
|
||||
scroll = Gtk.ScrolledWindow()
|
||||
scroll.add(view)
|
||||
dialog.get_content_area().add(scroll)
|
||||
dialog.set_default_size(600, 350)
|
||||
dialog.connect("response", lambda widget, response: widget.destroy())
|
||||
dialog.show_all()
|
||||
view.grab_focus()
|
||||
|
||||
def on_destroy(self, widget):
|
||||
if self.backend:
|
||||
self.backend.close()
|
||||
if self.passwordDialog:
|
||||
self.passwordDialog.response(Gtk.ResponseType.CANCEL)
|
||||
if self.resultDialog:
|
||||
self.resultDialog.destroy()
|
||||
|
||||
|
||||
class NetworkApplication(Gtk.Application):
|
||||
def __init__(self):
|
||||
super().__init__(application_id="org.stormux.I38.Network", flags=Gio.ApplicationFlags.FLAGS_NONE)
|
||||
|
||||
def do_activate(self):
|
||||
window = self.get_active_window()
|
||||
if window is None:
|
||||
window = NetworkWindow(self)
|
||||
window.present()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(NetworkApplication().run(sys.argv))
|
||||
@@ -0,0 +1,571 @@
|
||||
#!/usr/bin/env python3
|
||||
"""NetworkManager operations for I38; no elevated GUI or private password store.
|
||||
|
||||
Copyright (C) 2026 I38 contributors
|
||||
SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""
|
||||
|
||||
import grp
|
||||
import locale
|
||||
import os
|
||||
from collections import namedtuple
|
||||
from pathlib import Path
|
||||
import uuid
|
||||
|
||||
import gi
|
||||
|
||||
gi.require_version("NM", "1.0")
|
||||
from gi.repository import Gio, GLib, GObject, NM
|
||||
|
||||
if (NM.MAJOR_VERSION, NM.MINOR_VERSION) < (1, 30):
|
||||
raise ValueError("I38 Network requires libnm 1.30 or newer.")
|
||||
|
||||
|
||||
NM_NAME = "org.freedesktop.NetworkManager"
|
||||
NM_PATH = "/org/freedesktop/NetworkManager"
|
||||
POLKIT_NAME = "org.freedesktop.PolicyKit1"
|
||||
POLKIT_PATH = "/org/freedesktop/PolicyKit1/Authority"
|
||||
AUTH_PATH = "/org/stormux/I38/NetworkAuthentication"
|
||||
SECRET_PATH = NM_PATH + "/SecretAgent"
|
||||
SECRET_INTERFACE = NM_NAME + ".SecretAgent"
|
||||
SECURITY_SETTING = "802-11-wireless-security"
|
||||
NETWORK_CONTROL = NM.ClientPermission.NETWORK_CONTROL
|
||||
MODIFY_SYSTEM = NM.ClientPermission.SETTINGS_MODIFY_SYSTEM
|
||||
MODIFY_OWN = NM.ClientPermission.SETTINGS_MODIFY_OWN
|
||||
ApFlags = getattr(NM, "80211ApFlags")
|
||||
ApSecurityFlags = getattr(NM, "80211ApSecurityFlags")
|
||||
|
||||
NetworkRow = namedtuple("NetworkRow", "key name kind state signal security device profile ap active")
|
||||
|
||||
AUTH_XML = """<node><interface name="org.freedesktop.PolicyKit1.AuthenticationAgent">
|
||||
<method name="BeginAuthentication"><arg type="s" direction="in"/>
|
||||
<arg type="s" direction="in"/><arg type="s" direction="in"/>
|
||||
<arg type="a{ss}" direction="in"/><arg type="s" direction="in"/>
|
||||
<arg type="a(sa{sv})" direction="in"/></method>
|
||||
<method name="CancelAuthentication"><arg type="s" direction="in"/></method>
|
||||
</interface></node>"""
|
||||
|
||||
SECRET_XML = """<node><interface name="org.freedesktop.NetworkManager.SecretAgent">
|
||||
<method name="GetSecrets"><arg type="a{sa{sv}}" direction="in"/>
|
||||
<arg type="o" direction="in"/><arg type="s" direction="in"/>
|
||||
<arg type="as" direction="in"/><arg type="u" direction="in"/>
|
||||
<arg type="a{sa{sv}}" direction="out"/></method>
|
||||
<method name="CancelGetSecrets"><arg type="o" direction="in"/>
|
||||
<arg type="s" direction="in"/></method>
|
||||
<method name="SaveSecrets"><arg type="a{sa{sv}}" direction="in"/>
|
||||
<arg type="o" direction="in"/></method>
|
||||
<method name="DeleteSecrets"><arg type="a{sa{sv}}" direction="in"/>
|
||||
<arg type="o" direction="in"/></method>
|
||||
</interface></node>"""
|
||||
|
||||
|
||||
def network_group_member():
|
||||
"""Check the groups of this session, not newly edited account records."""
|
||||
try:
|
||||
groupId = grp.getgrnam("network").gr_gid
|
||||
except KeyError:
|
||||
return False
|
||||
return groupId in set(os.getgroups()) | {os.getgid()}
|
||||
|
||||
|
||||
def security_type(ap):
|
||||
flags = ap.get_wpa_flags() | ap.get_rsn_flags()
|
||||
if flags & (ApSecurityFlags.KEY_MGMT_802_1X | ApSecurityFlags.KEY_MGMT_EAP_SUITE_B_192):
|
||||
return "Enterprise", None
|
||||
if flags & ApSecurityFlags.KEY_MGMT_PSK:
|
||||
return "WPA Personal", "wpa-psk"
|
||||
if flags & ApSecurityFlags.KEY_MGMT_SAE:
|
||||
return "WPA3 Personal", "sae"
|
||||
if flags & (ApSecurityFlags.KEY_MGMT_OWE | ApSecurityFlags.KEY_MGMT_OWE_TM):
|
||||
return "Enhanced open", "owe"
|
||||
if ap.get_flags() & ApFlags.PRIVACY:
|
||||
return "WEP (unsupported)", None
|
||||
return "Open", ""
|
||||
|
||||
|
||||
def valid_password(password, keyMgmt):
|
||||
if keyMgmt == "wpa-psk":
|
||||
return NM.utils_wpa_psk_valid(password)
|
||||
return keyMgmt == "sae" and 1 <= len(password.encode("utf-8")) <= 63
|
||||
|
||||
|
||||
def new_wifi_connection(row, password, remember):
|
||||
connection = NM.SimpleConnection.new()
|
||||
setting = NM.SettingConnection.new()
|
||||
setting.props.id = row.name
|
||||
setting.props.uuid = str(uuid.uuid4())
|
||||
setting.props.type = "802-11-wireless"
|
||||
setting.props.autoconnect = remember
|
||||
connection.add_setting(setting)
|
||||
wireless = NM.SettingWireless.new()
|
||||
wireless.props.ssid = row.ap.get_ssid()
|
||||
wireless.props.mode = "infrastructure"
|
||||
connection.add_setting(wireless)
|
||||
keyMgmt = security_type(row.ap)[1]
|
||||
if keyMgmt is None:
|
||||
raise ValueError("This Wi-Fi security type is not supported by I38 Network.")
|
||||
if keyMgmt:
|
||||
security = NM.SettingWirelessSecurity.new()
|
||||
security.props.key_mgmt = keyMgmt
|
||||
if keyMgmt in ("wpa-psk", "sae"):
|
||||
if not valid_password(password, keyMgmt):
|
||||
raise ValueError("The Wi-Fi password has an invalid length or format.")
|
||||
security.props.psk = password
|
||||
security.props.psk_flags = NM.SettingSecretFlags.NONE if remember else NM.SettingSecretFlags.NOT_SAVED
|
||||
connection.add_setting(security)
|
||||
connection.normalize()
|
||||
return connection
|
||||
|
||||
|
||||
class PromptGuard:
|
||||
"""Decline polkit challenges for this process, without replacing the session agent.
|
||||
|
||||
NetworkManager permits interactive authorization internally. Preflight checks
|
||||
cannot prevent a late challenge after a policy change. A unix-process agent
|
||||
takes precedence over the session agent and closes that race.
|
||||
"""
|
||||
|
||||
def __init__(self, bus, changed, status):
|
||||
self.bus = bus
|
||||
self.changed = changed
|
||||
self.status = status
|
||||
self.ready = False
|
||||
self.error = ""
|
||||
self.owner = None
|
||||
self.closed = False
|
||||
self.objectId = bus.register_object(
|
||||
AUTH_PATH, Gio.DBusNodeInfo.new_for_xml(AUTH_XML).interfaces[0], self.handle_call, None, None)
|
||||
self.watchId = Gio.bus_watch_name_on_connection(
|
||||
bus, POLKIT_NAME, Gio.BusNameWatcherFlags.NONE, self.on_appeared, self.on_vanished)
|
||||
|
||||
def on_appeared(self, bus, name, owner):
|
||||
self.owner = owner
|
||||
startTime = int(Path("/proc/self/stat").read_text().rsplit(")", 1)[1].split()[19])
|
||||
self.subject = ("unix-process", {"pid": GLib.Variant("u", os.getpid()),
|
||||
"start-time": GLib.Variant("t", startTime),
|
||||
"uid": GLib.Variant("i", os.getuid())})
|
||||
bus.call(owner, POLKIT_PATH, POLKIT_NAME + ".Authority", "RegisterAuthenticationAgent",
|
||||
GLib.Variant("((sa{sv})ss)", (self.subject, locale.setlocale(locale.LC_MESSAGES), AUTH_PATH)),
|
||||
None, Gio.DBusCallFlags.NONE, 10000, None, self.on_registered, owner)
|
||||
|
||||
def on_registered(self, bus, result, owner):
|
||||
try:
|
||||
bus.call_finish(result)
|
||||
self.ready = not self.closed and self.owner == owner
|
||||
except GLib.Error as error:
|
||||
self.ready = False
|
||||
self.error = error.message
|
||||
self.status("Cannot protect against administrator password prompts. Connection changes are disabled.")
|
||||
self.changed()
|
||||
|
||||
def on_vanished(self, bus, name):
|
||||
self.owner = None
|
||||
self.ready = False
|
||||
self.changed()
|
||||
|
||||
def handle_call(self, bus, sender, path, interface, method, parameters, invocation):
|
||||
if sender != self.owner:
|
||||
invocation.return_dbus_error(POLKIT_NAME + ".Error.NotAuthorized", "Unexpected sender")
|
||||
elif method == "BeginAuthentication":
|
||||
invocation.return_dbus_error(POLKIT_NAME + ".Error.Cancelled", "I38 does not request administrator passwords")
|
||||
self.status("System policy requires administrator authorization. This operation was declined; no password dialog was opened.")
|
||||
else:
|
||||
invocation.return_value(GLib.Variant("()", ()))
|
||||
|
||||
def close(self):
|
||||
self.closed = True
|
||||
self.ready = False
|
||||
Gio.bus_unwatch_name(self.watchId)
|
||||
if self.owner and hasattr(self, "subject"):
|
||||
self.bus.call(self.owner, POLKIT_PATH, POLKIT_NAME + ".Authority", "UnregisterAuthenticationAgent",
|
||||
GLib.Variant("((sa{sv})s)", (self.subject, AUTH_PATH)), None,
|
||||
Gio.DBusCallFlags.NONE, 5000, None, None, None)
|
||||
self.bus.unregister_object(self.objectId)
|
||||
|
||||
|
||||
class NetworkBackend:
|
||||
def __init__(self, changed, status, password):
|
||||
self.changed = changed
|
||||
self.status = status
|
||||
self.password = password
|
||||
self.client = None
|
||||
self.guard = None
|
||||
self.agentReady = False
|
||||
self.closed = False
|
||||
self.busy = False
|
||||
self.pending = None
|
||||
self.secretRequest = None
|
||||
self.signalHandlers = []
|
||||
self.cancellable = Gio.Cancellable()
|
||||
self.refreshSource = 0
|
||||
self.member = network_group_member()
|
||||
NM.Client.new_async(self.cancellable, self.on_client_ready, None)
|
||||
|
||||
def on_client_ready(self, source, result, data):
|
||||
try:
|
||||
self.client = NM.Client.new_finish(result)
|
||||
except GLib.Error:
|
||||
self.status("Cannot contact NetworkManager. Check that the NetworkManager service is running.")
|
||||
return
|
||||
if self.closed:
|
||||
return
|
||||
self.bus = self.client.get_dbus_connection()
|
||||
self.guard = PromptGuard(self.bus, self.changed, self.status)
|
||||
self.secretObjectId = self.bus.register_object(
|
||||
SECRET_PATH, Gio.DBusNodeInfo.new_for_xml(SECRET_XML).interfaces[0], self.handle_secret, None, None)
|
||||
self.client.connect("notify", self.on_notify)
|
||||
self.register_agent()
|
||||
self.changed()
|
||||
|
||||
def register_agent(self):
|
||||
self.agentReady = False
|
||||
owner = self.client.get_dbus_name_owner()
|
||||
if not owner:
|
||||
return
|
||||
self.bus.call(owner, NM_PATH + "/AgentManager", NM_NAME + ".AgentManager", "RegisterWithCapabilities",
|
||||
GLib.Variant("(su)", ("org.stormux.I38.Network", 0)), None,
|
||||
Gio.DBusCallFlags.NONE, 10000, self.cancellable, self.on_agent_registered, owner)
|
||||
|
||||
def on_agent_registered(self, bus, result, owner):
|
||||
try:
|
||||
bus.call_finish(result)
|
||||
self.agentReady = not self.closed and self.client.get_dbus_name_owner() == owner
|
||||
except GLib.Error:
|
||||
self.status("Cannot register I38's Wi-Fi password dialog. Connecting is disabled.")
|
||||
self.changed()
|
||||
|
||||
def on_notify(self, obj, prop):
|
||||
if self.closed:
|
||||
return
|
||||
if prop.name == "dbus-name-owner":
|
||||
self.finish_operation("NetworkManager restarted. Review the connection state before trying again.")
|
||||
self.register_agent()
|
||||
self.schedule_refresh()
|
||||
|
||||
def schedule_refresh(self, *args):
|
||||
if not self.closed and not self.refreshSource:
|
||||
self.refreshSource = GLib.timeout_add(250, self.flush_refresh)
|
||||
|
||||
def flush_refresh(self):
|
||||
self.refreshSource = 0
|
||||
self.changed()
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
def permission_problem(self, permissions):
|
||||
if not self.client or not self.client.get_nm_running():
|
||||
return "NetworkManager is not running."
|
||||
if not self.member:
|
||||
return ("Managing connections requires membership in the network group on Stormux. "
|
||||
"Ask your administrator to add your account, then log out and back in.")
|
||||
if not self.guard or not self.guard.ready:
|
||||
if self.guard and self.guard.error:
|
||||
return "Connection changes are disabled: " + self.guard.error
|
||||
return "Connection changes are disabled until protection against administrator prompts is ready."
|
||||
if self.client.get_permissions_state() != NM.Ternary.TRUE:
|
||||
return "Waiting for NetworkManager permissions."
|
||||
for permission in permissions:
|
||||
result = self.client.get_permission_result(permission)
|
||||
if result != NM.ClientPermissionResult.YES:
|
||||
action = permission.value_nick.replace("-", " ")
|
||||
if result == NM.ClientPermissionResult.AUTH:
|
||||
return "System policy requires administrator authorization for {}. Ask your administrator to check the network group policy.".format(action)
|
||||
if result == NM.ClientPermissionResult.UNKNOWN:
|
||||
return "NetworkManager has not provided permission for {}.".format(action)
|
||||
return "System policy denies {} for this account.".format(action)
|
||||
return ""
|
||||
|
||||
def authorize(self, permissions):
|
||||
problem = self.permission_problem(permissions)
|
||||
if problem:
|
||||
self.status(problem)
|
||||
return False
|
||||
return not self.closed
|
||||
|
||||
def modification_permission(self, profile):
|
||||
setting = profile.get_setting_connection()
|
||||
return MODIFY_OWN if setting.get_num_permissions() else MODIFY_SYSTEM
|
||||
|
||||
def rows(self):
|
||||
for obj, handler in self.signalHandlers:
|
||||
GObject.signal_handler_disconnect(obj, handler)
|
||||
self.signalHandlers = []
|
||||
if not self.client or not self.client.get_nm_running():
|
||||
return []
|
||||
rows = []
|
||||
for device in self.client.get_devices():
|
||||
if device.get_device_type() not in (NM.DeviceType.WIFI, NM.DeviceType.ETHERNET):
|
||||
continue
|
||||
self.watch(device)
|
||||
# filter_connections() has a documented GI ownership bug. Validate
|
||||
# individually so Python never unrefs NetworkManager's profiles.
|
||||
profiles = [p for p in self.client.get_connections() if device.connection_valid(p)]
|
||||
for profile in profiles:
|
||||
self.watch(profile)
|
||||
active = device.get_active_connection()
|
||||
activeProfile = active.get_connection() if active else None
|
||||
if active:
|
||||
self.watch(active)
|
||||
if device.get_device_type() == NM.DeviceType.WIFI:
|
||||
groups = {}
|
||||
activeAp = device.get_active_access_point()
|
||||
for ap in device.get_access_points():
|
||||
self.watch(ap)
|
||||
ssid = ap.get_ssid()
|
||||
if not ssid or not ssid.get_data():
|
||||
continue
|
||||
groupKey = (bytes(ssid.get_data()), security_type(ap)[1], security_type(ap)[0])
|
||||
previous = groups.get(groupKey)
|
||||
if previous is None or ap == activeAp or (previous != activeAp and ap.get_strength() > previous.get_strength()):
|
||||
groups[groupKey] = ap
|
||||
represented = set()
|
||||
for groupKey, ap in groups.items():
|
||||
matches = [p for p in profiles if ap.connection_valid(p)]
|
||||
if activeProfile in matches:
|
||||
matches.remove(activeProfile)
|
||||
matches.insert(0, activeProfile)
|
||||
# Keep distinct saved profiles selectable even for identical SSIDs.
|
||||
for profile in matches or [None]:
|
||||
if profile:
|
||||
represented.add(profile.get_uuid())
|
||||
connected = active if profile and activeProfile == profile else None
|
||||
rows.append(self.make_row(device, profile, ap, connected))
|
||||
for profile in profiles:
|
||||
if profile.get_uuid() not in represented:
|
||||
rows.append(self.make_row(device, profile, None, active if profile == activeProfile else None))
|
||||
else:
|
||||
for profile in profiles:
|
||||
rows.append(self.make_row(device, profile, None, active if profile == activeProfile else None))
|
||||
return sorted(rows, key=lambda row: (row.name.casefold(), row.kind, row.key))
|
||||
|
||||
def watch(self, obj):
|
||||
self.signalHandlers.append((obj, obj.connect("notify", self.schedule_refresh)))
|
||||
|
||||
def make_row(self, device, profile, ap, active):
|
||||
wifi = device.get_device_type() == NM.DeviceType.WIFI
|
||||
name = profile.get_id() if profile else NM.utils_ssid_to_utf8(ap.get_ssid().get_data())
|
||||
key = (device.get_path(), profile.get_uuid() if profile else bytes(ap.get_ssid().get_data()).hex() + ":" + security_type(ap)[0])
|
||||
state = "Connected" if active and active.get_state() == NM.ActiveConnectionState.ACTIVATED else "Disconnected"
|
||||
if active and active.get_state() == NM.ActiveConnectionState.ACTIVATING:
|
||||
state = "Connecting"
|
||||
elif active and active.get_state() == NM.ActiveConnectionState.DEACTIVATING:
|
||||
state = "Disconnecting"
|
||||
elif wifi and not ap and not active:
|
||||
state = "Unavailable"
|
||||
elif not device.get_managed():
|
||||
state = "Unmanaged"
|
||||
elif wifi and not self.client.wireless_hardware_get_enabled():
|
||||
state = "Hardware blocked"
|
||||
elif wifi and not self.client.wireless_get_enabled():
|
||||
state = "Wi-Fi disabled"
|
||||
elif not wifi and not device.get_carrier():
|
||||
state = "Cable unplugged"
|
||||
security = security_type(ap)[0] if ap else ""
|
||||
return NetworkRow(key, name, "Wi-Fi" if wifi else "Ethernet", state,
|
||||
"{}%".format(ap.get_strength()) if ap else "", security,
|
||||
device, profile, ap, active)
|
||||
|
||||
def connect_row(self, row, password="", remember=True):
|
||||
permissions = [NETWORK_CONTROL] + ([] if row.profile else [MODIFY_SYSTEM])
|
||||
if self.busy or not self.authorize(permissions):
|
||||
return
|
||||
if not self.agentReady:
|
||||
self.status("The Wi-Fi password dialog is not ready. Try again shortly.")
|
||||
return
|
||||
try:
|
||||
connection = row.profile or new_wifi_connection(row, password, remember)
|
||||
except (ValueError, GLib.Error) as error:
|
||||
self.status(str(error))
|
||||
return
|
||||
self.busy = True
|
||||
self.pending = {"uuid": connection.get_uuid(), "profile": row.profile, "device": row.device,
|
||||
"active": None, "name": row.name}
|
||||
self.status("Connecting to {}…".format(row.name))
|
||||
if row.profile:
|
||||
self.client.activate_connection_async(connection, row.device, row.ap.get_path() if row.ap else None,
|
||||
self.cancellable, self.on_activated, False)
|
||||
else:
|
||||
self.client.add_and_activate_connection_async(connection, row.device, row.ap.get_path(),
|
||||
self.cancellable, self.on_activated, True)
|
||||
self.changed()
|
||||
|
||||
def on_activated(self, client, result, isNew):
|
||||
try:
|
||||
active = client.add_and_activate_connection_finish(result) if isNew else client.activate_connection_finish(result)
|
||||
except GLib.Error:
|
||||
self.finish_operation("Connection could not be started. Check permissions, availability, and the Wi-Fi password.")
|
||||
return
|
||||
if not self.pending or self.closed:
|
||||
return
|
||||
self.pending["active"] = active
|
||||
active.connect("state-changed", self.on_activation_state)
|
||||
self.activationTimeout = GLib.timeout_add_seconds(120, self.on_activation_timeout)
|
||||
self.check_activation(active)
|
||||
|
||||
def on_activation_state(self, active, state, reason):
|
||||
self.check_activation(active, reason)
|
||||
|
||||
def check_activation(self, active, reason=0):
|
||||
if not self.pending or self.pending["active"] != active:
|
||||
return
|
||||
state = active.get_state()
|
||||
if state == NM.ActiveConnectionState.ACTIVATED:
|
||||
self.finish_operation("Connected to {}.".format(self.pending["name"]))
|
||||
elif state == NM.ActiveConnectionState.DEACTIVATED:
|
||||
detail = reason.value_nick.replace("-", " ") if hasattr(reason, "value_nick") else "unknown reason"
|
||||
self.finish_operation("Connection failed or was cancelled: {}. Check the password and network availability.".format(detail))
|
||||
|
||||
def on_activation_timeout(self):
|
||||
self.activationTimeout = 0
|
||||
self.finish_operation("Connection is taking longer than expected. NetworkManager may still be trying; review its current state.")
|
||||
return GLib.SOURCE_REMOVE
|
||||
|
||||
def finish_operation(self, message):
|
||||
if getattr(self, "activationTimeout", 0):
|
||||
GLib.source_remove(self.activationTimeout)
|
||||
self.activationTimeout = 0
|
||||
self.cancel_secret()
|
||||
self.pending = None
|
||||
self.busy = False
|
||||
if not self.closed:
|
||||
self.status(message)
|
||||
self.changed()
|
||||
|
||||
def handle_secret(self, bus, sender, path, interface, method, parameters, invocation):
|
||||
if sender != self.client.get_dbus_name_owner():
|
||||
invocation.return_dbus_error(SECRET_INTERFACE + ".PermissionDenied", "Unexpected sender")
|
||||
return
|
||||
args = parameters.unpack()
|
||||
if method == "CancelGetSecrets":
|
||||
if self.secretRequest and self.secretRequest[0] == args[0]:
|
||||
self.cancel_secret()
|
||||
invocation.return_value(GLib.Variant("()", ()))
|
||||
return
|
||||
if method in ("SaveSecrets", "DeleteSecrets"):
|
||||
# This agent never owns or stores passwords. NetworkManager owns saved secrets.
|
||||
invocation.return_value(GLib.Variant("()", ()))
|
||||
return
|
||||
settings, connectionPath, settingName, hints, flags = args
|
||||
connectionId = settings.get("connection", {}).get("uuid")
|
||||
keyMgmt = settings.get(SECURITY_SETTING, {}).get("key-mgmt")
|
||||
if (not self.pending or connectionId != self.pending["uuid"] or
|
||||
settingName != SECURITY_SETTING or keyMgmt not in ("wpa-psk", "sae") or
|
||||
not flags & int(NM.SecretAgentGetSecretsFlags.ALLOW_INTERACTION)):
|
||||
invocation.return_dbus_error(SECRET_INTERFACE + ".NoSecrets", "No credentials available for this request")
|
||||
return
|
||||
self.cancel_secret()
|
||||
self.secretRequest = (connectionPath, invocation)
|
||||
profile = self.client.get_connection_by_uuid(connectionId)
|
||||
canRemember = bool(profile and not self.permission_problem([self.modification_permission(profile)]))
|
||||
|
||||
def answered(password, remember):
|
||||
if not self.secretRequest or self.secretRequest[1] != invocation:
|
||||
return
|
||||
if password is None:
|
||||
self.cancel_secret()
|
||||
return
|
||||
if not valid_password(password, keyMgmt):
|
||||
self.cancel_secret()
|
||||
self.status("The Wi-Fi password has an invalid length or format.")
|
||||
return
|
||||
if remember and profile:
|
||||
if not self.authorize([self.modification_permission(profile)]):
|
||||
self.cancel_secret()
|
||||
return
|
||||
clone = NM.SimpleConnection.new_clone(profile)
|
||||
security = clone.get_setting_wireless_security()
|
||||
security.props.psk = password
|
||||
security.props.psk_flags = NM.SettingSecretFlags.NONE
|
||||
profile.update2(clone.to_dbus(NM.ConnectionSerializationFlags.ALL), NM.SettingsUpdate2Flags.TO_DISK,
|
||||
GLib.Variant("a{sv}", {}), self.cancellable, saved, password)
|
||||
else:
|
||||
reply(password)
|
||||
|
||||
def saved(profile, result, password):
|
||||
try:
|
||||
profile.update2_finish(result)
|
||||
except GLib.Error:
|
||||
self.cancel_secret()
|
||||
self.status("The password could not be saved. Check your connection permissions.")
|
||||
return
|
||||
reply(password)
|
||||
|
||||
def reply(password):
|
||||
if self.secretRequest and self.secretRequest[1] == invocation:
|
||||
self.secretRequest = None
|
||||
invocation.return_value(GLib.Variant("(a{sa{sv}})", ({SECURITY_SETTING: {"psk": GLib.Variant("s", password)}},)))
|
||||
|
||||
self.password(self.pending["name"], keyMgmt, canRemember, answered)
|
||||
|
||||
def cancel_secret(self):
|
||||
if self.secretRequest:
|
||||
request = self.secretRequest
|
||||
self.secretRequest = None
|
||||
request[1].return_dbus_error(SECRET_INTERFACE + ".UserCanceled", "Password entry cancelled")
|
||||
|
||||
def disconnect_row(self, row):
|
||||
if self.busy or not row.active or not self.authorize([NETWORK_CONTROL]):
|
||||
return
|
||||
self.busy = True
|
||||
self.client.deactivate_connection_async(row.active, self.cancellable, self.on_disconnected, None)
|
||||
self.changed()
|
||||
|
||||
def on_disconnected(self, client, result, data):
|
||||
try:
|
||||
client.deactivate_connection_finish(result)
|
||||
self.finish_operation("Disconnected. NetworkManager may automatically activate another saved connection.")
|
||||
except GLib.Error:
|
||||
self.finish_operation("Could not disconnect. Check your network permissions.")
|
||||
|
||||
def set_wireless(self, enabled):
|
||||
if self.busy or not self.authorize([NM.ClientPermission.ENABLE_DISABLE_WIFI]):
|
||||
return
|
||||
self.busy = True
|
||||
self.bus.call(self.client.get_dbus_name_owner(), NM_PATH, "org.freedesktop.DBus.Properties", "Set",
|
||||
GLib.Variant("(ssv)", (NM_NAME, "WirelessEnabled", GLib.Variant("b", enabled))), None,
|
||||
Gio.DBusCallFlags.NONE, 10000, self.cancellable, self.on_wireless_set, enabled)
|
||||
self.changed()
|
||||
|
||||
def on_wireless_set(self, bus, result, enabled):
|
||||
try:
|
||||
bus.call_finish(result)
|
||||
self.finish_operation("Wi-Fi enabled." if enabled else "Wi-Fi disabled.")
|
||||
except GLib.Error:
|
||||
self.finish_operation("Could not change Wi-Fi state. Check permissions and the hardware wireless switch.")
|
||||
|
||||
def scan(self):
|
||||
if not self.authorize([NM.ClientPermission.WIFI_SCAN]):
|
||||
return
|
||||
devices = [d for d in self.client.get_devices() if isinstance(d, NM.DeviceWifi) and d.get_managed()]
|
||||
if not devices or not self.client.wireless_get_enabled():
|
||||
self.status("No enabled Wi-Fi adapter is available.")
|
||||
return
|
||||
self.status("Refreshing Wi-Fi networks…")
|
||||
for device in devices:
|
||||
device.request_scan_async(self.cancellable, self.on_scanned, None)
|
||||
|
||||
def on_scanned(self, device, result, data):
|
||||
try:
|
||||
device.request_scan_finish(result)
|
||||
self.status("Wi-Fi scan requested. The list will update as results arrive.")
|
||||
except GLib.Error:
|
||||
self.status("Wi-Fi scan could not be requested. Try again shortly.")
|
||||
self.changed()
|
||||
|
||||
def close(self):
|
||||
self.closed = True
|
||||
self.cancellable.cancel()
|
||||
self.cancel_secret()
|
||||
if self.refreshSource:
|
||||
GLib.source_remove(self.refreshSource)
|
||||
if getattr(self, "activationTimeout", 0):
|
||||
GLib.source_remove(self.activationTimeout)
|
||||
for obj, handler in self.signalHandlers:
|
||||
GObject.signal_handler_disconnect(obj, handler)
|
||||
if self.client and hasattr(self, "secretObjectId"):
|
||||
owner = self.client.get_dbus_name_owner()
|
||||
if owner:
|
||||
self.bus.call(owner, NM_PATH + "/AgentManager", NM_NAME + ".AgentManager", "Unregister", None,
|
||||
None, Gio.DBusCallFlags.NONE, 5000, None, None, None)
|
||||
self.bus.unregister_object(self.secretObjectId)
|
||||
if self.guard:
|
||||
self.guard.close()
|
||||
Reference in new issue
Block a user