Compare commits

...

18 Commits

Author SHA1 Message Date
nevaforget de0b1d40ba Make login box background transparent 2026-03-26 16:31:54 +01:00
nevaforget 6907db0c2a Clean dist/ before wheel build to prevent stale artifacts 2026-03-26 16:26:56 +01:00
nevaforget 64f08d7e8b Harden greeter against threading issues, path traversal, and edge cases
Security:
- Fix path traversal in _save/_load_last_session by rejecting usernames
  starting with dot (blocks '..' and hidden file creation)
- Add avatar file size limit (10 MB) to prevent DoS via large ~/.face
- Add session_name length validation on write (symmetric with read)
- Add payload size check to send_message (symmetric with recv_message)
- Set log level to INFO in production (was DEBUG)

Quality:
- Eliminate main-thread blocking on user switch: _cancel_pending_session
  now sets a cancellation event and closes the socket instead of doing
  blocking IPC. The login worker checks the event after each step.
- Move power actions (reboot/shutdown) to background threads
- Catch TimeoutExpired in addition to CalledProcessError for power actions
- Consolidate socket cleanup in _login_worker via finally block, remove
  redundant _close_greetd_sock calls from error callbacks
- Fix _select_initial_user to return False for GLib.idle_add deregistration
- Fix context manager leak in resolve_wallpaper_path on exception
- Pass Config object to GreeterWindow instead of loading it twice
2026-03-26 16:25:13 +01:00
nevaforget cab1997dff fix: GTK-Theme-Validierung entfernt — GTK löst Theme-Namen selbst auf
Die Regex VALID_THEME_NAME blockierte Theme-Namen mit '+' (z.B.
catppuccin-mocha-lavender-standard+default). Da GTK den Theme-Namen
intern über Standardverzeichnisse auflöst, ist eigene Validierung
unnötig und kontraproduktiv.
2026-03-26 15:37:02 +01:00
nevaforget 3dfa596f9a fix: greetd-Session nach Auth-Fehler sauber canceln
Nach fehlgeschlagenem Login (falsches Passwort) wurde die greetd-Session
nicht gecancelt — beim nächsten Versuch kam "a session is already being
configured". Jetzt wird cancel_session gesendet nach Auth-Fehler, und
bei create_session-Fehler wird einmal cancel + retry versucht.

Außerdem: GTK-Theme-Name und PKGBUILD-pkgver aktualisiert.
2026-03-26 15:26:12 +01:00
nevaforget 357d2459cf fix: IPC byte order, globales GTK-Theme, Session-Vorauswahl
- ipc.py: !I (Big-Endian) → =I (Native Byte Order) für greetd-Protokoll
- Per-User GTK-Theme entfernt, stattdessen globales Theme aus moongreet.toml
- Last-Session pro User in /var/cache/moongreet/last-session/ speichern/laden
- PKGBUILD und install-Hook für last-session-Cache erweitert
2026-03-26 14:51:23 +01:00
nevaforget ba4f30f254 fix: Niri-Greeter-Config mit Retry-Loop gegen offene Session bei Crash
Wenn moongreet crasht bevor Niri bereit ist, blieb eine offene
Niri-Session als greeter-User zurueck. Retry-Loop stellt sicher,
dass niri sich in jedem Fall beendet. Beispiel-Configs korrigiert
(moongreet braucht Niri als Compositor).
2026-03-26 14:29:49 +01:00
nevaforget e37b273913 fix: Display-Null-Check und File-Logging in main.py
Gdk.Display.get_default() kann None zurueckgeben wenn der Compositor
noch nicht bereit ist. Vorher crashte moongreet mit TypeError, ohne
dass der Fehler irgendwo geloggt wurde. Display wird jetzt einmal
geholt, geprueft und an _register_icons/_load_css durchgereicht.
Logging geht nach /var/cache/moongreet/moongreet.log und stderr.
2026-03-26 14:16:38 +01:00
nevaforget ecd89f5b10 Simplify pkgver() to require tags 2026-03-26 14:02:38 +01:00
nevaforget d089fa201c fix: Build-Artefakte aus Repo entfernt, .gitignore ergänzt
makepkg-Artefakte (pkg/src, pkg/pkg, .pkg.tar.zst) waren
versehentlich committed. Entfernt und per .gitignore geschützt.
2026-03-26 13:44:07 +01:00
nevaforget 6400270a50 fix: PKGBUILD compositor-agnostisch, Beispiel-Configs bereinigt
Moongreet ist ein reiner GTK4-Greeter ohne eigenen Compositor.
niri-greeter.kdl entfernt — der User konfiguriert seinen
Compositor selbst (regreet → moongreet tauschen).
2026-03-26 13:38:27 +01:00
nevaforget 10b613b50b fix: PKGBUILD als -git Paket mit automatischer Versionierung
pkgver() generiert Version aus git describe, makepkg -si
aktualisiert automatisch ohne manuelles Version-Bumpen.
2026-03-26 13:30:36 +01:00
nevaforget 99c016adbc fix: PKGBUILD auf Git-Source umstellen (clone + makepkg Workflow) 2026-03-26 13:09:28 +01:00
nevaforget 9738e71ecc feat: Multi-Monitor-Support — Wallpaper auf Sekundärmonitoren
- WallpaperWindow für Sekundärmonitore (nur Hintergrundbild)
- GreeterWindow bekommt bg_path als Parameter
- resolve_wallpaper_path() aus config.py extrahiert (wiederverwendbar)
- main.py: Monitor-Enumeration, Layer-Shell pro Monitor
- Keyboard-Exclusive nur auf dem primären Monitor
- CSS: ungültige max-width/max-height Properties entfernt
2026-03-26 13:05:29 +01:00
nevaforget 8f2540024d fix: do_unrealize durch Signal-Handler ersetzen (PyGObject VFunc-Kompatibilität)
do_unrealize als GObject-VFunc-Override crashte beim super()-Chain-Up.
Signal-basierter _on_unrealize-Handler wie bei _on_realize.
2026-03-26 12:59:45 +01:00
nevaforget 4cd73a430b fix: Audit-Findings — Realize-Handler, Thread-Safety, Input-Validierung
- _on_realize implementiert (war nur connected, nicht definiert)
- do_unrealize für as_file() Context-Manager-Cleanup
- threading.Lock für _greetd_sock Race Condition
- TOML-Parsing-Fehler abfangen statt Crash
- last-user Datei: Längen- und Zeichenvalidierung
- detect_locale: non-alpha LANG-Werte abweisen
- exec_cmd Plausibility-Check mit shutil.which
- Exception-Details ins Log statt in die UI
- subprocess.run Timeout für Power-Actions
- Sequence[Path] statt tuple[Path, ...] in get_sessions
- Mock-Server _recvall für fragmentierte Reads
- [behavior]-Config-Sektion entfernt (unimplementiert)
- Design Decisions in CLAUDE.md dokumentiert
2026-03-26 12:56:52 +01:00
nevaforget 8b1608f99d fix: Audit-Findings — Theme-Validierung, locale-unabhängige Tests
- Theme-Name aus settings.ini gegen Regex validiert (nur [A-Za-z0-9_-]),
  verhindert Path-Traversal über GTK-Theme-Loading (S-05)
- Faillock-Tests nutzen expliziten strings-Parameter statt System-Locale,
  Tests laufen jetzt auch auf EN-Systemen (MAINT-4)
- Test für Path-Traversal im Theme-Namen ergänzt
2026-03-26 12:36:16 +01:00
nevaforget 65d3ba64f9 fix: Audit-Findings — Login-Thread, Traversable-Assets, Session-Exec
- Login-IPC in Background-Thread ausgelagert, UI friert nicht mehr ein
  bei langsamem PAM/greetd (PERF-2/BUG-4)
- importlib.resources korrekt via as_file()/read_text() statt
  Path(str()) — funktioniert in ZIP-Wheels (BUG-1)
- is_absolute()-Check für Session-Exec entfernt, greetd löst PATH
  selbst auf — relative Executables wie 'sway' blockieren nicht mehr (LOGIC-1)
- ValueError (json.JSONDecodeError) im except abgefangen (BUG-2)
2026-03-26 12:34:19 +01:00
23 changed files with 874 additions and 179 deletions
+6
View File
@@ -8,3 +8,9 @@ build/
.pytest_cache/ .pytest_cache/
.pyright/ .pyright/
*.egg *.egg
# makepkg build artifacts
pkg/src/
pkg/pkg/
pkg/*.pkg.tar*
pkg/greetd-moongreet/
+7 -2
View File
@@ -42,5 +42,10 @@ uv run moongreet
- `sessions.py` — Wayland/X11 Sessions aus .desktop Files - `sessions.py` — Wayland/X11 Sessions aus .desktop Files
- `power.py` — Reboot/Shutdown via loginctl - `power.py` — Reboot/Shutdown via loginctl
- `i18n.py` — Locale-Erkennung (LANG / /etc/locale.conf) und String-Tabellen (DE/EN) - `i18n.py` — Locale-Erkennung (LANG / /etc/locale.conf) und String-Tabellen (DE/EN)
- `greeter.py` — GTK4 UI (Overlay-Layout), Faillock-Warnung nach 2 Fehlversuchen - `greeter.py` — GTK4 UI (Overlay-Layout), Faillock-Warnung nach 2 Fehlversuchen, WallpaperWindow für Sekundärmonitore
- `main.py` — Entry Point, GTK App, Layer Shell Setup - `main.py` — Entry Point, GTK App, Layer Shell Setup, Multi-Monitor-Orchestrierung
## Design Decisions
- **Synchrones I/O im GTK-Konstruktor**: `load_config`, `load_strings`, `get_users` und `get_sessions` laufen synchron in `GreeterWindow.__init__`. Async Loading mit Placeholder-UI wäre möglich, erhöht aber die Komplexität erheblich. Der Greeter startet 1x pro Boot auf lokaler Hardware — die Daten sind klein (passwd, locale.conf, wenige .desktop-Files), die Latenz im Normalfall vernachlässigbar.
- **Synchrones Avatar-Decoding**: `GdkPixbuf.Pixbuf.new_from_file_at_scale` läuft synchron auf dem Main Thread. Bei großen Bildern als `.face`-Datei kann die UI kurz stocken. Der Avatar-Cache (`_avatar_cache`) federt das nach dem ersten Laden ab. Async Decoding per Worker-Thread + `GLib.idle_add` wäre die Alternative, rechtfertigt aber den Aufwand nicht für einen Single-User-Greeter.
+3 -1
View File
@@ -5,5 +5,7 @@
vt = 1 vt = 1
[default_session] [default_session]
command = "moongreet" # Moongreet braucht einen Wayland-Compositor — niri stellt diesen bereit.
# Siehe niri-greeter.kdl fuer die Compositor-Konfiguration.
command = "niri -c /etc/greetd/niri-greeter.kdl"
user = "greeter" user = "greeter"
+2 -4
View File
@@ -4,7 +4,5 @@
[appearance] [appearance]
# Absolute path to wallpaper image # Absolute path to wallpaper image
background = "/usr/share/backgrounds/wallpaper.jpg" background = "/usr/share/backgrounds/wallpaper.jpg"
# GTK theme for the greeter UI
[behavior] gtk-theme = "catppuccin-mocha-lavender-standard+default"
# show_user_list = true
# default_session = "Hyprland"
+59
View File
@@ -0,0 +1,59 @@
// ABOUTME: Niri-Konfiguration fuer den Moongreet Login-Greeter.
// ABOUTME: Wird von greetd gestartet — minimale Config ohne Keybinds fuer Sicherheit.
input {
keyboard {
xkb {
layout "de"
}
numlock
}
touchpad {
tap
natural-scroll
}
mouse {
accel-profile "flat"
}
}
cursor {
xcursor-theme "Sweet-cursors"
}
layout {
gaps 0
focus-ring {
off
}
border {
off
}
}
// Moongreet starten und niri beenden, sobald moongreet sich schliesst.
// Retry-Loop stellt sicher, dass niri auch bei fruehen Crashes von moongreet beendet wird.
spawn-sh-at-startup "moongreet; while ! niri msg action quit --skip-confirmation 2>/dev/null; do sleep 0.5; done"
// Greeter-Fenster maximiert darstellen
window-rule {
open-maximized true
}
hotkey-overlay {
skip-at-startup
}
prefer-no-csd
animations {
off
}
binds {
// Keine Keybinds — verhindert Zugriff auf Terminals oder andere Aktionen
}
+20 -9
View File
@@ -1,13 +1,14 @@
# ABOUTME: AUR PKGBUILD for Moongreet — greetd greeter for Wayland. # ABOUTME: PKGBUILD for Moongreet — greetd greeter for Wayland.
# ABOUTME: Builds from local source, installs config and cache directory. # ABOUTME: Builds from git source with automatic version detection.
# Maintainer: Dominik Kressler # Maintainer: Dominik Kressler
pkgname=moongreet pkgname=moongreet-git
pkgver=0.1.0 pkgver=0.2.0.r0.g64f08d7
pkgrel=1 pkgrel=1
pkgdesc="A greetd greeter for Wayland, built with Python + GTK4 + gtk4-layer-shell" pkgdesc="A greetd greeter for Wayland, built with Python + GTK4 + gtk4-layer-shell"
arch=('any') arch=('any')
url="https://gitea.moonarch.de/nevaforget/greetd-moongreet"
license=('MIT') license=('MIT')
depends=( depends=(
'python' 'python'
@@ -17,26 +18,36 @@ depends=(
'greetd' 'greetd'
) )
makedepends=( makedepends=(
'git'
'python-build' 'python-build'
'python-installer' 'python-installer'
'python-hatchling' 'python-hatchling'
) )
provides=('moongreet')
conflicts=('moongreet')
install=moongreet.install install=moongreet.install
source=("$pkgname-$pkgver.tar.gz") source=("git+${url}.git")
sha256sums=('SKIP') sha256sums=('SKIP')
pkgver() {
cd "$srcdir/greetd-moongreet"
git describe --long --tags | sed 's/^v//;s/-/.r/;s/-/./'
}
build() { build() {
cd "$srcdir/$pkgname-$pkgver" cd "$srcdir/greetd-moongreet"
rm -rf dist/
python -m build --wheel --no-isolation python -m build --wheel --no-isolation
} }
package() { package() {
cd "$srcdir/$pkgname-$pkgver" cd "$srcdir/greetd-moongreet"
python -m installer --destdir="$pkgdir" dist/*.whl python -m installer --destdir="$pkgdir" dist/*.whl
# Example config # Greeter config
install -Dm644 config/moongreet.toml "$pkgdir/etc/moongreet/moongreet.toml" install -Dm644 config/moongreet.toml "$pkgdir/etc/moongreet/moongreet.toml"
# Cache directory # Cache directories
install -dm755 "$pkgdir/var/cache/moongreet" install -dm755 "$pkgdir/var/cache/moongreet"
install -dm755 "$pkgdir/var/cache/moongreet/last-session"
} }
+5 -6
View File
@@ -1,16 +1,15 @@
# ABOUTME: pacman install hooks for Moongreet. # ABOUTME: pacman install hooks for Moongreet.
# ABOUTME: Sets ownership on cache directory for the greeter user. # ABOUTME: Sets ownership on cache directory and prints setup instructions.
post_install() { post_install() {
if getent passwd greeter > /dev/null 2>&1; then if getent passwd greeter > /dev/null 2>&1; then
chown greeter:greeter /var/cache/moongreet chown greeter:greeter /var/cache/moongreet
chown greeter:greeter /var/cache/moongreet/last-session
fi fi
echo "==> Copy /etc/moongreet/moongreet.toml and adjust the wallpaper path." echo "==> Moongreet installed."
echo "==> Configure greetd to use moongreet:" echo "==> Add moongreet to your greeter compositor command in /etc/greetd/config.toml."
echo " [default_session]" echo "==> Adjust wallpaper: /etc/moongreet/moongreet.toml"
echo " command = \"moongreet\""
echo " user = \"greeter\""
} }
post_upgrade() { post_upgrade() {
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "moongreet" name = "moongreet"
version = "0.1.0" version = "0.2.1"
description = "A greetd greeter for Wayland with GTK4" description = "A greetd greeter for Wayland with GTK4"
requires-python = ">=3.11" requires-python = ">=3.11"
license = "MIT" license = "MIT"
+36 -1
View File
@@ -1,8 +1,10 @@
# ABOUTME: Configuration loading from moongreet.toml. # ABOUTME: Configuration loading from moongreet.toml.
# ABOUTME: Parses appearance and behavior settings for the greeter. # ABOUTME: Parses appearance and behavior settings with wallpaper path resolution.
import tomllib import tomllib
from contextlib import AbstractContextManager
from dataclasses import dataclass from dataclasses import dataclass
from importlib.resources import as_file, files
from pathlib import Path from pathlib import Path
DEFAULT_CONFIG_PATHS = [ DEFAULT_CONFIG_PATHS = [
@@ -15,6 +17,7 @@ class Config:
"""Greeter configuration loaded from moongreet.toml.""" """Greeter configuration loaded from moongreet.toml."""
background: Path | None = None background: Path | None = None
gtk_theme: str | None = None
def load_config(config_path: Path | None = None) -> Config: def load_config(config_path: Path | None = None) -> Config:
@@ -33,8 +36,11 @@ def load_config(config_path: Path | None = None) -> Config:
if not config_path.exists(): if not config_path.exists():
return Config() return Config()
try:
with open(config_path, "rb") as f: with open(config_path, "rb") as f:
data = tomllib.load(f) data = tomllib.load(f)
except (tomllib.TOMLDecodeError, OSError):
return Config()
config = Config() config = Config()
appearance = data.get("appearance", {}) appearance = data.get("appearance", {})
@@ -46,4 +52,33 @@ def load_config(config_path: Path | None = None) -> Config:
bg_path = config_path.parent / bg_path bg_path = config_path.parent / bg_path
config.background = bg_path config.background = bg_path
gtk_theme = appearance.get("gtk-theme")
if gtk_theme:
config.gtk_theme = gtk_theme
return config return config
_PACKAGE_DATA = files("moongreet") / "data"
_DEFAULT_WALLPAPER_PATH = _PACKAGE_DATA / "wallpaper.jpg"
def resolve_wallpaper_path(
config: Config,
) -> tuple[Path, AbstractContextManager | None]:
"""Resolve the wallpaper path from config or fall back to the package default.
Returns (path, context_manager). The context_manager is non-None when a
package resource was extracted to a temporary file — the caller must keep
it alive and call __exit__ when done.
"""
if config.background and config.background.exists():
return config.background, None
ctx = as_file(_DEFAULT_WALLPAPER_PATH)
try:
path = ctx.__enter__()
except Exception:
ctx.__exit__(None, None, None)
raise
return path, ctx
+217 -73
View File
@@ -1,11 +1,15 @@
# ABOUTME: Main greeter window — builds the GTK4 UI layout for the Moongreet greeter. # ABOUTME: Main greeter window — builds the GTK4 UI layout for the Moongreet greeter.
# ABOUTME: Handles user selection, session choice, password entry, and power actions. # ABOUTME: Handles user selection, session choice, password entry, and power actions.
import logging
import os import os
import re
import shlex import shlex
import shutil
import socket import socket
import stat import stat
import subprocess import subprocess
import threading
from importlib.resources import files from importlib.resources import files
from pathlib import Path from pathlib import Path
@@ -14,19 +18,24 @@ gi.require_version("Gtk", "4.0")
gi.require_version("Gdk", "4.0") gi.require_version("Gdk", "4.0")
from gi.repository import Gtk, Gdk, GLib, GdkPixbuf from gi.repository import Gtk, Gdk, GLib, GdkPixbuf
from moongreet.config import load_config from moongreet.config import Config, load_config, resolve_wallpaper_path
from moongreet.i18n import load_strings, Strings from moongreet.i18n import load_strings, Strings
from moongreet.ipc import create_session, post_auth_response, start_session, cancel_session from moongreet.ipc import create_session, post_auth_response, start_session, cancel_session
from moongreet.users import User, get_users, get_avatar_path, get_user_gtk_theme from moongreet.users import User, get_users, get_avatar_path
from moongreet.sessions import Session, get_sessions from moongreet.sessions import Session, get_sessions
from moongreet.power import reboot, shutdown from moongreet.power import reboot, shutdown
logger = logging.getLogger(__name__)
LAST_USER_PATH = Path("/var/cache/moongreet/last-user") LAST_USER_PATH = Path("/var/cache/moongreet/last-user")
LAST_SESSION_DIR = Path("/var/cache/moongreet/last-session")
FAILLOCK_MAX_ATTEMPTS = 3 FAILLOCK_MAX_ATTEMPTS = 3
VALID_USERNAME = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_.-]*$")
MAX_USERNAME_LENGTH = 256
PACKAGE_DATA = files("moongreet") / "data" PACKAGE_DATA = files("moongreet") / "data"
DEFAULT_AVATAR_PATH = PACKAGE_DATA / "default-avatar.svg" DEFAULT_AVATAR_PATH = PACKAGE_DATA / "default-avatar.svg"
DEFAULT_WALLPAPER_PATH = PACKAGE_DATA / "wallpaper.jpg"
AVATAR_SIZE = 128 AVATAR_SIZE = 128
MAX_AVATAR_FILE_SIZE = 10 * 1024 * 1024 # 10 MB
def faillock_warning(attempt_count: int, strings: Strings | None = None) -> str | None: def faillock_warning(attempt_count: int, strings: Strings | None = None) -> str | None:
@@ -41,27 +50,67 @@ def faillock_warning(attempt_count: int, strings: Strings | None = None) -> str
return None return None
def _build_wallpaper_widget(bg_path: Path | None) -> Gtk.Widget:
"""Create a wallpaper widget that fills the available space."""
if bg_path and bg_path.exists():
background = Gtk.Picture()
background.set_filename(str(bg_path))
background.set_content_fit(Gtk.ContentFit.COVER)
background.set_hexpand(True)
background.set_vexpand(True)
return background
background = Gtk.Box()
background.set_hexpand(True)
background.set_vexpand(True)
return background
class WallpaperWindow(Gtk.ApplicationWindow):
"""A window that shows only the wallpaper — used for secondary monitors."""
def __init__(self, bg_path: Path | None = None, **kwargs) -> None:
super().__init__(**kwargs)
self.add_css_class("greeter")
self.set_default_size(1920, 1080)
self.set_child(_build_wallpaper_widget(bg_path))
class GreeterWindow(Gtk.ApplicationWindow): class GreeterWindow(Gtk.ApplicationWindow):
"""The main greeter window with login UI.""" """The main greeter window with login UI."""
def __init__(self, **kwargs) -> None: def __init__(self, bg_path: Path | None = None, config: Config | None = None, **kwargs) -> None:
super().__init__(**kwargs) super().__init__(**kwargs)
self.add_css_class("greeter") self.add_css_class("greeter")
self.set_default_size(1920, 1080) self.set_default_size(1920, 1080)
self._config = load_config() self._config = config if config is not None else load_config()
self._strings = load_strings() self._strings = load_strings()
self._users = get_users() self._users = get_users()
self._sessions = get_sessions() self._sessions = get_sessions()
self._selected_user: User | None = None self._selected_user: User | None = None
self._greetd_sock: socket.socket | None = None self._greetd_sock: socket.socket | None = None
self._greetd_sock_lock = threading.Lock()
self._login_cancelled = threading.Event()
self._default_avatar_pixbuf: GdkPixbuf.Pixbuf | None = None self._default_avatar_pixbuf: GdkPixbuf.Pixbuf | None = None
self._avatar_cache: dict[str, GdkPixbuf.Pixbuf] = {} self._avatar_cache: dict[str, GdkPixbuf.Pixbuf] = {}
self._failed_attempts: dict[str, int] = {} self._failed_attempts: dict[str, int] = {}
self._bg_path = bg_path
self._apply_global_theme()
self._build_ui() self._build_ui()
self._select_initial_user()
self._setup_keyboard_navigation() self._setup_keyboard_navigation()
# Defer initial user selection until the window is realized,
# so get_color() returns the actual theme foreground for SVG tinting
self.connect("realize", self._on_realize)
def _on_realize(self, widget: Gtk.Widget) -> None:
"""Called when the window is realized — select initial user.
Deferred from __init__ so get_color() returns actual theme values
for SVG tinting. Uses idle_add so the first frame renders before
avatar loading blocks the main loop.
"""
GLib.idle_add(self._select_initial_user)
def _build_ui(self) -> None: def _build_ui(self) -> None:
"""Build the complete greeter UI layout.""" """Build the complete greeter UI layout."""
@@ -69,21 +118,8 @@ class GreeterWindow(Gtk.ApplicationWindow):
overlay = Gtk.Overlay() overlay = Gtk.Overlay()
self.set_child(overlay) self.set_child(overlay)
# Background wallpaper (blurred and darkened) # Background wallpaper
bg_path = self._config.background overlay.set_child(_build_wallpaper_widget(self._bg_path))
if not bg_path or not bg_path.exists():
bg_path = Path(str(DEFAULT_WALLPAPER_PATH))
if bg_path.exists():
background = Gtk.Picture()
background.set_filename(str(bg_path))
background.set_content_fit(Gtk.ContentFit.COVER)
background.set_hexpand(True)
background.set_vexpand(True)
else:
background = Gtk.Box()
background.set_hexpand(True)
background.set_vexpand(True)
overlay.set_child(background)
# Main layout: 3 rows (top spacer, center login, bottom bar) # Main layout: 3 rows (top spacer, center login, bottom bar)
main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL) main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
@@ -213,10 +249,13 @@ class GreeterWindow(Gtk.ApplicationWindow):
return bar return bar
def _select_initial_user(self) -> None: def _select_initial_user(self) -> bool:
"""Select the last user or the first available user.""" """Select the last user or the first available user.
Returns False to deregister from GLib.idle_add after a single invocation.
"""
if not self._users: if not self._users:
return return False
# Try to load last user # Try to load last user
last_username = self._load_last_user() last_username = self._load_last_user()
@@ -232,6 +271,7 @@ class GreeterWindow(Gtk.ApplicationWindow):
target_user = self._users[0] target_user = self._users[0]
self._switch_to_user(target_user) self._switch_to_user(target_user)
return False
def _switch_to_user(self, user: User) -> None: def _switch_to_user(self, user: User) -> None:
"""Update the UI to show the selected user.""" """Update the UI to show the selected user."""
@@ -249,31 +289,28 @@ class GreeterWindow(Gtk.ApplicationWindow):
) )
if avatar_path and avatar_path.exists(): if avatar_path and avatar_path.exists():
self._set_avatar_from_file(avatar_path, user.username) self._set_avatar_from_file(avatar_path, user.username)
elif DEFAULT_AVATAR_PATH.exists():
self._set_default_avatar()
else: else:
self._avatar_image.set_from_icon_name("avatar-default-symbolic") # Default avatar — _set_default_avatar uses Traversable.read_text()
# which works in ZIP wheels too, no exists() check needed
self._set_default_avatar()
# Apply user's GTK theme if available # Pre-select last used session for this user
self._apply_user_theme(user) self._select_last_session(user)
# Focus password entry # Focus password entry
self._password_entry.grab_focus() self._password_entry.grab_focus()
def _apply_user_theme(self, user: User) -> None: def _apply_global_theme(self) -> None:
"""Load the user's preferred GTK theme from their settings.ini.""" """Apply the GTK theme from moongreet.toml configuration."""
gtk_config_dir = user.home / ".config" / "gtk-4.0" theme_name = self._config.gtk_theme
theme_name = get_user_gtk_theme(config_dir=gtk_config_dir) if not theme_name:
return
settings = Gtk.Settings.get_default() settings = Gtk.Settings.get_default()
if settings is None: if settings is None:
return return
current = settings.get_property("gtk-theme-name")
if theme_name and current != theme_name:
settings.set_property("gtk-theme-name", theme_name) settings.set_property("gtk-theme-name", theme_name)
elif not theme_name and current:
settings.reset_property("gtk-theme-name")
def _get_foreground_color(self) -> str: def _get_foreground_color(self) -> str:
"""Get the current GTK theme foreground color as a hex string.""" """Get the current GTK theme foreground color as a hex string."""
@@ -307,6 +344,9 @@ class GreeterWindow(Gtk.ApplicationWindow):
def _set_avatar_from_file(self, path: Path, username: str | None = None) -> None: def _set_avatar_from_file(self, path: Path, username: str | None = None) -> None:
"""Load an image file and set it as the avatar, scaled to AVATAR_SIZE.""" """Load an image file and set it as the avatar, scaled to AVATAR_SIZE."""
try: try:
if path.stat().st_size > MAX_AVATAR_FILE_SIZE:
self._avatar_image.set_from_icon_name("avatar-default-symbolic")
return
pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale( pixbuf = GdkPixbuf.Pixbuf.new_from_file_at_scale(
str(path), AVATAR_SIZE, AVATAR_SIZE, True str(path), AVATAR_SIZE, AVATAR_SIZE, True
) )
@@ -372,6 +412,7 @@ class GreeterWindow(Gtk.ApplicationWindow):
def _close_greetd_sock(self) -> None: def _close_greetd_sock(self) -> None:
"""Close the greetd socket and reset the reference.""" """Close the greetd socket and reset the reference."""
with self._greetd_sock_lock:
if self._greetd_sock: if self._greetd_sock:
try: try:
self._greetd_sock.close() self._greetd_sock.close()
@@ -379,6 +420,11 @@ class GreeterWindow(Gtk.ApplicationWindow):
pass pass
self._greetd_sock = None self._greetd_sock = None
def _set_login_sensitive(self, sensitive: bool) -> None:
"""Enable or disable login controls during authentication."""
self._password_entry.set_sensitive(sensitive)
self._session_dropdown.set_sensitive(sensitive)
def _attempt_login(self, user: User, password: str, session: Session) -> None: def _attempt_login(self, user: User, password: str, session: Session) -> None:
"""Attempt to authenticate and start a session via greetd IPC.""" """Attempt to authenticate and start a session via greetd IPC."""
sock_path = os.environ.get("GREETD_SOCK") sock_path = os.environ.get("GREETD_SOCK")
@@ -389,75 +435,120 @@ class GreeterWindow(Gtk.ApplicationWindow):
if not self._validate_greetd_sock(sock_path): if not self._validate_greetd_sock(sock_path):
return return
# Disable UI while authenticating — the IPC runs in a background thread
self._login_cancelled.clear()
self._set_login_sensitive(False)
thread = threading.Thread(
target=self._login_worker,
args=(user, password, session, sock_path),
daemon=True,
)
thread.start()
def _login_worker(self, user: User, password: str, session: Session, sock_path: str) -> None:
"""Run greetd IPC in a background thread to avoid blocking the GTK main loop."""
try: try:
if self._login_cancelled.is_set():
return
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(10.0) sock.settimeout(10.0)
sock.connect(sock_path) sock.connect(sock_path)
with self._greetd_sock_lock:
self._greetd_sock = sock self._greetd_sock = sock
# Step 1: Create session # Step 1: Create session — if a stale session exists, cancel it and retry
response = create_session(sock, user.username) response = create_session(sock, user.username)
if self._login_cancelled.is_set():
return
if response.get("type") == "error": if response.get("type") == "error":
self._show_greetd_error(response, self._strings.auth_failed) cancel_session(sock)
self._close_greetd_sock() response = create_session(sock, user.username)
if self._login_cancelled.is_set():
return
if response.get("type") == "error":
GLib.idle_add(self._on_login_error, response, self._strings.auth_failed)
return return
# Step 2: Send password if auth message received # Step 2: Send password if auth message received
if response.get("type") == "auth_message": if response.get("type") == "auth_message":
response = post_auth_response(sock, password) response = post_auth_response(sock, password)
if self._login_cancelled.is_set():
return
if response.get("type") == "error": if response.get("type") == "error":
self._failed_attempts[user.username] = self._failed_attempts.get(user.username, 0) + 1 self._failed_attempts[user.username] = self._failed_attempts.get(user.username, 0) + 1
self._show_greetd_error(response, self._strings.wrong_password)
warning = faillock_warning(self._failed_attempts[user.username], self._strings) warning = faillock_warning(self._failed_attempts[user.username], self._strings)
if warning: cancel_session(sock)
current = self._error_label.get_text() GLib.idle_add(self._on_login_auth_error, response, warning)
self._error_label.set_text(f"{current}\n{warning}")
self._close_greetd_sock()
return return
if response.get("type") == "auth_message": if response.get("type") == "auth_message":
# Multi-stage auth (e.g. TOTP) is not supported # Multi-stage auth (e.g. TOTP) is not supported
cancel_session(sock) cancel_session(sock)
self._close_greetd_sock() GLib.idle_add(self._on_login_error, None, self._strings.multi_stage_unsupported)
self._show_error(self._strings.multi_stage_unsupported)
return return
# Step 3: Start session # Step 3: Start session
if response.get("type") == "success": if response.get("type") == "success":
cmd = shlex.split(session.exec_cmd) cmd = shlex.split(session.exec_cmd)
if not cmd or not Path(cmd[0]).is_absolute(): if not cmd or not shutil.which(cmd[0]):
self._show_error(self._strings.invalid_session_command)
cancel_session(sock) cancel_session(sock)
self._close_greetd_sock() GLib.idle_add(self._on_login_error, None, self._strings.invalid_session_command)
return return
response = start_session(sock, cmd) response = start_session(sock, cmd)
if self._login_cancelled.is_set():
return
if response.get("type") == "success": if response.get("type") == "success":
self._save_last_user(user.username) self._save_last_user(user.username)
self._close_greetd_sock() self._save_last_session(user.username, session.name)
self.get_application().quit() GLib.idle_add(self.get_application().quit)
return return
else: else:
self._show_greetd_error(response, self._strings.session_start_failed) GLib.idle_add(self._on_login_error, response, self._strings.session_start_failed)
return
except (ConnectionError, OSError, ValueError) as e:
if self._login_cancelled.is_set():
# Socket was closed by _cancel_pending_session — exit silently
return
logger.error("greetd IPC error: %s", e)
if isinstance(e, ConnectionError):
GLib.idle_add(self._on_login_error, None, self._strings.connection_error)
else:
GLib.idle_add(self._on_login_error, None, self._strings.socket_error)
finally:
self._close_greetd_sock() self._close_greetd_sock()
except ConnectionError as e: def _on_login_error(self, response: dict | None, message: str) -> None:
self._close_greetd_sock() """Handle login error on the GTK main thread."""
self._show_error(self._strings.connection_error.format(error=e)) if response:
except OSError as e: self._show_greetd_error(response, message)
self._close_greetd_sock() else:
self._show_error(self._strings.socket_error.format(error=e)) self._show_error(message)
self._set_login_sensitive(True)
def _on_login_auth_error(self, response: dict, warning: str | None) -> None:
"""Handle authentication failure with optional faillock warning on the GTK main thread."""
self._show_greetd_error(response, self._strings.wrong_password)
if warning:
current = self._error_label.get_text()
self._error_label.set_text(f"{current}\n{warning}")
self._set_login_sensitive(True)
def _cancel_pending_session(self) -> None: def _cancel_pending_session(self) -> None:
"""Cancel any in-progress greetd session.""" """Cancel any in-progress greetd session.
if self._greetd_sock:
try: Sets the cancellation event and closes the socket to interrupt
cancel_session(self._greetd_sock) any blocking I/O in the login worker. The worker checks the
except (ConnectionError, OSError): event and exits silently instead of showing an error.
pass """
self._login_cancelled.set()
self._close_greetd_sock() self._close_greetd_sock()
def _get_selected_session(self) -> Session | None: def _get_selected_session(self) -> Session | None:
@@ -469,6 +560,18 @@ class GreeterWindow(Gtk.ApplicationWindow):
return self._sessions[idx] return self._sessions[idx]
return None return None
def _select_last_session(self, user: User) -> None:
"""Pre-select the last used session for a user in the dropdown."""
if not self._sessions:
return
last_session_name = self._load_last_session(user.username)
if not last_session_name:
return
for i, session in enumerate(self._sessions):
if session.name == last_session_name:
self._session_dropdown.set_selected(i)
return
MAX_GREETD_ERROR_LENGTH = 200 MAX_GREETD_ERROR_LENGTH = 200
def _show_greetd_error(self, response: dict, fallback: str) -> None: def _show_greetd_error(self, response: dict, fallback: str) -> None:
@@ -488,26 +591,38 @@ class GreeterWindow(Gtk.ApplicationWindow):
def _on_reboot_clicked(self, button: Gtk.Button) -> None: def _on_reboot_clicked(self, button: Gtk.Button) -> None:
"""Handle reboot button click.""" """Handle reboot button click."""
try: button.set_sensitive(False)
reboot() threading.Thread(
except subprocess.CalledProcessError: target=self._power_worker, args=(reboot, self._strings.reboot_failed),
self._show_error(self._strings.reboot_failed) daemon=True,
).start()
def _on_shutdown_clicked(self, button: Gtk.Button) -> None: def _on_shutdown_clicked(self, button: Gtk.Button) -> None:
"""Handle shutdown button click.""" """Handle shutdown button click."""
button.set_sensitive(False)
threading.Thread(
target=self._power_worker, args=(shutdown, self._strings.shutdown_failed),
daemon=True,
).start()
def _power_worker(self, action, error_msg: str) -> None:
"""Run a power action in a background thread to avoid blocking the GTK main loop."""
try: try:
shutdown() action()
except subprocess.CalledProcessError: except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
self._show_error(self._strings.shutdown_failed) GLib.idle_add(self._show_error, error_msg)
@staticmethod @staticmethod
def _load_last_user() -> str | None: def _load_last_user() -> str | None:
"""Load the last logged-in username from cache.""" """Load the last logged-in username from cache."""
if LAST_USER_PATH.exists(): if LAST_USER_PATH.exists():
try: try:
return LAST_USER_PATH.read_text().strip() username = LAST_USER_PATH.read_text().strip()
except OSError: except OSError:
return None return None
if len(username) > MAX_USERNAME_LENGTH or not VALID_USERNAME.match(username):
return None
return username
return None return None
@staticmethod @staticmethod
@@ -518,3 +633,32 @@ class GreeterWindow(Gtk.ApplicationWindow):
LAST_USER_PATH.write_text(username) LAST_USER_PATH.write_text(username)
except OSError: except OSError:
pass # Non-critical — cache dir may not be writable pass # Non-critical — cache dir may not be writable
MAX_SESSION_NAME_LENGTH = 256
@staticmethod
def _save_last_session(username: str, session_name: str) -> None:
"""Save the last used session name for a user to cache."""
if not VALID_USERNAME.match(username) or len(username) > MAX_USERNAME_LENGTH:
return
if not session_name or len(session_name) > GreeterWindow.MAX_SESSION_NAME_LENGTH:
return
try:
LAST_SESSION_DIR.mkdir(parents=True, exist_ok=True)
(LAST_SESSION_DIR / username).write_text(session_name)
except OSError:
pass # Non-critical — cache dir may not be writable
@staticmethod
def _load_last_session(username: str) -> str | None:
"""Load the last used session name for a user from cache."""
session_file = LAST_SESSION_DIR / username
if not session_file.exists():
return None
try:
name = session_file.read_text().strip()
except OSError:
return None
if not name or len(name) > GreeterWindow.MAX_SESSION_NAME_LENGTH:
return None
return name
+10 -6
View File
@@ -31,9 +31,11 @@ class Strings:
reboot_failed: str reboot_failed: str
shutdown_failed: str shutdown_failed: str
# Templates (use .format()) # Error messages (continued)
connection_error: str connection_error: str
socket_error: str socket_error: str
# Templates (use .format())
faillock_attempts_remaining: str faillock_attempts_remaining: str
faillock_locked: str faillock_locked: str
@@ -54,8 +56,8 @@ _STRINGS_DE = Strings(
session_start_failed="Session konnte nicht gestartet werden", session_start_failed="Session konnte nicht gestartet werden",
reboot_failed="Neustart fehlgeschlagen", reboot_failed="Neustart fehlgeschlagen",
shutdown_failed="Herunterfahren fehlgeschlagen", shutdown_failed="Herunterfahren fehlgeschlagen",
connection_error="Verbindungsfehler: {error}", connection_error="Verbindungsfehler",
socket_error="Socket-Fehler: {error}", socket_error="Socket-Fehler",
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",
) )
@@ -76,8 +78,8 @@ _STRINGS_EN = Strings(
session_start_failed="Failed to start session", session_start_failed="Failed to start session",
reboot_failed="Reboot failed", reboot_failed="Reboot failed",
shutdown_failed="Shutdown failed", shutdown_failed="Shutdown failed",
connection_error="Connection error: {error}", connection_error="Connection error",
socket_error="Socket error: {error}", socket_error="Socket error",
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",
) )
@@ -102,7 +104,9 @@ def detect_locale(locale_conf_path: Path = DEFAULT_LOCALE_CONF) -> str:
return "en" return "en"
# Extract language prefix: "de_DE.UTF-8" → "de" # Extract language prefix: "de_DE.UTF-8" → "de"
lang = lang.split("_")[0].split(".")[0] lang = lang.split("_")[0].split(".")[0].lower()
if not lang.isalpha():
return "en"
return lang return lang
+4 -2
View File
@@ -22,14 +22,16 @@ def _recvall(sock: Any, n: int) -> bytes:
def send_message(sock: Any, msg: dict) -> None: def send_message(sock: Any, msg: dict) -> None:
"""Send a length-prefixed JSON message to the greetd socket.""" """Send a length-prefixed JSON message to the greetd socket."""
payload = json.dumps(msg).encode("utf-8") payload = json.dumps(msg).encode("utf-8")
header = struct.pack("!I", len(payload)) if len(payload) > MAX_PAYLOAD_SIZE:
raise ValueError(f"Payload too large: {len(payload)} bytes (max {MAX_PAYLOAD_SIZE})")
header = struct.pack("=I", len(payload))
sock.sendall(header + payload) sock.sendall(header + payload)
def recv_message(sock: Any) -> dict: def recv_message(sock: Any) -> dict:
"""Receive a length-prefixed JSON message from the greetd socket.""" """Receive a length-prefixed JSON message from the greetd socket."""
header = _recvall(sock, 4) header = _recvall(sock, 4)
length = struct.unpack("!I", header)[0] length = struct.unpack("=I", header)[0]
if length > MAX_PAYLOAD_SIZE: if length > MAX_PAYLOAD_SIZE:
raise ConnectionError(f"Payload too large: {length} bytes (max {MAX_PAYLOAD_SIZE})") raise ConnectionError(f"Payload too large: {length} bytes (max {MAX_PAYLOAD_SIZE})")
+95 -14
View File
@@ -1,15 +1,18 @@
# ABOUTME: Entry point for Moongreet — sets up GTK Application and Layer Shell. # ABOUTME: Entry point for Moongreet — sets up GTK Application and Layer Shell.
# ABOUTME: Handles CLI invocation and initializes the greeter window. # ABOUTME: Handles multi-monitor setup: login UI on primary, wallpaper on secondary monitors.
import logging
import sys import sys
from importlib.resources import files from importlib.resources import files
from pathlib import Path
import gi import gi
gi.require_version("Gtk", "4.0") gi.require_version("Gtk", "4.0")
gi.require_version("Gdk", "4.0") gi.require_version("Gdk", "4.0")
from gi.repository import Gtk, Gdk from gi.repository import Gtk, Gdk
from moongreet.greeter import GreeterWindow from moongreet.config import load_config, resolve_wallpaper_path
from moongreet.greeter import GreeterWindow, WallpaperWindow
# gtk4-layer-shell is optional for development/testing # gtk4-layer-shell is optional for development/testing
try: try:
@@ -19,45 +22,121 @@ try:
except (ValueError, ImportError): except (ValueError, ImportError):
HAS_LAYER_SHELL = False HAS_LAYER_SHELL = False
LOG_DIR = Path("/var/cache/moongreet")
LOG_FILE = LOG_DIR / "moongreet.log"
logger = logging.getLogger(__name__)
def _setup_logging() -> None:
"""Configure logging to file and stderr."""
root = logging.getLogger()
root.setLevel(logging.INFO)
formatter = logging.Formatter(
"%(asctime)s %(levelname)s %(name)s: %(message)s"
)
# Always log to stderr
stderr_handler = logging.StreamHandler(sys.stderr)
stderr_handler.setLevel(logging.INFO)
stderr_handler.setFormatter(formatter)
root.addHandler(stderr_handler)
# Log to file if the directory is writable
if LOG_DIR.is_dir():
try:
file_handler = logging.FileHandler(LOG_FILE)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(formatter)
root.addHandler(file_handler)
except PermissionError:
logger.warning("Cannot write to %s", LOG_FILE)
class MoongreetApp(Gtk.Application): class MoongreetApp(Gtk.Application):
"""GTK Application for the Moongreet greeter.""" """GTK Application for the Moongreet greeter."""
def __init__(self) -> None: def __init__(self) -> None:
super().__init__(application_id="dev.moonarch.moongreet") super().__init__(application_id="dev.moonarch.moongreet")
self._wallpaper_ctx = None
self._secondary_windows: list[WallpaperWindow] = []
def do_activate(self) -> None: def do_activate(self) -> None:
"""Create and present the greeter window.""" """Create and present greeter windows on all monitors."""
self._register_icons() display = Gdk.Display.get_default()
self._load_css() if display is None:
window = GreeterWindow(application=self) logger.error("No display available — cannot start greeter UI")
return
self._register_icons(display)
self._load_css(display)
# Resolve wallpaper once, share across all windows
config = load_config()
bg_path, self._wallpaper_ctx = resolve_wallpaper_path(config)
monitors = display.get_monitors()
primary_monitor = None
# Find primary monitor — fall back to first available
for i in range(monitors.get_n_items()):
monitor = monitors.get_item(i)
if hasattr(monitor, 'is_primary') and monitor.is_primary():
primary_monitor = monitor
break
if primary_monitor is None and monitors.get_n_items() > 0:
primary_monitor = monitors.get_item(0)
# Main greeter window (login UI) on primary monitor
greeter = GreeterWindow(bg_path=bg_path, config=config, application=self)
if HAS_LAYER_SHELL: if HAS_LAYER_SHELL:
self._setup_layer_shell(window) self._setup_layer_shell(greeter, keyboard=True)
if primary_monitor is not None:
Gtk4LayerShell.set_monitor(greeter, primary_monitor)
greeter.present()
window.present() # Wallpaper-only windows on secondary monitors
for i in range(monitors.get_n_items()):
monitor = monitors.get_item(i)
if monitor == primary_monitor:
continue
wallpaper_win = WallpaperWindow(bg_path=bg_path, application=self)
if HAS_LAYER_SHELL:
self._setup_layer_shell(wallpaper_win, keyboard=False)
Gtk4LayerShell.set_monitor(wallpaper_win, monitor)
wallpaper_win.present()
self._secondary_windows.append(wallpaper_win)
def _register_icons(self) -> None: def do_shutdown(self) -> None:
"""Clean up wallpaper context manager on exit."""
if self._wallpaper_ctx is not None:
self._wallpaper_ctx.__exit__(None, None, None)
self._wallpaper_ctx = None
Gtk.Application.do_shutdown(self)
def _register_icons(self, display: Gdk.Display) -> None:
"""Register custom icons from the package data/icons directory.""" """Register custom icons from the package data/icons directory."""
icons_dir = files("moongreet") / "data" / "icons" icons_dir = files("moongreet") / "data" / "icons"
icon_theme = Gtk.IconTheme.get_for_display(Gdk.Display.get_default()) icon_theme = Gtk.IconTheme.get_for_display(display)
icon_theme.add_search_path(str(icons_dir)) icon_theme.add_search_path(str(icons_dir))
def _load_css(self) -> None: def _load_css(self, display: Gdk.Display) -> None:
"""Load the CSS stylesheet for the greeter.""" """Load the CSS stylesheet for the greeter."""
css_provider = Gtk.CssProvider() css_provider = Gtk.CssProvider()
css_path = files("moongreet") / "style.css" css_path = files("moongreet") / "style.css"
css_provider.load_from_path(str(css_path)) css_provider.load_from_path(str(css_path))
Gtk.StyleContext.add_provider_for_display( Gtk.StyleContext.add_provider_for_display(
Gdk.Display.get_default(), display,
css_provider, css_provider,
Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
) )
def _setup_layer_shell(self, window: Gtk.Window) -> None: def _setup_layer_shell(self, window: Gtk.Window, keyboard: bool = False) -> None:
"""Configure gtk4-layer-shell for fullscreen greeter display.""" """Configure gtk4-layer-shell for fullscreen display."""
Gtk4LayerShell.init_for_window(window) Gtk4LayerShell.init_for_window(window)
Gtk4LayerShell.set_layer(window, Gtk4LayerShell.Layer.TOP) Gtk4LayerShell.set_layer(window, Gtk4LayerShell.Layer.TOP)
if keyboard:
Gtk4LayerShell.set_keyboard_mode( Gtk4LayerShell.set_keyboard_mode(
window, Gtk4LayerShell.KeyboardMode.EXCLUSIVE window, Gtk4LayerShell.KeyboardMode.EXCLUSIVE
) )
@@ -73,6 +152,8 @@ class MoongreetApp(Gtk.Application):
def main() -> None: def main() -> None:
"""Run the Moongreet application.""" """Run the Moongreet application."""
_setup_logging()
logger.info("Moongreet starting")
app = MoongreetApp() app = MoongreetApp()
app.run(sys.argv) app.run(sys.argv)
+5 -2
View File
@@ -4,11 +4,14 @@
import subprocess import subprocess
POWER_TIMEOUT = 30
def reboot() -> None: def reboot() -> None:
"""Reboot the system via loginctl.""" """Reboot the system via loginctl."""
subprocess.run(["loginctl", "reboot"], check=True) subprocess.run(["loginctl", "reboot"], check=True, timeout=POWER_TIMEOUT)
def shutdown() -> None: def shutdown() -> None:
"""Shut down the system via loginctl.""" """Shut down the system via loginctl."""
subprocess.run(["loginctl", "poweroff"], check=True) subprocess.run(["loginctl", "poweroff"], check=True, timeout=POWER_TIMEOUT)
+3 -2
View File
@@ -2,6 +2,7 @@
# ABOUTME: Parses .desktop files from standard session directories. # ABOUTME: Parses .desktop files from standard session directories.
import configparser import configparser
from collections.abc import Sequence
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -37,8 +38,8 @@ def _parse_desktop_file(path: Path, session_type: str) -> Session | None:
def get_sessions( def get_sessions(
wayland_dirs: tuple[Path, ...] = DEFAULT_WAYLAND_DIRS, wayland_dirs: Sequence[Path] = DEFAULT_WAYLAND_DIRS,
xsession_dirs: tuple[Path, ...] = DEFAULT_XSESSION_DIRS, xsession_dirs: Sequence[Path] = DEFAULT_XSESSION_DIRS,
) -> list[Session]: ) -> list[Session]:
"""Discover available sessions from .desktop files.""" """Discover available sessions from .desktop files."""
sessions: list[Session] = [] sessions: list[Session] = []
+2 -4
View File
@@ -12,16 +12,14 @@ window.greeter {
.login-box { .login-box {
padding: 40px; padding: 40px;
border-radius: 12px; border-radius: 12px;
background-color: alpha(@theme_bg_color, 0.7); background-color: transparent;
} }
/* Round avatar image */ /* Round avatar image — size is set via set_size_request() in code */
.avatar { .avatar {
border-radius: 50%; border-radius: 50%;
min-width: 128px; min-width: 128px;
min-height: 128px; min-height: 128px;
max-width: 128px;
max-height: 128px;
background-color: @theme_selected_bg_color; background-color: @theme_selected_bg_color;
border: 3px solid alpha(white, 0.3); border: 3px solid alpha(white, 0.3);
} }
+3 -1
View File
@@ -102,6 +102,8 @@ def get_user_gtk_theme(config_dir: Path | None = None) -> str | None:
return None return None
if config.has_option("Settings", "gtk-theme-name"): if config.has_option("Settings", "gtk-theme-name"):
return config.get("Settings", "gtk-theme-name") theme = config.get("Settings", "gtk-theme-name")
if theme:
return theme
return None return None
+64 -1
View File
@@ -5,7 +5,7 @@ from pathlib import Path
import pytest import pytest
from moongreet.config import load_config, Config from moongreet.config import load_config, resolve_wallpaper_path, Config
class TestLoadConfig: class TestLoadConfig:
@@ -35,6 +35,34 @@ class TestLoadConfig:
assert config.background is None assert config.background is None
def test_returns_defaults_for_corrupt_toml(self, tmp_path: Path) -> None:
toml_file = tmp_path / "moongreet.toml"
toml_file.write_text("this is not valid [[[ toml !!!")
config = load_config(toml_file)
assert config.background is None
def test_loads_gtk_theme(self, tmp_path: Path) -> None:
toml_file = tmp_path / "moongreet.toml"
toml_file.write_text(
"[appearance]\n"
'gtk-theme = "Catppuccin-Mocha-Standard-Blue-Dark"\n'
)
config = load_config(toml_file)
assert config.gtk_theme == "Catppuccin-Mocha-Standard-Blue-Dark"
def test_returns_none_gtk_theme_when_missing(self, tmp_path: Path) -> None:
toml_file = tmp_path / "moongreet.toml"
toml_file.write_text("[appearance]\n")
config = load_config(toml_file)
assert config.gtk_theme is None
def test_resolves_relative_path_against_config_dir(self, tmp_path: Path) -> None: def test_resolves_relative_path_against_config_dir(self, tmp_path: Path) -> None:
toml_file = tmp_path / "moongreet.toml" toml_file = tmp_path / "moongreet.toml"
toml_file.write_text( toml_file.write_text(
@@ -45,3 +73,38 @@ class TestLoadConfig:
config = load_config(toml_file) config = load_config(toml_file)
assert config.background == tmp_path / "wallpaper.jpg" assert config.background == tmp_path / "wallpaper.jpg"
class TestResolveWallpaperPath:
"""Tests for resolving the wallpaper path from config or package default."""
def test_uses_configured_path_when_exists(self, tmp_path: Path) -> None:
wallpaper = tmp_path / "custom.jpg"
wallpaper.write_bytes(b"fake-image")
config = Config(background=wallpaper)
path, ctx = resolve_wallpaper_path(config)
assert path == wallpaper
assert ctx is None
def test_falls_back_to_package_default(self) -> None:
config = Config(background=None)
path, ctx = resolve_wallpaper_path(config)
assert path is not None
assert path.exists()
assert ctx is not None
# Clean up context manager
ctx.__exit__(None, None, None)
def test_falls_back_when_configured_path_missing(self, tmp_path: Path) -> None:
config = Config(background=tmp_path / "nonexistent.jpg")
path, ctx = resolve_wallpaper_path(config)
assert path is not None
assert path.exists()
assert ctx is not None
ctx.__exit__(None, None, None)
+11 -3
View File
@@ -63,6 +63,13 @@ class TestDetectLocale:
assert result == "en" assert result == "en"
def test_rejects_non_alpha_lang(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LANG", "../../etc")
result = detect_locale()
assert result == "en"
class TestLoadStrings: class TestLoadStrings:
"""Tests for loading the correct string table.""" """Tests for loading the correct string table."""
@@ -111,8 +118,9 @@ class TestLoadStrings:
result = strings.faillock_attempts_remaining.format(n=1) result = strings.faillock_attempts_remaining.format(n=1)
assert "1" in result assert "1" in result
def test_connection_error_template(self) -> None: def test_connection_error_is_generic(self) -> None:
strings = load_strings("en") strings = load_strings("en")
result = strings.connection_error.format(error="timeout") # Error messages should not contain format placeholders (no info leakage)
assert "timeout" in result assert "{" not in strings.connection_error
assert "{" not in strings.socket_error
+263 -13
View File
@@ -10,7 +10,8 @@ from pathlib import Path
import pytest import pytest
from moongreet.greeter import faillock_warning, FAILLOCK_MAX_ATTEMPTS from moongreet.greeter import faillock_warning, FAILLOCK_MAX_ATTEMPTS, LAST_SESSION_DIR
from moongreet.i18n import load_strings
from moongreet.ipc import create_session, post_auth_response, start_session, cancel_session from moongreet.ipc import create_session, post_auth_response, start_session, cancel_session
@@ -39,22 +40,33 @@ class MockGreetd:
self._thread = threading.Thread(target=self._serve, daemon=True) self._thread = threading.Thread(target=self._serve, daemon=True)
self._thread.start() self._thread.start()
@staticmethod
def _recvall(conn: socket.socket, n: int) -> bytes:
"""Receive exactly n bytes from a socket, handling fragmented reads."""
buf = bytearray()
while len(buf) < n:
chunk = conn.recv(n - len(buf))
if not chunk:
break
buf.extend(chunk)
return bytes(buf)
def _serve(self) -> None: def _serve(self) -> None:
conn, _ = self._server.accept() conn, _ = self._server.accept()
try: try:
for response in self._responses: for response in self._responses:
# Receive a message # Receive a message
header = conn.recv(4) header = self._recvall(conn, 4)
if len(header) < 4: if len(header) < 4:
break break
length = struct.unpack("!I", header)[0] length = struct.unpack("=I", header)[0]
payload = conn.recv(length) payload = self._recvall(conn, length)
msg = json.loads(payload.decode("utf-8")) msg = json.loads(payload.decode("utf-8"))
self._received.append(msg) self._received.append(msg)
# Send response # Send response
resp_payload = json.dumps(response).encode("utf-8") resp_payload = json.dumps(response).encode("utf-8")
conn.sendall(struct.pack("!I", len(resp_payload)) + resp_payload) conn.sendall(struct.pack("=I", len(resp_payload)) + resp_payload)
finally: finally:
conn.close() conn.close()
@@ -100,12 +112,13 @@ class TestLoginFlow:
assert mock.received[1] == {"type": "post_auth_message_response", "response": "geheim"} assert mock.received[1] == {"type": "post_auth_message_response", "response": "geheim"}
assert mock.received[2] == {"type": "start_session", "cmd": ["Hyprland"]} assert mock.received[2] == {"type": "start_session", "cmd": ["Hyprland"]}
def test_wrong_password(self, tmp_path: Path) -> None: def test_wrong_password_sends_cancel(self, tmp_path: Path) -> None:
"""Simulate a failed login due to wrong password.""" """After a failed login, cancel_session must be sent to free the greetd session."""
sock_path = tmp_path / "greetd.sock" sock_path = tmp_path / "greetd.sock"
mock = MockGreetd(sock_path) mock = MockGreetd(sock_path)
mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"}) mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
mock.expect({"type": "error", "error_type": "auth_error", "description": "Authentication failed"}) mock.expect({"type": "error", "error_type": "auth_error", "description": "Authentication failed"})
mock.expect({"type": "success"}) # Response to cancel_session
mock.start() mock.start()
try: try:
@@ -119,10 +132,64 @@ class TestLoginFlow:
assert response["type"] == "error" assert response["type"] == "error"
assert response["description"] == "Authentication failed" assert response["description"] == "Authentication failed"
# The greeter must cancel the session after auth failure
response = cancel_session(sock)
assert response["type"] == "success"
sock.close() sock.close()
finally: finally:
mock.stop() mock.stop()
assert mock.received[2] == {"type": "cancel_session"}
def test_stale_session_cancel_and_retry(self, tmp_path: Path) -> None:
"""When create_session fails due to a stale session, cancel and retry."""
sock_path = tmp_path / "greetd.sock"
mock = MockGreetd(sock_path)
# First create_session → error (stale session)
mock.expect({"type": "error", "error_type": "error", "description": "a session is already being configured"})
# cancel_session → success
mock.expect({"type": "success"})
# Second create_session → auth_message (retry succeeds)
mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
# post_auth_response → success
mock.expect({"type": "success"})
# start_session → success
mock.expect({"type": "success"})
mock.start()
try:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(str(sock_path))
# Step 1: Create session fails
response = create_session(sock, "dominik")
assert response["type"] == "error"
# Step 2: Cancel stale session
response = cancel_session(sock)
assert response["type"] == "success"
# Step 3: Retry create session
response = create_session(sock, "dominik")
assert response["type"] == "auth_message"
# Step 4: Send password
response = post_auth_response(sock, "geheim")
assert response["type"] == "success"
# Step 5: Start session
response = start_session(sock, ["niri-session"])
assert response["type"] == "success"
sock.close()
finally:
mock.stop()
assert mock.received[0] == {"type": "create_session", "username": "dominik"}
assert mock.received[1] == {"type": "cancel_session"}
assert mock.received[2] == {"type": "create_session", "username": "dominik"}
def test_multi_stage_auth_sends_cancel(self, tmp_path: Path) -> None: def test_multi_stage_auth_sends_cancel(self, tmp_path: Path) -> None:
"""When greetd sends a second auth_message after password, cancel the session.""" """When greetd sends a second auth_message after password, cancel the session."""
sock_path = tmp_path / "greetd.sock" sock_path = tmp_path / "greetd.sock"
@@ -178,26 +245,129 @@ class TestLoginFlow:
assert mock.received[1] == {"type": "cancel_session"} assert mock.received[1] == {"type": "cancel_session"}
class TestSessionCancellation:
"""Tests for cancelling an in-progress greetd session during user switch."""
def test_cancel_closes_socket_and_sets_event(self, tmp_path: Path) -> None:
"""_cancel_pending_session should close the socket and set the cancelled event."""
from moongreet.greeter import GreeterWindow
win = GreeterWindow.__new__(GreeterWindow)
win._greetd_sock_lock = threading.Lock()
win._login_cancelled = threading.Event()
# Create a real socket pair to verify close
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock_path = tmp_path / "test.sock"
server.bind(str(sock_path))
server.listen(1)
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client.connect(str(sock_path))
server.close()
win._greetd_sock = client
win._cancel_pending_session()
assert win._login_cancelled.is_set()
assert win._greetd_sock is None
def test_cancel_is_noop_without_socket(self) -> None:
"""_cancel_pending_session should be safe to call when no socket exists."""
from moongreet.greeter import GreeterWindow
win = GreeterWindow.__new__(GreeterWindow)
win._greetd_sock_lock = threading.Lock()
win._login_cancelled = threading.Event()
win._greetd_sock = None
win._cancel_pending_session()
assert win._login_cancelled.is_set()
assert win._greetd_sock is None
def test_cancel_does_not_block_main_thread(self, tmp_path: Path) -> None:
"""_cancel_pending_session must not do blocking I/O — only close the socket."""
from moongreet.greeter import GreeterWindow
win = GreeterWindow.__new__(GreeterWindow)
win._greetd_sock_lock = threading.Lock()
win._login_cancelled = threading.Event()
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
win._greetd_sock = sock
# Should complete nearly instantly (no IPC calls)
import time
start = time.monotonic()
win._cancel_pending_session()
elapsed = time.monotonic() - start
assert elapsed < 0.1 # No blocking I/O
def test_worker_exits_silently_when_cancelled(self, tmp_path: Path) -> None:
"""_login_worker should exit without showing an error when cancelled mid-flight."""
from unittest.mock import MagicMock, patch
from moongreet.greeter import GreeterWindow
from moongreet.users import User
win = GreeterWindow.__new__(GreeterWindow)
win._greetd_sock_lock = threading.Lock()
win._login_cancelled = threading.Event()
win._greetd_sock = None
win._failed_attempts = {}
win._strings = MagicMock()
# Set cancelled before the worker runs
win._login_cancelled.set()
# Create a socket that will fail (simulating closed socket)
sock_path = tmp_path / "greetd.sock"
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server.bind(str(sock_path))
server.listen(1)
user = User(username="dom", uid=1000, gecos="Dominik", home=Path("/home/dom"), shell="/bin/zsh")
with patch("moongreet.greeter.GLib.idle_add") as mock_idle:
win._login_worker(user, "pw", MagicMock(exec_cmd="niri-session"), str(sock_path))
# Should NOT have scheduled any error callback
for call in mock_idle.call_args_list:
func = call[0][0]
assert func != win._on_login_error, "Worker should not show error when cancelled"
assert func != win._on_login_auth_error, "Worker should not show auth error when cancelled"
server.close()
class TestFaillockWarning: class TestFaillockWarning:
"""Tests for the faillock warning message logic.""" """Tests for the faillock warning message logic."""
def test_no_warning_on_zero_attempts(self) -> None:
strings = load_strings("de")
assert faillock_warning(0, strings) is None
def test_no_warning_on_first_attempt(self) -> None: def test_no_warning_on_first_attempt(self) -> None:
assert faillock_warning(1) is None strings = load_strings("de")
assert faillock_warning(1, strings) is None
def test_warning_on_second_attempt(self) -> None: def test_warning_on_second_attempt(self) -> None:
warning = faillock_warning(2) strings = load_strings("de")
warning = faillock_warning(2, strings)
assert warning is not None assert warning is not None
assert "1" in warning # 1 Versuch übrig assert "1" in warning # 1 Versuch übrig
def test_warning_on_third_attempt(self) -> None: def test_warning_on_third_attempt(self) -> None:
warning = faillock_warning(3) strings = load_strings("de")
warning = faillock_warning(3, strings)
assert warning is not None assert warning is not None
assert "gesperrt" in warning.lower() assert warning == strings.faillock_locked
def test_warning_beyond_max_attempts(self) -> None: def test_warning_beyond_max_attempts(self) -> None:
warning = faillock_warning(4) strings = load_strings("de")
warning = faillock_warning(4, strings)
assert warning is not None assert warning is not None
assert "gesperrt" in warning.lower() assert warning == strings.faillock_locked
def test_max_attempts_constant_is_three(self) -> None: def test_max_attempts_constant_is_three(self) -> None:
assert FAILLOCK_MAX_ATTEMPTS == 3 assert FAILLOCK_MAX_ATTEMPTS == 3
@@ -226,3 +396,83 @@ class TestLastUser:
from moongreet.greeter import GreeterWindow from moongreet.greeter import GreeterWindow
result = GreeterWindow._load_last_user() result = GreeterWindow._load_last_user()
assert result is None assert result is None
def test_load_last_user_rejects_oversized_username(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
cache_path = tmp_path / "last-user"
cache_path.write_text("a" * 300)
monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)
from moongreet.greeter import GreeterWindow
result = GreeterWindow._load_last_user()
assert result is None
def test_load_last_user_rejects_invalid_characters(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
cache_path = tmp_path / "last-user"
cache_path.write_text("../../etc/passwd")
monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)
from moongreet.greeter import GreeterWindow
result = GreeterWindow._load_last_user()
assert result is None
class TestLastSession:
"""Tests for saving and loading the last session per user."""
def test_save_and_load_last_session(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)
from moongreet.greeter import GreeterWindow
GreeterWindow._save_last_session("dominik", "Niri")
session_file = tmp_path / "dominik"
assert session_file.exists()
assert session_file.read_text() == "Niri"
result = GreeterWindow._load_last_session("dominik")
assert result == "Niri"
def test_load_last_session_missing_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)
from moongreet.greeter import GreeterWindow
result = GreeterWindow._load_last_session("nobody")
assert result is None
def test_load_last_session_rejects_oversized_name(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)
(tmp_path / "dominik").write_text("A" * 300)
from moongreet.greeter import GreeterWindow
result = GreeterWindow._load_last_session("dominik")
assert result is None
def test_save_last_session_validates_username(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Usernames with path traversal should not create files outside the cache dir."""
monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)
from moongreet.greeter import GreeterWindow
GreeterWindow._save_last_session("../../etc/evil", "Niri")
# Should not have created any file
assert not (tmp_path / "../../etc/evil").exists()
def test_regex_rejects_dot_dot_username(self) -> None:
"""Username '..' must not pass VALID_USERNAME validation."""
from moongreet.greeter import VALID_USERNAME
assert VALID_USERNAME.match("..") is None
def test_regex_rejects_dot_username(self) -> None:
"""Username '.' must not pass VALID_USERNAME validation."""
from moongreet.greeter import VALID_USERNAME
assert VALID_USERNAME.match(".") is None
def test_regex_allows_dot_in_middle(self) -> None:
"""Usernames like 'first.last' must still be valid."""
from moongreet.greeter import VALID_USERNAME
assert VALID_USERNAME.match("first.last") is not None
def test_regex_rejects_leading_dot(self) -> None:
"""Usernames starting with '.' are rejected (hidden files)."""
from moongreet.greeter import VALID_USERNAME
assert VALID_USERNAME.match(".hidden") is None
+20 -12
View File
@@ -38,7 +38,7 @@ class FakeSocket:
def with_response(cls, response: dict) -> "FakeSocket": def with_response(cls, response: dict) -> "FakeSocket":
"""Create a FakeSocket pre-loaded with a length-prefixed JSON response.""" """Create a FakeSocket pre-loaded with a length-prefixed JSON response."""
payload = json.dumps(response).encode("utf-8") payload = json.dumps(response).encode("utf-8")
data = struct.pack("!I", len(payload)) + payload data = struct.pack("=I", len(payload)) + payload
return cls(recv_data=data) return cls(recv_data=data)
@@ -73,7 +73,7 @@ class TestSendMessage:
send_message(sock, msg) send_message(sock, msg)
payload = json.dumps(msg).encode("utf-8") payload = json.dumps(msg).encode("utf-8")
expected = struct.pack("!I", len(payload)) + payload expected = struct.pack("=I", len(payload)) + payload
assert bytes(sock.sent) == expected assert bytes(sock.sent) == expected
def test_sends_empty_dict(self) -> None: def test_sends_empty_dict(self) -> None:
@@ -82,7 +82,7 @@ class TestSendMessage:
send_message(sock, {}) send_message(sock, {})
payload = json.dumps({}).encode("utf-8") payload = json.dumps({}).encode("utf-8")
expected = struct.pack("!I", len(payload)) + payload expected = struct.pack("=I", len(payload)) + payload
assert bytes(sock.sent) == expected assert bytes(sock.sent) == expected
def test_sends_nested_message(self) -> None: def test_sends_nested_message(self) -> None:
@@ -93,12 +93,20 @@ class TestSendMessage:
# Verify the payload is correctly length-prefixed # Verify the payload is correctly length-prefixed
length_bytes = bytes(sock.sent[:4]) length_bytes = bytes(sock.sent[:4])
length = struct.unpack("!I", length_bytes)[0] length = struct.unpack("=I", length_bytes)[0]
decoded = json.loads(sock.sent[4:]) decoded = json.loads(sock.sent[4:])
assert length == len(json.dumps(msg).encode("utf-8")) assert length == len(json.dumps(msg).encode("utf-8"))
assert decoded == msg assert decoded == msg
def test_rejects_oversized_payload(self) -> None:
sock = FakeSocket()
msg = {"type": "huge", "data": "x" * 100000}
with pytest.raises(ValueError, match="Payload too large"):
send_message(sock, msg)
class TestRecvMessage: class TestRecvMessage:
"""Tests for receiving and decoding length-prefixed JSON messages.""" """Tests for receiving and decoding length-prefixed JSON messages."""
@@ -132,7 +140,7 @@ class TestRecvMessage:
"""recv() may return fewer bytes than requested — must loop.""" """recv() may return fewer bytes than requested — must loop."""
response = {"type": "success"} response = {"type": "success"}
payload = json.dumps(response).encode("utf-8") payload = json.dumps(response).encode("utf-8")
data = struct.pack("!I", len(payload)) + payload data = struct.pack("=I", len(payload)) + payload
sock = FragmentingSocket(data, chunk_size=3) sock = FragmentingSocket(data, chunk_size=3)
result = recv_message(sock) result = recv_message(sock)
@@ -141,7 +149,7 @@ class TestRecvMessage:
def test_rejects_oversized_payload(self) -> None: def test_rejects_oversized_payload(self) -> None:
"""Payloads exceeding the size limit must be rejected.""" """Payloads exceeding the size limit must be rejected."""
header = struct.pack("!I", 10_000_000) header = struct.pack("=I", 10_000_000)
sock = FakeSocket(recv_data=header) sock = FakeSocket(recv_data=header)
with pytest.raises(ConnectionError, match="too large"): with pytest.raises(ConnectionError, match="too large"):
@@ -162,7 +170,7 @@ class TestCreateSession:
result = create_session(sock, "dominik") result = create_session(sock, "dominik")
# Verify sent message # Verify sent message
length = struct.unpack("!I", bytes(sock.sent[:4]))[0] length = struct.unpack("=I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length]) sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {"type": "create_session", "username": "dominik"} assert sent_msg == {"type": "create_session", "username": "dominik"}
assert result == response assert result == response
@@ -177,7 +185,7 @@ class TestPostAuthResponse:
result = post_auth_response(sock, "mypassword") result = post_auth_response(sock, "mypassword")
length = struct.unpack("!I", bytes(sock.sent[:4]))[0] length = struct.unpack("=I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length]) sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == { assert sent_msg == {
"type": "post_auth_message_response", "type": "post_auth_message_response",
@@ -192,7 +200,7 @@ class TestPostAuthResponse:
result = post_auth_response(sock, None) result = post_auth_response(sock, None)
length = struct.unpack("!I", bytes(sock.sent[:4]))[0] length = struct.unpack("=I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length]) sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == { assert sent_msg == {
"type": "post_auth_message_response", "type": "post_auth_message_response",
@@ -209,7 +217,7 @@ class TestStartSession:
result = start_session(sock, ["Hyprland"]) result = start_session(sock, ["Hyprland"])
length = struct.unpack("!I", bytes(sock.sent[:4]))[0] length = struct.unpack("=I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length]) sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {"type": "start_session", "cmd": ["Hyprland"]} assert sent_msg == {"type": "start_session", "cmd": ["Hyprland"]}
assert result == response assert result == response
@@ -220,7 +228,7 @@ class TestStartSession:
result = start_session(sock, ["sway", "--config", "/etc/sway/config"]) result = start_session(sock, ["sway", "--config", "/etc/sway/config"])
length = struct.unpack("!I", bytes(sock.sent[:4]))[0] length = struct.unpack("=I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length]) sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == { assert sent_msg == {
"type": "start_session", "type": "start_session",
@@ -237,7 +245,7 @@ class TestCancelSession:
result = cancel_session(sock) result = cancel_session(sock)
length = struct.unpack("!I", bytes(sock.sent[:4]))[0] length = struct.unpack("=I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length]) sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {"type": "cancel_session"} assert sent_msg == {"type": "cancel_session"}
assert result == response assert result == response
+17 -3
View File
@@ -6,7 +6,7 @@ from unittest.mock import patch, call
import pytest import pytest
from moongreet.power import reboot, shutdown from moongreet.power import reboot, shutdown, POWER_TIMEOUT
class TestReboot: class TestReboot:
@@ -17,7 +17,7 @@ class TestReboot:
reboot() reboot()
mock_run.assert_called_once_with( mock_run.assert_called_once_with(
["loginctl", "reboot"], check=True ["loginctl", "reboot"], check=True, timeout=POWER_TIMEOUT
) )
@patch("moongreet.power.subprocess.run") @patch("moongreet.power.subprocess.run")
@@ -27,6 +27,13 @@ class TestReboot:
with pytest.raises(subprocess.CalledProcessError): with pytest.raises(subprocess.CalledProcessError):
reboot() reboot()
@patch("moongreet.power.subprocess.run")
def test_raises_on_timeout(self, mock_run) -> None:
mock_run.side_effect = subprocess.TimeoutExpired("loginctl", POWER_TIMEOUT)
with pytest.raises(subprocess.TimeoutExpired):
reboot()
class TestShutdown: class TestShutdown:
"""Tests for the shutdown power action.""" """Tests for the shutdown power action."""
@@ -36,7 +43,7 @@ class TestShutdown:
shutdown() shutdown()
mock_run.assert_called_once_with( mock_run.assert_called_once_with(
["loginctl", "poweroff"], check=True ["loginctl", "poweroff"], check=True, timeout=POWER_TIMEOUT
) )
@patch("moongreet.power.subprocess.run") @patch("moongreet.power.subprocess.run")
@@ -45,3 +52,10 @@ class TestShutdown:
with pytest.raises(subprocess.CalledProcessError): with pytest.raises(subprocess.CalledProcessError):
shutdown() shutdown()
@patch("moongreet.power.subprocess.run")
def test_raises_on_timeout(self, mock_run) -> None:
mock_run.side_effect = subprocess.TimeoutExpired("loginctl", POWER_TIMEOUT)
with pytest.raises(subprocess.TimeoutExpired):
shutdown()
+6 -4
View File
@@ -186,16 +186,18 @@ class TestGetUserGtkTheme:
assert result is None assert result is None
def test_handles_interpolation_characters(self, tmp_path: Path) -> None: def test_passes_theme_with_special_characters(self, tmp_path: Path) -> None:
"""Theme names with % characters should not trigger interpolation errors.""" """Theme names with special characters are passed through to GTK."""
gtk_dir = tmp_path / ".config" / "gtk-4.0" gtk_dir = tmp_path / ".config" / "gtk-4.0"
gtk_dir.mkdir(parents=True) gtk_dir.mkdir(parents=True)
settings = gtk_dir / "settings.ini" settings = gtk_dir / "settings.ini"
settings.write_text("[Settings]\ngtk-theme-name=My%Theme\n") settings.write_text(
"[Settings]\ngtk-theme-name=catppuccin-mocha-lavender-standard+default\n"
)
result = get_user_gtk_theme(config_dir=gtk_dir) result = get_user_gtk_theme(config_dir=gtk_dir)
assert result == "My%Theme" assert result == "catppuccin-mocha-lavender-standard+default"
def test_ignores_symlinked_accountsservice_icon(self, tmp_path: Path) -> None: def test_ignores_symlinked_accountsservice_icon(self, tmp_path: Path) -> None:
"""AccountsService icon as symlink should be ignored to prevent traversal.""" """AccountsService icon as symlink should be ignored to prevent traversal."""