diff --git a/src/vocalinux/ui/keyboard_backends/evdev_backend.py b/src/vocalinux/ui/keyboard_backends/evdev_backend.py index 564f70e..53a302c 100644 --- a/src/vocalinux/ui/keyboard_backends/evdev_backend.py +++ b/src/vocalinux/ui/keyboard_backends/evdev_backend.py @@ -30,6 +30,7 @@ from typing import Any, Optional, Sequence, TextIO # Try to import evdev try: import evdev + from evdev import _uinput # type: ignore[attr-defined] # C extension, unseen by mypy from evdev import InputDevice, InputEvent, UInput, UInputError, ecodes EVDEV_AVAILABLE = True @@ -39,6 +40,7 @@ except ImportError: InputEvent = None # type: ignore UInput = None # type: ignore UInputError = OSError # type: ignore + _uinput = None # type: ignore ecodes = None # type: ignore EVDEV_AVAILABLE = False @@ -82,6 +84,99 @@ if EVDEV_AVAILABLE: else: _FORWARDED_EVENT_TYPES = frozenset() +# Suffix on every uinput clone name. ``_open_keyboard_device`` uses it to +# skip clones (also ones created by other Vocalinux processes), so it must +# survive the kernel's uinput name-length truncation. +_CLONE_NAME_SUFFIX = " (vocalinux)" + + +def _clone_device_name(device: InputDevice) -> str: + """Clone name that keeps the suffix rescan checks rely on. + + The kernel stores at most ``_uinput.maxnamelen - 1`` name *bytes*, so + the base is truncated after encoding — a long multibyte device name + can otherwise push the suffix past the kernel's cut, and rescan would + no longer recognize (and skip) the clone. + """ + base = str(getattr(device, "name", "") or "") + limit = _uinput.maxnamelen - 1 - len(_CLONE_NAME_SUFFIX) # suffix is ASCII + base = base.encode("utf-8", errors="replace")[:limit].decode("utf-8", errors="ignore") + return f"{base}{_CLONE_NAME_SUFFIX}" + + +if EVDEV_AVAILABLE: + + class _WriteOnlyUInput(UInput): + """A ``UInput`` built on a write-only ``/dev/uinput`` descriptor. + + python-evdev opens the uinput node ``O_RDWR``, although the kernel + only needs write access for the setup ioctls and event injection. + Installs still carrying the original udev rule (``MODE=0620``, + write-only for the ``input`` group) therefore fail ``UInput()`` + with EACCES and fall back to ungrabbed monitoring, where the + dictation shortcut also reaches the focused app. Opening the node + ``O_WRONLY`` keeps key suppression working there. + """ + + def __init__( + self, + events: Optional[dict[int, Sequence[int]]], + name: str, + bustype: int, + devnode: str = "/dev/uinput", + ) -> None: + self.name: str = name + self.vendor: int = 0x1 + self.product: int = 0x1 + self.version: int = 0x1 + self.bustype: int = bustype + self.phys: str = "py-evdev-uinput" + self.devnode: str = devnode + self._verify() + self.fd = os.open(devnode, os.O_WRONLY | os.O_NONBLOCK) + try: + absinfo, prepared_events = self._prepare_events(events) + except Exception: + os.close(self.fd) + raise + # A failing _uinput helper destroys the device and closes fd + # itself (uinput.c on_err), so the calls below need no cleanup. + _uinput.set_phys(self.fd, self.phys) + for event_type, code in prepared_events: + _uinput.enable(self.fd, event_type, code) + _uinput.setup( + self.fd, + name, + self.vendor, + self.product, + self.version, + bustype, + absinfo, + ecodes.ecodes.get("FF_MAX_EFFECTS", 96), + ) + _uinput.create(self.fd) + try: + self.device = self._find_device(self.fd) + except Exception: + _uinput.close(self.fd) + raise + + # EventIO.need_write insists on O_RDWR; a write-only uinput fd + # still takes writes, so emit through _uinput.write directly. + def write(self, etype: int, code: int, value: int) -> None: + _uinput.write(self.fd, etype, code, value) + + def write_event(self, event: InputEvent) -> None: + if hasattr(event, "event"): + event = event.event + _uinput.write(self.fd, event.type, event.code, event.value) + + def syn(self) -> None: + _uinput.write(self.fd, ecodes.EV_SYN, ecodes.SYN_REPORT, 0) + +else: + _WriteOnlyUInput = None # type: ignore + # Map modifier key names to evdev key codes MODIFIER_KEY_CODES: dict[str, set[int]] = { "ctrl": {KEY_LEFTCTRL, KEY_RIGHTCTRL}, @@ -793,7 +888,7 @@ class EvdevDeviceHub: # to them, and grabbing is what lets the shortcut reach us at all # when a remapper has already grabbed the physical device. device_name = str(getattr(device, "name", "") or "") - if device_path in self._clone_paths or device_name.endswith(" (vocalinux)"): + if device_path in self._clone_paths or device_name.endswith(_CLONE_NAME_SUFFIX): logger.debug(f"Skipping Vocalinux clone device: {device_path} ({device_name})") try: device.close() @@ -884,11 +979,28 @@ class EvdevDeviceHub: } for event_type in (ecodes.EV_SYN, ecodes.EV_FF, ecodes.EV_REP): capabilities.pop(event_type, None) - return UInput( - events=capabilities, - name=f"{device.name} (vocalinux)", - bustype=ecodes.BUS_VIRTUAL, - ) + name = _clone_device_name(device) + try: + return UInput( + events=capabilities, + name=name, + bustype=ecodes.BUS_VIRTUAL, + ) + except OSError as e: + # evdev's _uinput.open insists on O_RDWR (its failure has no + # errno), but uinput only needs write access. Installs that + # still carry the original write-only udev rule (MODE=0620) + # get a working clone through an O_WRONLY fd instead of + # silently falling back to ungrabbed monitoring, where the + # dictation shortcut also reaches the focused app. + if e.errno not in (None, errno.EACCES, errno.EPERM): + raise + logger.debug(f"O_RDWR /dev/uinput refused ({e}); retrying write-only") + return _WriteOnlyUInput( + events=capabilities, + name=name, + bustype=ecodes.BUS_VIRTUAL, + ) except (OSError, IOError, TypeError, ValueError, UInputError) as e: if not self._uinput_warned: self._uinput_warned = True @@ -1422,10 +1534,10 @@ class EvdevKeyboardBackend(KeyboardBackend): # so a press is withheld rather than dropped outright — a later # non-target press replays it (composition), while a release # after a real gesture stays consumed (the clone never saw it). - withheld = self._withheld_modifier.get(fd) - if event.value == 1 and withheld is not None: + withheld = self._withheld_modifier.setdefault(fd, {}) + if event.value == 1: withheld[event.code] = [event, False] - elif event.value == 0 and withheld is not None: + elif event.value == 0: entry = withheld.pop(event.code, None) if entry is not None and entry[1]: # Replayed as composition — the clone believes the diff --git a/tests/test_evdev_backend_ext.py b/tests/test_evdev_backend_ext.py index 21faca9..4ba4f6e 100644 --- a/tests/test_evdev_backend_ext.py +++ b/tests/test_evdev_backend_ext.py @@ -11,11 +11,12 @@ This module tests: - _monitor_devices() thread """ +import contextlib import errno import os import threading import time -from unittest.mock import MagicMock, Mock, mock_open, patch +from unittest.mock import MagicMock, Mock, call, mock_open, patch import pytest @@ -25,6 +26,7 @@ from vocalinux.ui.keyboard_backends.evdev_backend import ( MODIFIER_KEY_CODES, EvdevDeviceHub, EvdevKeyboardBackend, + _clone_device_name, device_has_key, device_has_modifier_key, ecodes, @@ -1150,20 +1152,76 @@ class TestEvdevGrabAndForwarding: device.grab.assert_not_called() assert backend.devices == [] + @patch("vocalinux.ui.keyboard_backends.evdev_backend._WriteOnlyUInput") @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") - def test_open_falls_back_when_uinput_unavailable(self, mock_input_device, mock_uinput): + def test_open_falls_back_when_uinput_unavailable( + self, mock_input_device, mock_uinput, mock_write_only + ): """Without /dev/uinput the device still works ungrabbed (keys leak).""" device = MagicMock() device.fileno.return_value = 10 device.info.bustype = 0x03 mock_input_device.return_value = device mock_uinput.side_effect = OSError("uinput denied") + mock_write_only.side_effect = OSError(errno.EACCES, "uinput denied") backend = EvdevKeyboardBackend() opened = backend._open_keyboard_device("/dev/input/event0") assert opened is True + mock_write_only.assert_called_once() + device.grab.assert_not_called() + assert backend._forwarders == {} + assert device in backend.devices + + @patch("vocalinux.ui.keyboard_backends.evdev_backend._WriteOnlyUInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_retries_uinput_write_only_on_eacces( + self, mock_input_device, mock_uinput, mock_write_only + ): + """An O_RDWR refusal retries O_WRONLY so the device still gets grabbed. + + Installs carrying the original write-only udev rule (MODE=0620) + denied evdev's O_RDWR open of /dev/uinput; a write-only descriptor + satisfies the kernel and keeps key suppression working. + """ + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x03 + mock_input_device.return_value = device + mock_uinput.side_effect = OSError("could not open uinput device in write mode") + forwarder = MagicMock() + forwarder.device.path = "/dev/input/event20" + mock_write_only.return_value = forwarder + + backend = EvdevKeyboardBackend() + opened = backend._open_keyboard_device("/dev/input/event0") + + assert opened is True + mock_write_only.assert_called_once() + device.grab.assert_called_once() + assert backend._forwarders[10] is forwarder + + @patch("vocalinux.ui.keyboard_backends.evdev_backend._WriteOnlyUInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_does_not_retry_write_only_on_other_oserror( + self, mock_input_device, mock_uinput, mock_write_only + ): + """A UInput failure with a real errno is not a permission retry.""" + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x03 + mock_input_device.return_value = device + mock_uinput.side_effect = OSError(errno.EIO, "i/o error") + + backend = EvdevKeyboardBackend() + opened = backend._open_keyboard_device("/dev/input/event0") + + assert opened is True + mock_write_only.assert_not_called() device.grab.assert_not_called() assert backend._forwarders == {} assert device in backend.devices @@ -1607,12 +1665,18 @@ class TestEvdevGrabAndForwarding: assert backend._forwarders == {} def test_monitor_forwards_survivors_during_syn_dropped(self): - """Mid-drop: shortcut keys are eaten, surviving other keys forwarded.""" + """Mid-drop: a withheld modifier press replays ahead of other keys. + + RIGHTALT surviving with KEY_A reads as AltGr composition, not a + gesture, so the withheld press is replayed before the letter — + the app sees exactly what the user typed. + """ backend = EvdevKeyboardBackend() device = MagicMock() device.fileno.return_value = 10 forwarder = MagicMock() backend._forwarders = {10: forwarder} + backend._forwarded_held = {10: set()} backend._dropped_devices = {10} shortcut_event = self._key_event(100, 1) # KEY_RIGHTALT press other_event = self._key_event(30, 1) # KEY_A @@ -1643,7 +1707,33 @@ class TestEvdevGrabAndForwarding: backend._monitor_devices() backend._handle_key_event.assert_not_called() - forwarder.write_event.assert_called_once_with(other_event) + assert forwarder.write_event.call_args_list == [ + call(shortcut_event), + call(other_event), + ] + + def test_pure_modifier_press_withheld_lazily(self) -> None: + """Production path: the withheld map is created on first modifier event. + + The AltGr-replay machinery only works if the per-fd map exists; it + used to be populated by nothing in production, so withheld presses + were silently dropped and composition lost its modifier. + """ + backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt") + fd = 10 + forwarder = MagicMock() + backend._forwarders[fd] = forwarder + backend._forwarded_held[fd] = set() + + press = self._key_event(100, 1) + key_e = self._key_event(18, 1) + + assert backend._event_is_shortcut(fd, press) is True + assert 100 in backend._withheld_modifier[fd] + + assert backend._event_is_shortcut(fd, key_e) is False + forwarder.write_event.assert_called_once_with(press) + assert 100 in backend._forwarded_held[fd] def test_pure_modifier_press_stays_consumed(self) -> None: """A real PTT hold never reaches the application.""" @@ -1747,6 +1837,98 @@ class TestEvdevGrabAndForwarding: assert backend._withheld_modifier[fd] == {} +class TestWriteOnlyUInput: + """The write-only clone drives the real _uinput setup on an O_WRONLY fd.""" + + @contextlib.contextmanager + def _built_uinput(self, fd: int = 42): + """Instantiate the real _WriteOnlyUInput with os/evdev calls mocked.""" + from vocalinux.ui.keyboard_backends import evdev_backend as eb + + device = MagicMock() + with ( + patch.object(eb._WriteOnlyUInput, "_verify"), + patch.object(eb.os, "open", return_value=fd) as mock_open, + patch.object(eb._uinput, "set_phys") as set_phys, + patch.object(eb._uinput, "enable") as enable, + patch.object(eb._uinput, "setup") as setup, + patch.object(eb._uinput, "create") as create, + patch.object(eb._WriteOnlyUInput, "_find_device", return_value=device), + patch.object(eb._uinput, "write") as write, + ): + uinput = eb._WriteOnlyUInput(events={1: [30]}, name="kbd (vocalinux)", bustype=0x06) + yield uinput, mock_open, set_phys, enable, setup, create, write, device + + def test_opens_wronly_and_runs_setup(self) -> None: + """Constructor opens O_WRONLY and performs the full uinput setup.""" + with self._built_uinput(fd=42) as ( + uinput, + mock_open, + set_phys, + enable, + setup, + create, + _, + device, + ): + mock_open.assert_called_once_with("/dev/uinput", os.O_WRONLY | os.O_NONBLOCK) + set_phys.assert_called_once_with(42, "py-evdev-uinput") + enable.assert_called_once_with(42, 1, 30) + assert setup.call_args[0][0] == 42 + create.assert_called_once_with(42) + assert uinput.fd == 42 + assert uinput.device is device + + def test_emits_through_uinput_write(self) -> None: + """write/write_event/syn bypass need_write's O_RDWR check.""" + import evdev as real_evdev + + with self._built_uinput(fd=42) as (uinput, _, _, _, _, _, write, _): + uinput.write(1, 30, 1) + uinput.write_event(real_evdev.InputEvent(0, 0, 1, 30, 0)) + uinput.syn() + + assert write.call_args_list == [ + call(42, 1, 30, 1), + call(42, 1, 30, 0), + call(42, 0, 0, 0), # EV_SYN / SYN_REPORT + ] + + def test_failed_event_prep_closes_descriptor(self) -> None: + """A failure before the _uinput calls still releases our fd.""" + from vocalinux.ui.keyboard_backends import evdev_backend as eb + + with ( + patch.object(eb._WriteOnlyUInput, "_verify"), + patch.object(eb.os, "open", return_value=42), + patch.object(eb.os, "close") as mock_close, + patch.object( + eb._WriteOnlyUInput, "_prepare_events", side_effect=TypeError("bad events") + ), + ): + with pytest.raises(TypeError): + eb._WriteOnlyUInput(events=None, name="kbd (vocalinux)", bustype=0x06) + + mock_close.assert_called_once_with(42) + + def test_clone_device_name_caps_encoded_bytes(self) -> None: + """A long multibyte device name truncates by bytes, keeping the suffix.""" + device = MagicMock() + device.name = "\u2328" * 40 # 3 bytes each, 120 bytes total + + name = _clone_device_name(device) + + assert name.endswith(" (vocalinux)") + assert len(name.encode("utf-8")) <= 79 + + def test_clone_device_name_keeps_short_names(self) -> None: + """Ordinary names keep the full base plus suffix.""" + device = MagicMock() + device.name = "AT Translated Set 2 keyboard" + + assert _clone_device_name(device) == "AT Translated Set 2 keyboard (vocalinux)" + + class TestSharedEvdevDeviceLayer: """Several backends share one evdev reader instead of competing grabs.