feat: add sysauth service for Polkit authentication
- Add SysauthService with D-Bus agent registration - Add AuthDialog with centered overlay password prompt - Add password utils (PAM verification, keyrings.alt caching) - Integrate sysauth into main app initialization - Update WiFi dialog to use sysauth for password prompts - Add sysauth.css with dialog styling - Remove standalone WiFi password dialog
This commit is contained in:
11 files changed
+1100
-297
No files matched your search
@@ -17,6 +17,7 @@ from fabric.utils import get_relative_path, logger, monitor_file
|
||||
|
||||
from services.config import config
|
||||
from services.keyboard_layout import KeyboardLayout
|
||||
from services.sysauth import SysauthService
|
||||
from shared.data import APP_NAME, load_config
|
||||
from utils.functions import set_process_name
|
||||
from utils.gtk_utils import generate_colors_from_wallpaper
|
||||
@@ -96,6 +97,10 @@ def main():
|
||||
# Set up styling before creating windows to prevent unstyled flashes
|
||||
setup_css(app)
|
||||
|
||||
# Initialize Polkit authentication agent
|
||||
sysauth = SysauthService()
|
||||
sysauth.start()
|
||||
|
||||
switcher = ApplicationSwitcher()
|
||||
panel = Panel()
|
||||
modusnoti = ModusNoti()
|
||||
|
||||
@@ -20,5 +20,6 @@ from .network import ( # noqa: F401
|
||||
Wifi,
|
||||
)
|
||||
from .numlock import NumLock # noqa: F401
|
||||
from .sysauth import SysauthService, get_sysauth_service # noqa: F401
|
||||
from .todo import TodoService # noqa: F401
|
||||
from .wallpaper import WallpaperService, create_thumbnail # noqa: F401
|
||||
@@ -0,0 +1,350 @@
|
||||
"""Polkit authentication agent service for Modus.
|
||||
|
||||
Registers as a Polkit authentication agent on the D-Bus session bus
|
||||
and listens for authentication requests from polkitd. Emits signals
|
||||
so the UI layer can show a password dialog.
|
||||
|
||||
D-Bus flow:
|
||||
1. Export org.freedesktop.PolicyKit1.AuthenticationAgent on session bus
|
||||
2. Register with Polkit Authority on system bus
|
||||
3. polkitd calls BeginAuthentication → service emits signal
|
||||
4. UI responds via respond(cookie, password)
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
from fabric.core.service import Service, Signal
|
||||
from fabric.utils import Gio, GLib, logger
|
||||
|
||||
POLKIT_BUS_NAME = "org.freedesktop.PolicyKit1"
|
||||
POLKIT_AUTHORITY_PATH = "/org/freedesktop/PolicyKit1/Authority"
|
||||
POLKIT_AUTHORITY_IFACE = "org.freedesktop.PolicyKit1.Authority"
|
||||
|
||||
AGENT_OBJECT_PATH = "/org/freedesktop/PolicyKit1/AuthenticationAgent"
|
||||
AGENT_INTERFACE = "org.freedesktop.PolicyKit1.AuthenticationAgent"
|
||||
|
||||
AGENT_INTROSPECTION = """<!DOCTYPE node>
|
||||
<node>
|
||||
<interface name="org.freedesktop.PolicyKit1.AuthenticationAgent">
|
||||
<method name="BeginAuthentication">
|
||||
<arg name="action_id" type="s" direction="in"/>
|
||||
<arg name="message" type="s" direction="in"/>
|
||||
<arg name="icon_name" type="s" direction="in"/>
|
||||
<arg name="details" type="a{ss}" direction="in"/>
|
||||
<arg name="cookie" type="s" direction="in"/>
|
||||
<arg name="identities" type="a(si)" direction="in"/>
|
||||
</method>
|
||||
<method name="CancelAuthentication">
|
||||
<arg name="cookie" type="s" direction="in"/>
|
||||
</method>
|
||||
</interface>
|
||||
</node>
|
||||
"""
|
||||
|
||||
|
||||
class SysauthService(Service):
|
||||
"""Polkit authentication agent as a Modus service.
|
||||
|
||||
Registers with polkitd and emits signals when authentication
|
||||
is requested. The UI layer connects to these signals to show
|
||||
a password dialog.
|
||||
"""
|
||||
|
||||
_instance = None
|
||||
|
||||
def __new__(cls, *args, **kwargs):
|
||||
if cls._instance is None:
|
||||
cls._instance = super().__new__(cls)
|
||||
return cls._instance
|
||||
|
||||
@Signal
|
||||
def begin_authentication(
|
||||
self, action_id: str, message: str, icon_name: str, cookie: str, uid: int
|
||||
) -> None:
|
||||
"""Emitted when polkitd requests authentication.
|
||||
|
||||
Args:
|
||||
action_id: Processed action identifier (human-readable).
|
||||
message: Description of the action requiring auth.
|
||||
icon_name: Icon name for the action.
|
||||
cookie: Session cookie for responding.
|
||||
uid: User ID being authenticated.
|
||||
"""
|
||||
|
||||
@Signal
|
||||
def authentication_cancelled(self, cookie: str) -> None:
|
||||
"""Emitted when polkitd cancels an authentication request."""
|
||||
|
||||
@Signal
|
||||
def authentication_completed(self, cookie: str, success: bool) -> None:
|
||||
"""Emitted when authentication completes (success or failure)."""
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
if getattr(self, "_initialized", False):
|
||||
return
|
||||
self._initialized = True
|
||||
super().__init__(**kwargs)
|
||||
|
||||
self._session_bus_id = None
|
||||
self._authority_proxy = None
|
||||
self._registered = False
|
||||
|
||||
def start(self) -> bool:
|
||||
"""Start the agent: export on session bus and register with Authority.
|
||||
|
||||
Returns:
|
||||
True if registration succeeded, False otherwise.
|
||||
"""
|
||||
try:
|
||||
bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
||||
if bus is None:
|
||||
logger.error("[SysauthService] Cannot get session bus")
|
||||
return False
|
||||
|
||||
node_info = Gio.DBusNodeInfo.new_for_xml(AGENT_INTROSPECTION)
|
||||
interface_info = node_info.lookup_interface(AGENT_INTERFACE)
|
||||
|
||||
self._session_bus_id = bus.register_object_with_closures2(
|
||||
AGENT_OBJECT_PATH,
|
||||
interface_info,
|
||||
self._handle_method_call,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"[SysauthService] Exported agent on session bus at {AGENT_OBJECT_PATH}"
|
||||
)
|
||||
|
||||
if not self._register_with_authority():
|
||||
logger.warning(
|
||||
"[SysauthService] Failed to register with Authority, "
|
||||
"running without polkit integration"
|
||||
)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except GLib.Error as e:
|
||||
logger.error(f"[SysauthService] Failed to start: {e}")
|
||||
return False
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Unregister and clean up D-Bus resources."""
|
||||
if self._session_bus_id is not None:
|
||||
try:
|
||||
bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
||||
if bus is not None:
|
||||
bus.unregister_object(self._session_bus_id)
|
||||
except GLib.Error:
|
||||
pass
|
||||
self._session_bus_id = None
|
||||
|
||||
if self._registered:
|
||||
self._unregister_from_authority()
|
||||
self._registered = False
|
||||
|
||||
logger.info("[SysauthService] Stopped")
|
||||
|
||||
def respond(self, cookie: str, password: str) -> None:
|
||||
"""Send authentication response back to polkitd.
|
||||
|
||||
Args:
|
||||
cookie: The authentication session cookie.
|
||||
password: The password to send.
|
||||
"""
|
||||
try:
|
||||
bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
||||
if bus is None:
|
||||
logger.error("[SysauthService] Cannot get session bus for response")
|
||||
return
|
||||
|
||||
bus.call_sync(
|
||||
POLKIT_BUS_NAME,
|
||||
AGENT_OBJECT_PATH,
|
||||
AGENT_INTERFACE,
|
||||
"Respond",
|
||||
GLib.Variant("(ss)", (cookie, password)),
|
||||
None,
|
||||
Gio.DBusCallFlags.NONE,
|
||||
-1,
|
||||
None,
|
||||
)
|
||||
except GLib.Error as e:
|
||||
logger.error(f"[SysauthService] Failed to send response: {e}")
|
||||
|
||||
def _register_with_authority(self) -> bool:
|
||||
"""Register this agent with the Polkit Authority on the system bus."""
|
||||
try:
|
||||
system_bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
|
||||
if system_bus is None:
|
||||
logger.error("[SysauthService] Cannot get system bus")
|
||||
return False
|
||||
|
||||
self._authority_proxy = Gio.DBusProxy.new_sync(
|
||||
system_bus,
|
||||
Gio.DBusProxyFlags.NONE,
|
||||
None,
|
||||
POLKIT_BUS_NAME,
|
||||
POLKIT_AUTHORITY_PATH,
|
||||
POLKIT_AUTHORITY_IFACE,
|
||||
None,
|
||||
)
|
||||
|
||||
pid = os.getpid()
|
||||
locale = os.environ.get("LANG", "en_US.UTF-8")
|
||||
|
||||
builder = GLib.VariantBuilder(GLib.VariantType("(sa{sv}so)"))
|
||||
builder.add_value(GLib.Variant("s", "unix-process"))
|
||||
details = GLib.VariantBuilder(GLib.VariantType("a{sv}"))
|
||||
details.add_value(GLib.Variant("{sv}", ("pid", GLib.Variant("u", pid))))
|
||||
builder.add_value(details.end())
|
||||
builder.add_value(GLib.Variant("s", locale))
|
||||
builder.add_value(GLib.Variant("o", AGENT_OBJECT_PATH))
|
||||
|
||||
self._authority_proxy.call_sync(
|
||||
"RegisterAuthenticationAgent",
|
||||
builder.end(),
|
||||
Gio.DBusCallFlags.NONE,
|
||||
-1,
|
||||
None,
|
||||
)
|
||||
|
||||
self._registered = True
|
||||
logger.info(f"[SysauthService] Registered with Authority (PID {pid})")
|
||||
return True
|
||||
|
||||
except GLib.Error as e:
|
||||
logger.error(f"[SysauthService] Registration failed: {e}")
|
||||
return False
|
||||
|
||||
def _unregister_from_authority(self) -> None:
|
||||
"""Unregister this agent from the Polkit Authority."""
|
||||
if self._authority_proxy is None:
|
||||
return
|
||||
|
||||
try:
|
||||
pid = os.getpid()
|
||||
|
||||
builder = GLib.VariantBuilder(GLib.VariantType("(sa{sv}o)"))
|
||||
builder.add_value(GLib.Variant("s", "unix-process"))
|
||||
details = GLib.VariantBuilder(GLib.VariantType("a{sv}"))
|
||||
details.add_value(GLib.Variant("{sv}", ("pid", GLib.Variant("u", pid))))
|
||||
builder.add_value(details.end())
|
||||
builder.add_value(GLib.Variant("o", AGENT_OBJECT_PATH))
|
||||
|
||||
self._authority_proxy.call_sync(
|
||||
"UnregisterAuthenticationAgent",
|
||||
builder.end(),
|
||||
Gio.DBusCallFlags.NONE,
|
||||
-1,
|
||||
None,
|
||||
)
|
||||
|
||||
logger.info("[SysauthService] Unregistered from Authority")
|
||||
except GLib.Error as e:
|
||||
logger.warning(f"[SysauthService] Unregister failed: {e}")
|
||||
|
||||
def _handle_method_call(
|
||||
self,
|
||||
connection: Gio.DBusConnection,
|
||||
sender: str,
|
||||
object_path: str,
|
||||
interface_name: str,
|
||||
method_name: str,
|
||||
parameters: GLib.Variant,
|
||||
invocation: Gio.DBusMethodInvocation,
|
||||
) -> None:
|
||||
"""Handle incoming D-Bus method calls from polkitd."""
|
||||
if method_name == "BeginAuthentication":
|
||||
self._handle_begin_authentication(parameters, invocation)
|
||||
elif method_name == "CancelAuthentication":
|
||||
self._handle_cancel_authentication(parameters, invocation)
|
||||
else:
|
||||
invocation.return_error_literal(
|
||||
Gio.DBusError,
|
||||
Gio.DBusError.UNKNOWN_METHOD,
|
||||
f"Unknown method: {method_name}",
|
||||
)
|
||||
|
||||
def _handle_begin_authentication(
|
||||
self,
|
||||
parameters: GLib.Variant,
|
||||
invocation: Gio.DBusMethodInvocation,
|
||||
) -> None:
|
||||
"""Handle BeginAuthentication call from polkitd."""
|
||||
action_id, message, icon_name, _details, cookie, identities = (
|
||||
parameters.unpack()
|
||||
)
|
||||
|
||||
uid = -1
|
||||
if identities:
|
||||
for kind, value in identities:
|
||||
if kind == "unix-user":
|
||||
if isinstance(value, dict):
|
||||
uid = value.get("uid", -1)
|
||||
elif isinstance(value, int):
|
||||
uid = value
|
||||
elif isinstance(value, tuple):
|
||||
uid = value[0] if value else -1
|
||||
break
|
||||
|
||||
if uid == -1:
|
||||
uid = os.getuid()
|
||||
|
||||
processed_action = _process_action_id(action_id, message)
|
||||
|
||||
logger.info(
|
||||
f"[SysauthService] BeginAuthentication: action={processed_action} "
|
||||
f"uid={uid} cookie={cookie[:16]}..."
|
||||
)
|
||||
|
||||
invocation.return_value(None)
|
||||
|
||||
GLib.idle_add(
|
||||
lambda: self.emit(
|
||||
"begin-authentication",
|
||||
processed_action,
|
||||
message,
|
||||
icon_name,
|
||||
cookie,
|
||||
uid,
|
||||
)
|
||||
)
|
||||
|
||||
def _handle_cancel_authentication(
|
||||
self,
|
||||
parameters: GLib.Variant,
|
||||
invocation: Gio.DBusMethodInvocation,
|
||||
) -> None:
|
||||
"""Handle CancelAuthentication call from polkitd."""
|
||||
(cookie,) = parameters.unpack()
|
||||
logger.info(f"[SysauthService] CancelAuthentication: cookie={cookie[:16]}")
|
||||
invocation.return_value(None)
|
||||
GLib.idle_add(lambda: self.emit("authentication-cancelled", cookie))
|
||||
|
||||
|
||||
def _process_action_id(action_id: str, message: str) -> str:
|
||||
"""Process action_id to extract a human-readable name.
|
||||
|
||||
Matches the C++ sysauth behavior:
|
||||
- For org.freedesktop.policykit.exec: extracts program name from message
|
||||
- For other actions: strips everything before the last dot
|
||||
"""
|
||||
if action_id == "org.freedesktop.policykit.exec":
|
||||
try:
|
||||
start = message.rfind("/") + 1
|
||||
end = message.find("'", start)
|
||||
if start > 0 and end > start:
|
||||
return message[start:end]
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
return action_id
|
||||
|
||||
if "." in action_id:
|
||||
return action_id.rsplit(".", 1)[-1]
|
||||
|
||||
return action_id
|
||||
|
||||
|
||||
def get_sysauth_service() -> SysauthService:
|
||||
"""Get the singleton sysauth service instance."""
|
||||
return SysauthService()
|
||||
@@ -0,0 +1,277 @@
|
||||
"""Fabric-based authentication dialog for sysauth.
|
||||
|
||||
Provides a centered overlay password prompt matching the WiFi dialog pattern.
|
||||
Uses Fabric widgets and integrates with the SysauthService for
|
||||
Polkit authentication requests.
|
||||
"""
|
||||
|
||||
import threading
|
||||
|
||||
from fabric.utils import Gdk, GLib
|
||||
from fabric.widgets.box import Box
|
||||
from fabric.widgets.button import Button
|
||||
from fabric.widgets.entry import Entry
|
||||
from fabric.widgets.image import Image
|
||||
from fabric.widgets.label import Label
|
||||
from fabric.widgets.window import Window
|
||||
|
||||
|
||||
class AuthDialog(Window):
|
||||
"""Centered authentication dialog matching the WiFi password dialog pattern."""
|
||||
|
||||
def __init__(self, action_id: str, message: str, icon_name: str):
|
||||
self._result_accepted = False
|
||||
self._main_loop = None
|
||||
|
||||
super().__init__(
|
||||
title="modus-dialog",
|
||||
layer="overlay",
|
||||
anchor="center",
|
||||
keyboard_mode="on-demand",
|
||||
visible=False,
|
||||
name="sysauth-dialog",
|
||||
)
|
||||
|
||||
self.set_type_hint(Gdk.WindowTypeHint.DIALOG)
|
||||
self.set_keep_above(True)
|
||||
self.set_modal(True)
|
||||
self.set_skip_taskbar_hint(True)
|
||||
self.set_skip_pager_hint(True)
|
||||
self.set_resizable(False)
|
||||
|
||||
self._action_id = action_id
|
||||
self._message = message
|
||||
self._icon_name = icon_name
|
||||
|
||||
self._build_ui()
|
||||
self._connect_signals()
|
||||
|
||||
def _build_ui(self):
|
||||
"""Construct the dialog UI matching the WiFi dialog layout."""
|
||||
self._image = Image(
|
||||
icon_name=self._icon_name,
|
||||
size=48,
|
||||
name="sysauth-icon",
|
||||
)
|
||||
|
||||
self._label_title = Label(
|
||||
label=self._action_id,
|
||||
name="sysauth-title",
|
||||
h_align="start",
|
||||
wrap=True,
|
||||
max_width_chars=40,
|
||||
)
|
||||
|
||||
self._title_container = Box(
|
||||
orientation="h",
|
||||
spacing=8,
|
||||
children=[self._image, self._label_title],
|
||||
name="sysauth-title-container",
|
||||
h_align="start",
|
||||
)
|
||||
|
||||
self._label_description = Label(
|
||||
label=self._message,
|
||||
name="sysauth-description",
|
||||
h_align="start",
|
||||
wrap=True,
|
||||
max_width_chars=40,
|
||||
lines=3,
|
||||
)
|
||||
|
||||
self._error_label = Label(
|
||||
label="Incorrect password. Please try again.",
|
||||
name="sysauth-error",
|
||||
h_align="center",
|
||||
visible=False,
|
||||
)
|
||||
|
||||
self._password_label = Label(
|
||||
label="Password:",
|
||||
name="sysauth-password-label",
|
||||
h_align="start",
|
||||
)
|
||||
|
||||
self._entry_password = Entry(
|
||||
placeholder_text="Enter password",
|
||||
name="sysauth-password-entry",
|
||||
visibility=False,
|
||||
h_expand=True,
|
||||
h_align="fill",
|
||||
)
|
||||
|
||||
self._password_container = Box(
|
||||
orientation="v",
|
||||
h_expand=True,
|
||||
spacing=6,
|
||||
children=[self._password_label, self._entry_password],
|
||||
name="sysauth-password-container",
|
||||
)
|
||||
|
||||
self._button_cancel = Button(
|
||||
label="Cancel",
|
||||
name="sysauth-cancel-button",
|
||||
on_clicked=self._on_cancel_clicked,
|
||||
)
|
||||
|
||||
self._button_ok = Button(
|
||||
label="Authenticate",
|
||||
name="sysauth-ok-button",
|
||||
on_clicked=self._on_ok_clicked,
|
||||
)
|
||||
|
||||
self._button_box = Box(
|
||||
orientation="h",
|
||||
spacing=12,
|
||||
h_expand=True,
|
||||
children=[self._button_cancel, self._button_ok],
|
||||
name="sysauth-button-box",
|
||||
h_align="end",
|
||||
)
|
||||
|
||||
self._content_box = Box(
|
||||
orientation="v",
|
||||
h_expand=True,
|
||||
spacing=12,
|
||||
children=[
|
||||
self._title_container,
|
||||
self._label_description,
|
||||
self._error_label,
|
||||
self._password_container,
|
||||
self._button_box,
|
||||
],
|
||||
name="sysauth-content",
|
||||
h_align="fill",
|
||||
v_align="center",
|
||||
)
|
||||
|
||||
self._dialog_background = Box(
|
||||
children=[self._content_box],
|
||||
name="sysauth-dialog-background",
|
||||
h_align="center",
|
||||
v_align="center",
|
||||
)
|
||||
|
||||
self.children = self._dialog_background
|
||||
|
||||
def _connect_signals(self):
|
||||
"""Connect button and entry signals."""
|
||||
self._entry_password.connect("activate", lambda *_: self._on_ok_clicked())
|
||||
self.connect("key-press-event", self._on_key_press)
|
||||
self.connect("notify::visible", self._on_visibility_changed)
|
||||
|
||||
def run(self) -> bool:
|
||||
"""Show the dialog and run a nested main loop until dismissed.
|
||||
|
||||
Returns:
|
||||
True if the user accepted (entered password), False if cancelled.
|
||||
"""
|
||||
self._main_loop = GLib.MainLoop()
|
||||
|
||||
self.show_all()
|
||||
self._error_label.set_visible(False)
|
||||
self._entry_password.grab_focus()
|
||||
|
||||
self._main_loop.run()
|
||||
|
||||
return self._result_accepted
|
||||
|
||||
def _on_key_press(self, _widget, event):
|
||||
"""Handle keyboard input."""
|
||||
keyval = event.keyval
|
||||
|
||||
if keyval == Gdk.KEY_Return or keyval == Gdk.KEY_KP_Enter:
|
||||
self._on_ok_clicked()
|
||||
return True
|
||||
elif keyval == Gdk.KEY_Escape:
|
||||
self._on_cancel_clicked()
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _on_visibility_changed(self, _widget, *_args):
|
||||
"""Focus password entry when dialog becomes visible."""
|
||||
if self.get_visible():
|
||||
GLib.timeout_add(100, lambda: self._entry_password.grab_focus())
|
||||
|
||||
def _on_ok_clicked(self, *_args):
|
||||
"""Handle OK button press or Enter key."""
|
||||
password = self._entry_password.get_text().strip()
|
||||
if not password:
|
||||
self._entry_password.grab_focus()
|
||||
return
|
||||
|
||||
self._result_accepted = True
|
||||
self._dismiss()
|
||||
|
||||
def _on_cancel_clicked(self, *_args):
|
||||
"""Handle Cancel button press or Escape key."""
|
||||
self._result_accepted = False
|
||||
self._dismiss()
|
||||
|
||||
def _dismiss(self):
|
||||
"""Dismiss the dialog."""
|
||||
self.hide()
|
||||
GLib.timeout_add(100, self._quit_main_loop)
|
||||
|
||||
def _quit_main_loop(self):
|
||||
"""Quit the nested main loop."""
|
||||
if self._main_loop is not None and self._main_loop.is_running():
|
||||
self._main_loop.quit()
|
||||
return False
|
||||
|
||||
def show_error(self, message: str = "Incorrect password. Please try again."):
|
||||
"""Show an error message in the dialog."""
|
||||
self._error_label.set_text(message)
|
||||
self._error_label.set_visible(True)
|
||||
self._entry_password.set_text("")
|
||||
GLib.timeout_add(10, lambda: self._entry_password.grab_focus())
|
||||
|
||||
def reset(self):
|
||||
"""Reset the dialog for reuse."""
|
||||
self._result_accepted = False
|
||||
self._error_label.set_visible(False)
|
||||
self._entry_password.set_text("")
|
||||
|
||||
@property
|
||||
def password(self) -> str:
|
||||
"""Get the entered password."""
|
||||
return self._entry_password.get_text()
|
||||
|
||||
|
||||
def run_auth_dialog(
|
||||
action_id: str,
|
||||
message: str,
|
||||
icon_name: str,
|
||||
on_result=None,
|
||||
error_message: "str | None" = None,
|
||||
):
|
||||
"""Run an authentication dialog in a background thread.
|
||||
|
||||
Args:
|
||||
action_id: The action identifier to display.
|
||||
message: Description of the action.
|
||||
icon_name: Icon name for the action.
|
||||
on_result: Callback(password: str | None) when dialog closes.
|
||||
error_message: Optional error message to show on first display.
|
||||
"""
|
||||
|
||||
def _run():
|
||||
dialog = AuthDialog(action_id, message, icon_name)
|
||||
|
||||
if error_message:
|
||||
GLib.idle_add(lambda: dialog.show_error(error_message))
|
||||
|
||||
accepted = dialog.run()
|
||||
|
||||
def _callback():
|
||||
if accepted and dialog.password:
|
||||
on_result(dialog.password)
|
||||
else:
|
||||
on_result(None)
|
||||
return False
|
||||
|
||||
GLib.idle_add(_callback)
|
||||
|
||||
thread = threading.Thread(target=_run, daemon=True)
|
||||
thread.start()
|
||||
@@ -1,269 +0,0 @@
|
||||
from fabric.utils import Gdk, GLib, logger
|
||||
from fabric.widgets.box import Box
|
||||
from fabric.widgets.button import Button
|
||||
from fabric.widgets.entry import Entry
|
||||
from fabric.widgets.image import Image
|
||||
from fabric.widgets.label import Label
|
||||
from fabric.widgets.window import Window
|
||||
|
||||
from utils.gtk_utils import svg_file
|
||||
|
||||
|
||||
class WiFiPasswordDialog(Window):
|
||||
def __init__(
|
||||
self,
|
||||
ssid: str,
|
||||
on_connect_callback=None,
|
||||
on_cancel_callback=None,
|
||||
on_dialog_closed=None,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(
|
||||
title="modus-dialog",
|
||||
layer="overlay",
|
||||
anchor="center",
|
||||
keyboard_mode="on-demand",
|
||||
visible=False,
|
||||
name="wifi-password-dialog",
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
self.ssid = ssid
|
||||
self.on_connect_callback = on_connect_callback
|
||||
self.on_cancel_callback = on_cancel_callback
|
||||
# self.set_size_request(400, 300)
|
||||
self.set_resizable(False)
|
||||
self.on_dialog_closed = on_dialog_closed
|
||||
self.is_connecting = False
|
||||
|
||||
self._create_dialog_content()
|
||||
self.connect("key-press-event", self._on_key_press)
|
||||
self.connect("notify::visible", self._on_visibility_changed)
|
||||
|
||||
def _create_dialog_content(self):
|
||||
self.wifi_icon = svg_file(
|
||||
"wifi/network-wireless-100.svg", size=20, name="wifi-dialog-icon"
|
||||
)
|
||||
|
||||
self.title_label = Label(
|
||||
label=f'The Wi-Fi network "{self.ssid}" requires a password.',
|
||||
name="wifi-dialog-title",
|
||||
h_align="start",
|
||||
wrap=True,
|
||||
max_width_chars=40,
|
||||
)
|
||||
|
||||
self.title_container = Box(
|
||||
orientation="h",
|
||||
spacing=8,
|
||||
children=[self.wifi_icon, self.title_label],
|
||||
name="wifi-dialog-title-container",
|
||||
h_align="start",
|
||||
)
|
||||
|
||||
self.error_label = Label(
|
||||
label="Incorrect password. Please try again.",
|
||||
name="wifi-dialog-error",
|
||||
h_align="center",
|
||||
visible=False,
|
||||
)
|
||||
|
||||
self.password_label = Label(
|
||||
label="Password:", name="wifi-dialog-password-label", h_align="start"
|
||||
)
|
||||
|
||||
self.password_entry = Entry(
|
||||
placeholder_text="Enter password",
|
||||
name="wifi-dialog-password-entry",
|
||||
visibility=False,
|
||||
h_expand=True,
|
||||
h_align="fill",
|
||||
)
|
||||
|
||||
self.password_entry.connect("activate", lambda *_: self._on_join_clicked())
|
||||
self.password_entry.connect("changed", self._on_password_changed)
|
||||
|
||||
self.password_visible = False
|
||||
self.show_password_button = Button(
|
||||
image=Image(icon_name="view-conceal-symbolic", size=16),
|
||||
name="wifi-dialog-show-password-button",
|
||||
on_clicked=self._on_show_password_clicked,
|
||||
)
|
||||
|
||||
self.show_password_label = Label(
|
||||
label="Show password", name="wifi-dialog-show-password-label"
|
||||
)
|
||||
|
||||
self.show_password_box = Box(
|
||||
orientation="h",
|
||||
spacing=8,
|
||||
children=[self.show_password_button, self.show_password_label],
|
||||
name="wifi-dialog-show-password-box",
|
||||
)
|
||||
|
||||
self.cancel_button = Button(
|
||||
label="Cancel",
|
||||
name="wifi-dialog-cancel-button",
|
||||
on_clicked=self._on_cancel_clicked,
|
||||
)
|
||||
|
||||
self.join_button = Button(
|
||||
label="Join",
|
||||
name="wifi-dialog-join-button",
|
||||
on_clicked=self._on_join_clicked,
|
||||
)
|
||||
|
||||
self.button_box = Box(
|
||||
orientation="h",
|
||||
spacing=12,
|
||||
h_expand=True,
|
||||
children=[self.cancel_button, self.join_button],
|
||||
name="wifi-dialog-button-box",
|
||||
h_align="end",
|
||||
)
|
||||
|
||||
self.password_container = Box(
|
||||
orientation="v",
|
||||
h_expand=True,
|
||||
spacing=6,
|
||||
children=[self.password_label, self.password_entry, self.show_password_box],
|
||||
name="wifi-dialog-password-container",
|
||||
)
|
||||
|
||||
self.content_box = Box(
|
||||
orientation="v",
|
||||
h_expand=True,
|
||||
spacing=12,
|
||||
children=[
|
||||
self.title_container,
|
||||
self.error_label,
|
||||
self.password_container,
|
||||
self.button_box,
|
||||
],
|
||||
name="wifi-dialog-content",
|
||||
h_align="fill",
|
||||
v_align="center",
|
||||
)
|
||||
|
||||
self.dialog_background = Box(
|
||||
children=[self.content_box],
|
||||
name="wifi-dialog-background",
|
||||
h_align="center",
|
||||
v_align="center",
|
||||
)
|
||||
|
||||
self.children = self.dialog_background
|
||||
|
||||
self._update_join_button_state()
|
||||
|
||||
def _on_password_changed(self, entry):
|
||||
self._update_join_button_state()
|
||||
|
||||
def _update_join_button_state(self):
|
||||
password = self.password_entry.get_text().strip()
|
||||
has_password = len(password) > 0
|
||||
|
||||
if has_password:
|
||||
self.join_button.set_opacity(1.0)
|
||||
self.join_button.remove_style_class("disabled")
|
||||
else:
|
||||
self.join_button.set_opacity(0.5)
|
||||
self.join_button.add_style_class("disabled")
|
||||
|
||||
def _on_show_password_clicked(self, *args):
|
||||
self.password_visible = not self.password_visible
|
||||
self.password_entry.set_visibility(self.password_visible)
|
||||
|
||||
icon_name = (
|
||||
"view-reveal-symbolic" if self.password_visible else "view-conceal-symbolic"
|
||||
)
|
||||
self.show_password_button.get_image().set_property("icon-name", icon_name)
|
||||
|
||||
def _on_key_press(self, widget, event):
|
||||
keyval = event.keyval
|
||||
|
||||
if keyval == Gdk.KEY_Return or keyval == Gdk.KEY_KP_Enter:
|
||||
self._on_join_clicked()
|
||||
return True
|
||||
elif keyval == Gdk.KEY_Escape:
|
||||
self._on_cancel_clicked()
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _on_visibility_changed(self, widget, *args):
|
||||
"""Handle visibility changes"""
|
||||
if self.get_visible():
|
||||
GLib.timeout_add(100, lambda: self.password_entry.grab_focus())
|
||||
|
||||
def _on_cancel_clicked(self, *args):
|
||||
self.hide()
|
||||
if self.on_cancel_callback:
|
||||
self.on_cancel_callback()
|
||||
if self.on_dialog_closed:
|
||||
self.on_dialog_closed()
|
||||
|
||||
def _on_join_clicked(self, *args):
|
||||
if self.is_connecting:
|
||||
return
|
||||
|
||||
password = self.password_entry.get_text().strip()
|
||||
if not password:
|
||||
self.password_entry.grab_focus()
|
||||
return
|
||||
|
||||
self.is_connecting = True
|
||||
self.join_button.set_sensitive(False)
|
||||
|
||||
self.error_label.set_visible(False)
|
||||
|
||||
self.hide()
|
||||
if self.on_connect_callback:
|
||||
self.on_connect_callback(self.ssid, password)
|
||||
if self.on_dialog_closed:
|
||||
self.on_dialog_closed()
|
||||
|
||||
def show_dialog(self):
|
||||
self.show_all()
|
||||
self.password_entry.set_text("")
|
||||
self.error_label.set_visible(False)
|
||||
|
||||
self.password_visible = False
|
||||
self.password_entry.set_visibility(False)
|
||||
self.show_password_button.get_image().set_property(
|
||||
"icon-name", "view-conceal-symbolic"
|
||||
)
|
||||
|
||||
self.is_connecting = False
|
||||
self.join_button.set_sensitive(True)
|
||||
|
||||
self._update_join_button_state()
|
||||
|
||||
def show_error(self, message="Incorrect password. Please try again."):
|
||||
self.is_connecting = False
|
||||
self.join_button.set_sensitive(True)
|
||||
|
||||
if not self.get_visible():
|
||||
self.error_label.set_text(message)
|
||||
self.error_label.set_visible(True)
|
||||
self.show_all()
|
||||
GLib.timeout_add(10, lambda: self._focus_and_select_password())
|
||||
else:
|
||||
self.error_label.set_text(message)
|
||||
self.error_label.set_visible(True)
|
||||
self._focus_and_select_password()
|
||||
|
||||
def _focus_and_select_password(self):
|
||||
try:
|
||||
self.password_entry.grab_focus()
|
||||
self.password_entry.select_region(0, -1)
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"[wifi_password_dialog] self.password_entry.grab_focus() failed: {e}"
|
||||
)
|
||||
return False
|
||||
|
||||
def destroy_dialog(self):
|
||||
self.hide()
|
||||
self.destroy()
|
||||
@@ -15,3 +15,4 @@
|
||||
|
||||
@import url("widgets/pomodoro.css");
|
||||
@import url("widgets/todo.css");
|
||||
@import url("widgets/sysauth.css");
|
||||
@@ -0,0 +1,98 @@
|
||||
/* Sysauth Dialog */
|
||||
|
||||
#sysauth-dialog {
|
||||
background-color: transparent;
|
||||
}
|
||||
|
||||
#sysauth-dialog-background {
|
||||
background-color: @glass-dialog;
|
||||
border-radius: apply(radius-md);
|
||||
border: 1px solid @glass-border;
|
||||
padding: 20px;
|
||||
min-width: 400px;
|
||||
}
|
||||
|
||||
#sysauth-title-container {
|
||||
margin-bottom: 8px;
|
||||
}
|
||||
|
||||
#sysauth-icon {
|
||||
color: @glass-accent;
|
||||
margin-right: 4px;
|
||||
}
|
||||
|
||||
#sysauth-title {
|
||||
color: @glass-text;
|
||||
font-size: apply(font-base);
|
||||
font-weight: 500;
|
||||
}
|
||||
|
||||
#sysauth-description {
|
||||
color: @glass-text-secondary;
|
||||
font-size: apply(font-sm);
|
||||
margin-bottom: 8px;
|
||||
}
|
||||
|
||||
#sysauth-error {
|
||||
color: @glass-error;
|
||||
font-size: apply(font-sm);
|
||||
font-weight: 500;
|
||||
margin-bottom: 8px;
|
||||
}
|
||||
|
||||
#sysauth-password-label {
|
||||
color: @glass-text;
|
||||
font-size: apply(font-md);
|
||||
font-weight: 500;
|
||||
margin-bottom: 4px;
|
||||
}
|
||||
|
||||
#sysauth-password-entry {
|
||||
background-color: @glass-surface-1;
|
||||
border: 1px solid @glass-border;
|
||||
border-radius: apply(radius-xs);
|
||||
padding: 8px 12px;
|
||||
color: @glass-text;
|
||||
font-size: apply(font-md);
|
||||
}
|
||||
|
||||
#sysauth-password-entry:focus {
|
||||
border-color: @glass-ring;
|
||||
box-shadow: 0 0 0 2px @glass-focus;
|
||||
background-color: @glass-hover;
|
||||
}
|
||||
|
||||
#sysauth-button-box {
|
||||
margin-top: 8px;
|
||||
}
|
||||
|
||||
#sysauth-cancel-button,
|
||||
#sysauth-ok-button {
|
||||
border-radius: apply(radius-xs);
|
||||
font-size: apply(font-md);
|
||||
font-weight: 500;
|
||||
min-width: 80px;
|
||||
min-height: 30px;
|
||||
padding: 6px 12px;
|
||||
}
|
||||
|
||||
#sysauth-cancel-button {
|
||||
background-color: @glass-hover;
|
||||
border: 1px solid @glass-border;
|
||||
color: @glass-text;
|
||||
}
|
||||
|
||||
#sysauth-cancel-button:hover {
|
||||
background-color: @glass-active;
|
||||
}
|
||||
|
||||
#sysauth-ok-button {
|
||||
background-color: @glass-accent;
|
||||
border: 1px solid @glass-accent;
|
||||
color: @glass-text;
|
||||
}
|
||||
|
||||
#sysauth-ok-button:hover {
|
||||
background-color: @glass-accent;
|
||||
border-color: @glass-accent;
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Password verification and keyring caching utilities.
|
||||
|
||||
Provides PAM-based password verification with su fallback,
|
||||
and system keyring integration for caching authenticated passwords.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
|
||||
from fabric.utils import logger
|
||||
|
||||
KEYRING_SERVICE = "sysauth"
|
||||
|
||||
|
||||
def verify_password(username: str, password: str) -> bool:
|
||||
"""Verify password using PAM with su fallback.
|
||||
|
||||
Args:
|
||||
username: The username to authenticate.
|
||||
password: The password to verify.
|
||||
|
||||
Returns:
|
||||
True if authentication succeeded, False otherwise.
|
||||
"""
|
||||
if _verify_pam(username, password):
|
||||
return True
|
||||
return _verify_su(username, password)
|
||||
|
||||
|
||||
def _verify_pam(username: str, password: str) -> bool:
|
||||
"""Verify password using python-pam."""
|
||||
try:
|
||||
import pam as pam_module
|
||||
|
||||
p = pam_module.pam()
|
||||
return p.authenticate(username, password)
|
||||
except ImportError:
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.warning(f"[password] PAM verification failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def _verify_su(username: str, password: str) -> bool:
|
||||
"""Verify password using the su command as fallback."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["su", "-c", "true", username],
|
||||
input=password.encode(),
|
||||
capture_output=True,
|
||||
timeout=5,
|
||||
)
|
||||
return result.returncode == 0
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError, PermissionError) as e:
|
||||
logger.warning(f"[password] su verification failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def cache_password(action_id: str, uid: int, password: str) -> None:
|
||||
"""Cache a password in the system keyring.
|
||||
|
||||
Args:
|
||||
action_id: The action identifier.
|
||||
uid: The user ID being authenticated.
|
||||
"""
|
||||
kr = _get_keyring()
|
||||
if kr is None:
|
||||
return
|
||||
|
||||
try:
|
||||
cache_key = f"{action_id}:{uid}"
|
||||
kr.set_password(KEYRING_SERVICE, cache_key, password)
|
||||
logger.debug(f"[password] Cached password for {action_id} (uid={uid})")
|
||||
except Exception as e:
|
||||
logger.debug(f"[password] Failed to cache password: {e}")
|
||||
|
||||
|
||||
def get_cached_password(action_id: str, uid: int) -> "str | None":
|
||||
"""Retrieve a cached password from the system keyring.
|
||||
|
||||
Args:
|
||||
action_id: The action identifier.
|
||||
uid: The user ID being authenticated.
|
||||
|
||||
Returns:
|
||||
The cached password, or None if not found.
|
||||
"""
|
||||
kr = _get_keyring()
|
||||
if kr is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
cache_key = f"{action_id}:{uid}"
|
||||
password = kr.get_password(KEYRING_SERVICE, cache_key)
|
||||
if password:
|
||||
logger.debug(
|
||||
f"[password] Retrieved cached password for {action_id} (uid={uid})"
|
||||
)
|
||||
return password
|
||||
except Exception as e:
|
||||
logger.debug(f"[password] Failed to retrieve cached password: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def get_username_from_uid(uid: int) -> "str":
|
||||
"""Get the username for a given UID.
|
||||
|
||||
Args:
|
||||
uid: The user ID.
|
||||
|
||||
Returns:
|
||||
The username, or 'root' if lookup fails.
|
||||
"""
|
||||
try:
|
||||
import pwd
|
||||
|
||||
return pwd.getpwuid(uid).pw_name
|
||||
except (KeyError, ImportError):
|
||||
return "root"
|
||||
|
||||
|
||||
def _get_keyring():
|
||||
"""Get the keyring backend, using keyrings.alt file-based backend."""
|
||||
try:
|
||||
import keyring as keyring_module
|
||||
|
||||
try:
|
||||
from keyrings.alt.file import PlaintextKeyring
|
||||
|
||||
kb = PlaintextKeyring()
|
||||
keyring_module.set_keyring(kb)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
return keyring_module
|
||||
except ImportError:
|
||||
logger.debug("[password] keyring module not available")
|
||||
return None
|
||||
@@ -7,7 +7,7 @@ from fabric.widgets.label import Label
|
||||
from fabric.widgets.separator import Separator
|
||||
|
||||
from services.network import NetworkClient
|
||||
from shared.dialogs.wifi_password_dialog import WiFiPasswordDialog
|
||||
from shared.dialogs.sysauth_dialog import run_auth_dialog
|
||||
from shared.widgets.smooth_switch import SmoothSwitch
|
||||
from shared.window.animated_scrollwindow import AnimatedScrollable
|
||||
from utils.functions import (
|
||||
@@ -62,8 +62,6 @@ class WifiNetworkSlot(Box):
|
||||
name="wifi-lock-icon",
|
||||
)
|
||||
|
||||
self.password_dialog = None
|
||||
|
||||
start_box = Box(
|
||||
orientation="h",
|
||||
spacing=8,
|
||||
@@ -184,19 +182,19 @@ class WifiNetworkSlot(Box):
|
||||
if self.parent and hasattr(self.parent, "hide_controlcenter"):
|
||||
self.parent.hide_controlcenter()
|
||||
|
||||
if self.password_dialog:
|
||||
self.password_dialog.destroy_dialog()
|
||||
|
||||
self.password_dialog = WiFiPasswordDialog(
|
||||
ssid=self.ssid,
|
||||
on_connect_callback=self._on_password_connect,
|
||||
on_cancel_callback=self._on_password_cancel,
|
||||
run_auth_dialog(
|
||||
action_id=self.ssid,
|
||||
message=f'Wi-Fi network "{self.ssid}" requires a password.',
|
||||
icon_name="network-wireless-symbolic",
|
||||
on_result=self._on_password_result,
|
||||
)
|
||||
|
||||
self.password_dialog.show_dialog()
|
||||
def _on_password_result(self, password):
|
||||
"""Handle password dialog result"""
|
||||
if password is None:
|
||||
self._reset_connect_state()
|
||||
return
|
||||
|
||||
def _on_password_connect(self, ssid, password):
|
||||
"""Handle password dialog connect action"""
|
||||
if password.strip():
|
||||
connecting_icon = get_wifi_connecting_icon()
|
||||
self.dimage.set_from_file(connecting_icon)
|
||||
@@ -209,15 +207,10 @@ class WifiNetworkSlot(Box):
|
||||
return
|
||||
if success:
|
||||
self.is_connected = True
|
||||
|
||||
GLib.timeout_add(500, lambda: self._reset_connect_state())
|
||||
|
||||
if self.password_dialog:
|
||||
self.password_dialog.is_connecting = False
|
||||
else:
|
||||
self._reset_connect_state()
|
||||
if self.password_dialog:
|
||||
self._show_connection_error(message)
|
||||
self._show_connection_error(message)
|
||||
|
||||
self.on_changed()
|
||||
|
||||
@@ -231,20 +224,20 @@ class WifiNetworkSlot(Box):
|
||||
f"[WiFi] Failed to initiate connection to {self.access_point.ssid}: {e}"
|
||||
)
|
||||
self._reset_connect_state()
|
||||
if self.password_dialog:
|
||||
self._show_connection_error("Connection failed. Please try again.")
|
||||
self._show_connection_error("Connection failed. Please try again.")
|
||||
self.on_changed()
|
||||
|
||||
def _show_connection_error(self, message="Incorrect password. Please try again."):
|
||||
"""Show connection error in a separate thread to prevent UI blocking"""
|
||||
if self.password_dialog:
|
||||
self.password_dialog.show_error(message)
|
||||
"""Show connection error and re-prompt for password"""
|
||||
run_auth_dialog(
|
||||
action_id=self.ssid,
|
||||
message=f'Wi-Fi network "{self.ssid}" requires a password.',
|
||||
icon_name="network-wireless-symbolic",
|
||||
on_result=self._on_password_result,
|
||||
error_message=message,
|
||||
)
|
||||
return False # Don't repeat if called from GLib.timeout_add
|
||||
|
||||
def _on_password_cancel(self):
|
||||
"""Handle password dialog cancel action"""
|
||||
self._reset_connect_state()
|
||||
|
||||
|
||||
class WifiConnections(Box):
|
||||
def __init__(self, parent, show_back_button=True, network_service=None, **kwargs):
|
||||
|
||||
Reference in new issue
Block a user