From 96e012d9bd4e6d34d53cbf696dfbcce85fd492dc Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 1 Oct 2026 23:45:49 -0700 Subject: [PATCH] fix(evdev): keep key suppression when /dev/uinput is write-only (#893) * fix(evdev): keep key suppression when /dev/uinput is write-only python-evdev opens /dev/uinput O_RDWR, but the kernel only needs write access for the uinput setup ioctls and event injection. Installs that still carry the original 80-dotool.rules line (MODE=0620, write-only for the input group) therefore fail UInput() with EACCES and fall back to ungrabbed monitoring, where the dictation shortcut is detected but also reaches the focused application. Retry the clone on an O_WRONLY descriptor via a small _WriteOnlyUInput subclass so suppression keeps working there (refs #871). Also populate _withheld_modifier lazily in _event_is_shortcut: the AltGr-replay machinery was dead code because nothing created the per-fd map, so a withheld modifier press was never replayed and composition (e.g. RightAlt+e) lost its modifier. * fix(evdev): byte-truncate clone names and cover the write-only path - _clone_device_name now truncates the encoded name by bytes: the kernel limit is 79 bytes, and a long multibyte device name could push the " (vocalinux)" suffix past the cut so rescan would not skip the clone. - _WriteOnlyUInput closes the fd explicitly when _prepare_events fails before the _uinput calls; the helpers close it themselves on any later failure. - Tests now instantiate the real subclass with mocked OS/native calls (open flags, setup sequence, write/write_event/syn ordering, fd cleanup, name truncation). --------- Co-authored-by: Jatin K Malik --- .../ui/keyboard_backends/evdev_backend.py | 130 +++++++++++- tests/test_evdev_backend_ext.py | 190 +++++++++++++++++- 2 files changed, 307 insertions(+), 13 deletions(-) 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.