From 40aeadb68a8bb60656ce89b6ccb7f27bd28be6b2 Mon Sep 17 00:00:00 2001 From: Storm Dragon Date: Fri, 17 Jul 2026 13:38:01 -0400 Subject: [PATCH] Tighter integration with i38lock. Found and squished a bug with sound sometimes not loading. --- src/cthulhu/cthulhu.py | 14 +- src/cthulhu/cthulhu_state.py | 4 + src/cthulhu/event_manager.py | 69 +++++- src/cthulhu/input_event_manager.py | 18 ++ src/cthulhu/meson.build | 1 + src/cthulhu/screen_lock_monitor.py | 155 +++++++++++++ src/cthulhu/sound.py | 36 +++ src/cthulhu/speech.py | 4 +- ...resentation_interrupt_sound_regressions.py | 14 ++ tests/test_screen_lock_monitor.py | 211 ++++++++++++++++++ tests/test_sound_recovery.py | 13 ++ 11 files changed, 536 insertions(+), 3 deletions(-) create mode 100644 src/cthulhu/screen_lock_monitor.py create mode 100644 tests/test_screen_lock_monitor.py diff --git a/src/cthulhu/cthulhu.py b/src/cthulhu/cthulhu.py index 9d06531..ec7d8e8 100644 --- a/src/cthulhu/cthulhu.py +++ b/src/cthulhu/cthulhu.py @@ -257,6 +257,7 @@ from . import keybindings from . import learn_mode_presenter from . import logger from . import mako_notification_monitor +from . import screen_lock_monitor from . import messages from . import notification_presenter from . import focus_manager @@ -724,6 +725,7 @@ def start() -> None: # Start D-Bus remote controller service after ATSPI is ready with debug.startup_timing("startup idle services"): cthulhuApp.getMakoNotificationMonitor().start() + cthulhuApp.getScreenLockMonitor().start() GObject.idle_add(_start_dbus_service) debug.print_startup_timing("AT-SPI main loop start") @@ -772,11 +774,13 @@ def shutdown(script: Optional[Any] = None, inputEvent: Optional[Any] = None) -> signal.signal(signal.SIGALRM, settings.timeoutCallback) signal.alarm(settings.timeoutTime) - cthulhu_state.activeScript.presentationInterrupt() + if cthulhu_state.activeScript is not None: + cthulhu_state.activeScript.presentationInterrupt() cthulhuApp.getSignalManager().emitSignal('stop-application-completed') sound_theme_manager.getManager().playStopSound(wait=True, timeoutSeconds=1) cthulhuApp.getPluginSystemManager().unloadAllPlugins(ForceAllPlugins=True) + cthulhuApp.getScreenLockMonitor().stop() cthulhuApp.getMakoNotificationMonitor().stop() # Deactivate the event manager first so that it clears its queue and will not @@ -991,6 +995,9 @@ class Cthulhu(GObject.Object): with debug.startup_timing("notification monitor construction"): self.makoNotificationMonitor: mako_notification_monitor.MakoNotificationMonitor = \ mako_notification_monitor.MakoNotificationMonitor(self) + with debug.startup_timing("screen lock monitor construction"): + self.screenLockMonitor: screen_lock_monitor.ScreenLockMonitor = \ + screen_lock_monitor.ScreenLockMonitor(self) with debug.startup_timing("compat API registration"): self.createCompatAPI() with debug.startup_timing("plugin manager construction"): @@ -1009,6 +1016,9 @@ class Cthulhu(GObject.Object): def getMakoNotificationMonitor(self) -> mako_notification_monitor.MakoNotificationMonitor: return self.makoNotificationMonitor + def getScreenLockMonitor(self) -> screen_lock_monitor.ScreenLockMonitor: + return self.screenLockMonitor + def getDynamicApiManager(self) -> DynamicApiManager: return self.dynamicApiManager @@ -1061,6 +1071,7 @@ class Cthulhu(GObject.Object): "speech_server_present": speechServer is not None, "speech_server": speechServer.__class__.__name__ if speechServer else "", }, + "sound": sound.get_debug_snapshot(), "event_manager": self.eventManager.get_debug_snapshot(), "script_manager": self.scriptManager.get_debug_snapshot(), "input_event_manager": input_event_manager.get_manager().get_debug_snapshot(), @@ -1097,6 +1108,7 @@ class Cthulhu(GObject.Object): self.getDynamicApiManager().registerAPI('Cmdnames', cmdnames) self.getDynamicApiManager().registerAPI('NotificationPresenter', notification_presenter) self.getDynamicApiManager().registerAPI('MakoNotificationMonitor', self.makoNotificationMonitor) + self.getDynamicApiManager().registerAPI('ScreenLockMonitor', self.screenLockMonitor) self.getDynamicApiManager().registerAPI('CthulhuState', cthulhu_state) self.getDynamicApiManager().registerAPI('CthulhuPlatform', cthulhu_platform) self.getDynamicApiManager().registerAPI('Settings', settings) diff --git a/src/cthulhu/cthulhu_state.py b/src/cthulhu/cthulhu_state.py index 8611b19..de5ed82 100644 --- a/src/cthulhu/cthulhu_state.py +++ b/src/cthulhu/cthulhu_state.py @@ -64,6 +64,10 @@ compositorSnapshot: Optional[Any] = None # pauseAtspiChurn: bool = False +# Whether a trusted screen locker has requested secure-input isolation. +# +screenLockActive: bool = False + # The desktop-context token that should be prioritized next. # prioritizedDesktopContextToken: Optional[str] = None diff --git a/src/cthulhu/event_manager.py b/src/cthulhu/event_manager.py index 930eaf4..ba27685 100644 --- a/src/cthulhu/event_manager.py +++ b/src/cthulhu/event_manager.py @@ -41,7 +41,9 @@ import time from typing import TYPE_CHECKING, Optional, Dict, List, Tuple, Any, Union from . import cthulhu +from . import braille from . import debug +from . import focus_manager from . import input_event from . import input_event_manager from . import cthulhu_state @@ -132,6 +134,9 @@ class EventManager: def _sync_focus_on_startup(self) -> bool: """Initialize active window and focus when startup missed focus events.""" + if cthulhu_state.screenLockActive: + return False + focus = cthulhu_state.locusOfFocus if focus and not AXObject.is_dead(focus): return False @@ -216,8 +221,49 @@ class EventManager: "prioritized_context_token_present": self._prioritizedContextToken is not None, "key_handling_active": self._keyHandlingActive, "input_event_manager_present": self._inputEventManager is not None, + "screen_lock_active": cthulhu_state.screenLockActive, } + def set_screen_lock_active(self, active: bool, schedule_focus_resync: bool = True) -> None: + """Isolates screen-reader state while I38Lock owns secure input.""" + + if cthulhu_state.screenLockActive == active: + return + + cthulhu_state.screenLockActive = active + reason = ( + "I38Lock secure screen lock active" + if active + else "I38Lock secure screen lock released" + ) + input_event_manager.get_manager().set_secure_input_active(active, reason) + self._discard_queued_events_for_screen_lock() + + active_script = self.app.scriptManager.get_active_script() + if active and active_script is not None: + active_script.presentationInterrupt() + if active: + braille.clear() + + self.app.scriptManager.set_active_script(None, reason) + manager = focus_manager.get_manager() + if manager is not None: + manager.clear_state(reason) + cthulhu_state.pendingSelfHostedFocus = None + + if not active and schedule_focus_resync: + GLib.idle_add(self._sync_focus_on_startup) + + def _discard_queued_events_for_screen_lock(self) -> None: + """Drops queued accessibility events at both lock-state boundaries.""" + + self._gidleLock.acquire() + try: + self._eventQueue = queue.Queue(0) + self._prioritizedEvent = None + finally: + self._gidleLock.release() + def set_compositor_state_adapter(self, adapter: Any) -> None: """Stores the compositor state adapter used for startup wiring.""" @@ -604,6 +650,16 @@ class EventManager: def _ignore(self, event: Atspi.Event) -> bool: """Returns True if this event should be ignored.""" + if cthulhu_state.screenLockActive: + debug.print_log( + debug.LEVEL_INFO, + "EVENT MANAGER", + "Ignoring - secure screen lock is active", + "screen-lock-active", + True, + ) + return True + app = AXObject.get_application(event.source) debug.printMessage(debug.LEVEL_INFO, '') tokens = ["EVENT MANAGER:", event.type, "from", app] @@ -965,6 +1021,11 @@ class EventManager: - e: an at-spi event. """ + if cthulhu_state.screenLockActive: + msg = "EVENT MANAGER: Dropping event during secure screen lock." + debug.printMessage(debug.LEVEL_INFO, msg, True) + return + if debug.debugEventQueue: if self._enqueueCount: msg = f"EVENT MANAGER: _enqueue entered before exiting (count={self._enqueueCount})" @@ -1098,6 +1159,9 @@ class EventManager: return True def _onNoFocus(self) -> bool: + if cthulhu_state.screenLockActive: + return False + if not self._isNoFocus(): return False @@ -1300,7 +1364,7 @@ class EventManager: - event: an instance of BrailleEvent or a KeyboardEvent """ - if not cthulhu_state.activeScript: + if cthulhu_state.screenLockActive or not cthulhu_state.activeScript: return if not isinstance(event, input_event.BrailleEvent): @@ -1658,6 +1722,9 @@ class EventManager: - e: an at-spi event. """ + if cthulhu_state.screenLockActive: + return + debug.printObjectEvent(debug.LEVEL_INFO, event, timestamp=True) eType = event.type diff --git a/src/cthulhu/input_event_manager.py b/src/cthulhu/input_event_manager.py index 4c78470..15d5874 100644 --- a/src/cthulhu/input_event_manager.py +++ b/src/cthulhu/input_event_manager.py @@ -73,6 +73,7 @@ class InputEventManager: self._mapped_keysyms: List[int] = [] self._grabbed_bindings: Dict[int, keybindings.KeyBinding] = {} self._paused: bool = False + self._secure_input_active: bool = False self._wnck = None self._did_attempt_wnck_load: bool = False self._scriptWithSuspendedGrabsForXterm = None @@ -119,6 +120,7 @@ class InputEventManager: "mapped_keysyms": len(self._mapped_keysyms), "grabbed_bindings": len(self._grabbed_bindings), "paused": self._paused, + "secure_input_active": self._secure_input_active, "suspended_xterm_script_present": self._scriptWithSuspendedGrabsForXterm is not None, "last_input_event_present": self._last_input_event is not None, "last_non_modifier_key_event_present": self._last_non_modifier_key_event is not None, @@ -188,6 +190,18 @@ class InputEventManager: debug.print_message(debug.LEVEL_INFO, msg, True) self._paused = pause + def set_secure_input_active(self, active: bool, reason: str = "") -> None: + """Suppresses ordinary keyboard processing while a trusted lock screen is active.""" + + if self._secure_input_active == active: + return + + self._secure_input_active = active + self._last_input_event = None + self._last_non_modifier_key_event = None + msg = f"INPUT EVENT MANAGER: Secure input active: {active}. {reason}" + debug.print_message(debug.LEVEL_INFO, msg, True) + def check_grabbed_bindings(self) -> None: """Checks the grabbed key bindings.""" @@ -753,6 +767,10 @@ class InputEventManager: ) -> bool: """Processes this Atspi keyboard event.""" + if self._secure_input_active: + msg = "INPUT EVENT MANAGER: Ignoring keyboard event during secure screen lock." + debug.print_message(debug.LEVEL_INFO, msg, True) + return False if self._paused: msg = "INPUT EVENT MANAGER: Keyboard event processing is paused." debug.print_message(debug.LEVEL_INFO, msg, True) diff --git a/src/cthulhu/meson.build b/src/cthulhu/meson.build index 9eaf271..d84960a 100644 --- a/src/cthulhu/meson.build +++ b/src/cthulhu/meson.build @@ -87,6 +87,7 @@ cthulhu_python_sources = files([ 'script.py', 'script_manager.py', 'script_utilities.py', + 'screen_lock_monitor.py', 'settings.py', 'settings_manager.py', 'signal_manager.py', diff --git a/src/cthulhu/screen_lock_monitor.py b/src/cthulhu/screen_lock_monitor.py new file mode 100644 index 0000000..7b41e58 --- /dev/null +++ b/src/cthulhu/screen_lock_monitor.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2026 Stormux +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. + +"""Tracks the lifetime of I38Lock's secure accessibility-feedback worker.""" + +from __future__ import annotations + +import os +from typing import Any, Optional + +from gi.repository import Gio, GLib + +from . import debug + + +class ScreenLockMonitor: + """Enters secure screen-reader mode while I38Lock owns its session-bus name.""" + + BUS_NAME = "org.stormux.I38Lock1" + TRUSTED_EXECUTABLE = "i38lock" + + def __init__(self, app: Any) -> None: + self._app = app + self._connection: Optional[Gio.DBusConnection] = None + self._watch_id: int = 0 + self._is_running: bool = False + self._is_active: bool = False + + def start(self) -> None: + """Starts watching the session bus for I38Lock.""" + + if self._is_running: + return + + try: + self._connection = Gio.bus_get_sync(Gio.BusType.SESSION, None) + except Exception as error: + msg = f"SCREEN LOCK MONITOR: Failed to connect to session bus: {error}" + debug.printMessage(debug.LEVEL_WARNING, msg, True) + return + + self._watch_id = Gio.bus_watch_name_on_connection( + self._connection, + self.BUS_NAME, + Gio.BusNameWatcherFlags.NONE, + self._on_name_appeared, + self._on_name_vanished, + ) + self._is_running = True + debug.printMessage(debug.LEVEL_INFO, "SCREEN LOCK MONITOR: Started", True) + + def stop(self) -> None: + """Stops watching I38Lock and releases any active secure state.""" + + if self._watch_id: + Gio.bus_unwatch_name(self._watch_id) + self._watch_id = 0 + + if self._is_active: + self._set_active(False, schedule_focus_resync=False) + + self._connection = None + self._is_running = False + debug.printMessage(debug.LEVEL_INFO, "SCREEN LOCK MONITOR: Stopped", True) + + def is_running(self) -> bool: + """Returns whether the D-Bus name watcher is running.""" + + return self._is_running + + def is_active(self) -> bool: + """Returns whether I38Lock currently owns its secure-state bus name.""" + + return self._is_active + + def _set_active(self, active: bool, schedule_focus_resync: bool = True) -> None: + if self._is_active == active: + return + + self._is_active = active + self._app.eventManager.set_screen_lock_active( + active, + schedule_focus_resync=schedule_focus_resync, + ) + + def _get_name_owner_pid( + self, + connection: Gio.DBusConnection, + name_owner: str, + ) -> Optional[int]: + try: + result = connection.call_sync( + "org.freedesktop.DBus", + "/org/freedesktop/DBus", + "org.freedesktop.DBus", + "GetConnectionUnixProcessID", + GLib.Variant("(s)", (name_owner,)), + GLib.VariantType("(u)"), + Gio.DBusCallFlags.NONE, + 500, + None, + ) + return result.unpack()[0] + except Exception as error: + msg = f"SCREEN LOCK MONITOR: Failed to get owner PID for {name_owner}: {error}" + debug.printMessage(debug.LEVEL_WARNING, msg, True) + return None + + def _is_trusted_owner(self, connection: Gio.DBusConnection, name_owner: str) -> bool: + pid = self._get_name_owner_pid(connection, name_owner) + if pid is None: + return False + + try: + executable = os.path.basename(os.readlink(f"/proc/{pid}/exe")) + except OSError as error: + msg = f"SCREEN LOCK MONITOR: Failed to inspect owner PID {pid}: {error}" + debug.printMessage(debug.LEVEL_WARNING, msg, True) + return False + + if executable != self.TRUSTED_EXECUTABLE: + msg = f"SCREEN LOCK MONITOR: Ignoring untrusted owner executable {executable}" + debug.printMessage(debug.LEVEL_WARNING, msg, True) + return False + + return True + + def _on_name_appeared( + self, + connection: Gio.DBusConnection, + name: str, + name_owner: str, + _user_data: Optional[Any] = None, + ) -> None: + msg = f"SCREEN LOCK MONITOR: {name} appeared as {name_owner}" + debug.printMessage(debug.LEVEL_INFO, msg, True) + if not self._is_trusted_owner(connection, name_owner): + return + self._set_active(True) + + def _on_name_vanished( + self, + _connection: Gio.DBusConnection, + name: str, + _user_data: Optional[Any] = None, + ) -> None: + msg = f"SCREEN LOCK MONITOR: {name} vanished" + debug.printMessage(debug.LEVEL_INFO, msg, True) + self._set_active(False) diff --git a/src/cthulhu/sound.py b/src/cthulhu/sound.py index 9e47e9c..bf44a6a 100644 --- a/src/cthulhu/sound.py +++ b/src/cthulhu/sound.py @@ -144,6 +144,37 @@ class Player: self._stopWorkerLocked("Sound system shutdown") self._initialized = False + def get_debug_snapshot(self) -> dict[str, Any]: + """Returns read-only sound worker state for runtime diagnostics.""" + + with self._workerLock: + process = self._workerProcess + workerExitStatus: Any = "" + workerAlive = False + workerPid = 0 + if process is not None: + workerPid = process.pid + workerExitStatus = process.poll() + workerAlive = workerExitStatus is None + + with self._responseLock: + pendingResponseCount = len(self._pendingResponses) + + return { + "gstreamer_available": _gstreamerAvailable, + "sound_system_available": isSoundSystemAvailable(), + "failure_reason": getSoundSystemFailureReason() or "", + "initialized": self._initialized, + "worker_present": process is not None, + "worker_alive": workerAlive, + "worker_pid": workerPid, + "worker_exit_status": workerExitStatus if workerExitStatus is not None else "", + "worker_sink": self._workerSink or "", + "worker_restart_required": self._workerRestartRequired, + "worker_restart_reason": self._workerRestartReason or "", + "pending_responses": pendingResponseCount, + } + def _playIcon(self, icon: Icon, interrupt: bool = True) -> None: if not icon.isValid(): return @@ -555,6 +586,7 @@ def _buildSoundHelperEnvironment() -> dict[str, str]: environment = os.environ.copy() if pythonPathEntries: environment["PYTHONPATH"] = os.pathsep.join(pythonPathEntries) + environment.setdefault("ORC_CODE", "backup") return environment @@ -571,6 +603,10 @@ def getPlayer() -> Player: return _player +def get_debug_snapshot() -> dict[str, Any]: + return _player.get_debug_snapshot() + + def play(item: Any, interrupt: bool = True) -> None: _player.play(item, interrupt) diff --git a/src/cthulhu/speech.py b/src/cthulhu/speech.py index 60c3e7c..5226c38 100644 --- a/src/cthulhu/speech.py +++ b/src/cthulhu/speech.py @@ -540,7 +540,9 @@ def speak(content: Union[str, List[Any]], acss: Optional[Any] = None, interrupt: if element.isValid(): player = sound.getPlayer() if player: - player.play(element, interrupt=interrupt) + # Role/state icons are short UI cues. Speech can be queued, but + # sound cues should not pile up behind stale or previous sounds. + player.play(element, interrupt=True) elif toSpeak: newVoice = ACSS(acss) newItemsToSpeak = [] diff --git a/tests/test_presentation_interrupt_sound_regressions.py b/tests/test_presentation_interrupt_sound_regressions.py index 7a06267..e02d141 100644 --- a/tests/test_presentation_interrupt_sound_regressions.py +++ b/tests/test_presentation_interrupt_sound_regressions.py @@ -1,6 +1,7 @@ import sys import unittest import importlib +import tempfile from pathlib import Path from unittest import mock @@ -27,6 +28,19 @@ class PresentationInterruptSoundRegressionTests(unittest.TestCase): echoServer.stop.assert_called_once_with() soundPlayer.stop.assert_called_once_with() + def test_speech_icons_interrupt_previous_sound_when_speech_is_queued(self): + soundPlayer = mock.Mock() + + with tempfile.TemporaryDirectory() as temporaryDirectory: + soundPath = Path(temporaryDirectory) / "button.wav" + soundPath.write_bytes(b"RIFF") + icon = speech.Icon(temporaryDirectory, "button.wav") + + with mock.patch.object(speech.sound, "getPlayer", return_value=soundPlayer): + speech.speak([icon], interrupt=False) + + soundPlayer.play.assert_called_once_with(icon, interrupt=True) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_screen_lock_monitor.py b/tests/test_screen_lock_monitor.py new file mode 100644 index 0000000..a320568 --- /dev/null +++ b/tests/test_screen_lock_monitor.py @@ -0,0 +1,211 @@ +import sys +import types +import unittest +from unittest import mock + +import cthulhu as cthulhu_package +from cthulhu import cthulhu_state + +stubCthulhu = types.ModuleType("cthulhu.cthulhu") +stubCthulhu.cthulhuApp = mock.Mock() +stubCthulhu.cthulhuApp.getPluginSystemManager.return_value = None +previousCthulhuModule = sys.modules.get("cthulhu.cthulhu") +sys.modules.setdefault("cthulhu.cthulhu", stubCthulhu) + +from cthulhu import event_manager +from cthulhu import input_event_manager +from cthulhu.screen_lock_monitor import ScreenLockMonitor + +if previousCthulhuModule is None: + sys.modules.pop("cthulhu.cthulhu", None) + if getattr(cthulhu_package, "cthulhu", None) is stubCthulhu: + delattr(cthulhu_package, "cthulhu") +else: + sys.modules["cthulhu.cthulhu"] = previousCthulhuModule + + +class ScreenLockMonitorTests(unittest.TestCase): + def setUp(self): + self.app = mock.Mock() + self.monitor = ScreenLockMonitor(self.app) + + def test_name_appearance_enters_secure_lock_mode(self): + with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=True): + self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42") + + self.app.eventManager.set_screen_lock_active.assert_called_once_with( + True, + schedule_focus_resync=True, + ) + self.assertTrue(self.monitor.is_active()) + + def test_untrusted_name_owner_does_not_enter_secure_lock_mode(self): + with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=False): + self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42") + + self.app.eventManager.set_screen_lock_active.assert_not_called() + self.assertFalse(self.monitor.is_active()) + + def test_name_disappearance_leaves_secure_lock_mode(self): + with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=True): + self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42") + self.monitor._on_name_vanished(mock.Mock(), self.monitor.BUS_NAME) + + self.assertEqual( + self.app.eventManager.set_screen_lock_active.call_args_list, + [ + mock.call(True, schedule_focus_resync=True), + mock.call(False, schedule_focus_resync=True), + ], + ) + self.assertFalse(self.monitor.is_active()) + + def test_start_watches_i38lock_bus_name(self): + connection = mock.Mock() + with ( + mock.patch("cthulhu.screen_lock_monitor.Gio.bus_get_sync", return_value=connection), + mock.patch( + "cthulhu.screen_lock_monitor.Gio.bus_watch_name_on_connection", + return_value=23, + ) as watch_name, + ): + self.monitor.start() + + watch_name.assert_called_once_with( + connection, + self.monitor.BUS_NAME, + mock.ANY, + self.monitor._on_name_appeared, + self.monitor._on_name_vanished, + ) + self.assertTrue(self.monitor.is_running()) + + def test_stop_releases_secure_lock_without_focus_resync(self): + with mock.patch.object(self.monitor, "_is_trusted_owner", return_value=True): + self.monitor._on_name_appeared(mock.Mock(), self.monitor.BUS_NAME, ":1.42") + + with mock.patch("cthulhu.screen_lock_monitor.Gio.bus_unwatch_name"): + self.monitor.stop() + + self.assertEqual( + self.app.eventManager.set_screen_lock_active.call_args_list, + [ + mock.call(True, schedule_focus_resync=True), + mock.call(False, schedule_focus_resync=False), + ], + ) + self.assertFalse(self.monitor.is_active()) + + def test_i38lock_process_owner_is_trusted(self): + connection = mock.Mock() + result = mock.Mock() + result.unpack.return_value = (1234,) + connection.call_sync.return_value = result + + with mock.patch("cthulhu.screen_lock_monitor.os.readlink", return_value="/usr/bin/i38lock"): + self.assertTrue(self.monitor._is_trusted_owner(connection, ":1.42")) + + def test_non_i38lock_process_owner_is_not_trusted(self): + connection = mock.Mock() + result = mock.Mock() + result.unpack.return_value = (1234,) + connection.call_sync.return_value = result + + with mock.patch("cthulhu.screen_lock_monitor.os.readlink", return_value="/usr/bin/python"): + self.assertFalse(self.monitor._is_trusted_owner(connection, ":1.42")) + + +class SecureScreenLockStateTests(unittest.TestCase): + def setUp(self): + self.original_screen_lock_active = cthulhu_state.screenLockActive + cthulhu_state.screenLockActive = False + + def tearDown(self): + cthulhu_state.screenLockActive = self.original_screen_lock_active + + def test_entering_secure_lock_clears_context_and_pauses_input(self): + app = mock.Mock() + active_script = mock.Mock() + app.scriptManager.get_active_script.return_value = active_script + manager = event_manager.EventManager(app, asyncMode=True) + keyboard_manager = mock.Mock() + focus_manager = mock.Mock() + + with ( + mock.patch.object(input_event_manager, "get_manager", return_value=keyboard_manager), + mock.patch("cthulhu.event_manager.focus_manager.get_manager", return_value=focus_manager), + mock.patch("cthulhu.event_manager.braille.clear") as clear_braille, + ): + manager.set_screen_lock_active(True) + + self.assertTrue(cthulhu_state.screenLockActive) + active_script.presentationInterrupt.assert_called_once_with() + app.scriptManager.set_active_script.assert_called_once_with( + None, + "I38Lock secure screen lock active", + ) + focus_manager.clear_state.assert_called_once_with( + "I38Lock secure screen lock active" + ) + keyboard_manager.set_secure_input_active.assert_called_once_with( + True, + "I38Lock secure screen lock active", + ) + clear_braille.assert_called_once_with() + + def test_object_events_are_dropped_without_inspecting_locked_content(self): + app = mock.Mock() + manager = event_manager.EventManager(app, asyncMode=True) + cthulhu_state.screenLockActive = True + event = mock.Mock() + + with ( + mock.patch.object(manager, "_addToQueue") as add_to_queue, + mock.patch("cthulhu.event_manager.AXObject.get_application") as get_application, + ): + manager._enqueue(event) + + add_to_queue.assert_not_called() + get_application.assert_not_called() + + def test_unlock_resumes_input_and_schedules_focus_resync(self): + app = mock.Mock() + manager = event_manager.EventManager(app, asyncMode=True) + keyboard_manager = mock.Mock() + focus_manager = mock.Mock() + cthulhu_state.screenLockActive = True + + with ( + mock.patch.object(input_event_manager, "get_manager", return_value=keyboard_manager), + mock.patch("cthulhu.event_manager.focus_manager.get_manager", return_value=focus_manager), + mock.patch("cthulhu.event_manager.GLib.idle_add") as idle_add, + ): + manager.set_screen_lock_active(False) + + self.assertFalse(cthulhu_state.screenLockActive) + keyboard_manager.set_secure_input_active.assert_called_once_with( + False, + "I38Lock secure screen lock released", + ) + idle_add.assert_called_once_with(manager._sync_focus_on_startup) + + def test_secure_input_blocks_key_echo_before_event_creation(self): + manager = input_event_manager.InputEventManager() + manager.set_secure_input_active(True, "test lock") + + with mock.patch("cthulhu.input_event_manager.input_event.KeyboardEvent") as event_class: + consumed = manager.process_keyboard_event( + mock.Mock(), + True, + 25, + 112, + 0, + "p", + ) + + self.assertFalse(consumed) + event_class.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sound_recovery.py b/tests/test_sound_recovery.py index b6d1b33..7506360 100644 --- a/tests/test_sound_recovery.py +++ b/tests/test_sound_recovery.py @@ -2,6 +2,7 @@ import sys import types import unittest import importlib +import os from pathlib import Path from unittest import mock @@ -68,6 +69,18 @@ class SoundSinkTests(unittest.TestCase): for module in reloadedModules: importlib.reload(module) + def test_sound_helper_environment_defaults_to_orc_backup(self): + with mock.patch.dict(os.environ, {}, clear=True): + environment = sound._buildSoundHelperEnvironment() + + self.assertEqual("backup", environment["ORC_CODE"]) + + def test_sound_helper_environment_preserves_explicit_orc_code(self): + with mock.patch.dict(os.environ, {"ORC_CODE": "debug"}, clear=True): + environment = sound._buildSoundHelperEnvironment() + + self.assertEqual("debug", environment["ORC_CODE"]) + class PlayerRecoveryTests(unittest.TestCase): def test_worker_diagnostic_marks_restart_required(self):