Security hardening based on triple audit (security, quality, performance)
- Remove SIGUSR1 unlock handler (unauthenticated unlock vector) - Sanitize LD_PRELOAD (discard inherited environment) - Refuse to run as root - Validate D-Bus signal sender against fprintd proxy owner - Pass bytearray (not str) to PAM conversation callback for wipeable password - Resolve libc before returning CFUNCTYPE callback - Bundle fingerprint success idle_add into single atomic callback - Add running/device_proxy guards to VerifyStart retries with error handling - Add fingerprint attempt counter (max 10 before disabling) - Add power button confirmation dialog (inline yes/cancel) - Move fingerprint D-Bus init before session lock to avoid mainloop blocking - Resolve wallpaper path once, share across all monitor windows - Document faillock as UI-only (pam_faillock handles real brute-force protection) - Fix type hints (Callable), remove dead import (c_char), fix import order
This commit is contained in:
parent
5fda0dce0c
commit
fb11c551bd
@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "moonlock"
|
name = "moonlock"
|
||||||
version = "0.2.2"
|
version = "0.3.0"
|
||||||
description = "A secure Wayland lockscreen with GTK4, PAM and fingerprint support"
|
description = "A secure Wayland lockscreen with GTK4, PAM and fingerprint support"
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.11"
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
|
|||||||
@ -3,7 +3,7 @@
|
|||||||
|
|
||||||
import ctypes
|
import ctypes
|
||||||
import ctypes.util
|
import ctypes.util
|
||||||
from ctypes import POINTER, Structure, c_char_p, c_int, c_void_p, CFUNCTYPE, pointer, c_char
|
from ctypes import POINTER, Structure, c_char_p, c_int, c_void_p, CFUNCTYPE, pointer
|
||||||
|
|
||||||
# PAM return codes
|
# PAM return codes
|
||||||
PAM_SUCCESS = 0
|
PAM_SUCCESS = 0
|
||||||
@ -81,8 +81,14 @@ def _get_libc() -> ctypes.CDLL:
|
|||||||
return _cached_libc
|
return _cached_libc
|
||||||
|
|
||||||
|
|
||||||
def _make_conv_func(password: str) -> PamConvFunc:
|
def _make_conv_func(password_bytes: bytearray) -> PamConvFunc:
|
||||||
"""Create a PAM conversation callback that provides the password."""
|
"""Create a PAM conversation callback that provides the password.
|
||||||
|
|
||||||
|
Takes a bytearray (not str) so the caller can wipe it after use.
|
||||||
|
The callback creates a temporary bytes copy for strdup, then the
|
||||||
|
bytearray remains the single wipeable source of truth.
|
||||||
|
"""
|
||||||
|
libc = _get_libc()
|
||||||
|
|
||||||
def _conv(
|
def _conv(
|
||||||
num_msg: int,
|
num_msg: int,
|
||||||
@ -91,7 +97,6 @@ def _make_conv_func(password: str) -> PamConvFunc:
|
|||||||
appdata_ptr: c_void_p,
|
appdata_ptr: c_void_p,
|
||||||
) -> int:
|
) -> int:
|
||||||
# PAM expects malloc'd memory — it will free() the responses and resp strings
|
# PAM expects malloc'd memory — it will free() the responses and resp strings
|
||||||
libc = _get_libc()
|
|
||||||
resp_array = libc.calloc(num_msg, ctypes.sizeof(PamResponse))
|
resp_array = libc.calloc(num_msg, ctypes.sizeof(PamResponse))
|
||||||
if not resp_array:
|
if not resp_array:
|
||||||
return PAM_AUTH_ERR
|
return PAM_AUTH_ERR
|
||||||
@ -99,7 +104,7 @@ def _make_conv_func(password: str) -> PamConvFunc:
|
|||||||
resp_ptr = ctypes.cast(resp_array, POINTER(PamResponse))
|
resp_ptr = ctypes.cast(resp_array, POINTER(PamResponse))
|
||||||
for i in range(num_msg):
|
for i in range(num_msg):
|
||||||
# strdup allocates with malloc, which PAM can safely free()
|
# strdup allocates with malloc, which PAM can safely free()
|
||||||
resp_ptr[i].resp = libc.strdup(password.encode("utf-8"))
|
resp_ptr[i].resp = libc.strdup(bytes(password_bytes))
|
||||||
resp_ptr[i].resp_retcode = 0
|
resp_ptr[i].resp_retcode = 0
|
||||||
|
|
||||||
resp[0] = resp_ptr
|
resp[0] = resp_ptr
|
||||||
@ -122,8 +127,8 @@ def authenticate(username: str, password: str) -> bool:
|
|||||||
# Use a mutable bytearray so we can wipe the password after use
|
# Use a mutable bytearray so we can wipe the password after use
|
||||||
password_bytes = bytearray(password.encode("utf-8"))
|
password_bytes = bytearray(password.encode("utf-8"))
|
||||||
|
|
||||||
# Set up conversation
|
# Set up conversation — pass bytearray, not the str
|
||||||
conv_func = _make_conv_func(password)
|
conv_func = _make_conv_func(password_bytes)
|
||||||
conv = PamConv(conv=conv_func, appdata_ptr=None)
|
conv = PamConv(conv=conv_func, appdata_ptr=None)
|
||||||
|
|
||||||
# PAM handle
|
# PAM handle
|
||||||
|
|||||||
@ -4,17 +4,20 @@
|
|||||||
import logging
|
import logging
|
||||||
from typing import Callable
|
from typing import Callable
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
import gi
|
import gi
|
||||||
gi.require_version("Gio", "2.0")
|
gi.require_version("Gio", "2.0")
|
||||||
from gi.repository import Gio, GLib
|
from gi.repository import Gio, GLib
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
FPRINTD_BUS_NAME = "net.reactivated.Fprint"
|
FPRINTD_BUS_NAME = "net.reactivated.Fprint"
|
||||||
FPRINTD_MANAGER_PATH = "/net/reactivated/Fprint/Manager"
|
FPRINTD_MANAGER_PATH = "/net/reactivated/Fprint/Manager"
|
||||||
FPRINTD_MANAGER_IFACE = "net.reactivated.Fprint.Manager"
|
FPRINTD_MANAGER_IFACE = "net.reactivated.Fprint.Manager"
|
||||||
FPRINTD_DEVICE_IFACE = "net.reactivated.Fprint.Device"
|
FPRINTD_DEVICE_IFACE = "net.reactivated.Fprint.Device"
|
||||||
|
|
||||||
|
# Maximum fingerprint verification attempts before disabling
|
||||||
|
_MAX_FP_ATTEMPTS = 10
|
||||||
|
|
||||||
# Retry-able statuses (finger not read properly, try again)
|
# Retry-able statuses (finger not read properly, try again)
|
||||||
_RETRY_STATUSES = {
|
_RETRY_STATUSES = {
|
||||||
"verify-swipe-too-short",
|
"verify-swipe-too-short",
|
||||||
@ -32,13 +35,18 @@ class FingerprintListener:
|
|||||||
self._device_path: str | None = None
|
self._device_path: str | None = None
|
||||||
self._signal_id: int | None = None
|
self._signal_id: int | None = None
|
||||||
self._running: bool = False
|
self._running: bool = False
|
||||||
|
self._failed_attempts: int = 0
|
||||||
self._on_success: Callable[[], None] | None = None
|
self._on_success: Callable[[], None] | None = None
|
||||||
self._on_failure: Callable[[], None] | None = None
|
self._on_failure: Callable[[], None] | None = None
|
||||||
|
|
||||||
self._init_device()
|
self._init_device()
|
||||||
|
|
||||||
def _init_device(self) -> None:
|
def _init_device(self) -> None:
|
||||||
"""Connect to fprintd and get the default device."""
|
"""Connect to fprintd and get the default device.
|
||||||
|
|
||||||
|
This uses synchronous D-Bus calls — call before creating GTK windows
|
||||||
|
to avoid blocking the mainloop.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
manager = Gio.DBusProxy.new_for_bus_sync(
|
manager = Gio.DBusProxy.new_for_bus_sync(
|
||||||
Gio.BusType.SYSTEM,
|
Gio.BusType.SYSTEM,
|
||||||
@ -146,6 +154,13 @@ class FingerprintListener:
|
|||||||
if signal_name != "VerifyStatus":
|
if signal_name != "VerifyStatus":
|
||||||
return
|
return
|
||||||
|
|
||||||
|
# Validate signal origin — only accept signals from fprintd
|
||||||
|
if sender_name and not sender_name.startswith(FPRINTD_BUS_NAME):
|
||||||
|
expected_sender = proxy.get_name_owner()
|
||||||
|
if sender_name != expected_sender:
|
||||||
|
logger.warning("Ignoring VerifyStatus from unexpected sender: %s", sender_name)
|
||||||
|
return
|
||||||
|
|
||||||
status = parameters[0]
|
status = parameters[0]
|
||||||
done = parameters[1]
|
done = parameters[1]
|
||||||
self._on_verify_status(status, done)
|
self._on_verify_status(status, done)
|
||||||
@ -162,14 +177,25 @@ class FingerprintListener:
|
|||||||
|
|
||||||
if status in _RETRY_STATUSES:
|
if status in _RETRY_STATUSES:
|
||||||
# Retry — finger wasn't read properly
|
# Retry — finger wasn't read properly
|
||||||
if done:
|
if done and self._running and self._device_proxy:
|
||||||
|
try:
|
||||||
self._device_proxy.VerifyStart("(s)", "any")
|
self._device_proxy.VerifyStart("(s)", "any")
|
||||||
|
except GLib.Error as e:
|
||||||
|
logger.error("Failed to restart fingerprint verification: %s", e.message)
|
||||||
return
|
return
|
||||||
|
|
||||||
if status == "verify-no-match":
|
if status == "verify-no-match":
|
||||||
|
self._failed_attempts += 1
|
||||||
if self._on_failure:
|
if self._on_failure:
|
||||||
self._on_failure()
|
self._on_failure()
|
||||||
# Restart verification for another attempt
|
if self._failed_attempts >= _MAX_FP_ATTEMPTS:
|
||||||
if done:
|
logger.warning("Fingerprint max attempts (%d) reached, stopping listener", _MAX_FP_ATTEMPTS)
|
||||||
self._device_proxy.VerifyStart("(s)", "any")
|
self.stop()
|
||||||
|
return
|
||||||
|
# Restart verification for another attempt
|
||||||
|
if done and self._running and self._device_proxy:
|
||||||
|
try:
|
||||||
|
self._device_proxy.VerifyStart("(s)", "any")
|
||||||
|
except GLib.Error as e:
|
||||||
|
logger.error("Failed to restart fingerprint verification: %s", e.message)
|
||||||
return
|
return
|
||||||
|
|||||||
@ -29,6 +29,12 @@ class Strings:
|
|||||||
reboot_failed: str
|
reboot_failed: str
|
||||||
shutdown_failed: str
|
shutdown_failed: str
|
||||||
|
|
||||||
|
# Power confirmation
|
||||||
|
reboot_confirm: str
|
||||||
|
shutdown_confirm: str
|
||||||
|
confirm_yes: str
|
||||||
|
confirm_no: str
|
||||||
|
|
||||||
# Templates (use .format())
|
# Templates (use .format())
|
||||||
faillock_attempts_remaining: str
|
faillock_attempts_remaining: str
|
||||||
faillock_locked: str
|
faillock_locked: str
|
||||||
@ -46,6 +52,10 @@ _STRINGS_DE = Strings(
|
|||||||
wrong_password="Falsches Passwort",
|
wrong_password="Falsches Passwort",
|
||||||
reboot_failed="Neustart fehlgeschlagen",
|
reboot_failed="Neustart fehlgeschlagen",
|
||||||
shutdown_failed="Herunterfahren fehlgeschlagen",
|
shutdown_failed="Herunterfahren fehlgeschlagen",
|
||||||
|
reboot_confirm="Wirklich neu starten?",
|
||||||
|
shutdown_confirm="Wirklich herunterfahren?",
|
||||||
|
confirm_yes="Ja",
|
||||||
|
confirm_no="Abbrechen",
|
||||||
faillock_attempts_remaining="Noch {n} Versuch(e) vor Kontosperrung!",
|
faillock_attempts_remaining="Noch {n} Versuch(e) vor Kontosperrung!",
|
||||||
faillock_locked="Konto ist möglicherweise gesperrt",
|
faillock_locked="Konto ist möglicherweise gesperrt",
|
||||||
)
|
)
|
||||||
@ -62,6 +72,10 @@ _STRINGS_EN = Strings(
|
|||||||
wrong_password="Wrong password",
|
wrong_password="Wrong password",
|
||||||
reboot_failed="Reboot failed",
|
reboot_failed="Reboot failed",
|
||||||
shutdown_failed="Shutdown failed",
|
shutdown_failed="Shutdown failed",
|
||||||
|
reboot_confirm="Really reboot?",
|
||||||
|
shutdown_confirm="Really shut down?",
|
||||||
|
confirm_yes="Yes",
|
||||||
|
confirm_no="Cancel",
|
||||||
faillock_attempts_remaining="{n} attempt(s) remaining before lockout!",
|
faillock_attempts_remaining="{n} attempt(s) remaining before lockout!",
|
||||||
faillock_locked="Account may be locked",
|
faillock_locked="Account may be locked",
|
||||||
)
|
)
|
||||||
|
|||||||
@ -6,8 +6,12 @@ gi.require_version("Gtk", "4.0")
|
|||||||
gi.require_version("Gdk", "4.0")
|
gi.require_version("Gdk", "4.0")
|
||||||
from gi.repository import Gtk, Gdk, GdkPixbuf, GLib
|
from gi.repository import Gtk, Gdk, GdkPixbuf, GLib
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from collections.abc import Callable
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
from moonlock.auth import authenticate
|
from moonlock.auth import authenticate
|
||||||
from moonlock.config import Config, load_config, resolve_background_path
|
from moonlock.config import Config, load_config, resolve_background_path
|
||||||
from moonlock.fingerprint import FingerprintListener
|
from moonlock.fingerprint import FingerprintListener
|
||||||
@ -15,6 +19,8 @@ from moonlock.i18n import Strings, load_strings
|
|||||||
from moonlock.users import get_current_user, get_avatar_path, get_default_avatar_path, User
|
from moonlock.users import get_current_user, get_avatar_path, get_default_avatar_path, User
|
||||||
from moonlock import power
|
from moonlock import power
|
||||||
|
|
||||||
|
# UI-only attempt counter — the real brute-force protection is pam_faillock
|
||||||
|
# in the system-auth PAM stack, which persists across process restarts.
|
||||||
FAILLOCK_MAX_ATTEMPTS = 3
|
FAILLOCK_MAX_ATTEMPTS = 3
|
||||||
AVATAR_SIZE = 128
|
AVATAR_SIZE = 128
|
||||||
|
|
||||||
@ -22,9 +28,10 @@ AVATAR_SIZE = 128
|
|||||||
class LockscreenWindow(Gtk.ApplicationWindow):
|
class LockscreenWindow(Gtk.ApplicationWindow):
|
||||||
"""Fullscreen lockscreen window with password and fingerprint auth."""
|
"""Fullscreen lockscreen window with password and fingerprint auth."""
|
||||||
|
|
||||||
def __init__(self, application: Gtk.Application, unlock_callback: callable | None = None,
|
def __init__(self, application: Gtk.Application, unlock_callback: Callable[[], None] | None = None,
|
||||||
config: Config | None = None,
|
config: Config | None = None,
|
||||||
fingerprint_listener: FingerprintListener | None = None) -> None:
|
fingerprint_listener: FingerprintListener | None = None,
|
||||||
|
wallpaper_path: Path | None = None) -> None:
|
||||||
super().__init__(application=application)
|
super().__init__(application=application)
|
||||||
self.add_css_class("lockscreen")
|
self.add_css_class("lockscreen")
|
||||||
|
|
||||||
@ -33,6 +40,7 @@ class LockscreenWindow(Gtk.ApplicationWindow):
|
|||||||
self._user = get_current_user()
|
self._user = get_current_user()
|
||||||
self._failed_attempts = 0
|
self._failed_attempts = 0
|
||||||
self._unlock_callback = unlock_callback
|
self._unlock_callback = unlock_callback
|
||||||
|
self._wallpaper_path = wallpaper_path or resolve_background_path(self._config)
|
||||||
|
|
||||||
# Fingerprint listener (shared across windows to avoid multiple device claims)
|
# Fingerprint listener (shared across windows to avoid multiple device claims)
|
||||||
self._fp_listener = fingerprint_listener or FingerprintListener()
|
self._fp_listener = fingerprint_listener or FingerprintListener()
|
||||||
@ -59,9 +67,8 @@ class LockscreenWindow(Gtk.ApplicationWindow):
|
|||||||
overlay = Gtk.Overlay()
|
overlay = Gtk.Overlay()
|
||||||
self.set_child(overlay)
|
self.set_child(overlay)
|
||||||
|
|
||||||
# Background wallpaper
|
# Background wallpaper (path resolved once, shared across monitors)
|
||||||
wallpaper_path = resolve_background_path(self._config)
|
background = Gtk.Picture.new_for_filename(str(self._wallpaper_path))
|
||||||
background = Gtk.Picture.new_for_filename(str(wallpaper_path))
|
|
||||||
background.set_content_fit(Gtk.ContentFit.COVER)
|
background.set_content_fit(Gtk.ContentFit.COVER)
|
||||||
background.set_hexpand(True)
|
background.set_hexpand(True)
|
||||||
background.set_vexpand(True)
|
background.set_vexpand(True)
|
||||||
@ -211,9 +218,12 @@ class LockscreenWindow(Gtk.ApplicationWindow):
|
|||||||
|
|
||||||
def _on_fingerprint_success(self) -> None:
|
def _on_fingerprint_success(self) -> None:
|
||||||
"""Called when fingerprint verification succeeds."""
|
"""Called when fingerprint verification succeeds."""
|
||||||
GLib.idle_add(self._fp_label.set_text, self._strings.fingerprint_success)
|
def _handle_success():
|
||||||
GLib.idle_add(self._fp_label.add_css_class, "success")
|
self._fp_label.set_text(self._strings.fingerprint_success)
|
||||||
GLib.idle_add(self._unlock)
|
self._fp_label.add_css_class("success")
|
||||||
|
self._unlock()
|
||||||
|
return GLib.SOURCE_REMOVE
|
||||||
|
GLib.idle_add(_handle_success)
|
||||||
|
|
||||||
def _on_fingerprint_failure(self) -> None:
|
def _on_fingerprint_failure(self) -> None:
|
||||||
"""Called when fingerprint verification fails (no match)."""
|
"""Called when fingerprint verification fails (no match)."""
|
||||||
@ -283,14 +293,58 @@ class LockscreenWindow(Gtk.ApplicationWindow):
|
|||||||
if self._unlock_callback:
|
if self._unlock_callback:
|
||||||
self._unlock_callback()
|
self._unlock_callback()
|
||||||
|
|
||||||
def _on_power_action(self, action: callable) -> None:
|
def _on_power_action(self, action: Callable[[], None]) -> None:
|
||||||
"""Execute a power action (reboot/shutdown)."""
|
"""Request a power action with confirmation."""
|
||||||
|
confirm_msg = (
|
||||||
|
self._strings.reboot_confirm
|
||||||
|
if action == power.reboot
|
||||||
|
else self._strings.shutdown_confirm
|
||||||
|
)
|
||||||
|
self._show_power_confirm(confirm_msg, action)
|
||||||
|
|
||||||
|
def _show_power_confirm(self, message: str, action: Callable[[], None]) -> None:
|
||||||
|
"""Show inline confirmation buttons for a power action."""
|
||||||
|
# Replace error label with confirmation prompt
|
||||||
|
self._error_label.set_text(message)
|
||||||
|
self._error_label.set_visible(True)
|
||||||
|
self._error_label.remove_css_class("error-label")
|
||||||
|
self._error_label.add_css_class("confirm-label")
|
||||||
|
|
||||||
|
# Add confirm buttons below the error label
|
||||||
|
self._confirm_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
|
||||||
|
self._confirm_box.set_halign(Gtk.Align.CENTER)
|
||||||
|
|
||||||
|
yes_btn = Gtk.Button(label=self._strings.confirm_yes)
|
||||||
|
yes_btn.add_css_class("confirm-yes")
|
||||||
|
yes_btn.connect("clicked", lambda _: self._execute_power_action(action))
|
||||||
|
self._confirm_box.append(yes_btn)
|
||||||
|
|
||||||
|
no_btn = Gtk.Button(label=self._strings.confirm_no)
|
||||||
|
no_btn.add_css_class("confirm-no")
|
||||||
|
no_btn.connect("clicked", lambda _: self._dismiss_power_confirm())
|
||||||
|
self._confirm_box.append(no_btn)
|
||||||
|
|
||||||
|
self._login_box.append(self._confirm_box)
|
||||||
|
|
||||||
|
def _execute_power_action(self, action: Callable[[], None]) -> None:
|
||||||
|
"""Execute the confirmed power action."""
|
||||||
|
self._dismiss_power_confirm()
|
||||||
try:
|
try:
|
||||||
action()
|
action()
|
||||||
except Exception:
|
except Exception:
|
||||||
|
logger.exception("Power action failed")
|
||||||
error_msg = (
|
error_msg = (
|
||||||
self._strings.reboot_failed
|
self._strings.reboot_failed
|
||||||
if action == power.reboot
|
if action == power.reboot
|
||||||
else self._strings.shutdown_failed
|
else self._strings.shutdown_failed
|
||||||
)
|
)
|
||||||
self._show_error(error_msg)
|
self._show_error(error_msg)
|
||||||
|
|
||||||
|
def _dismiss_power_confirm(self) -> None:
|
||||||
|
"""Remove the confirmation prompt."""
|
||||||
|
if hasattr(self, "_confirm_box"):
|
||||||
|
self._login_box.remove(self._confirm_box)
|
||||||
|
del self._confirm_box
|
||||||
|
self._error_label.set_visible(False)
|
||||||
|
self._error_label.remove_css_class("confirm-label")
|
||||||
|
self._error_label.add_css_class("error-label")
|
||||||
|
|||||||
@ -7,7 +7,8 @@ import sys
|
|||||||
from importlib.resources import files
|
from importlib.resources import files
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
# gtk4-layer-shell must be loaded before libwayland-client
|
# gtk4-layer-shell must be loaded before libwayland-client.
|
||||||
|
# Only allow our own library in LD_PRELOAD — discard anything inherited from the environment.
|
||||||
_LAYER_SHELL_LIB = "/usr/lib/libgtk4-layer-shell.so"
|
_LAYER_SHELL_LIB = "/usr/lib/libgtk4-layer-shell.so"
|
||||||
_existing_preload = os.environ.get("LD_PRELOAD", "")
|
_existing_preload = os.environ.get("LD_PRELOAD", "")
|
||||||
_is_testing = "pytest" in sys.modules or "unittest" in sys.modules
|
_is_testing = "pytest" in sys.modules or "unittest" in sys.modules
|
||||||
@ -16,7 +17,7 @@ if (
|
|||||||
and _LAYER_SHELL_LIB not in _existing_preload
|
and _LAYER_SHELL_LIB not in _existing_preload
|
||||||
and os.path.exists(_LAYER_SHELL_LIB)
|
and os.path.exists(_LAYER_SHELL_LIB)
|
||||||
):
|
):
|
||||||
os.environ["LD_PRELOAD"] = f"{_existing_preload}:{_LAYER_SHELL_LIB}".lstrip(":")
|
os.environ["LD_PRELOAD"] = _LAYER_SHELL_LIB
|
||||||
os.execvp(sys.executable, [sys.executable, "-m", "moonlock.main"] + sys.argv[1:])
|
os.execvp(sys.executable, [sys.executable, "-m", "moonlock.main"] + sys.argv[1:])
|
||||||
|
|
||||||
import gi
|
import gi
|
||||||
@ -87,15 +88,20 @@ class MoonlockApp(Gtk.Application):
|
|||||||
|
|
||||||
def _activate_with_session_lock(self) -> None:
|
def _activate_with_session_lock(self) -> None:
|
||||||
"""Lock the session using ext-session-lock-v1 protocol."""
|
"""Lock the session using ext-session-lock-v1 protocol."""
|
||||||
|
# Init fingerprint D-Bus before locking — sync D-Bus calls would block
|
||||||
|
# the GTK mainloop if done after lock when the UI needs to be responsive.
|
||||||
|
fp_listener = FingerprintListener()
|
||||||
|
|
||||||
|
# Resolve wallpaper once, share across all monitors
|
||||||
|
from moonlock.config import resolve_background_path
|
||||||
|
wallpaper_path = resolve_background_path(self._config)
|
||||||
|
|
||||||
self._lock_instance = Gtk4SessionLock.Instance.new()
|
self._lock_instance = Gtk4SessionLock.Instance.new()
|
||||||
self._lock_instance.lock()
|
self._lock_instance.lock()
|
||||||
|
|
||||||
display = Gdk.Display.get_default()
|
display = Gdk.Display.get_default()
|
||||||
monitors = display.get_monitors()
|
monitors = display.get_monitors()
|
||||||
|
|
||||||
# Shared fingerprint listener across all windows (only one can claim the device)
|
|
||||||
fp_listener = FingerprintListener()
|
|
||||||
|
|
||||||
for i in range(monitors.get_n_items()):
|
for i in range(monitors.get_n_items()):
|
||||||
monitor = monitors.get_item(i)
|
monitor = monitors.get_item(i)
|
||||||
try:
|
try:
|
||||||
@ -104,6 +110,7 @@ class MoonlockApp(Gtk.Application):
|
|||||||
unlock_callback=self._unlock,
|
unlock_callback=self._unlock,
|
||||||
config=self._config,
|
config=self._config,
|
||||||
fingerprint_listener=fp_listener,
|
fingerprint_listener=fp_listener,
|
||||||
|
wallpaper_path=wallpaper_path,
|
||||||
)
|
)
|
||||||
self._lock_instance.assign_window_to_monitor(window, monitor)
|
self._lock_instance.assign_window_to_monitor(window, monitor)
|
||||||
window.present()
|
window.present()
|
||||||
@ -145,21 +152,8 @@ class MoonlockApp(Gtk.Application):
|
|||||||
logger.exception("Failed to load CSS stylesheet")
|
logger.exception("Failed to load CSS stylesheet")
|
||||||
|
|
||||||
|
|
||||||
def _install_signal_handlers(app: MoonlockApp) -> None:
|
def _install_excepthook() -> None:
|
||||||
"""Install signal handlers for external unlock (SIGUSR1) and crash logging."""
|
"""Install a global exception handler that logs crashes without unlocking."""
|
||||||
import signal
|
|
||||||
from gi.repository import GLib
|
|
||||||
|
|
||||||
def _handle_unlock(signum, frame):
|
|
||||||
"""SIGUSR1: External unlock request (e.g. from recovery wrapper)."""
|
|
||||||
logger.info("Received SIGUSR1 — unlocking session")
|
|
||||||
GLib.idle_add(app._unlock)
|
|
||||||
|
|
||||||
def _handle_crash_log(signum, frame):
|
|
||||||
"""Log unhandled exceptions but do NOT unlock — compositor keeps screen locked."""
|
|
||||||
logger.critical("Unhandled exception in moonlock", exc_info=True)
|
|
||||||
|
|
||||||
signal.signal(signal.SIGUSR1, _handle_unlock)
|
|
||||||
sys.excepthook = lambda exc_type, exc_value, exc_tb: (
|
sys.excepthook = lambda exc_type, exc_value, exc_tb: (
|
||||||
logger.critical("Unhandled exception — screen stays locked (compositor policy)",
|
logger.critical("Unhandled exception — screen stays locked (compositor policy)",
|
||||||
exc_info=(exc_type, exc_value, exc_tb)),
|
exc_info=(exc_type, exc_value, exc_tb)),
|
||||||
@ -170,9 +164,14 @@ def _install_signal_handlers(app: MoonlockApp) -> None:
|
|||||||
def main() -> None:
|
def main() -> None:
|
||||||
"""Run the Moonlock application."""
|
"""Run the Moonlock application."""
|
||||||
_setup_logging()
|
_setup_logging()
|
||||||
|
|
||||||
|
if os.getuid() == 0:
|
||||||
|
logger.critical("Moonlock should not run as root")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
logger.info("Moonlock starting")
|
logger.info("Moonlock starting")
|
||||||
app = MoonlockApp()
|
app = MoonlockApp()
|
||||||
_install_signal_handlers(app)
|
_install_excepthook()
|
||||||
app.run(sys.argv)
|
app.run(sys.argv)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@ -56,6 +56,7 @@ class TestFingerprintListenerLifecycle:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
listener._signal_id = 42
|
listener._signal_id = 42
|
||||||
|
|
||||||
listener.stop()
|
listener.stop()
|
||||||
@ -81,6 +82,7 @@ class TestFingerprintSignalHandling:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
on_success = MagicMock()
|
on_success = MagicMock()
|
||||||
on_failure = MagicMock()
|
on_failure = MagicMock()
|
||||||
listener._on_success = on_success
|
listener._on_success = on_success
|
||||||
@ -94,6 +96,7 @@ class TestFingerprintSignalHandling:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
on_success = MagicMock()
|
on_success = MagicMock()
|
||||||
on_failure = MagicMock()
|
on_failure = MagicMock()
|
||||||
listener._on_success = on_success
|
listener._on_success = on_success
|
||||||
@ -109,6 +112,7 @@ class TestFingerprintSignalHandling:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
on_success = MagicMock()
|
on_success = MagicMock()
|
||||||
on_failure = MagicMock()
|
on_failure = MagicMock()
|
||||||
listener._on_success = on_success
|
listener._on_success = on_success
|
||||||
@ -124,6 +128,7 @@ class TestFingerprintSignalHandling:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
on_success = MagicMock()
|
on_success = MagicMock()
|
||||||
on_failure = MagicMock()
|
on_failure = MagicMock()
|
||||||
listener._on_success = on_success
|
listener._on_success = on_success
|
||||||
@ -139,6 +144,7 @@ class TestFingerprintSignalHandling:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
on_success = MagicMock()
|
on_success = MagicMock()
|
||||||
on_failure = MagicMock()
|
on_failure = MagicMock()
|
||||||
listener._on_success = on_success
|
listener._on_success = on_success
|
||||||
|
|||||||
@ -66,6 +66,7 @@ class TestFingerprintAuthFlow:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
|
|
||||||
unlock_called = []
|
unlock_called = []
|
||||||
listener._on_success = lambda: unlock_called.append(True)
|
listener._on_success = lambda: unlock_called.append(True)
|
||||||
@ -81,6 +82,7 @@ class TestFingerprintAuthFlow:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
listener._on_success = MagicMock()
|
listener._on_success = MagicMock()
|
||||||
listener._on_failure = MagicMock()
|
listener._on_failure = MagicMock()
|
||||||
|
|
||||||
@ -96,6 +98,7 @@ class TestFingerprintAuthFlow:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
listener._on_success = MagicMock()
|
listener._on_success = MagicMock()
|
||||||
listener._on_failure = MagicMock()
|
listener._on_failure = MagicMock()
|
||||||
|
|
||||||
@ -113,6 +116,7 @@ class TestFingerprintAuthFlow:
|
|||||||
listener = FingerprintListener.__new__(FingerprintListener)
|
listener = FingerprintListener.__new__(FingerprintListener)
|
||||||
listener._device_proxy = MagicMock()
|
listener._device_proxy = MagicMock()
|
||||||
listener._running = True
|
listener._running = True
|
||||||
|
listener._failed_attempts = 0
|
||||||
fp_unlock = []
|
fp_unlock = []
|
||||||
listener._on_success = lambda: fp_unlock.append(True)
|
listener._on_success = lambda: fp_unlock.append(True)
|
||||||
listener._on_failure = MagicMock()
|
listener._on_failure = MagicMock()
|
||||||
|
|||||||
@ -199,33 +199,19 @@ class TestDefensiveWindowCreation:
|
|||||||
lock_instance.unlock.assert_not_called()
|
lock_instance.unlock.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
class TestSignalHandlers:
|
class TestExcepthook:
|
||||||
"""Tests for signal handlers and excepthook behavior."""
|
"""Tests for the global exception handler."""
|
||||||
|
|
||||||
def test_sigusr1_triggers_unlock(self):
|
|
||||||
"""SIGUSR1 should schedule an unlock via GLib.idle_add."""
|
|
||||||
MoonlockApp, _ = _import_main()
|
|
||||||
from moonlock.main import _install_signal_handlers
|
|
||||||
import signal
|
|
||||||
|
|
||||||
app = MoonlockApp.__new__(MoonlockApp)
|
|
||||||
app._lock_instance = MagicMock()
|
|
||||||
|
|
||||||
_install_signal_handlers(app)
|
|
||||||
|
|
||||||
handler = signal.getsignal(signal.SIGUSR1)
|
|
||||||
assert handler is not signal.SIG_DFL
|
|
||||||
|
|
||||||
def test_excepthook_does_not_unlock(self):
|
def test_excepthook_does_not_unlock(self):
|
||||||
"""Unhandled exceptions must NOT unlock the session."""
|
"""Unhandled exceptions must NOT unlock the session."""
|
||||||
MoonlockApp, _ = _import_main()
|
MoonlockApp, _ = _import_main()
|
||||||
from moonlock.main import _install_signal_handlers
|
from moonlock.main import _install_excepthook
|
||||||
|
|
||||||
app = MoonlockApp.__new__(MoonlockApp)
|
app = MoonlockApp.__new__(MoonlockApp)
|
||||||
app._lock_instance = MagicMock()
|
app._lock_instance = MagicMock()
|
||||||
|
|
||||||
original_hook = sys.excepthook
|
original_hook = sys.excepthook
|
||||||
_install_signal_handlers(app)
|
_install_excepthook()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
with patch("moonlock.main.logger"):
|
with patch("moonlock.main.logger"):
|
||||||
@ -235,3 +221,9 @@ class TestSignalHandlers:
|
|||||||
app._lock_instance.unlock.assert_not_called()
|
app._lock_instance.unlock.assert_not_called()
|
||||||
finally:
|
finally:
|
||||||
sys.excepthook = original_hook
|
sys.excepthook = original_hook
|
||||||
|
|
||||||
|
def test_no_sigusr1_handler(self):
|
||||||
|
"""SIGUSR1 must NOT be handled — signal-based unlock is a security hole."""
|
||||||
|
import signal
|
||||||
|
handler = signal.getsignal(signal.SIGUSR1)
|
||||||
|
assert handler is signal.SIG_DFL
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user