fix(injection): release ydotool Ctrl after a timed-out paste (#675)
subprocess.run(..., timeout=3) SIGKILLs the ydotool client. ydotoold keeps any keys it already applied, so virtual Left Ctrl can stay held until logout. Send a best-effort V-up/Ctrl-up (or a legacy ctrl tap) after paste failure. Fixes #658
This commit is contained in:
@@ -1433,20 +1433,25 @@ class TextInjector:
|
||||
# and the evdev code that produces Latin 'v' on the active layout —
|
||||
# KEY_V=47 only on QWERTY). Passing 1.x codes to 0.1.x does not
|
||||
# paste; it types garbage (e.g. "2442").
|
||||
paste_cmd: list = []
|
||||
try:
|
||||
use_terminal_paste = self._should_use_terminal_paste()
|
||||
cmd = self._clipboard_paste_command(terminal=use_terminal_paste)
|
||||
paste_cmd = self._clipboard_paste_command(terminal=use_terminal_paste)
|
||||
logger.debug(
|
||||
"Simulating %s paste with: %s",
|
||||
"terminal" if use_terminal_paste else "standard",
|
||||
cmd,
|
||||
paste_cmd,
|
||||
)
|
||||
subprocess.run(
|
||||
cmd, check=True, stderr=subprocess.PIPE, text=True, timeout=3, env=host_env()
|
||||
paste_cmd, check=True, stderr=subprocess.PIPE, text=True, timeout=3, env=host_env()
|
||||
)
|
||||
logger.info(f"Text injected via clipboard paste: '{text[:20]}...' ({len(text)} chars)")
|
||||
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
|
||||
logger.warning(f"Paste simulation failed: {e}")
|
||||
# timeout=3 SIGKILLs the ydotool client. ydotoold keeps any keys it
|
||||
# already applied, so a virtual Ctrl can stay held until logout (#658).
|
||||
if paste_cmd:
|
||||
self._ydotool_release_paste_keys(paste_cmd)
|
||||
with self._state_lock:
|
||||
if generation == self._clipboard_restore_generation:
|
||||
self._clipboard_restore_target = None
|
||||
@@ -1745,6 +1750,63 @@ class TextInjector:
|
||||
setattr(self, cache_attr, cmd)
|
||||
return list(cmd)
|
||||
|
||||
def _ydotool_paste_release_command(self, paste_cmd: list) -> list:
|
||||
"""Return argv that lifts keys held by a ydotool paste chord.
|
||||
|
||||
wtype chords are ignored: they do not go through ydotoold, so a killed
|
||||
client cannot leave a virtual modifier down.
|
||||
"""
|
||||
if not paste_cmd or paste_cmd[0] != "ydotool":
|
||||
return []
|
||||
|
||||
pressed: list = []
|
||||
for token in paste_cmd:
|
||||
if isinstance(token, str) and token.endswith(":1"):
|
||||
code = token[:-2]
|
||||
if code:
|
||||
pressed.append(code)
|
||||
if pressed:
|
||||
# Reverse press order so the letter lifts before Shift/Ctrl.
|
||||
return ["ydotool", "key"] + [f"{code}:0" for code in reversed(pressed)]
|
||||
|
||||
token = ""
|
||||
for part in reversed(paste_cmd):
|
||||
if isinstance(part, str) and part not in ("ydotool", "key"):
|
||||
token = part.lower()
|
||||
break
|
||||
has_ctrl = "ctrl" in token
|
||||
has_shift = "shift" in token
|
||||
if has_ctrl and has_shift:
|
||||
# 0.1.x has no keycode:value form. A named tap is down+up; if the
|
||||
# modifier is already down, the up half should clear it.
|
||||
return ["ydotool", "key", "ctrl+shift"]
|
||||
if has_shift:
|
||||
return ["ydotool", "key", "shift"]
|
||||
if has_ctrl:
|
||||
return ["ydotool", "key", "ctrl"]
|
||||
return []
|
||||
|
||||
def _ydotool_release_paste_keys(self, paste_cmd: list) -> None:
|
||||
"""Best-effort Ctrl/Shift/letter up after a failed ydotool paste (#658).
|
||||
|
||||
Releasing an already-up key is a no-op. If ydotoold is still wedged the
|
||||
follow-up may also time out; we swallow that so paste failure stays False.
|
||||
"""
|
||||
release = self._ydotool_paste_release_command(paste_cmd)
|
||||
if not release:
|
||||
return
|
||||
try:
|
||||
subprocess.run(
|
||||
release,
|
||||
check=False,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=1,
|
||||
env=host_env(),
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired, subprocess.CalledProcessError):
|
||||
logger.debug("ydotool paste-key release did not complete")
|
||||
|
||||
# evdev keycodes for modifier keys. If any of these is still physically held
|
||||
# when a Wayland injection fires, the injected keystrokes are modified: a
|
||||
# held Alt turns the Ctrl+V paste into Ctrl+Alt+V (nothing pastes), and a
|
||||
|
||||
@@ -857,6 +857,153 @@ class TestTextInjector(unittest.TestCase):
|
||||
cmd = injector._ydotool_ctrl_v_command()
|
||||
self.assertEqual(cmd, ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"])
|
||||
|
||||
def _minimal_paste_injector(self) -> TextInjector:
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
injector._state_lock = threading.Lock()
|
||||
injector._clipboard_restore_generation = 0
|
||||
injector._clipboard_restore_target = None
|
||||
injector.wayland_tool = "ydotool"
|
||||
injector._wtype_paste_usable = False
|
||||
return injector
|
||||
|
||||
def _inject_failing_paste(
|
||||
self, injector: TextInjector, mock_run: MagicMock, paste_cmd: list, error: BaseException
|
||||
) -> bool:
|
||||
def run_side_effect(cmd, **kwargs):
|
||||
if cmd == paste_cmd:
|
||||
raise error
|
||||
return MagicMock(returncode=0, stdout="", stderr="")
|
||||
|
||||
mock_run.side_effect = run_side_effect
|
||||
with patch.object(injector, "_copy_to_clipboard", return_value=True):
|
||||
with patch.object(injector, "_read_clipboard", return_value=None):
|
||||
with patch.object(injector, "_should_copy_to_clipboard", return_value=False):
|
||||
with patch.object(injector, "_should_use_terminal_paste", return_value=False):
|
||||
with patch.object(
|
||||
injector, "_clipboard_paste_command", return_value=paste_cmd
|
||||
):
|
||||
return injector._inject_via_clipboard_paste("hello")
|
||||
|
||||
def test_ydotool_paste_release_command_lifts_held_keys(self):
|
||||
"""Release argv follows the actual paste chord, including Shift and layout 'v'."""
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
self.assertEqual(
|
||||
injector._ydotool_paste_release_command(
|
||||
["ydotool", "key", "29:1", "47:1", "47:0", "29:0"]
|
||||
),
|
||||
["ydotool", "key", "47:0", "29:0"],
|
||||
)
|
||||
self.assertEqual(
|
||||
injector._ydotool_paste_release_command(
|
||||
["ydotool", "key", "29:1", "42:1", "47:1", "47:0", "42:0", "29:0"]
|
||||
),
|
||||
["ydotool", "key", "47:0", "42:0", "29:0"],
|
||||
)
|
||||
self.assertEqual(
|
||||
injector._ydotool_paste_release_command(
|
||||
["ydotool", "key", "29:1", "17:1", "17:0", "29:0"]
|
||||
),
|
||||
["ydotool", "key", "17:0", "29:0"],
|
||||
)
|
||||
self.assertEqual(
|
||||
injector._ydotool_paste_release_command(["ydotool", "key", "ctrl+v"]),
|
||||
["ydotool", "key", "ctrl"],
|
||||
)
|
||||
self.assertEqual(
|
||||
injector._ydotool_paste_release_command(["ydotool", "key", "ctrl+shift+v"]),
|
||||
["ydotool", "key", "ctrl+shift"],
|
||||
)
|
||||
self.assertEqual(
|
||||
injector._ydotool_paste_release_command(["wtype", "-M", "ctrl", "v"]),
|
||||
[],
|
||||
)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.subprocess.run")
|
||||
def test_ydotool_paste_timeout_releases_v1_modifiers(self, mock_run):
|
||||
"""Timeout mid Ctrl+V must send V-up and Ctrl-up (#658)."""
|
||||
injector = self._minimal_paste_injector()
|
||||
paste_cmd = ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"]
|
||||
result = self._inject_failing_paste(
|
||||
injector, mock_run, paste_cmd, subprocess.TimeoutExpired(paste_cmd, 3)
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
cmds = [c.args[0] for c in mock_run.call_args_list if c.args]
|
||||
release_cmd = ["ydotool", "key", "47:0", "29:0"]
|
||||
self.assertIn(release_cmd, cmds)
|
||||
release_kwargs = next(
|
||||
c.kwargs for c in mock_run.call_args_list if c.args and c.args[0] == release_cmd
|
||||
)
|
||||
self.assertIn("env", release_kwargs)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.subprocess.run")
|
||||
def test_ydotool_paste_timeout_releases_terminal_shift(self, mock_run):
|
||||
"""A timed-out Ctrl+Shift+V chord must also lift Shift (#658)."""
|
||||
injector = self._minimal_paste_injector()
|
||||
paste_cmd = ["ydotool", "key", "29:1", "42:1", "47:1", "47:0", "42:0", "29:0"]
|
||||
result = self._inject_failing_paste(
|
||||
injector, mock_run, paste_cmd, subprocess.TimeoutExpired(paste_cmd, 3)
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
cmds = [c.args[0] for c in mock_run.call_args_list if c.args]
|
||||
self.assertIn(["ydotool", "key", "47:0", "42:0", "29:0"], cmds)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.subprocess.run")
|
||||
def test_ydotool_paste_error_releases_legacy_ctrl(self, mock_run):
|
||||
"""A failed 0.1.x ctrl+v paste must tap ctrl so a stuck modifier can lift (#658)."""
|
||||
injector = self._minimal_paste_injector()
|
||||
paste_cmd = list(TextInjector._YDOTOOL_LEGACY_CTRL_V)
|
||||
result = self._inject_failing_paste(
|
||||
injector, mock_run, paste_cmd, subprocess.CalledProcessError(1, paste_cmd)
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
cmds = [c.args[0] for c in mock_run.call_args_list if c.args]
|
||||
self.assertIn(["ydotool", "key", "ctrl"], cmds)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.subprocess.run")
|
||||
def test_ydotool_paste_error_releases_legacy_ctrl_shift(self, mock_run):
|
||||
"""A failed 0.1.x ctrl+shift+v paste must tap ctrl+shift (#658)."""
|
||||
injector = self._minimal_paste_injector()
|
||||
paste_cmd = list(TextInjector._YDOTOOL_LEGACY_CTRL_SHIFT_V)
|
||||
result = self._inject_failing_paste(
|
||||
injector, mock_run, paste_cmd, subprocess.CalledProcessError(1, paste_cmd)
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
cmds = [c.args[0] for c in mock_run.call_args_list if c.args]
|
||||
self.assertIn(["ydotool", "key", "ctrl+shift"], cmds)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.subprocess.run")
|
||||
def test_ydotool_paste_release_timeout_still_returns_false(self, mock_run):
|
||||
"""A wedged daemon on the release path must not raise out of paste (#658)."""
|
||||
injector = self._minimal_paste_injector()
|
||||
paste_cmd = ["ydotool", "key", "29:1", "47:1", "47:0", "29:0"]
|
||||
mock_run.side_effect = subprocess.TimeoutExpired("ydotool", 3)
|
||||
with patch.object(injector, "_copy_to_clipboard", return_value=True):
|
||||
with patch.object(injector, "_read_clipboard", return_value=None):
|
||||
with patch.object(injector, "_should_copy_to_clipboard", return_value=False):
|
||||
with patch.object(injector, "_should_use_terminal_paste", return_value=False):
|
||||
with patch.object(
|
||||
injector, "_clipboard_paste_command", return_value=paste_cmd
|
||||
):
|
||||
result = injector._inject_via_clipboard_paste("hello")
|
||||
self.assertFalse(result)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.subprocess.run")
|
||||
def test_wtype_paste_timeout_does_not_release_ydotool_keys(self, mock_run):
|
||||
"""wtype paste failure must not send a ydotool key-up (#658)."""
|
||||
injector = self._minimal_paste_injector()
|
||||
paste_cmd = ["wtype", "-M", "ctrl", "v"]
|
||||
result = self._inject_failing_paste(
|
||||
injector, mock_run, paste_cmd, subprocess.TimeoutExpired(paste_cmd, 3)
|
||||
)
|
||||
|
||||
self.assertFalse(result)
|
||||
cmds = [c.args[0] for c in mock_run.call_args_list if c.args]
|
||||
self.assertFalse(any(isinstance(c, list) and c and c[0] == "ydotool" for c in cmds))
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.shutil.which")
|
||||
@patch("vocalinux.text_injection.text_injector.subprocess.run")
|
||||
def test_clipboard_paste_returns_false_on_paste_failure(self, mock_run, mock_which):
|
||||
|
||||
Reference in New Issue
Block a user