diff --git a/src/axidev_osk/app.py b/src/axidev_osk/app.py index 690100f..68fe203 100644 --- a/src/axidev_osk/app.py +++ b/src/axidev_osk/app.py @@ -15,7 +15,7 @@ from .runtime.application import ApplicationRuntime from .runtime.registries import ServiceRegistry from .services.keyboard import KeyboardService -from .services.kwin_lock import KWinLockService +from .services.secure_input_panel import SecureInputPanelService from .services.single_instance import ExistingInstanceActivated from .windows.overlay import OverlayBackend, prepare_always_on_top_window_environment @@ -63,7 +63,7 @@ def _input_panel_services( services = ServiceRegistry() services.register("keyboard", KeyboardService(), autostart=not lock_lifecycle) if lock_lifecycle: - services.register("kwin_lock", KWinLockService(parent=app)) + services.register("secure_input_panel", SecureInputPanelService(parent=app)) return services diff --git a/src/axidev_osk/cli/linux_greeter.py b/src/axidev_osk/cli/linux_greeter.py index 2656650..bf68246 100644 --- a/src/axidev_osk/cli/linux_greeter.py +++ b/src/axidev_osk/cli/linux_greeter.py @@ -62,11 +62,17 @@ HEALTHY_RUNTIME_SECONDS = 60.0 POLL_SECONDS = 0.1 -PLASMA_LOCK_SCREEN_PATCH_START = "// BEGIN AXIDEV OSK MANAGED" -PLASMA_LOCK_SCREEN_PATCH_END = "// END AXIDEV OSK MANAGED" +PLASMA_LOCK_SCREEN_V0173_PATCH_START = "// BEGIN AXIDEV OSK MANAGED" +PLASMA_LOCK_SCREEN_V0173_PATCH_END = "// END AXIDEV OSK MANAGED" +PLASMA_LOCK_SCREEN_ROOT_PATCH_START = "// BEGIN AXIDEV OSK ROOT MANAGED" +PLASMA_LOCK_SCREEN_ROOT_PATCH_END = "// END AXIDEV OSK ROOT MANAGED" +PLASMA_LOCK_SCREEN_BUTTON_PATCH_START = "// BEGIN AXIDEV OSK BUTTON MANAGED" +PLASMA_LOCK_SCREEN_BUTTON_PATCH_END = "// END AXIDEV OSK BUTTON MANAGED" +PLASMA_LOCK_SCREEN_IMPORT_PATCH_START = "// BEGIN AXIDEV OSK IMPORT MANAGED" +PLASMA_LOCK_SCREEN_IMPORT_PATCH_END = "// END AXIDEV OSK IMPORT MANAGED" PLASMA_LOCK_SCREEN_MIN_VERSION = (6, 7, 0) PLASMA_LOCK_SCREEN_MAX_VERSION = (7, 0, 0) -PLASMA_LOCK_SCREEN_PATCH = ( +PLASMA_LOCK_SCREEN_V0173_PATCH = ( " // BEGIN AXIDEV OSK MANAGED\n" " Connections {\n" " target: lockScreenRoot\n" @@ -79,6 +85,95 @@ " }\n" " // END AXIDEV OSK MANAGED\n" ) +PLASMA_LOCK_SCREEN_ROOT_PATCH = ( + " // BEGIN AXIDEV OSK ROOT MANAGED\n" + " Connections {\n" + " target: lockScreenRoot\n" + " Component.onCompleted: Qt.callLater(function() {\n" + " lockScreenRoot.uiVisible = true;\n" + " })\n\n" + " function onUiVisibleChanged() {\n" + " if (!lockScreenRoot.uiVisible) {\n" + " lockScreenRoot.uiVisible = true;\n" + " }\n" + " }\n" + " }\n" + " // END AXIDEV OSK ROOT MANAGED\n" +) +PLASMA_LOCK_SCREEN_IMPORT_PATCH = ( + "// BEGIN AXIDEV OSK IMPORT MANAGED\n" + "import org.kde.plasma.workspace.keyboardlayout as Keyboards\n" + "import org.kde.plasma.workspace.dbus as DBus\n" + "// END AXIDEV OSK IMPORT MANAGED\n" +) +PLASMA_LOCK_SCREEN_BUTTON_PATCH = ( + " // BEGIN AXIDEV OSK BUTTON MANAGED\n" + " PlasmaComponents3.ToolButton {\n" + " id: axidevOskButton\n\n" + " property int previousVirtualKeyboardMode: -1\n\n" + " focusPolicy: Qt.TabFocus\n" + " text: \"Axidev OSK\"\n" + " icon.name: \"input-keyboard-virtual-on\"\n" + " visible: inputPanel.status === Loader.Ready\n" + " Layout.fillHeight: true\n\n" + " function restoreVirtualKeyboardMode() {\n" + " if (previousVirtualKeyboardMode >= 0) {\n" + " Keyboards.KWinVirtualKeyboard.mode = previousVirtualKeyboardMode;\n" + " previousVirtualKeyboardMode = -1;\n" + " }\n" + " }\n\n" + " function showPreparedKeyboard() {\n" + " if (inputPanel.keyboardActive || previousVirtualKeyboardMode >= 0) {\n" + " return;\n" + " }\n" + " previousVirtualKeyboardMode = Keyboards.KWinVirtualKeyboard.mode;\n" + " Keyboards.KWinVirtualKeyboard.mode = 2;\n" + " mainBlock.mainPasswordBox.forceActiveFocus();\n" + " inputPanel.showHide();\n" + " }\n\n" + " Connections {\n" + " target: Keyboards.KWinVirtualKeyboard\n\n" + " function onVisibleChanged() {\n" + " if (Keyboards.KWinVirtualKeyboard.visible) {\n" + " axidevOskButton.restoreVirtualKeyboardMode();\n" + " }\n" + " }\n" + " }\n\n" + " function releasePreparedKeyboard() {\n" + " restoreVirtualKeyboardMode();\n" + " DBus.SessionBus.asyncCall({\n" + " service: \"org.axidev.OSK.LockScreen\",\n" + " path: \"/org/axidev/OSK/LockScreen\",\n" + " member: \"release\"\n" + " });\n" + " }\n\n" + " Connections {\n" + " target: authenticator\n\n" + " function onSucceeded() {\n" + " axidevOskButton.releasePreparedKeyboard();\n" + " }\n" + " }\n\n" + " Connections {\n" + " target: mainBlock\n\n" + " function onPasswordResult(password) {\n" + " axidevOskButton.releasePreparedKeyboard();\n" + " }\n" + " }\n\n" + " onClicked: {\n" + " mainBlock.mainPasswordBox.forceActiveFocus();\n" + " DBus.SessionBus.asyncCall({\n" + " service: \"org.axidev.OSK.LockScreen\",\n" + " path: \"/org/axidev/OSK/LockScreen\",\n" + " member: \"prepare\"\n" + " }, function() {\n" + " axidevOskButton.showPreparedKeyboard();\n" + " }, function(error) {\n" + " console.warn(\"Cannot prepare Axidev OSK:\", error.message);\n" + " });\n" + " }\n" + " }\n" + " // END AXIDEV OSK BUTTON MANAGED\n" +) @dataclass(frozen=True) class GreetdConfig: @@ -111,12 +206,6 @@ def write(self, path: Path, contents: str, mode: int = 0o644) -> None: self._remember(path) linux._write_atomic(path, contents, mode) - def symlink(self, path: Path, target: Path) -> None: - self._remember(path) - path.parent.mkdir(parents=True, exist_ok=True) - path.unlink(missing_ok=True) - path.symlink_to(target) - def remove(self, path: Path) -> None: self._remember(path) path.unlink(missing_ok=True) @@ -190,7 +279,7 @@ def run_runtime_command(namespace: argparse.Namespace, argv: list[str]) -> int: def _setup(requested_manager: str | None) -> int: existing = _load_state(required=False) - legacy_plasma = existing is not None and _is_legacy_plasma_state(existing) + legacy_plasma = existing is not None and _is_v0173_plasma_state(existing) if existing is not None and not legacy_plasma: if requested_manager is not None and existing["manager"] != requested_manager: raise linux.LinuxSetupError( @@ -207,7 +296,6 @@ def _setup(requested_manager: str | None) -> int: raise linux.LinuxSetupError("managed greeter state is incomplete; remove it before setup") if legacy_plasma: - assert existing is not None if requested_manager is not None and requested_manager != "plasma-login": raise linux.LinuxSetupError( "greeter integration already manages plasma-login; remove it first" @@ -221,8 +309,8 @@ def _setup(requested_manager: str | None) -> int: launcher = _installed_launcher() account, details = adapter.prepare(launcher) - details["legacy_plasma"] = legacy_plasma - + if legacy_plasma: + details["v0173_plasma"] = True linux._setup_permissions(account) _install_manager(manager, adapter, account, launcher, details) print( @@ -474,7 +562,7 @@ def _install_plasma( launcher: Path, details: dict[str, Any], ) -> dict[str, Any]: - if bool(details.get("legacy_plasma")): + if bool(details.get("v0173_plasma")): _require_removable_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) _require_removable_file(PLASMA_SERVICE_PATH, _plasma_service_text()) _require_removable_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) @@ -528,20 +616,22 @@ def _check_plasma(launcher: Path, state: dict[str, Any]) -> list[tuple[str, bool "Plasma version >=6.7.0,<7.0.0", _plasma_lock_screen_version_supported(), ) - if _is_legacy_plasma_state(state): - service_ok = linux._read_text(PLASMA_SERVICE_PATH) == _plasma_service_text() - link_ok = ( - PLASMA_WANTS_PATH.is_symlink() - and PLASMA_WANTS_PATH.resolve() == PLASMA_SERVICE_PATH.resolve() - ) + if _is_v0173_plasma_state(state): return [ version_check, ( str(NATIVE_SUPERVISOR_PATH), linux._read_text(NATIVE_SUPERVISOR_PATH) == _native_supervisor_text(launcher), ), - (str(PLASMA_SERVICE_PATH), service_ok), - (str(PLASMA_WANTS_PATH), link_ok), + ( + str(PLASMA_SERVICE_PATH), + linux._read_text(PLASMA_SERVICE_PATH) == _plasma_service_text(), + ), + ( + str(PLASMA_WANTS_PATH), + PLASMA_WANTS_PATH.is_symlink() + and PLASMA_WANTS_PATH.resolve() == PLASMA_SERVICE_PATH.resolve(), + ), ] original_kwinrc = _state_text(state, "original_kwinrc") return [ @@ -592,13 +682,25 @@ def _check_greetd(launcher: Path, state: dict[str, Any]) -> list[tuple[str, bool def _remove_plasma(launcher: Path, state: dict[str, Any]) -> None: - if _is_legacy_plasma_state(state): + if _is_v0173_plasma_state(state): _require_removable_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) _require_removable_file(PLASMA_SERVICE_PATH, _plasma_service_text()) _require_removable_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) + lock_screen_ui = linux._read_text(PLASMA_LOCK_SCREEN_UI_PATH) + unmanaged_lock_screen_ui = ( + _plasma_lock_screen_ui_without_patch(lock_screen_ui) + if lock_screen_ui is not None + else None + ) _remove_owned_symlink(PLASMA_WANTS_PATH, PLASMA_SERVICE_PATH) linux._remove_owned_file(PLASMA_SERVICE_PATH, _plasma_service_text()) linux._remove_owned_file(NATIVE_SUPERVISOR_PATH, _native_supervisor_text(launcher)) + if lock_screen_ui is not None and unmanaged_lock_screen_ui != lock_screen_ui: + linux._write_atomic( + PLASMA_LOCK_SCREEN_UI_PATH, + unmanaged_lock_screen_ui, + PLASMA_LOCK_SCREEN_UI_PATH.stat().st_mode & 0o777, + ) return original_kwinrc = _state_text(state, "original_kwinrc") managed_kwinrc = _plasma_kwin_config_text(original_kwinrc or None) @@ -816,16 +918,61 @@ def _plasma_kwin_config_text(original: str | None) -> str: def _plasma_lock_screen_patch_is_current(text: str | None) -> bool: - """Return whether QML contains exactly one unmodified managed block.""" + """Return whether QML contains both unmodified managed blocks.""" return bool( text is not None - and text.count(PLASMA_LOCK_SCREEN_PATCH) == 1 - and text.count(PLASMA_LOCK_SCREEN_PATCH_START) == 1 - and text.count(PLASMA_LOCK_SCREEN_PATCH_END) == 1 + and text.count(PLASMA_LOCK_SCREEN_IMPORT_PATCH) == 1 + and text.count(PLASMA_LOCK_SCREEN_ROOT_PATCH) == 1 + and text.count(PLASMA_LOCK_SCREEN_BUTTON_PATCH) == 1 + and text.count(PLASMA_LOCK_SCREEN_IMPORT_PATCH_START) == 1 + and text.count(PLASMA_LOCK_SCREEN_IMPORT_PATCH_END) == 1 + and text.count(PLASMA_LOCK_SCREEN_ROOT_PATCH_START) == 1 + and text.count(PLASMA_LOCK_SCREEN_ROOT_PATCH_END) == 1 + and text.count(PLASMA_LOCK_SCREEN_BUTTON_PATCH_START) == 1 + and text.count(PLASMA_LOCK_SCREEN_BUTTON_PATCH_END) == 1 ) +def _managed_block_span(text: str, start: str, end: str) -> tuple[int, int] | None: + """Locate one complete line-delimited managed block.""" + + if start not in text and end not in text: + return None + if text.count(start) != 1 or text.count(end) != 1: + raise linux.LinuxSetupError("invalid Axidev lock marker pair") + start_index = text.rfind("\n", 0, text.index(start)) + 1 + end_marker = text.index(end) + if end_marker < start_index: + raise linux.LinuxSetupError("invalid Axidev lock marker order") + end_index = text.find("\n", end_marker) + return start_index, len(text) if end_index < 0 else end_index + 1 + + +def _replace_managed_block( + text: str, + start: str, + end: str, + replacement: str, +) -> tuple[str, bool]: + span = _managed_block_span(text, start, end) + if span is None: + return text, False + return text[: span[0]] + replacement + text[span[1] :], True + + +def _remove_managed_block(text: str, start: str, end: str) -> tuple[str, bool]: + span = _managed_block_span(text, start, end) + if span is None: + return text, False + block_start, block_end = span + if block_start >= 2 and text[block_start - 2 : block_start] == "\n\n": + block_start -= 1 + elif text[block_end : block_end + 1] == "\n": + block_end += 1 + return text[:block_start] + text[block_end:], True + + def _plasma_version() -> tuple[int, int, int] | None: """Return the version of the package that owns Plasma's lock-screen QML.""" @@ -895,26 +1042,93 @@ def _require_supported_plasma_lock_screen_version() -> None: def _plasma_lock_screen_ui_text(original: str) -> str: """Add the managed always-visible unlock UI block to Plasma QML.""" - if _plasma_lock_screen_patch_is_current(original): - return original - if PLASMA_LOCK_SCREEN_PATCH_START in original or PLASMA_LOCK_SCREEN_PATCH_END in original: - raise linux.LinuxSetupError("refusing to replace a changed Axidev lock-screen QML block") - anchor = " MouseArea {\n id: lockScreenRoot\n" - if original.count(anchor) != 1: + if PLASMA_LOCK_SCREEN_V0173_PATCH in original: + original = original.replace("\n" + PLASMA_LOCK_SCREEN_V0173_PATCH, "", 1) + elif ( + PLASMA_LOCK_SCREEN_V0173_PATCH_START in original + or PLASMA_LOCK_SCREEN_V0173_PATCH_END in original + ): + raise linux.LinuxSetupError("refusing to replace a changed Axidev 0.17.3 QML block") + + managed, has_import = _replace_managed_block( + original, + PLASMA_LOCK_SCREEN_IMPORT_PATCH_START, + PLASMA_LOCK_SCREEN_IMPORT_PATCH_END, + PLASMA_LOCK_SCREEN_IMPORT_PATCH, + ) + managed, has_root = _replace_managed_block( + managed, + PLASMA_LOCK_SCREEN_ROOT_PATCH_START, + PLASMA_LOCK_SCREEN_ROOT_PATCH_END, + PLASMA_LOCK_SCREEN_ROOT_PATCH, + ) + managed, has_button = _replace_managed_block( + managed, + PLASMA_LOCK_SCREEN_BUTTON_PATCH_START, + PLASMA_LOCK_SCREEN_BUTTON_PATCH_END, + PLASMA_LOCK_SCREEN_BUTTON_PATCH, + ) + root_anchor = " MouseArea {\n id: lockScreenRoot\n" + button_anchor = " PlasmaComponents3.ToolButton {\n id: virtualKeyboardButton\n" + if not has_root and managed.count(root_anchor) != 1: raise linux.LinuxSetupError( "Plasma lock-screen QML does not contain the supported lockScreenRoot structure" ) - return original.replace(anchor, anchor + "\n" + PLASMA_LOCK_SCREEN_PATCH, 1) + if not has_button and managed.count(button_anchor) != 1: + raise linux.LinuxSetupError( + "Plasma lock-screen QML does not contain the supported virtualKeyboardButton structure" + ) + if not has_root: + managed = managed.replace( + root_anchor, + root_anchor + "\n" + PLASMA_LOCK_SCREEN_ROOT_PATCH, + 1, + ) + if not has_button: + managed = managed.replace( + button_anchor, + PLASMA_LOCK_SCREEN_BUTTON_PATCH + "\n" + button_anchor, + 1, + ) + return managed if has_import else _plasma_lock_screen_ui_with_import(managed) + + +def _plasma_lock_screen_ui_with_import(text: str) -> str: + root_items = tuple(re.finditer(r"(?m)^Item \{\n", text)) + if len(root_items) != 1: + raise linux.LinuxSetupError( + "Plasma lock-screen QML does not contain the supported root Item structure" + ) + root_start = root_items[0].start() + return text[:root_start] + PLASMA_LOCK_SCREEN_IMPORT_PATCH + "\n" + text[root_start:] def _plasma_lock_screen_ui_without_patch(managed: str) -> str: - """Remove only the exact managed block from Plasma QML.""" + """Remove the line-delimited managed blocks from Plasma QML.""" - if _plasma_lock_screen_patch_is_current(managed): - return managed.replace("\n" + PLASMA_LOCK_SCREEN_PATCH, "", 1) - if PLASMA_LOCK_SCREEN_PATCH_START in managed or PLASMA_LOCK_SCREEN_PATCH_END in managed: - raise linux.LinuxSetupError("refusing to remove a changed Axidev lock-screen QML block") - return managed + unmanaged, _ = _remove_managed_block( + managed, + PLASMA_LOCK_SCREEN_IMPORT_PATCH_START, + PLASMA_LOCK_SCREEN_IMPORT_PATCH_END, + ) + unmanaged, _ = _remove_managed_block( + unmanaged, + PLASMA_LOCK_SCREEN_ROOT_PATCH_START, + PLASMA_LOCK_SCREEN_ROOT_PATCH_END, + ) + unmanaged, _ = _remove_managed_block( + unmanaged, + PLASMA_LOCK_SCREEN_BUTTON_PATCH_START, + PLASMA_LOCK_SCREEN_BUTTON_PATCH_END, + ) + if PLASMA_LOCK_SCREEN_V0173_PATCH in unmanaged: + return unmanaged.replace("\n" + PLASMA_LOCK_SCREEN_V0173_PATCH, "", 1) + if ( + PLASMA_LOCK_SCREEN_V0173_PATCH_START in unmanaged + or PLASMA_LOCK_SCREEN_V0173_PATCH_END in unmanaged + ): + raise linux.LinuxSetupError("refusing to remove a changed Axidev 0.17.3 QML block") + return unmanaged def _lightdm_config_text() -> str: @@ -982,7 +1196,7 @@ def _native_supervisor_text(launcher: Path) -> str: f' "{launcher}" linux run-greeter-keyboard --manager "${{manager}}"\n' " status=$?\n" ' message=$(printf \'axidev-osk greeter error: manager=%s account=%s protocol=%s ' - "stage=supervisor-exit detail=status=%s retry_seconds=%s\' \"${manager}\" " + "stage=supervisor-exit detail=status=%s retry_seconds=%s' \"${manager}\" " '"${account}" "${protocol}" "${status}" "${delay}")\n' ' printf \'%s\\n\' "${message}" >&2\n' " command -v systemd-cat >/dev/null 2>&1 && " @@ -1171,7 +1385,7 @@ def _state_mode(state: dict[str, Any], key: str) -> int: return value -def _is_legacy_plasma_state(state: dict[str, Any]) -> bool: +def _is_v0173_plasma_state(state: dict[str, Any]) -> bool: return state.get("manager") == "plasma-login" and "original_kwinrc" not in state diff --git a/src/axidev_osk/runtime/application.py b/src/axidev_osk/runtime/application.py index 9e02fd8..f1e26e7 100644 --- a/src/axidev_osk/runtime/application.py +++ b/src/axidev_osk/runtime/application.py @@ -15,7 +15,6 @@ from ..config.models import AppConfig, ChromeConfig, PromptConfig, SurfaceConfig, WindowConfig from ..services import register_services from ..services.keyboard import KeyboardService -from ..services.kwin_lock import KWinLockService from ..styles.theme import apply_theme from ..windows.surface import register_surfaces from .context import Context @@ -26,7 +25,7 @@ route_component_pressed, route_hot_corner_triggered, ) -from .events import ScreenLockStateChanged, WindowCloseRequested +from .events import WindowCloseRequested from .prompt import PromptResolutionWaiter from .registries import ( ComponentRegistry, @@ -72,7 +71,7 @@ def __init__( self._app = app self._show_startup_windows = show_startup_windows - self._screen_locked: bool | None = None + self._secure_input_panel_prepared = False self._config = config or build_default_app_config() self._dispatcher = Dispatcher() self._services = services or ServiceRegistry() @@ -157,38 +156,40 @@ def _handle_window_close_requested(self, event: object) -> None: if isinstance(event, WindowCloseRequested): self._quit_controller.request_quit() - def _handle_screen_lock_state_changed(self, event: object) -> None: - """Create or destroy secure runtime resources as KDE locks and unlocks.""" + def _prepare_secure_input_panel(self) -> None: + """Create the keyboard window and backend requested by the lock-screen button.""" - if not isinstance(event, ScreenLockStateChanged): - return - if event.locked == self._screen_locked: - if event.locked: - self._services.get("kwin_lock", KWinLockService).activate() + if self._secure_input_panel_prepared: return window_id = self._config.keyboard_window_id - if event.locked: + try: + self._keyboard.start(self.context) + window = self._window_manager.show(window_id) + window.set_close_enabled(False) + except Exception: try: - self._keyboard.start(self.context) - window = self._window_manager.show(window_id) - window.set_close_enabled(False) - self._services.get("kwin_lock", KWinLockService).activate() + self._window_manager.destroy(window_id) except Exception: - try: - self._window_manager.destroy(window_id) - except Exception: - _logger.exception("Failed to destroy a partially started lock window") - try: - self._keyboard.shutdown() - except Exception: - _logger.exception("Failed to shut down keyboard output after lock startup failed") - raise - else: + _logger.exception("Failed to destroy a partially prepared secure input panel") + try: + self._keyboard.shutdown() + except Exception: + _logger.exception("Failed to shut down keyboard output after panel preparation failed") + raise + self._secure_input_panel_prepared = True + + def _release_secure_input_panel(self) -> None: + """Destroy runtime resources released by the lock-screen QML.""" + + if not self._secure_input_panel_prepared: + return + try: + self._window_manager.destroy(self._config.keyboard_window_id) + finally: try: - self._window_manager.destroy(window_id) - finally: self._keyboard.shutdown() - self._screen_locked = event.locked + finally: + self._secure_input_panel_prepared = False def _handle_hot_corner_triggered(self, event: object) -> None: """Map hot-corner events to managed window visibility commands.""" diff --git a/src/axidev_osk/runtime/commands.py b/src/axidev_osk/runtime/commands.py index 584fa69..749b26f 100644 --- a/src/axidev_osk/runtime/commands.py +++ b/src/axidev_osk/runtime/commands.py @@ -81,6 +81,16 @@ class StateSet: value: object +@dataclass(frozen=True, slots=True) +class SecureInputPanelPrepare: + """Command requesting creation of secure input-panel runtime resources.""" + + +@dataclass(frozen=True, slots=True) +class SecureInputPanelRelease: + """Command requesting cleanup of secure input-panel runtime resources.""" + + @dataclass(frozen=True, slots=True) class WindowShow: """Command requesting a managed window to be shown. @@ -134,4 +144,4 @@ class AppQuit: exit_code: int = 0 -RuntimeCommand = KeyboardRegisterKeySpec | KeyboardKeyDown | KeyboardKeyUp | KeyboardSyncLatchedKey | StateSet | WindowShow | WindowHide | WindowToggleOpacity | WindowClose | AppQuit +RuntimeCommand = KeyboardRegisterKeySpec | KeyboardKeyDown | KeyboardKeyUp | KeyboardSyncLatchedKey | StateSet | SecureInputPanelPrepare | SecureInputPanelRelease | WindowShow | WindowHide | WindowToggleOpacity | WindowClose | AppQuit diff --git a/src/axidev_osk/runtime/event_handlers.py b/src/axidev_osk/runtime/event_handlers.py index 3d6cbb7..005a090 100644 --- a/src/axidev_osk/runtime/event_handlers.py +++ b/src/axidev_osk/runtime/event_handlers.py @@ -10,6 +10,8 @@ KeyboardRegisterKeySpec, KeyboardKeyUp, KeyboardSyncLatchedKey, + SecureInputPanelPrepare, + SecureInputPanelRelease, StateSet, WindowClose, WindowHide, @@ -69,6 +71,14 @@ def register_context_command_handlers(registry: EventHandlerRegistry) -> None: def register_event_handlers(registry: EventHandlerRegistry) -> None: """Register application-level runtime handlers in deterministic order.""" + registry.register_command_handler( + SecureInputPanelPrepare, + lambda runtime: lambda command: runtime._prepare_secure_input_panel(), + ) + registry.register_command_handler( + SecureInputPanelRelease, + lambda runtime: lambda command: runtime._release_secure_input_panel(), + ) registry.register_command_handler( WindowShow, lambda runtime: lambda command: runtime._window_manager.show(command.window_id), @@ -94,7 +104,6 @@ def register_event_handlers(registry: EventHandlerRegistry) -> None: lambda runtime: lambda command: runtime._app.exit(command.exit_code), ) registry.register_event_handler(lambda runtime: runtime._handle_window_close_requested) - registry.register_event_handler(lambda runtime: runtime._handle_screen_lock_state_changed) registry.register_event_handler(lambda runtime: runtime._handle_hot_corner_triggered) registry.register_event_handler(lambda runtime: runtime._handle_component_pressed) diff --git a/src/axidev_osk/runtime/events.py b/src/axidev_osk/runtime/events.py index cc770e4..42f3ebe 100644 --- a/src/axidev_osk/runtime/events.py +++ b/src/axidev_osk/runtime/events.py @@ -104,13 +104,6 @@ class HotCornerTriggered: corner: str -@dataclass(frozen=True, slots=True) -class ScreenLockStateChanged: - """The desktop session entered or left its locked state.""" - - locked: bool - - @dataclass(frozen=True, slots=True) class WindowCloseRequested: """A managed window requested application shutdown confirmation. @@ -143,7 +136,6 @@ class PromptResolved: | BackendKeyStateChanged | KeyLatchChanged | HotCornerTriggered - | ScreenLockStateChanged | WindowCloseRequested | PromptResolved ) diff --git a/src/axidev_osk/runtime/testing.py b/src/axidev_osk/runtime/testing.py index 1634a6d..e063f60 100644 --- a/src/axidev_osk/runtime/testing.py +++ b/src/axidev_osk/runtime/testing.py @@ -70,11 +70,6 @@ def _handle_window_close_requested(self, event: object) -> None: if isinstance(event, WindowCloseRequested): self._dispatcher.dispatch_command(AppQuit()) - def _handle_screen_lock_state_changed(self, event: object) -> None: - """Ignore platform lock-state events in the generic test runtime.""" - - del event - def _handle_hot_corner_triggered(self, event: object) -> None: """Route hot-corner visibility commands through production helper.""" diff --git a/src/axidev_osk/services/kwin_lock.py b/src/axidev_osk/services/kwin_lock.py deleted file mode 100644 index 1a19b83..0000000 --- a/src/axidev_osk/services/kwin_lock.py +++ /dev/null @@ -1,110 +0,0 @@ -"""KWin screen-lock state integration for the secure input panel.""" - -from __future__ import annotations - -import logging -from typing import TYPE_CHECKING - -from PySide6.QtCore import QObject, SLOT, Slot -from PySide6.QtDBus import QDBusConnection, QDBusInterface, QDBusMessage - -from ..runtime.events import ScreenLockStateChanged - -if TYPE_CHECKING: - from ..runtime.context import Context - -_logger = logging.getLogger(__name__) - - -class KWinLockService(QObject): - """Observe KDE's lock state and expose KWin input-method activation.""" - - def __init__(self, parent: QObject | None = None) -> None: - super().__init__(parent) - self._context: Context | None = None - self._connection = QDBusConnection.sessionBus() - self._virtual_keyboard: QDBusInterface | None = None - - def start(self, context: Context) -> None: - """Connect lock-state signals and publish the current state.""" - - self._context = context - if not self._connection.isConnected(): - raise RuntimeError("KDE session bus is unavailable") - self._virtual_keyboard = QDBusInterface( - "org.kde.KWin", - "/VirtualKeyboard", - "org.kde.kwin.VirtualKeyboard", - self._connection, - ) - connected_about = self._connection.connect( - "org.kde.screensaver", - "/ScreenSaver", - "org.kde.screensaver", - "AboutToLock", - self, - SLOT("aboutToLock()"), - ) - connected_active = self._connection.connect( - "org.freedesktop.ScreenSaver", - "/ScreenSaver", - "org.freedesktop.ScreenSaver", - "ActiveChanged", - self, - SLOT("activeChanged(bool)"), - ) - if not connected_about or not connected_active: - raise RuntimeError("Cannot monitor KDE screen-lock state") - - screen_saver = QDBusInterface( - "org.freedesktop.ScreenSaver", - "/ScreenSaver", - "org.freedesktop.ScreenSaver", - self._connection, - ) - reply = screen_saver.call("GetActive") - if reply.type() == QDBusMessage.MessageType.ReplyMessage and reply.arguments(): - self._emit_state(bool(reply.arguments()[0])) - else: - _logger.warning("KDE screen-lock state is unavailable; keeping the secure panel hidden") - self._emit_state(False) - - def stop(self) -> None: - """Disconnect lock-state signals.""" - - self._connection.disconnect( - "org.kde.screensaver", - "/ScreenSaver", - "org.kde.screensaver", - "AboutToLock", - self, - SLOT("aboutToLock()"), - ) - self._connection.disconnect( - "org.freedesktop.ScreenSaver", - "/ScreenSaver", - "org.freedesktop.ScreenSaver", - "ActiveChanged", - self, - SLOT("activeChanged(bool)"), - ) - self._context = None - self._virtual_keyboard = None - - def activate(self) -> None: - """Ask KWin to activate its configured virtual keyboard.""" - - if self._virtual_keyboard is not None: - self._virtual_keyboard.call("forceActivate") - - @Slot() - def aboutToLock(self) -> None: - self._emit_state(True) - - @Slot(bool) - def activeChanged(self, active: bool) -> None: - self._emit_state(active) - - def _emit_state(self, locked: bool) -> None: - if self._context is not None: - self._context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=locked)) diff --git a/src/axidev_osk/services/secure_input_panel.py b/src/axidev_osk/services/secure_input_panel.py new file mode 100644 index 0000000..9d85ac2 --- /dev/null +++ b/src/axidev_osk/services/secure_input_panel.py @@ -0,0 +1,62 @@ +"""Session D-Bus control surface for the Plasma lock-screen button.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from PySide6.QtCore import QObject, Slot +from PySide6.QtDBus import QDBusConnection + +from ..runtime.commands import SecureInputPanelPrepare, SecureInputPanelRelease + +if TYPE_CHECKING: + from ..runtime.context import Context + +SERVICE_NAME = "org.axidev.OSK.LockScreen" +OBJECT_PATH = "/org/axidev/OSK/LockScreen" + + +class SecureInputPanelService(QObject): + """Translate lock-screen D-Bus requests into runtime commands.""" + + def __init__(self, parent: QObject | None = None) -> None: + super().__init__(parent) + self._connection = QDBusConnection.sessionBus() + self._context: Context | None = None + + def start(self, context: Context) -> None: + """Publish the lock-screen control object on the session bus.""" + + if not self._connection.isConnected(): + raise RuntimeError("KDE session bus is unavailable") + if not self._connection.registerService(SERVICE_NAME): + raise RuntimeError(f"Cannot register D-Bus service {SERVICE_NAME}") + if not self._connection.registerObject( + OBJECT_PATH, + self, + QDBusConnection.RegisterOption.ExportAllSlots, + ): + self._connection.unregisterService(SERVICE_NAME) + raise RuntimeError(f"Cannot register D-Bus object {OBJECT_PATH}") + self._context = context + + def stop(self) -> None: + """Remove the lock-screen control object from the session bus.""" + + self._context = None + self._connection.unregisterObject(OBJECT_PATH) + self._connection.unregisterService(SERVICE_NAME) + + @Slot() + def prepare(self) -> None: + """Request secure input-panel creation through the runtime queue.""" + + if self._context is not None: + self._context.dispatcher.dispatch_command(SecureInputPanelPrepare()) + + @Slot() + def release(self) -> None: + """Request secure input-panel cleanup through the runtime queue.""" + + if self._context is not None: + self._context.dispatcher.dispatch_command(SecureInputPanelRelease()) diff --git a/tests/test_application_runtime.py b/tests/test_application_runtime.py index e81b2b0..c36884a 100644 --- a/tests/test_application_runtime.py +++ b/tests/test_application_runtime.py @@ -11,10 +11,10 @@ from axidev_osk.config.defaults import build_default_app_config from axidev_osk.config.models import WindowConfig from axidev_osk.runtime.application import ApplicationRuntime -from axidev_osk.runtime.events import PromptResolved, ScreenLockStateChanged +from axidev_osk.runtime.commands import SecureInputPanelPrepare, SecureInputPanelRelease +from axidev_osk.runtime.events import PromptResolved from axidev_osk.runtime.registries import ServiceRegistry from axidev_osk.services.keyboard import KeyboardService -from axidev_osk.services.kwin_lock import KWinLockService def _app() -> QApplication: @@ -109,16 +109,13 @@ def create_transient(window_config: WindowConfig, *, parent: QWidget | None = No class SecureInputPanelLifecycleTests(unittest.TestCase): - def test_repeated_lock_cycles_rebuild_window_and_restart_keyboard(self) -> None: + def test_repeated_prepare_release_cycles_rebuild_window_and_restart_keyboard(self) -> None: backend = Mock() backend.initialize.return_value = True backend.add_key_state_listener.return_value = lambda: None keyboard = KeyboardService(backend) - kwin_lock = KWinLockService() - kwin_lock.activate = Mock() services = ServiceRegistry() services.register("keyboard", keyboard, autostart=False) - services.register("kwin_lock", kwin_lock, autostart=False) runtime = ApplicationRuntime(_app(), services=services, show_startup_windows=False) lock_window = Mock() @@ -126,9 +123,10 @@ def test_repeated_lock_cycles_rebuild_window_and_restart_keyboard(self) -> None: patch.object(runtime._window_manager, "show", return_value=lock_window) as show, patch.object(runtime._window_manager, "destroy") as destroy, ): - runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) - runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=False)) - runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + runtime.context.dispatcher.dispatch_command(SecureInputPanelPrepare()) + runtime.context.dispatcher.dispatch_command(SecureInputPanelPrepare()) + runtime.context.dispatcher.dispatch_command(SecureInputPanelRelease()) + runtime.context.dispatcher.dispatch_command(SecureInputPanelPrepare()) self.assertEqual(backend.initialize.call_count, 2) backend.shutdown.assert_called_once_with() @@ -138,18 +136,14 @@ def test_repeated_lock_cycles_rebuild_window_and_restart_keyboard(self) -> None: [unittest.mock.call(False), unittest.mock.call(False)], ) destroy.assert_called_once_with(runtime._config.keyboard_window_id) - self.assertEqual(kwin_lock.activate.call_count, 2) - def test_failed_lock_window_creation_rolls_back_and_remains_retryable(self) -> None: + def test_failed_panel_creation_rolls_back_and_remains_retryable(self) -> None: backend = Mock() backend.initialize.return_value = True backend.add_key_state_listener.return_value = lambda: None keyboard = KeyboardService(backend) - kwin_lock = KWinLockService() - kwin_lock.activate = Mock() services = ServiceRegistry() services.register("keyboard", keyboard, autostart=False) - services.register("kwin_lock", kwin_lock, autostart=False) runtime = ApplicationRuntime(_app(), services=services, show_startup_windows=False) with ( @@ -161,27 +155,24 @@ def test_failed_lock_window_creation_rolls_back_and_remains_retryable(self) -> N patch.object(runtime._window_manager, "destroy") as destroy, ): with self.assertRaisesRegex(RuntimeError, "window failed"): - runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) - self.assertIsNone(runtime._screen_locked) + runtime.context.dispatcher.dispatch_command(SecureInputPanelPrepare()) + self.assertFalse(runtime._secure_input_panel_prepared) - runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + runtime.context.dispatcher.dispatch_command(SecureInputPanelPrepare()) self.assertEqual(show.call_count, 2) destroy.assert_called_once_with(runtime._config.keyboard_window_id) self.assertEqual(backend.initialize.call_count, 2) backend.shutdown.assert_called_once_with() - kwin_lock.activate.assert_called_once_with() - self.assertTrue(runtime._screen_locked) + self.assertTrue(runtime._secure_input_panel_prepared) - def test_failed_lock_startup_preserves_error_when_cleanup_also_fails(self) -> None: + def test_failed_panel_prepare_preserves_error_when_cleanup_also_fails(self) -> None: backend = Mock() backend.initialize.return_value = True backend.add_key_state_listener.return_value = lambda: None keyboard = KeyboardService(backend) - kwin_lock = KWinLockService() services = ServiceRegistry() services.register("keyboard", keyboard, autostart=False) - services.register("kwin_lock", kwin_lock, autostart=False) runtime = ApplicationRuntime(_app(), services=services, show_startup_windows=False) with ( @@ -198,13 +189,13 @@ def test_failed_lock_startup_preserves_error_when_cleanup_also_fails(self) -> No patch("axidev_osk.runtime.application._logger") as logger, self.assertRaisesRegex(RuntimeError, "window failed"), ): - runtime.context.dispatcher.dispatch_event(ScreenLockStateChanged(locked=True)) + runtime.context.dispatcher.dispatch_command(SecureInputPanelPrepare()) backend.shutdown.assert_called_once_with() logger.exception.assert_called_once_with( - "Failed to destroy a partially started lock window" + "Failed to destroy a partially prepared secure input panel" ) - self.assertIsNone(runtime._screen_locked) + self.assertFalse(runtime._secure_input_panel_prepared) if __name__ == "__main__": diff --git a/tests/test_kwin_lock.py b/tests/test_kwin_lock.py deleted file mode 100644 index 5f55327..0000000 --- a/tests/test_kwin_lock.py +++ /dev/null @@ -1,47 +0,0 @@ -from __future__ import annotations - -import unittest -from unittest.mock import Mock, patch - -from PySide6.QtDBus import QDBusMessage - -from axidev_osk.services.kwin_lock import KWinLockService - - -class KWinLockServiceTests(unittest.TestCase): - def test_lock_signals_are_bound_to_screen_locker_services(self) -> None: - connection = Mock() - connection.isConnected.return_value = True - connection.connect.return_value = True - reply = Mock() - reply.type.return_value = QDBusMessage.MessageType.ReplyMessage - reply.arguments.return_value = [False] - screen_saver = Mock() - screen_saver.call.return_value = reply - - with ( - patch( - "axidev_osk.services.kwin_lock.QDBusConnection.sessionBus", - return_value=connection, - ), - patch( - "axidev_osk.services.kwin_lock.QDBusInterface", - side_effect=(Mock(), screen_saver), - ), - ): - service = KWinLockService() - service.start(Mock()) - service.stop() - - self.assertEqual( - [call.args[0] for call in connection.connect.call_args_list], - ["org.kde.screensaver", "org.freedesktop.ScreenSaver"], - ) - self.assertEqual( - [call.args[0] for call in connection.disconnect.call_args_list], - ["org.kde.screensaver", "org.freedesktop.ScreenSaver"], - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_linux_greeter.py b/tests/test_linux_greeter.py index dff46f3..ca95b1a 100644 --- a/tests/test_linux_greeter.py +++ b/tests/test_linux_greeter.py @@ -176,30 +176,101 @@ def test_plasma_lock_screen_patch_is_additive_and_reversible(self) -> None: " id: lockScreenRoot\n\n" " property bool uiVisible: false\n" " }\n" + " Item {\n" + " }\n" + " RowLayout {\n" + " PlasmaComponents3.ToolButton {\n" + " id: virtualKeyboardButton\n" + " }\n" + " }\n" "}\n" ) managed = linux_greeter._plasma_lock_screen_ui_text(original) - self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_PATCH, managed) + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_IMPORT_PATCH, managed) + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_ROOT_PATCH, managed) + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH, managed) + self.assertLess(managed.index("id: axidevOskButton"), managed.index("id: virtualKeyboardButton")) + self.assertEqual(linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH.count("inputPanel.showHide()"), 1) + self.assertIn("DBus.SessionBus.asyncCall", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertIn('member: "prepare"', linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertIn('member: "release"', linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertIn("target: mainBlock", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertIn("function onPasswordResult(password)", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertIn( + "Keyboards.KWinVirtualKeyboard.mode = 2", + linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH, + ) + self.assertIn("onVisibleChanged", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertIn( + "previousVirtualKeyboardMode", + linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH, + ) + self.assertIn( + "inputPanel.keyboardActive || previousVirtualKeyboardMode >= 0", + linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH, + ) + self.assertIn("target: authenticator", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertIn("function onSucceeded()", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertNotIn("target: root", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) + self.assertNotIn("Component.onDestruction", linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH) self.assertEqual(linux_greeter._plasma_lock_screen_ui_text(managed), managed) self.assertEqual(linux_greeter._plasma_lock_screen_ui_without_patch(managed), original) - def test_plasma_lock_screen_patch_rejects_changed_markers(self) -> None: + changed = managed.replace( + "import org.kde.plasma.workspace.dbus as DBus\n", + "changed import\n", + ).replace( + " target: lockScreenRoot\n", + " changed root\n", + ).replace( + " text: \"Axidev OSK\"\n", + " changed button\n", + ) + self.assertEqual(linux_greeter._plasma_lock_screen_ui_text(changed), managed) + self.assertEqual(linux_greeter._plasma_lock_screen_ui_without_patch(changed), original) + + def test_plasma_lock_screen_patch_migrates_v0173_block(self) -> None: + original = ( + "Item {\n" + " MouseArea {\n" + " id: lockScreenRoot\n" + " }\n" + " RowLayout {\n" + " PlasmaComponents3.ToolButton {\n" + " id: virtualKeyboardButton\n" + " }\n" + " }\n" + "}\n" + ) + v0173 = original.replace( + " id: lockScreenRoot\n", + " id: lockScreenRoot\n\n" + linux_greeter.PLASMA_LOCK_SCREEN_V0173_PATCH, + ) + + managed = linux_greeter._plasma_lock_screen_ui_text(v0173) + + self.assertNotIn(linux_greeter.PLASMA_LOCK_SCREEN_V0173_PATCH, managed) + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_IMPORT_PATCH, managed) + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_ROOT_PATCH, managed) + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH, managed) + self.assertEqual(linux_greeter._plasma_lock_screen_ui_without_patch(v0173), original) + + def test_plasma_lock_screen_patch_rejects_incomplete_markers(self) -> None: changed = ( "Item {\n" " MouseArea {\n" " id: lockScreenRoot\n" - " // BEGIN AXIDEV OSK MANAGED\n" + " // BEGIN AXIDEV OSK ROOT MANAGED\n" " changed content\n" - " // END AXIDEV OSK MANAGED\n" " }\n" "}\n" ) - with self.assertRaisesRegex(linux.LinuxSetupError, "changed Axidev"): + with self.assertRaisesRegex(linux.LinuxSetupError, "marker pair"): linux_greeter._plasma_lock_screen_ui_text(changed) - with self.assertRaisesRegex(linux.LinuxSetupError, "changed Axidev"): + with self.assertRaisesRegex(linux.LinuxSetupError, "marker pair"): linux_greeter._plasma_lock_screen_ui_without_patch(changed) def test_plasma_version_is_read_from_owning_rpm(self) -> None: @@ -242,6 +313,11 @@ def test_plasma_install_and_remove_restore_kwin_config(self) -> None: " id: lockScreenRoot\n\n" " property bool uiVisible: false\n" " }\n" + " RowLayout {\n" + " PlasmaComponents3.ToolButton {\n" + " id: virtualKeyboardButton\n" + " }\n" + " }\n" "}\n" ) kwinrc.write_text(original, encoding="utf-8") @@ -273,7 +349,11 @@ def test_plasma_install_and_remove_restore_kwin_config(self) -> None: self.assertTrue(input_method.is_file()) self.assertTrue(kwin_dropin.is_file()) self.assertIn( - linux_greeter.PLASMA_LOCK_SCREEN_PATCH, + linux_greeter.PLASMA_LOCK_SCREEN_ROOT_PATCH, + lock_screen_ui.read_text(encoding="utf-8"), + ) + self.assertIn( + linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH, lock_screen_ui.read_text(encoding="utf-8"), ) @@ -282,7 +362,11 @@ def test_plasma_install_and_remove_restore_kwin_config(self) -> None: with patch.object(linux_greeter, "_runtime_launcher", return_value=launcher): self.assertTrue(linux_greeter._repair_plasma_lock_screen_patch(state)) self.assertIn( - linux_greeter.PLASMA_LOCK_SCREEN_PATCH, + linux_greeter.PLASMA_LOCK_SCREEN_ROOT_PATCH, + lock_screen_ui.read_text(encoding="utf-8"), + ) + self.assertIn( + linux_greeter.PLASMA_LOCK_SCREEN_BUTTON_PATCH, lock_screen_ui.read_text(encoding="utf-8"), ) @@ -298,43 +382,115 @@ def test_plasma_install_and_remove_restore_kwin_config(self) -> None: self.assertFalse(kwin_dropin.exists()) self.assertEqual(lock_screen_ui.read_text(encoding="utf-8"), original_lock_screen_ui) - def test_legacy_plasma_remove_keeps_working(self) -> None: + def test_v0173_plasma_remove_keeps_working(self) -> None: launcher = Path("/opt/axidev-osk/bin/axidev-osk") - legacy_state = {"schema": 1, "manager": "plasma-login", "account": "plasmalogin"} + state = {"schema": 1, "manager": "plasma-login", "account": "plasmalogin"} with TemporaryDirectory() as temporary: root = Path(temporary) supervisor = root / "supervisor" service = root / "service" wants = root / "wants" + lock_screen_ui = root / "LockScreenUi.qml" + original_lock_screen_ui = "Item {\n id: lockScreenRoot\n}\n" with ( patch.object(linux_greeter, "NATIVE_SUPERVISOR_PATH", supervisor), patch.object(linux_greeter, "PLASMA_SERVICE_PATH", service), patch.object(linux_greeter, "PLASMA_WANTS_PATH", wants), + patch.object(linux_greeter, "PLASMA_LOCK_SCREEN_UI_PATH", lock_screen_ui), ): supervisor.write_text( linux_greeter._native_supervisor_text(launcher), encoding="utf-8" ) service.write_text(linux_greeter._plasma_service_text(), encoding="utf-8") wants.symlink_to(service) - linux_greeter._remove_plasma(launcher, legacy_state) + lock_screen_ui.write_text( + original_lock_screen_ui.replace( + " id: lockScreenRoot\n", + " id: lockScreenRoot\n\n" + + linux_greeter.PLASMA_LOCK_SCREEN_V0173_PATCH, + ), + encoding="utf-8", + ) + + linux_greeter._remove_plasma(launcher, state) self.assertFalse(supervisor.exists()) self.assertFalse(service.exists()) self.assertFalse(wants.exists()) + self.assertEqual(lock_screen_ui.read_text(encoding="utf-8"), original_lock_screen_ui) - def test_removable_symlink_accepts_an_equivalent_target_path(self) -> None: + def test_v0173_plasma_install_migrates_owned_files_and_qml(self) -> None: with TemporaryDirectory() as temporary: root = Path(temporary) - actual = root / "actual" - alias = root / "alias" - actual.mkdir() - alias.symlink_to(actual, target_is_directory=True) - target = alias / "service" - target.write_text("service", encoding="utf-8") - link = root / "wants" - link.symlink_to(target) - - linux_greeter._require_removable_symlink(link, target) + supervisor = root / "supervisor" + service = root / "service" + wants = root / "wants" + input_method = root / "axidev-osk-input-panel.desktop" + kwin_dropin = root / "50-axidev-osk.conf" + kwin_unit = root / "plasma-login-kwin_wayland.service" + kwinrc = root / "kwinrc" + lock_screen_ui = root / "LockScreenUi.qml" + state_path = root / "greeter.json" + launcher = Path("/opt/axidev-osk/bin/axidev-osk") + account = linux.Account("plasmalogin", 981, 981, root) + original_lock_screen_ui = ( + "Item {\n" + " MouseArea {\n" + " id: lockScreenRoot\n" + " }\n" + " RowLayout {\n" + " PlasmaComponents3.ToolButton {\n" + " id: virtualKeyboardButton\n" + " }\n" + " }\n" + "}\n" + ) + v0173_lock_screen_ui = original_lock_screen_ui.replace( + " id: lockScreenRoot\n", + " id: lockScreenRoot\n\n" + + linux_greeter.PLASMA_LOCK_SCREEN_V0173_PATCH, + ) + kwin_unit.write_text(self.PLASMA_KWIN_UNIT, encoding="utf-8") + lock_screen_ui.write_text(v0173_lock_screen_ui, encoding="utf-8") + + with ( + patch.object(linux_greeter, "NATIVE_SUPERVISOR_PATH", supervisor), + patch.object(linux_greeter, "PLASMA_SERVICE_PATH", service), + patch.object(linux_greeter, "PLASMA_WANTS_PATH", wants), + patch.object(linux_greeter, "PLASMA_INPUT_METHOD_PATH", input_method), + patch.object(linux_greeter, "PLASMA_KWIN_DROPIN_PATH", kwin_dropin), + patch.object(linux_greeter, "PLASMA_KWIN_UNIT_PATHS", (kwin_unit,)), + patch.object(linux_greeter, "KWIN_CONFIG_PATH", kwinrc), + patch.object(linux_greeter, "PLASMA_LOCK_SCREEN_UI_PATH", lock_screen_ui), + patch.object(linux_greeter, "STATE_PATH", state_path), + patch.object(linux_greeter, "_plasma_version", return_value=(6, 7, 4)), + patch.object(linux, "_resolve_account", return_value=account), + ): + supervisor.write_text( + linux_greeter._native_supervisor_text(launcher), encoding="utf-8" + ) + service.write_text(linux_greeter._plasma_service_text(), encoding="utf-8") + wants.symlink_to(service) + prepared_account, details = linux_greeter._prepare_plasma(launcher) + details["v0173_plasma"] = True + linux_greeter._install_manager( + "plasma-login", + linux_greeter._manager_adapter("plasma-login"), + prepared_account, + launcher, + details, + ) + + managed_lock_screen_ui = lock_screen_ui.read_text(encoding="utf-8") + state = json.loads(state_path.read_text(encoding="utf-8")) + self.assertFalse(supervisor.exists()) + self.assertFalse(service.exists()) + self.assertFalse(wants.exists()) + self.assertTrue(input_method.is_file()) + self.assertTrue(kwin_dropin.is_file()) + self.assertNotIn(linux_greeter.PLASMA_LOCK_SCREEN_V0173_PATCH, managed_lock_screen_ui) + self.assertIn(linux_greeter.PLASMA_LOCK_SCREEN_ROOT_PATCH, managed_lock_screen_ui) + self.assertIn("original_kwinrc", state) def test_lightdm_uses_native_greeter_wrapper(self) -> None: wrapper = linux_greeter._lightdm_wrapper_text(Path("/opt/axidev-osk/bin/axidev-osk")) diff --git a/tests/test_secure_input_panel.py b/tests/test_secure_input_panel.py new file mode 100644 index 0000000..5e0100c --- /dev/null +++ b/tests/test_secure_input_panel.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +import unittest +from unittest.mock import Mock, patch + +from axidev_osk.runtime.commands import SecureInputPanelPrepare, SecureInputPanelRelease +from axidev_osk.services.secure_input_panel import OBJECT_PATH, SERVICE_NAME, SecureInputPanelService + + +class SecureInputPanelServiceTests(unittest.TestCase): + def test_dbus_methods_dispatch_runtime_commands(self) -> None: + connection = Mock() + connection.isConnected.return_value = True + connection.registerService.return_value = True + connection.registerObject.return_value = True + context = Mock() + + with patch( + "axidev_osk.services.secure_input_panel.QDBusConnection.sessionBus", + return_value=connection, + ): + service = SecureInputPanelService() + service.start(context) + service.prepare() + service.release() + service.stop() + + connection.registerService.assert_called_once_with(SERVICE_NAME) + self.assertEqual(connection.registerObject.call_args.args[:2], (OBJECT_PATH, service)) + dispatched = [call.args[0] for call in context.dispatcher.dispatch_command.call_args_list] + self.assertEqual(dispatched, [SecureInputPanelPrepare(), SecureInputPanelRelease()]) + connection.unregisterObject.assert_called_once_with(OBJECT_PATH) + connection.unregisterService.assert_called_once_with(SERVICE_NAME) + + def test_failed_object_registration_releases_service_name(self) -> None: + connection = Mock() + connection.isConnected.return_value = True + connection.registerService.return_value = True + connection.registerObject.return_value = False + + with patch( + "axidev_osk.services.secure_input_panel.QDBusConnection.sessionBus", + return_value=connection, + ): + service = SecureInputPanelService() + with self.assertRaisesRegex(RuntimeError, OBJECT_PATH): + service.start(Mock()) + + connection.unregisterService.assert_called_once_with(SERVICE_NAME) + + +if __name__ == "__main__": + unittest.main()