diff --git a/install.d/system_dependencies.sh b/install.d/system_dependencies.sh index 86ac379..c6b5a39 100644 --- a/install.d/system_dependencies.sh +++ b/install.d/system_dependencies.sh @@ -752,11 +752,16 @@ install_text_input_tools() { print_warning "You will need to LOG OUT and back in for text injection to work with ydotool/dotool" fi - # Install udev rule for ydotool/dotool - if [ ! -f /etc/udev/rules.d/80-dotool.rules ]; then + # Install udev rule for ydotool/dotool — and for the evdev + # backend's uinput clone, which opens /dev/uinput O_RDWR, so the + # rule needs 0660 (0620 is write-only). Appending (not overwriting) + # preserves any rules the user already keeps in this file; when an + # older MODE=0620 line is present our later line still wins. + UINPUT_UDEV_RULE='KERNEL=="uinput", GROUP="input", MODE="0660", OPTIONS+="static_node=uinput"' + if ! grep -qxF "$UINPUT_UDEV_RULE" /etc/udev/rules.d/80-dotool.rules 2>/dev/null; then print_info "Installing udev rule for input device access..." - echo 'KERNEL=="uinput", GROUP="input", MODE="0620", OPTIONS+="static_node=uinput"' \ - | sudo tee /etc/udev/rules.d/80-dotool.rules >/dev/null 2>&1 || print_warning "Failed to install udev rule" + printf '\n%s\n' "$UINPUT_UDEV_RULE" \ + | sudo tee -a /etc/udev/rules.d/80-dotool.rules >/dev/null 2>&1 || print_warning "Failed to install udev rule" sudo udevadm control --reload 2>/dev/null || true sudo udevadm trigger 2>/dev/null || true fi diff --git a/src/vocalinux/ui/keyboard_backends/evdev_backend.py b/src/vocalinux/ui/keyboard_backends/evdev_backend.py index 62c4ef3..04f16ce 100644 --- a/src/vocalinux/ui/keyboard_backends/evdev_backend.py +++ b/src/vocalinux/ui/keyboard_backends/evdev_backend.py @@ -3,6 +3,12 @@ evdev keyboard backend for Wayland support. This backend uses python-evdev to read keyboard events directly from input devices, which works on both X11 and Wayland (with proper permissions). + +When /dev/uinput is writable, each opened keyboard is grabbed (EVIOCGRAB) +and its events are re-emitted through a paired uinput clone, so the keys +that make up the dictation shortcut are consumed here and never reach the +focused application. Without uinput access the backend still listens +ungrabbed, but shortcut presses also pass through to apps. """ import errno @@ -12,17 +18,20 @@ import re import select import threading import time -from typing import Optional, TextIO +from typing import Optional, Sequence, TextIO # Try to import evdev try: import evdev - from evdev import InputDevice, ecodes + from evdev import InputDevice, InputEvent, UInput, UInputError, ecodes EVDEV_AVAILABLE = True except ImportError: evdev = None # type: ignore InputDevice = None # type: ignore + InputEvent = None # type: ignore + UInput = None # type: ignore + UInputError = OSError # type: ignore ecodes = None # type: ignore EVDEV_AVAILABLE = False @@ -49,6 +58,23 @@ KEY_LEFTMETA = 125 # Super/Windows key KEY_RIGHTMETA = 126 DEVICE_RESCAN_SECONDS = 2.0 +# Event types re-emitted on the uinput clone. Feedback events (EV_LED, +# EV_SND, EV_FF) are kernel->device traffic already delivered to every +# capable device — re-injecting them could feed back into the LED layer. +if EVDEV_AVAILABLE: + _FORWARDED_EVENT_TYPES = frozenset( + { + ecodes.EV_KEY, + ecodes.EV_REL, + ecodes.EV_ABS, + ecodes.EV_MSC, + ecodes.EV_SW, + getattr(ecodes, "EV_ROT", -1), + } + ) +else: + _FORWARDED_EVENT_TYPES = frozenset() + # Map modifier key names to evdev key codes MODIFIER_KEY_CODES: dict[str, set[int]] = { "ctrl": {KEY_LEFTCTRL, KEY_RIGHTCTRL}, @@ -311,11 +337,25 @@ class EvdevKeyboardBackend(KeyboardBackend): self._combo_all_codes: set[int] = set() self._combo_pressed: set[int] = set() # currently-held combo-relevant codes self._combo_active = False # True while a push-to-talk combo hold is live + # fds whose combo main-key press was consumed but not yet released — + # per-device so one keyboard's release can't consume another's press. + self._combo_swallowed: set[int] = set() self._resolve_combo_targets() self._devices_lock = threading.Lock() self._dropped_devices: set[int] = set() # fds with SYN_DROPPED pending self._device_paths_by_fd: dict[int, str] = {} + # Grabbed fd -> uinput clone that re-emits events we do not consume. + self._forwarders: dict[int, UInput] = {} + # Grabbed fd -> its clone's /dev/input/eventN path, and the live set + # of clone paths. Clones are never monitored: their events are the + # ones we forwarded, so reading them would loop input back in. + self._forwarder_paths: dict[int, str] = {} + self._clone_paths: set[str] = set() + # Grabbed fd -> key codes the clone currently believes are held. + # Needed to reconcile after SYN_DROPPED eats release events. + self._forwarded_held: dict[int, set[int]] = {} + self._uinput_warned = False if not EVDEV_AVAILABLE: logger.error("python-evdev not available") @@ -336,6 +376,7 @@ class EvdevKeyboardBackend(KeyboardBackend): self._combo_all_codes = set() self._combo_pressed = set() self._combo_active = False + self._combo_swallowed = set() spec = getattr(self, "_spec", None) if spec is None or not spec.is_combo: @@ -374,6 +415,12 @@ class EvdevKeyboardBackend(KeyboardBackend): push-to-talk hold could stay stuck on. Clearing the pressed set fails safe (the user simply re-presses the modifier), and ending any live hold prevents a stuck session. + + ``_combo_swallowed`` is deliberately left alone: it records consumed + presses per device, so clearing it here would let the matching + release of a still-held key leak to the application without its + press. Stale entries are pruned per-device on removal and during + the SYN_DROPPED resync instead. """ self._combo_pressed = set() # Ends an active push-to-talk hold (fires the release callback); no-op @@ -486,6 +533,11 @@ class EvdevKeyboardBackend(KeyboardBackend): self.key_pressed_devices = set() self._dropped_devices = set() self._device_paths_by_fd = {} + self._forwarders = {} + self._forwarder_paths = {} + self._clone_paths = set() + self._forwarded_held = {} + self._uinput_warned = False # Refresh combo targets and clear any stale held-key state. self._resolve_combo_targets() @@ -520,20 +572,39 @@ class EvdevKeyboardBackend(KeyboardBackend): self.monitor_thread.join(timeout=2.0) self.monitor_thread = None - # Close devices + self._close_all_devices() + + def _close_all_devices(self) -> None: + """Close every monitored device and uinput clone. + + Closing a grabbed device releases its EVIOCGRAB, and closing a + uinput clone makes the kernel release any keys the clone still + holds — together they hand input delivery back to applications. + """ with self._devices_lock: devices = list(self.devices) + forwarders = list(self._forwarders.values()) self.devices = [] self.device_fds = [] self.device_paths = set() self._dropped_devices = set() self._device_paths_by_fd = {} + self._forwarders = {} + self._forwarder_paths = {} + self._clone_paths = set() + self._forwarded_held = {} + self._combo_swallowed = set() for device in devices: try: device.close() - except Exception: - pass + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring device close failure during cleanup: {e}") + for forwarder in forwarders: + try: + forwarder.close() + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring clone close failure during cleanup: {e}") def _open_keyboard_device(self, device_path: str) -> bool: """Open a keyboard device if it is not already monitored.""" @@ -548,22 +619,229 @@ class EvdevKeyboardBackend(KeyboardBackend): logger.warning(f"Cannot open {device_path}: {e}") return False + # Only Vocalinux's own uinput clones are never monitored — reading + # them would loop the events we forwarded straight back in. Clones + # are matched by their registered device path; the " (vocalinux)" + # name suffix also covers clones of other Vocalinux processes whose + # paths we never registered. Other virtual devices (key remappers, + # accessibility keyboards) ARE monitored: users can bind shortcuts + # 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)"): + logger.debug(f"Skipping Vocalinux clone device: {device_path} ({device_name})") + try: + device.close() + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring close failure for clone {device_path}: {e}") + return False + + # Exclusive grab hides the device from the compositor; the paired + # clone re-emits every event we do not consume so apps keep typing. + forwarder = self._create_forwarder(device) + if forwarder is not None: + try: + device.grab() + except (OSError, IOError) as e: + logger.warning( + f"Cannot grab {device_path} ({e}); " + "shortcut keys will also reach the focused app" + ) + try: + forwarder.close() + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring clone close failure for {device_path}: {e}") + forwarder = None + with self._devices_lock: if device_path in self.device_paths or fd in self.device_fds: try: device.close() - except Exception: - pass + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring close failure for duplicate {device_path}: {e}") + if forwarder is not None: + try: + forwarder.close() + except (OSError, IOError, RuntimeError) as e: + logger.debug( + f"Ignoring clone close failure for duplicate {device_path}: {e}" + ) return False self.devices.append(device) self.device_fds.append(fd) self.device_paths.add(device_path) self._device_paths_by_fd[fd] = device_path + if forwarder is not None: + self._forwarders[fd] = forwarder + clone_path = self._forwarder_device_path(forwarder) + if clone_path is not None: + self._forwarder_paths[fd] = clone_path + self._clone_paths.add(clone_path) + self._forwarded_held[fd] = set() - logger.debug(f"Opened keyboard device: {device_path} ({device.name})") + logger.debug( + f"Opened keyboard device: {device_path} ({device.name})" + f"{' [grabbed]' if forwarder is not None else ''}" + ) return True + @staticmethod + def _forwarder_device_path(forwarder: UInput) -> Optional[str]: + """Return the clone's /dev/input/eventN path, or None if unknowable. + + The path is what lets ``_open_keyboard_device`` recognize (and skip) + our clones even when the clone name was truncated to the uinput + 80-byte limit. + """ + try: + path = forwarder.device.path + except (AttributeError, OSError, IOError): + return None + return path if isinstance(path, str) else None + + def _create_forwarder(self, device: InputDevice) -> Optional[UInput]: + """Create a uinput clone of a keyboard for pass-through forwarding. + + EV_REP is filtered out so the clone never autorepeats: the physical + device's own repeat events are forwarded verbatim, preserving its + repeat rate without doubling repeats. + """ + try: + capabilities: dict[int, Sequence[int]] = { + k: v for k, v in device.capabilities().items() + } + 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, + ) + except (OSError, IOError, TypeError, ValueError, UInputError) as e: + if not self._uinput_warned: + self._uinput_warned = True + logger.warning( + "Cannot create a virtual keyboard for key suppression " + f"(/dev/uinput not writable?): {e}. The dictation shortcut " + "will also reach the focused application." + ) + return None + + def _release_failed_forwarder(self, fd: int) -> None: + """Drop a dead uinput clone and release the source device's grab. + + A keyboard that stays grabbed while its clone can no longer accept + events swallows all user input; releasing EVIOCGRAB hands delivery + back to the kernel so applications see keys again (the shortcut can + then no longer be suppressed, matching the ungrabbed fallback). If + the grab itself cannot be released, the device is removed outright + — a grabbed device with a dead clone is worse than no listener. + """ + with self._devices_lock: + device = next((d for d in self.devices if d.fileno() == fd), None) + forwarder = self._forwarders.pop(fd, None) + clone_path = self._forwarder_paths.pop(fd, None) + if clone_path is not None: + self._clone_paths.discard(clone_path) + self._forwarded_held.pop(fd, None) + + if forwarder is not None: + try: + forwarder.close() + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring clone close failure for fd {fd}: {e}") + if device is None: + return + try: + device.ungrab() + except (OSError, IOError) as e: + logger.error( + f"Cannot release grab on fd {fd} ({e}); " + "removing the device so it stops swallowing input" + ) + self._remove_keyboard_device(fd, device) + return + logger.warning( + f"Released grab on {getattr(device, 'name', 'device')} (fd={fd}) " + "after a forwarding failure; keys pass through to applications " + "but the dictation shortcut can no longer be suppressed" + ) + + def _forward_event(self, fd: int, event: InputEvent) -> None: + """Re-emit an event on the uinput clone paired with a grabbed device. + + Successful EV_KEY writes also update ``_forwarded_held`` so + ``_resync_clone_key_state`` can tell which keys the clone still + believes are held after a SYN_DROPPED burst. + """ + if event.type != ecodes.EV_SYN and event.type not in _FORWARDED_EVENT_TYPES: + return + forwarder = self._forwarders.get(fd) + if forwarder is None: + return + try: + forwarder.write_event(event) + except (OSError, IOError) as e: + logger.error(f"Failed to forward event on fd {fd}: {e}") + self._release_failed_forwarder(fd) + return + if event.type == ecodes.EV_KEY: + held = self._forwarded_held.get(fd) + if held is not None: + if event.value == 0: + held.discard(event.code) + else: + held.add(event.code) + + def _event_is_shortcut(self, fd: int, event: InputEvent) -> bool: + """Return True if an EV_KEY event belongs to the dictation shortcut. + + Consumed events are handled locally and never re-emitted, so the + focused application never sees them: + + - Pure-modifier gestures (``right_alt+right_alt``, ``ctrl+ctrl``): + every event of the configured modifier side(s) is consumed — the + modifier is reserved for Vocalinux while the listener runs. + - Bare function-key shortcuts (``f11``): every event of that key. + - Modifier+key combos (``alt+r``): the main key is consumed only + while the combo owns it (press with all modifiers held, plus the + matching repeat/release). Required modifiers and unrelated keys + pass through, so typing and chords like Alt+Tab keep working. + + Consumption of a combo main-key press is tracked per device fd: + when two keyboards both press the main key while the modifiers are + held, each device's release must still find its own consumed press + — a shared flag would let the first release expose the second + keyboard's release without its press. Stale entries are pruned on + device removal and by the post-SYN_DROPPED resync. + """ + if event.type != ecodes.EV_KEY: + return False + spec = getattr(self, "_spec", None) + if spec is None: + return False + + if not spec.is_combo: + return event.code in self._get_target_key_codes() + + main_code = self._combo_main_code + if event.code != main_code or main_code is None: + return False + if not spec.modifiers: + return True # bare function key + if event.value == 1 and self._required_modifiers_held(): + self._combo_swallowed.add(fd) + return True + if event.value == 0: + # Consume the release of a swallowed press on THIS device exactly + # once, so apps never see a press/release pair we half-forwarded. + if fd not in self._combo_swallowed: + return False + self._combo_swallowed.discard(fd) + return True + return fd in self._combo_swallowed + def _scan_for_new_devices(self) -> int: """Find and open keyboard devices that appeared after startup.""" new_device_count = 0 @@ -587,8 +865,8 @@ class EvdevKeyboardBackend(KeyboardBackend): """Close and forget a disconnected keyboard device.""" try: device.close() - except Exception: - pass + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring close failure for disconnected fd {fd}: {e}") with self._devices_lock: try: @@ -604,94 +882,182 @@ class EvdevKeyboardBackend(KeyboardBackend): self.device_paths.discard(device_path) self._dropped_devices.discard(fd) self.key_pressed_devices.discard(id(device)) - # A disconnect can swallow the modifier release (e.g. a wireless - # split half dropping mid-hold); don't leave the combo logically held. - self._reset_combo_state() + forwarder = self._forwarders.pop(fd, None) + clone_path = self._forwarder_paths.pop(fd, None) + if clone_path is not None: + self._clone_paths.discard(clone_path) + self._forwarded_held.pop(fd, None) + # The fd can be recycled for a newly plugged device; a stale + # swallowed press would then eat that device's first release. + self._combo_swallowed.discard(fd) + + if forwarder is not None: + try: + forwarder.close() + except (OSError, IOError, RuntimeError) as e: + logger.debug(f"Ignoring clone close failure for disconnected fd {fd}: {e}") + # A disconnect can swallow the modifier release (e.g. a wireless + # split half dropping mid-hold); don't leave the combo logically held. + self._reset_combo_state() + + def _resync_clone_key_state(self, fd: int, device: InputDevice) -> None: + """Release clone-side keys whose releases were lost to SYN_DROPPED. + + The kernel's dropped-event burst may have skipped release events, + but the paired clone only saw what we forwarded — a forwarded press + whose release was dropped leaves the key held on the clone forever, + and no later event supplies the missing release. The burst can also + eat the release of a swallowed combo press, which the clone never + saw at all. Compare what the clone believes (``_forwarded_held``) + and what we paired (``_combo_swallowed``) against the device's live + kernel key state and emit releases for the phantom keys. + """ + forwarder = self._forwarders.get(fd) + held = self._forwarded_held.get(fd) + if (forwarder is None or not held) and fd not in self._combo_swallowed: + return + try: + actually_held = set(device.active_keys()) + except (OSError, IOError) as e: + logger.warning(f"Cannot resync key state for fd {fd}: {e}") + return + if forwarder is not None and held: + for code in sorted(held - actually_held): + try: + forwarder.write(ecodes.EV_KEY, code, 0) + except (OSError, IOError) as e: + logger.error(f"Failed to release stuck key {code} on fd {fd}: {e}") + self._release_failed_forwarder(fd) + return + held.discard(code) + if self._combo_main_code is not None and self._combo_main_code not in actually_held: + # A dropped burst can also eat the release of a swallowed combo + # press — a key the clone never saw, so ``held`` cannot reflect + # it. Without this prune the next ordinary press of that key + # would be consumed too, eating a keystroke. + self._combo_swallowed.discard(fd) def _monitor_devices(self) -> None: """Monitor keyboard devices for events.""" logger.debug("Starting device monitor thread") last_scan = time.monotonic() - while self.running: - try: - now = time.monotonic() - if now - last_scan >= DEVICE_RESCAN_SECONDS: - self._scan_for_new_devices() - last_scan = now + try: + while self.running: + try: + now = time.monotonic() + if now - last_scan >= DEVICE_RESCAN_SECONDS: + self._scan_for_new_devices() + last_scan = now - # Use select to wait for events on any device - with self._devices_lock: - device_fds = list(self.device_fds) + # Use select to wait for events on any device + with self._devices_lock: + device_fds = list(self.device_fds) - if not device_fds: - time.sleep(1.0) - continue - - readable, _, _ = select.select(device_fds, [], [], 1.0) # 1 second timeout - - for fd in readable: - try: - # Find the device for this fd - with self._devices_lock: - device = None - for d in self.devices: - if d.fileno() == fd: - device = d - break - - if device is None: - continue - - # Read events from this device - for event in device.read(): - if event.type == ecodes.EV_SYN: - if event.code == ecodes.SYN_DROPPED: - # Kernel buffer overflowed — discard until SYN_REPORT - self._dropped_devices.add(fd) - logger.warning( - f"SYN_DROPPED on {device.name} (fd={fd}), " - "resetting key state" - ) - elif event.code == ecodes.SYN_REPORT: - if fd in self._dropped_devices: - # End of dropped sequence — clear stale state - self._dropped_devices.discard(fd) - self.key_pressed_devices.discard(id(device)) - # A dropped modifier release must not leave - # the combo logically held. - self._reset_combo_state() - continue - if fd in self._dropped_devices: - continue - if event.type == ecodes.EV_KEY: - self._handle_key_event(event, device) - - except (OSError, IOError) as e: - # Device was disconnected - remove it to avoid busy loop - device_name = ( - device.name if device and hasattr(device, "name") else "unknown" - ) - logger.info(f"Device disconnected: {device_name} (fd={fd})") - if device is not None: - self._remove_keyboard_device(fd, device) + if not device_fds: + time.sleep(1.0) continue - except (OSError, ValueError) as e: - if self.running: - logger.error(f"Error monitoring devices: {e}") - break + readable, _, _ = select.select(device_fds, [], [], 1.0) # 1 second timeout + + for fd in readable: + try: + # Find the device for this fd + with self._devices_lock: + device = None + for d in self.devices: + if d.fileno() == fd: + device = d + break + + if device is None: + continue + + # Read events from this device. On grabbed devices, + # everything we do not consume is re-emitted on the + # paired uinput clone so the focused app keeps typing. + for event in device.read(): + if event.type == ecodes.EV_SYN: + if event.code == ecodes.SYN_DROPPED: + # Kernel buffer overflowed — discard until SYN_REPORT + self._dropped_devices.add(fd) + logger.warning( + f"SYN_DROPPED on {device.name} (fd={fd}), " + "resetting key state" + ) + elif event.code == ecodes.SYN_REPORT: + if fd in self._dropped_devices: + # End of dropped sequence — clear stale state + self._dropped_devices.discard(fd) + self.key_pressed_devices.discard(id(device)) + # Release keys the clone still thinks + # are held before the SYN_REPORT + # reaches it, atomically. + self._resync_clone_key_state(fd, device) + # A dropped modifier release must not leave + # the combo logically held. + self._reset_combo_state() + self._forward_event(fd, event) + continue + if fd in self._dropped_devices: + # Handling shortcut state mid-drop is unsafe, + # but surviving non-shortcut events still pass + # through so the app sees what the kernel kept. + if not self._event_is_shortcut(fd, event): + self._forward_event(fd, event) + continue + if event.type == ecodes.EV_KEY: + # Every key event updates shortcut state (combo + # modifiers must be tracked AND forwarded); only + # the forwarding decision differs. + consumed = self._event_is_shortcut(fd, event) + self._handle_key_event(event, device) + if not consumed: + self._forward_event(fd, event) + else: + self._forward_event(fd, event) + + except (OSError, IOError) as e: + # Device was disconnected - remove it to avoid busy loop + device_name = ( + device.name if device and hasattr(device, "name") else "unknown" + ) + logger.info(f"Device disconnected: {device_name} (fd={fd})") + if device is not None: + self._remove_keyboard_device(fd, device) + continue + + except (OSError, ValueError) as e: + if self.running: + logger.error(f"Error monitoring devices: {e}") + break + finally: + if self.running: + # Exiting while still "running" means the loop died + # unexpectedly — a failed select, an fd closed underneath + # it, or an error escaping the handler. Nothing will ever + # read the grabbed keyboards again, so close every device + # and clone: closing releases each grab and hands input + # delivery back to the kernel instead of leaving the + # user's keyboards dead. + logger.error( + "Keyboard monitor exited unexpectedly; closing devices " + "to release grabs so keyboards keep working" + ) + self.running = False + self.active = False + self._close_all_devices() logger.debug("Device monitor thread stopped") - def _handle_key_event(self, event, device) -> None: + def _handle_key_event(self, event: InputEvent, device: InputDevice) -> None: """Handle a key event from evdev.""" if getattr(self, "_spec", None) is not None and self._spec.is_combo: self._handle_combo_key_event(event) return self._handle_modifier_key_event(event, device) - def _handle_combo_key_event(self, event) -> None: + def _handle_combo_key_event(self, event: InputEvent) -> None: """Handle a key event for a modifier+key combo (e.g. Alt+R). Combo state is tracked globally across all keyboard devices, which is @@ -748,7 +1114,7 @@ class EvdevKeyboardBackend(KeyboardBackend): logger.debug(f"Combo {self._shortcut} released (evdev)") threading.Thread(target=self.key_release_callback, daemon=True).start() - def _handle_modifier_key_event(self, event, device) -> None: + def _handle_modifier_key_event(self, event: InputEvent, device: InputDevice) -> None: """Handle a key event for a single-modifier gesture (double-tap / hold).""" try: code = event.code diff --git a/tests/test_evdev_backend_ext.py b/tests/test_evdev_backend_ext.py index 46613d1..c95f9b7 100644 --- a/tests/test_evdev_backend_ext.py +++ b/tests/test_evdev_backend_ext.py @@ -26,6 +26,7 @@ from vocalinux.ui.keyboard_backends.evdev_backend import ( EvdevKeyboardBackend, device_has_key, device_has_modifier_key, + ecodes, find_keyboard_devices, ) @@ -1051,3 +1052,592 @@ class TestEvdevKeyboardBackendGetTargetKeyCodes: assert 42 in result # KEY_LEFTSHIFT assert 54 in result # KEY_RIGHTSHIFT + + +class TestEvdevGrabAndForwarding: + """Test EVIOCGRAB + uinput pass-through that consumes shortcut keys (#871).""" + + def _key_event(self, code: int, value: int = 1) -> MagicMock: + return MagicMock(type=1, code=code, value=value) # EV_KEY + + def _syn_event(self) -> MagicMock: + return MagicMock(type=0, code=0) # EV_SYN / SYN_REPORT + + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_grabs_device_and_registers_forwarder(self, mock_input_device, mock_uinput): + """A keyboard that can be cloned is grabbed and paired with a forwarder.""" + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x03 # BUS_USB + mock_input_device.return_value = device + forwarder = MagicMock() + mock_uinput.return_value = forwarder + + backend = EvdevKeyboardBackend() + opened = backend._open_keyboard_device("/dev/input/event0") + + assert opened is True + mock_uinput.assert_called_once() + device.grab.assert_called_once() + assert backend._forwarders[10] is forwarder + assert device in backend.devices + + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_monitors_other_virtual_devices(self, mock_input_device, mock_uinput): + """Remapper/accessibility virtual devices are monitored like real ones. + + A user can record a shortcut on a keyd/kmonad-style virtual keyboard; + ignoring every BUS_VIRTUAL device would leave that shortcut dead. + """ + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x06 # BUS_VIRTUAL + device.name = "keyd virtual keyboard" + mock_input_device.return_value = device + forwarder = MagicMock() + forwarder.device.path = "/dev/input/event20" + mock_uinput.return_value = forwarder + + backend = EvdevKeyboardBackend() + opened = backend._open_keyboard_device("/dev/input/event5") + + assert opened is True + device.grab.assert_called_once() + assert device in backend.devices + + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_skips_own_clone_by_name(self, mock_input_device, mock_uinput): + """Our clones are skipped by name so forwarded events never loop back.""" + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x06 # BUS_VIRTUAL + device.name = "AT Translated Set 2 keyboard (vocalinux)" + mock_input_device.return_value = device + + backend = EvdevKeyboardBackend() + opened = backend._open_keyboard_device("/dev/input/event5") + + assert opened is False + device.close.assert_called_once() + mock_uinput.assert_not_called() + device.grab.assert_not_called() + assert backend.devices == [] + + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_skips_own_clone_by_path(self, mock_input_device, mock_uinput): + """A clone whose name was truncated is still skipped via its path.""" + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x06 + device.name = "AT Translated Set 2 keyboard (vocal" # truncated name + mock_input_device.return_value = device + + backend = EvdevKeyboardBackend() + backend._clone_paths.add("/dev/input/event20") + + opened = backend._open_keyboard_device("/dev/input/event20") + + assert opened is False + device.close.assert_called_once() + mock_uinput.assert_not_called() + device.grab.assert_not_called() + assert backend.devices == [] + + @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): + """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") + + backend = EvdevKeyboardBackend() + opened = backend._open_keyboard_device("/dev/input/event0") + + assert opened is True + device.grab.assert_not_called() + assert backend._forwarders == {} + assert device in backend.devices + + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_falls_back_when_grab_fails(self, mock_input_device, mock_uinput): + """A failed grab drops the forwarder but keeps the listener running.""" + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x03 + device.grab.side_effect = OSError(errno.EBUSY, "device grabbed") + mock_input_device.return_value = device + forwarder = MagicMock() + mock_uinput.return_value = forwarder + + backend = EvdevKeyboardBackend() + opened = backend._open_keyboard_device("/dev/input/event0") + + assert opened is True + forwarder.close.assert_called_once() + assert backend._forwarders == {} + assert device in backend.devices + + def test_event_is_shortcut_pure_modifier(self) -> None: + """The configured modifier is consumed entirely; other keys are not.""" + backend = EvdevKeyboardBackend(shortcut="right_alt+right_alt") # KEY_RIGHTALT=100 + + assert backend._event_is_shortcut(10, self._key_event(100, 1)) is True + assert backend._event_is_shortcut(10, self._key_event(100, 2)) is True + assert backend._event_is_shortcut(10, self._key_event(100, 0)) is True + assert backend._event_is_shortcut(10, self._key_event(56, 1)) is False # left alt + assert backend._event_is_shortcut(10, self._key_event(30, 1)) is False # KEY_A + assert backend._event_is_shortcut(10, self._syn_event()) is False + + def test_event_is_shortcut_bare_function_key(self) -> None: + """A bare function-key shortcut consumes only that key.""" + backend = EvdevKeyboardBackend(shortcut="f11") # KEY_F11=87 + + assert backend._event_is_shortcut(10, self._key_event(87, 1)) is True + assert backend._event_is_shortcut(10, self._key_event(87, 0)) is True + assert backend._event_is_shortcut(10, self._key_event(88, 1)) is False # KEY_F12 + assert backend._event_is_shortcut(10, self._key_event(30, 1)) is False + + def test_event_is_shortcut_combo_only_while_engaged(self) -> None: + """alt+r: the main key is consumed only while the modifier is held.""" + backend = EvdevKeyboardBackend(shortcut="alt+r") + key_r = backend._combo_main_code + assert key_r == 19 # KEY_R + + backend._combo_pressed = {56} # KEY_LEFTALT held + assert backend._event_is_shortcut(10, self._key_event(key_r, 1)) is True + # Matching repeat and release stay consumed while the combo owns the + # key — even if the modifier was already released first. + backend._combo_pressed = set() + assert backend._event_is_shortcut(10, self._key_event(key_r, 2)) is True + assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is True + assert backend._combo_swallowed == set() + + # 'r' typed without alt still reaches apps; the modifier always passes + assert backend._event_is_shortcut(10, self._key_event(key_r, 1)) is False + assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is False + assert backend._event_is_shortcut(10, self._key_event(56, 1)) is False + assert backend._event_is_shortcut(10, self._key_event(100, 1)) is False + + def test_event_is_shortcut_combo_tracks_presses_per_device(self) -> None: + """Two keyboards pressing the combo main key each pair their release.""" + backend = EvdevKeyboardBackend(shortcut="alt+r") + key_r = backend._combo_main_code + backend._combo_pressed = {56} # KEY_LEFTALT held + + # Both keyboards press the main key; both presses are consumed. + assert backend._event_is_shortcut(10, self._key_event(key_r, 1)) is True + assert backend._event_is_shortcut(11, self._key_event(key_r, 1)) is True + + # The first release must not clear the second keyboard's swallowed + # press — otherwise its release would reach the app without a press. + assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is True + assert backend._event_is_shortcut(11, self._key_event(key_r, 0)) is True + assert backend._combo_swallowed == set() + + def test_reset_combo_state_keeps_other_devices_swallowed(self) -> None: + """A dropped-event reset on one device can't orphan another's press.""" + backend = EvdevKeyboardBackend(shortcut="alt+r") + backend._combo_swallowed = {10, 11} + + backend._reset_combo_state() + + assert backend._combo_swallowed == {10, 11} + + def test_forward_event_filters_feedback_types(self): + """Only real input events and SYN are re-emitted — never EV_LED.""" + backend = EvdevKeyboardBackend() + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + + backend._forward_event(10, self._key_event(30, 1)) + backend._forward_event(10, self._syn_event()) + backend._forward_event(10, MagicMock(type=0x11, code=0, value=1)) # EV_LED + + assert forwarder.write_event.call_count == 2 + + def test_forward_event_noop_without_forwarder(self): + """Ungrabbed devices have no forwarder; nothing is written.""" + backend = EvdevKeyboardBackend() + backend._forward_event(10, self._key_event(30, 1)) # no crash, no write + + def test_monitor_consumes_shortcut_key_but_forwards_syn(self): + """Grabbed device: the shortcut press is handled locally, not re-emitted.""" + backend = EvdevKeyboardBackend() # default right_alt push-to-talk + device = MagicMock() + device.fileno.return_value = 10 + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + key_event = self._key_event(100, 1) # KEY_RIGHTALT press + syn_event = self._syn_event() + device.read.side_effect = lambda: setattr(backend, "running", False) or [ + key_event, + syn_event, + ] + backend.running = True + backend.devices = [device] + backend.device_fds = [10] + backend._handle_key_event = MagicMock() + + with ( + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic", + side_effect=[0.0, 0.0], + ), + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.select.select", + return_value=([10], [], []), + ), + patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes, + ): + mock_ecodes.EV_SYN = 0 + mock_ecodes.SYN_DROPPED = 3 + mock_ecodes.SYN_REPORT = 0 + mock_ecodes.EV_KEY = 1 + backend._monitor_devices() + + backend._handle_key_event.assert_called_once_with(key_event, device) + forwarder.write_event.assert_called_once_with(syn_event) + + def test_monitor_forwards_non_shortcut_key(self): + """Grabbed device: an unrelated key is re-emitted, not consumed.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + key_event = self._key_event(30, 1) # KEY_A + syn_event = self._syn_event() + device.read.side_effect = lambda: setattr(backend, "running", False) or [ + key_event, + syn_event, + ] + backend.running = True + backend.devices = [device] + backend.device_fds = [10] + backend._handle_key_event = MagicMock() + + with ( + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic", + side_effect=[0.0, 0.0], + ), + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.select.select", + return_value=([10], [], []), + ), + patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes, + ): + mock_ecodes.EV_SYN = 0 + mock_ecodes.SYN_DROPPED = 3 + mock_ecodes.SYN_REPORT = 0 + mock_ecodes.EV_KEY = 1 + backend._monitor_devices() + + backend._handle_key_event.assert_called_once_with(key_event, device) + assert forwarder.write_event.call_count == 2 + + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.find_keyboard_devices") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_stop_closes_forwarders(self, mock_input_device, mock_find_devices, mock_uinput): + """stop() releases grabs (via close) and closes every uinput clone.""" + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x03 + mock_input_device.return_value = device + forwarder = MagicMock() + mock_uinput.return_value = forwarder + mock_find_devices.return_value = ["/dev/input/event0"] + + backend = EvdevKeyboardBackend() + backend.start() + backend.stop() + + forwarder.close.assert_called_once() + assert backend._forwarders == {} + + def test_remove_device_closes_paired_forwarder(self): + """Disconnecting a keyboard closes its uinput clone too.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + forwarder = MagicMock() + backend.devices = [device] + backend.device_fds = [10] + backend.device_paths = {"/dev/input/event0"} + backend._device_paths_by_fd = {10: "/dev/input/event0"} + backend._forwarders = {10: forwarder} + + backend._remove_keyboard_device(10, device) + + forwarder.close.assert_called_once() + assert backend._forwarders == {} + + def test_event_is_shortcut_without_spec(self) -> None: + """No parsed spec means nothing is consumed.""" + backend = EvdevKeyboardBackend() + backend._spec = None + + assert backend._event_is_shortcut(10, self._key_event(100, 1)) is False + + def test_event_is_shortcut_combo_edges(self) -> None: + """Combo consumption only ever applies to the main key.""" + backend = EvdevKeyboardBackend(shortcut="alt+r") + key_r = backend._combo_main_code + + # Modifier and unrelated keys are never consumed, held or not. + backend._combo_pressed = {56} + assert backend._event_is_shortcut(10, self._key_event(56, 1)) is False + assert backend._event_is_shortcut(10, self._key_event(30, 1)) is False + # An unowned repeat is forwarded like any other event. + assert backend._event_is_shortcut(10, self._key_event(key_r, 2)) is False + # A release without a swallowed press is forwarded too. + assert backend._event_is_shortcut(10, self._key_event(key_r, 0)) is False + + # No resolvable main key -> nothing consumed. + backend._combo_main_code = None + backend._combo_pressed = {56} + assert backend._event_is_shortcut(10, self._key_event(19, 1)) is False + + def test_forward_event_logs_write_failure(self) -> None: + """A dead uinput clone logs the failure instead of crashing the loop.""" + backend = EvdevKeyboardBackend() + forwarder = MagicMock() + forwarder.write_event.side_effect = OSError("gone") + backend._forwarders = {10: forwarder} + + with patch("vocalinux.ui.keyboard_backends.evdev_backend.logger") as mock_logger: + backend._forward_event(10, self._key_event(30, 1)) + + mock_logger.error.assert_called_once() + assert backend._forwarders == {} + forwarder.close.assert_called_once() + + def test_forward_failure_releases_grab(self) -> None: + """A write failure on the clone releases EVIOCGRAB so keys reach apps.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + device.name = "Keyboard" + forwarder = MagicMock() + forwarder.write_event.side_effect = OSError("uinput gone") + backend.devices = [device] + backend.device_fds = [10] + backend._forwarders = {10: forwarder} + backend._forwarded_held = {10: set()} + + backend._forward_event(10, self._key_event(30, 1)) + + device.ungrab.assert_called_once() + forwarder.close.assert_called_once() + assert backend._forwarders == {} + assert 10 not in backend._forwarded_held + # The device stays monitored, ungrabbed — the shortcut still detects. + assert device in backend.devices + + def test_forward_failure_removes_device_when_ungrab_fails(self) -> None: + """If the grab cannot be released, the device is closed outright.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + device.name = "Keyboard" + device.ungrab.side_effect = OSError("ungrab failed") + forwarder = MagicMock() + forwarder.write_event.side_effect = OSError("uinput gone") + backend.devices = [device] + backend.device_fds = [10] + backend.device_paths = {"/dev/input/event0"} + backend._device_paths_by_fd = {10: "/dev/input/event0"} + backend._forwarders = {10: forwarder} + backend._forwarded_held = {10: set()} + + backend._forward_event(10, self._key_event(30, 1)) + + # Closed rather than left grabbed with a dead clone. + device.close.assert_called_once() + assert backend.devices == [] + assert backend._forwarders == {} + assert backend.device_paths == set() + + def test_forward_event_tracks_held_keys(self) -> None: + """Forwarded presses/releases keep the clone's believed key state.""" + backend = EvdevKeyboardBackend() + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + backend._forwarded_held = {10: set()} + + backend._forward_event(10, self._key_event(30, 1)) + backend._forward_event(10, self._key_event(56, 1)) + assert backend._forwarded_held[10] == {30, 56} + + backend._forward_event(10, self._key_event(30, 0)) + assert backend._forwarded_held[10] == {56} + + def test_resync_releases_phantom_keys_after_syn_dropped(self) -> None: + """A dropped release is rewritten to the clone from live key state.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + device.active_keys.return_value = [] # nothing actually held + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + backend._forwarded_held = {10: {30, 56}} + + backend._resync_clone_key_state(10, device) + + forwarder.write.assert_any_call(ecodes.EV_KEY, 30, 0) + forwarder.write.assert_any_call(ecodes.EV_KEY, 56, 0) + assert forwarder.write.call_count == 2 + assert backend._forwarded_held[10] == set() + + def test_resync_keeps_still_held_keys(self) -> None: + """Keys still physically held are not released on the clone.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + device.active_keys.return_value = [30, 56] + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + backend._forwarded_held = {10: {30, 56}} + + backend._resync_clone_key_state(10, device) + + forwarder.write.assert_not_called() + assert backend._forwarded_held[10] == {30, 56} + + def test_resync_prunes_swallowed_combo_when_clone_held_nothing(self) -> None: + """A dropped combo release is reconciled even when the clone holds no keys. + + The swallowed press was never forwarded, so ``_forwarded_held`` is + empty — but its dropped release still needs the pairing pruned or + the next ordinary press of that key is consumed too. + """ + backend = EvdevKeyboardBackend() + backend._combo_main_code = 30 + device = MagicMock() + device.fileno.return_value = 10 + device.active_keys.return_value = [] # combo key physically up + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + backend._forwarded_held = {10: set()} + backend._combo_swallowed = {10} + + backend._resync_clone_key_state(10, device) + + assert backend._combo_swallowed == set() + forwarder.write.assert_not_called() + + def test_resync_keeps_swallowed_combo_while_key_still_held(self) -> None: + """A swallowed combo press stays paired while the key is physically down.""" + backend = EvdevKeyboardBackend() + backend._combo_main_code = 30 + device = MagicMock() + device.fileno.return_value = 10 + device.active_keys.return_value = [30] + backend._combo_swallowed = {10} + + backend._resync_clone_key_state(10, device) + + assert backend._combo_swallowed == {10} + + def test_monitor_exit_releases_grabs_on_select_failure(self) -> None: + """An unexpected monitor exit closes devices, releasing every grab.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + forwarder = MagicMock() + backend.running = True + backend.active = True + backend.devices = [device] + backend.device_fds = [10] + backend.device_paths = {"/dev/input/event0"} + backend._device_paths_by_fd = {10: "/dev/input/event0"} + backend._forwarders = {10: forwarder} + + with ( + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic", + side_effect=[0.0, 0.0], + ), + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.select.select", + side_effect=OSError("select failed"), + ), + ): + backend._monitor_devices() + + device.close.assert_called_once() + forwarder.close.assert_called_once() + assert backend.devices == [] + assert backend._forwarders == {} + assert backend.running is False + assert backend.active is False + + @patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput") + @patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice") + def test_open_device_race_closes_device_and_forwarder(self, mock_input_device, mock_uinput): + """A device that loses the registration race is cleaned up fully.""" + device = MagicMock() + device.fileno.return_value = 10 + device.info.bustype = 0x03 + mock_input_device.return_value = device + forwarder = MagicMock() + mock_uinput.return_value = forwarder + + backend = EvdevKeyboardBackend() + backend.device_fds = [10] # same fd won by another open + + opened = backend._open_keyboard_device("/dev/input/event7") + + assert opened is False + device.close.assert_called_once() + forwarder.close.assert_called_once() + assert backend.devices == [] + assert backend._forwarders == {} + + def test_monitor_forwards_survivors_during_syn_dropped(self): + """Mid-drop: shortcut keys are eaten, surviving other keys forwarded.""" + backend = EvdevKeyboardBackend() + device = MagicMock() + device.fileno.return_value = 10 + forwarder = MagicMock() + backend._forwarders = {10: forwarder} + backend._dropped_devices = {10} + shortcut_event = self._key_event(100, 1) # KEY_RIGHTALT press + other_event = self._key_event(30, 1) # KEY_A + device.read.side_effect = lambda: setattr(backend, "running", False) or [ + shortcut_event, + other_event, + ] + backend.running = True + backend.devices = [device] + backend.device_fds = [10] + backend._handle_key_event = MagicMock() + + with ( + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.time.monotonic", + side_effect=[0.0, 0.0], + ), + patch( + "vocalinux.ui.keyboard_backends.evdev_backend.select.select", + return_value=([10], [], []), + ), + patch("vocalinux.ui.keyboard_backends.evdev_backend.ecodes") as mock_ecodes, + ): + mock_ecodes.EV_SYN = 0 + mock_ecodes.SYN_DROPPED = 3 + mock_ecodes.SYN_REPORT = 0 + mock_ecodes.EV_KEY = 1 + backend._monitor_devices() + + backend._handle_key_event.assert_not_called() + forwarder.write_event.assert_called_once_with(other_event)