From fe7d8cd1d9c133d683d3704b8d229f87de5dfbf5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 18 Jul 2026 00:28:50 +0000 Subject: [PATCH 1/2] Implement OSC 52 clipboard protocol via PTY proxy --- guake/data/org.guake.gschema.xml | 5 + guake/data/prefs.glade | 15 ++ guake/prefs.py | 8 + guake/terminal.py | 289 ++++++++++++++++++++++++++++++- guake/tests/test_osc52.py | 196 +++++++++++++++++++++ 5 files changed, 509 insertions(+), 4 deletions(-) create mode 100644 guake/tests/test_osc52.py diff --git a/guake/data/org.guake.gschema.xml b/guake/data/org.guake.gschema.xml index cad5252fa..081aa6e07 100644 --- a/guake/data/org.guake.gschema.xml +++ b/guake/data/org.guake.gschema.xml @@ -336,6 +336,11 @@ Copy selected text if enabled. Copy the selected text. + + true + Enable OSC 52 clipboard access + Allow applications running in the terminal to set the system clipboard using the OSC 52 escape sequence. + diff --git a/guake/data/prefs.glade b/guake/data/prefs.glade index 6e58140d5..2a0afb3ce 100644 --- a/guake/data/prefs.glade +++ b/guake/data/prefs.glade @@ -643,6 +643,21 @@ 4 + + + Enable OSC 52 clipboard access + True + True + False + start + True + + + + 0 + 5 + + diff --git a/guake/prefs.py b/guake/prefs.py index 5c3097a37..6c35d92d6 100644 --- a/guake/prefs.py +++ b/guake/prefs.py @@ -408,6 +408,10 @@ def on_copy_on_select_toggled(self, chk): """Changes the value of copy_on_select in dconf""" self.settings.general.set_boolean("copy-on-select", chk.get_active()) + def on_enable_osc52_toggled(self, chk): + """Changes the value of enable_osc52 in dconf""" + self.settings.general.set_boolean("enable-osc52", chk.get_active()) + def on_tab_name_display_changed(self, combo): """Save `display-tab-names` property value in dconf""" self.settings.general.set_int("display-tab-names", combo.get_active()) @@ -1274,6 +1278,10 @@ def load_configs(self): value = self.settings.general.get_boolean("copy-on-select") self.get_widget("copy_on_select").set_active(value) + # enable osc52 + value = self.settings.general.get_boolean("enable-osc52") + self.get_widget("enable_osc52").set_active(value) + # font value = self.settings.styleFont.get_string("style") if value: diff --git a/guake/terminal.py b/guake/terminal.py index 2e587d7c7..192a03bb6 100644 --- a/guake/terminal.py +++ b/guake/terminal.py @@ -18,13 +18,17 @@ Boston, MA 02110-1301 USA """ import code +import base64 +import fcntl import logging import os import re import shlex import signal +import struct import subprocess import sys +import termios import threading import uuid @@ -87,6 +91,17 @@ def halt(loc): # pylint: enable=anomalous-backslash-in-string +# OSC 52 escape sequence: ESC ] 52 ; ; ( BEL | ESC \ ) +# params: c = CLIPBOARD, p = PRIMARY, s = SECONDARY, q = all, 0-7 = cut buffers +# data: base64-encoded text, or "?" to query, or "" to clear +_OSC52_RE = re.compile( + b"\\x1b\\]52;([^;]*);([^\\x07\\x1b]*)" + b"(?:\\x07|\\x1b\\\\)" +) + +# Maximum size of the incomplete-sequence buffer (1 MiB) to prevent runaway memory use +_OSC52_MAX_BUF = 1024 * 1024 + class DropTargets(IntEnum): URIS = 0 @@ -119,6 +134,13 @@ def __init__(self, guake): self.custom_fgcolor = None self.custom_palette = None + # OSC 52 proxy state (None when proxy is not active) + self._osc52_master_fd = None + self._osc52_buf = b"" + self._osc52_io_watch = None + self._osc52_commit_id = None + self._osc52_resize_id = None + self.setup_drag_and_drop() self.ENVV_EXCLUDE_LIST = ["GDK_BACKEND"] @@ -145,6 +167,18 @@ def pid(self, pid): self._pid = pid def feed_child(self, resolved_cmdline): + if self._osc52_master_fd is not None: + # OSC 52 proxy mode: write keyboard/programmatic input directly to the PTY master + data = ( + resolved_cmdline.encode("utf-8") + if isinstance(resolved_cmdline, str) + else resolved_cmdline + ) + try: + os.write(self._osc52_master_fd, data) + except OSError as e: + log.debug("OSC 52 proxy: feed_child write error: %s", e) + return if (Vte.MAJOR_VERSION, Vte.MINOR_VERSION) >= (0, 42): encoded = resolved_cmdline.encode("utf-8") try: @@ -555,8 +589,8 @@ def delete_shell(self, pid): except OSError: pass - def spawn_sync_pid(self, directory): - + def _get_shell_argv(self): + """Return the shell command argv based on current settings.""" argv = [] user_shell = self.guake.settings.general.get_string("default-shell") if user_shell and os.path.exists(user_shell): @@ -567,10 +601,24 @@ def spawn_sync_pid(self, directory): except KeyError: argv.append("/usr/bin/bash") - login_shell = self.guake.settings.general.get_boolean("use-login-shell") - if login_shell: + if self.guake.settings.general.get_boolean("use-login-shell"): argv.append("--login") + return argv + + def spawn_sync_pid(self, directory): + if self.guake.settings.general.get_boolean("enable-osc52"): + try: + return self._spawn_with_osc52_proxy(directory) + except Exception as e: # pylint: disable=broad-except + log.warning( + "OSC 52 proxy setup failed, falling back to normal spawn: %s", e + ) + return self._spawn_normal(directory) + + def _spawn_normal(self, directory): + """Original spawn logic using VTE's built-in PTY management.""" + argv = self._get_shell_argv() log.debug('Spawn command: "%s"', " ".join(argv)) pid = self.spawn_sync( @@ -600,6 +648,239 @@ def spawn_sync_pid(self, directory): self.pid = pid return pid + def _spawn_with_osc52_proxy(self, directory): + """Spawn the shell with a PTY proxy that intercepts OSC 52 clipboard sequences. + + Creates a PTY pair for shell communication. Shell output is read by + Guake, OSC 52 sequences are processed (clipboard is set) and stripped, + and the remaining data is fed into VTE via ``Vte.Terminal.feed()``. + Keyboard input from VTE (emitted via the ``commit`` signal because no + VTE-managed PTY is set) is written to the PTY master so the shell + receives it normally. + """ + argv = self._get_shell_argv() + log.debug('Spawn command (OSC 52 proxy): "%s"', " ".join(argv)) + + # Create a PTY for the shell process + master_fd, slave_fd = os.openpty() + + # Build environment dict from envv list + env_dict = {} + for entry in self.envv: + key, _, value = entry.partition("=") + if key: + env_dict[key] = value + + def _preexec(): + # Create a new session so the slave becomes the controlling terminal + os.setsid() + try: + fcntl.ioctl(0, termios.TIOCSCTTY, 0) + except Exception: # pylint: disable=broad-except + pass + + proc = subprocess.Popen( + argv, + stdin=slave_fd, + stdout=slave_fd, + stderr=slave_fd, + env=env_dict, + cwd=directory, + preexec_fn=_preexec, + close_fds=True, + ) + + # Close slave side in parent; child has it via dup2 to 0/1/2 + os.close(slave_fd) + + # Register the PTY session with utempter if available + if libutempter is not None: + libutempter.utempter_add_record(master_fd, os.uname()[1]) + + # Propagate the current terminal size to the new PTY + self._osc52_update_winsize(master_fd) + + # Store proxy state + self._osc52_master_fd = master_fd + self._osc52_buf = b"" + + # Watch the PTY master for shell output + self._osc52_io_watch = GLib.io_add_watch( + master_fd, + GLib.IOCondition.IN | GLib.IOCondition.HUP | GLib.IOCondition.ERR, + self._osc52_on_shell_output, + ) + + # VTE has no PTY, so it emits 'commit' for all keyboard/mouse input + self._osc52_commit_id = self.connect("commit", self._osc52_on_commit) + + # Keep terminal size in sync + self._osc52_resize_id = self.connect("size-allocate", self._osc52_on_resize) + + # Watch for child exit + GLib.child_watch_add(proc.pid, self._osc52_on_child_exit) + + self.pid = proc.pid + return proc.pid + + # ------------------------------------------------------------------ + # OSC 52 proxy helpers + # ------------------------------------------------------------------ + + def _osc52_update_winsize(self, master_fd): + """Push the current VTE terminal dimensions into the PTY.""" + try: + cols = self.get_column_count() + rows = self.get_row_count() + if cols > 0 and rows > 0: + winsize = struct.pack("HHHH", rows, cols, 0, 0) + fcntl.ioctl(master_fd, termios.TIOCSWINSZ, winsize) + except Exception as e: # pylint: disable=broad-except + log.debug("OSC 52 proxy: could not set terminal size: %s", e) + + def _osc52_on_shell_output(self, fd, condition): + """GLib IO watch callback: read shell output, strip OSC 52, feed to VTE.""" + if condition & GLib.IOCondition.IN: + try: + data = os.read(fd, 65536) + except OSError as e: + log.debug("OSC 52 proxy: read error: %s", e) + return False + if data: + filtered = self._osc52_process_data(data) + if filtered: + try: + self.feed(filtered) + except TypeError: + self.feed(filtered, len(filtered)) + + if condition & (GLib.IOCondition.HUP | GLib.IOCondition.ERR): + return False + + return True + + def _osc52_process_data(self, data): + """Strip OSC 52 sequences from *data*, handle clipboard side-effects. + + Maintains ``self._osc52_buf`` across calls so sequences that are split + across multiple ``read()`` calls are handled correctly. + """ + # Guard against a runaway buffer caused by a malformed/truncated sequence + if len(self._osc52_buf) > _OSC52_MAX_BUF: + log.warning("OSC 52: incomplete-sequence buffer exceeded limit, discarding") + self._osc52_buf = b"" + + combined = self._osc52_buf + data + self._osc52_buf = b"" + + # Process all complete OSC 52 sequences + filtered = _OSC52_RE.sub(self._osc52_handle_match, combined) + + # If there is a potential incomplete OSC 52 at the tail, buffer it + start = filtered.rfind(b"\x1b]52;") + if start != -1: + tail = filtered[start:] + # Only keep it buffered when the sequence has no terminator yet + if not _OSC52_RE.match(tail): + self._osc52_buf = tail + filtered = filtered[:start] + + return filtered + + def _osc52_handle_match(self, match): + """Regex substitution callback: set the clipboard and return empty bytes.""" + params = match.group(1).decode("ascii", errors="ignore") + encoded = match.group(2) + + if encoded == b"?": + # Query operation – not supported; silently ignore + return b"" + + try: + text = base64.b64decode(encoded).decode("utf-8", errors="replace") + except Exception as e: # pylint: disable=broad-except + log.debug("OSC 52: failed to decode base64 payload: %s", e) + return b"" + + self._osc52_set_clipboard(text, params) + return b"" + + def _osc52_set_clipboard(self, text, params): + """Write *text* to the clipboard targets indicated by *params*. + + Recognised params: ``c`` = CLIPBOARD, ``p`` = PRIMARY. An empty or + unrecognised params string defaults to CLIPBOARD. + """ + if not text: + return + + use_clipboard = "c" in params or not any(ch in params for ch in "ps") + use_primary = "p" in params + + if use_clipboard: + clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD) + clipboard.set_text(text, -1) + log.debug("OSC 52: set CLIPBOARD (%d chars)", len(text)) + + if use_primary: + primary = Gtk.Clipboard.get(Gdk.SELECTION_PRIMARY) + primary.set_text(text, -1) + log.debug("OSC 52: set PRIMARY selection (%d chars)", len(text)) + + def _osc52_on_commit(self, _terminal, text, _length): + """Forward keyboard/mouse input from VTE to the shell PTY.""" + if self._osc52_master_fd is None: + return + try: + data = text.encode("utf-8") if isinstance(text, str) else bytes(text) + if data: + os.write(self._osc52_master_fd, data) + except OSError as e: + log.debug("OSC 52 proxy: commit write error: %s", e) + + def _osc52_on_resize(self, _widget, _allocation): + """Propagate VTE size changes to the shell PTY.""" + if self._osc52_master_fd is not None: + self._osc52_update_winsize(self._osc52_master_fd) + + def _osc52_on_child_exit(self, pid, status): + """Handle shell process exit when running in OSC 52 proxy mode.""" + log.debug("OSC 52 proxy: process %d exited (status %d)", pid, status) + if libutempter is not None and self._osc52_master_fd is not None: + libutempter.utempter_remove_record(self._osc52_master_fd) + self._osc52_cleanup() + # Notify the rest of Guake (e.g. tab removal) via the standard signal + self.emit("child-exited", status) + + def _osc52_cleanup(self): + """Release all resources held by the OSC 52 proxy.""" + if self._osc52_io_watch is not None: + GLib.source_remove(self._osc52_io_watch) + self._osc52_io_watch = None + + if self._osc52_master_fd is not None: + try: + os.close(self._osc52_master_fd) + except OSError: + pass + self._osc52_master_fd = None + + self._osc52_buf = b"" + + if self._osc52_commit_id is not None: + try: + self.disconnect(self._osc52_commit_id) + except Exception: # pylint: disable=broad-except + pass + self._osc52_commit_id = None + + if self._osc52_resize_id is not None: + try: + self.disconnect(self._osc52_resize_id) + except Exception: # pylint: disable=broad-except + pass + self._osc52_resize_id = None + def set_color_foreground(self, font_color, *args, **kwargs): real_fgcolor = self.custom_fgcolor if self.custom_fgcolor else font_color super().set_color_foreground(real_fgcolor, *args, **kwargs) diff --git a/guake/tests/test_osc52.py b/guake/tests/test_osc52.py new file mode 100644 index 000000000..aba46da7e --- /dev/null +++ b/guake/tests/test_osc52.py @@ -0,0 +1,196 @@ +# -*- coding: utf-8 -*- +"""Tests for OSC 52 clipboard sequence filtering.""" + +import base64 +import re + +import pytest + +# Mirror the module-level constants from terminal.py so the tests are +# self-contained and do not require a running display or VTE. +_OSC52_RE = re.compile( + b"\\x1b\\]52;([^;]*);([^\\x07\\x1b]*)" + b"(?:\\x07|\\x1b\\\\)" +) +_OSC52_MAX_BUF = 1024 * 1024 + + +def _filter(data, buf=b""): + """Minimal re-implementation of GuakeTerminal._osc52_process_data.""" + clipboard_ops = [] + + def _handle(match): + params = match.group(1).decode("ascii", errors="ignore") + encoded = match.group(2) + if encoded == b"?": + return b"" + try: + text = base64.b64decode(encoded).decode("utf-8", errors="replace") + clipboard_ops.append((params, text)) + except Exception: + pass + return b"" + + combined = buf + data + filtered = _OSC52_RE.sub(_handle, combined) + + # Buffer any potential incomplete OSC 52 at the tail + new_buf = b"" + start = filtered.rfind(b"\x1b]52;") + if start != -1: + tail = filtered[start:] + if not _OSC52_RE.match(tail): + new_buf = tail + filtered = filtered[:start] + + return filtered, new_buf, clipboard_ops + + +# --------------------------------------------------------------------------- +# Basic filtering +# --------------------------------------------------------------------------- + + +def test_osc52_bel_terminator(): + """OSC 52 sequence ending with BEL (0x07) is stripped and clipboard set.""" + text = "hello clipboard" + encoded = base64.b64encode(text.encode()).decode() + osc = f"\x1b]52;c;{encoded}\x07".encode() + data = b"before" + osc + b"after" + + filtered, _, ops = _filter(data) + + assert filtered == b"beforeafter" + assert len(ops) == 1 + assert ops[0] == ("c", text) + + +def test_osc52_st_terminator(): + """OSC 52 sequence ending with ST (ESC \\) is stripped.""" + text = "st terminated" + encoded = base64.b64encode(text.encode()).decode() + osc = (f"\x1b]52;c;{encoded}\x1b\\").encode() + data = b"prefix" + osc + b"suffix" + + filtered, _, ops = _filter(data) + + assert filtered == b"prefixsuffix" + assert len(ops) == 1 + assert ops[0][1] == text + + +def test_osc52_primary_selection(): + """Params 'p' targets the PRIMARY selection.""" + text = "primary" + encoded = base64.b64encode(text.encode()).decode() + osc = f"\x1b]52;p;{encoded}\x07".encode() + + _, _, ops = _filter(osc) + + assert ops[0][0] == "p" + assert ops[0][1] == text + + +def test_osc52_empty_params(): + """Empty params field is preserved (defaults to CLIPBOARD in the handler).""" + text = "default" + encoded = base64.b64encode(text.encode()).decode() + osc = f"\x1b]52;;{encoded}\x07".encode() + + filtered, _, ops = _filter(osc) + + assert filtered == b"" + assert ops[0][0] == "" # empty params + assert ops[0][1] == text + + +def test_normal_output_unchanged(): + """Terminal output without OSC 52 passes through unmodified.""" + data = b"normal output\r\n\x1b[32mgreen\x1b[0m" + filtered, _, ops = _filter(data) + assert filtered == data + assert ops == [] + + +def test_multiple_osc52_sequences(): + """Multiple OSC 52 sequences in one chunk are all processed.""" + t1, t2 = "first", "second" + e1 = base64.b64encode(t1.encode()).decode() + e2 = base64.b64encode(t2.encode()).decode() + data = ( + b"A\x1b]52;c;" + e1.encode() + b"\x07" + b"B\x1b]52;p;" + e2.encode() + b"\x07C" + ) + + filtered, _, ops = _filter(data) + + assert filtered == b"ABC" + assert len(ops) == 2 + assert ops[0] == ("c", t1) + assert ops[1] == ("p", t2) + + +def test_query_operation_ignored(): + """OSC 52 with '?' data (clipboard query) is stripped but not added to ops.""" + osc = b"\x1b]52;c;?\x07" + filtered, _, ops = _filter(osc) + assert filtered == b"" + assert ops == [] + + +def test_incomplete_sequence_buffered(): + """An incomplete OSC 52 at the end of a chunk is held in the buffer.""" + text = "split" + encoded = base64.b64encode(text.encode()).decode() + full_osc = f"\x1b]52;c;{encoded}\x07".encode() + + # Send only the first half of the sequence + half = len(full_osc) // 2 + chunk1 = full_osc[:half] + chunk2 = full_osc[half:] + b"after" + + filtered1, buf, ops1 = _filter(chunk1) + assert ops1 == [] # sequence not yet complete + + filtered2, _, ops2 = _filter(chunk2, buf) + assert ops2[0][1] == text + assert b"after" in filtered2 + + +def test_utf8_content(): + """OSC 52 payload containing multibyte UTF-8 characters is decoded correctly.""" + text = "こんにちは" + encoded = base64.b64encode(text.encode("utf-8")).decode() + osc = f"\x1b]52;c;{encoded}\x07".encode() + + _, _, ops = _filter(osc) + + assert ops[0][1] == text + + +def test_osc52_default_clipboard_params(): + """When params contain neither 'c' nor 'p', the handler defaults to CLIPBOARD.""" + # Simulate the logic in _osc52_set_clipboard + params = "" + use_clipboard = "c" in params or not any(ch in params for ch in "ps") + use_primary = "p" in params + assert use_clipboard is True + assert use_primary is False + + +def test_osc52_primary_only(): + """When params is 'p', only PRIMARY is set.""" + params = "p" + use_clipboard = "c" in params or not any(ch in params for ch in "ps") + use_primary = "p" in params + assert use_clipboard is False + assert use_primary is True + + +def test_osc52_both_targets(): + """When params is 'cp', both CLIPBOARD and PRIMARY are set.""" + params = "cp" + use_clipboard = "c" in params or not any(ch in params for ch in "ps") + use_primary = "p" in params + assert use_clipboard is True + assert use_primary is True From 3e24e7490ced23b69dfc48fb3b31e6b09411106a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 18 Jul 2026 00:36:01 +0000 Subject: [PATCH 2/2] Address code review: fix EOF busy loop, false-positive buffering, error logging, and test clarity --- guake/terminal.py | 83 +++++++++++++++++++++++------ guake/tests/test_osc52.py | 106 +++++++++++++++++++++++++++++++++++++- 2 files changed, 170 insertions(+), 19 deletions(-) diff --git a/guake/terminal.py b/guake/terminal.py index 192a03bb6..5c3f390a3 100644 --- a/guake/terminal.py +++ b/guake/terminal.py @@ -99,6 +99,14 @@ def halt(loc): b"(?:\\x07|\\x1b\\\\)" ) +# Pattern for a *partial* (not yet terminated) OSC 52 sequence. Only data +# consisting of valid base64 characters (or "?") is accepted, so that +# unrelated byte sequences that happen to start with ESC]52; are not +# mistakenly buffered as incomplete OSC 52 sequences. +_OSC52_PARTIAL_RE = re.compile( + b"\\x1b\\]52;[^;]*;[A-Za-z0-9+/=?]*$" +) + # Maximum size of the incomplete-sequence buffer (1 MiB) to prevent runaway memory use _OSC52_MAX_BUF = 1024 * 1024 @@ -671,13 +679,30 @@ def _spawn_with_osc52_proxy(self, directory): if key: env_dict[key] = value + # Capture slave_fd in closure; pass_fds ensures it survives close_fds in + # the child so preexec_fn can reference it explicitly (rather than relying + # on fd 0 after the dup2, which is equivalent but less obvious). + _slave_fd = slave_fd + def _preexec(): - # Create a new session so the slave becomes the controlling terminal + # Create a new session so the slave becomes the controlling terminal. + # This runs in the child process; the parent logger is not available, + # so unexpected errors are printed to stderr (which at this point is + # already the slave PTY, visible in the terminal). os.setsid() try: - fcntl.ioctl(0, termios.TIOCSCTTY, 0) - except Exception: # pylint: disable=broad-except - pass + fcntl.ioctl(_slave_fd, termios.TIOCSCTTY, 0) + except OSError as _e: + import errno as _errno + # EPERM/ENOTTY can occur in restricted environments; treat as + # non-fatal because the terminal still functions without a + # controlling terminal. Anything else is unexpected. + if _e.errno not in (_errno.EPERM, _errno.ENOTTY): + import sys as _sys + print( + f"guake: OSC 52 proxy: TIOCSCTTY failed unexpectedly: {_e}", + file=_sys.stderr, + ) proc = subprocess.Popen( argv, @@ -688,6 +713,7 @@ def _preexec(): cwd=directory, preexec_fn=_preexec, close_fds=True, + pass_fds=(slave_fd,), ) # Close slave side in parent; child has it via dup2 to 0/1/2 @@ -746,13 +772,15 @@ def _osc52_on_shell_output(self, fd, condition): except OSError as e: log.debug("OSC 52 proxy: read error: %s", e) return False - if data: - filtered = self._osc52_process_data(data) - if filtered: - try: - self.feed(filtered) - except TypeError: - self.feed(filtered, len(filtered)) + if not data: + # EOF on the PTY master — stop the watch + return False + filtered = self._osc52_process_data(data) + if filtered: + try: + self.feed(filtered) + except TypeError: + self.feed(filtered, len(filtered)) if condition & (GLib.IOCondition.HUP | GLib.IOCondition.ERR): return False @@ -765,10 +793,22 @@ def _osc52_process_data(self, data): Maintains ``self._osc52_buf`` across calls so sequences that are split across multiple ``read()`` calls are handled correctly. """ - # Guard against a runaway buffer caused by a malformed/truncated sequence + # Guard against a runaway buffer caused by a malformed/truncated sequence. + # The buffer only ever holds an incomplete OSC 52 tail (i.e. everything + # from the last unmatched ESC]52; to the end of the previous chunk), so + # there is no valid terminal output to preserve – pass it through to VTE + # as raw bytes so the user sees the escape text rather than losing data. if len(self._osc52_buf) > _OSC52_MAX_BUF: - log.warning("OSC 52: incomplete-sequence buffer exceeded limit, discarding") + log.warning( + "OSC 52: buffer limit exceeded; displaying incomplete escape " + "sequence as raw text to avoid data loss" + ) + overflow = self._osc52_buf self._osc52_buf = b"" + try: + self.feed(overflow) + except TypeError: + self.feed(overflow, len(overflow)) combined = self._osc52_buf + data self._osc52_buf = b"" @@ -776,12 +816,14 @@ def _osc52_process_data(self, data): # Process all complete OSC 52 sequences filtered = _OSC52_RE.sub(self._osc52_handle_match, combined) - # If there is a potential incomplete OSC 52 at the tail, buffer it + # If there is a potential incomplete OSC 52 at the tail, buffer it. + # Use the stricter partial-pattern check to avoid buffering unrelated + # byte sequences that merely contain the ESC]52; prefix (e.g. binary + # output or help text that quotes escape sequences). start = filtered.rfind(b"\x1b]52;") if start != -1: tail = filtered[start:] - # Only keep it buffered when the sequence has no terminator yet - if not _OSC52_RE.match(tail): + if _OSC52_PARTIAL_RE.match(tail) and not _OSC52_RE.match(tail): self._osc52_buf = tail filtered = filtered[:start] @@ -810,10 +852,17 @@ def _osc52_set_clipboard(self, text, params): Recognised params: ``c`` = CLIPBOARD, ``p`` = PRIMARY. An empty or unrecognised params string defaults to CLIPBOARD. + + This method is always called from the GLib main-loop IO watch + (``_osc52_on_shell_output``), which is dispatched on the same thread as + the GTK main loop, so GTK clipboard calls are thread-safe here. """ if not text: return + # Use CLIPBOARD when 'c' is present, or when the params string contains + # neither 'p' (PRIMARY) nor 's' (SECONDARY) – an empty params string + # therefore also maps to CLIPBOARD. use_clipboard = "c" in params or not any(ch in params for ch in "ps") use_primary = "p" in params @@ -832,7 +881,7 @@ def _osc52_on_commit(self, _terminal, text, _length): if self._osc52_master_fd is None: return try: - data = text.encode("utf-8") if isinstance(text, str) else bytes(text) + data = text if isinstance(text, bytes) else text.encode("utf-8") if data: os.write(self._osc52_master_fd, data) except OSError as e: diff --git a/guake/tests/test_osc52.py b/guake/tests/test_osc52.py index aba46da7e..10719a70d 100644 --- a/guake/tests/test_osc52.py +++ b/guake/tests/test_osc52.py @@ -12,6 +12,9 @@ b"\\x1b\\]52;([^;]*);([^\\x07\\x1b]*)" b"(?:\\x07|\\x1b\\\\)" ) +_OSC52_PARTIAL_RE = re.compile( + b"\\x1b\\]52;[^;]*;[A-Za-z0-9+/=?]*$" +) _OSC52_MAX_BUF = 1024 * 1024 @@ -34,12 +37,13 @@ def _handle(match): combined = buf + data filtered = _OSC52_RE.sub(_handle, combined) - # Buffer any potential incomplete OSC 52 at the tail + # Buffer any potential incomplete OSC 52 at the tail (using the stricter + # partial-regex check to avoid false-positive buffering of unrelated data) new_buf = b"" start = filtered.rfind(b"\x1b]52;") if start != -1: tail = filtered[start:] - if not _OSC52_RE.match(tail): + if _OSC52_PARTIAL_RE.match(tail) and not _OSC52_RE.match(tail): new_buf = tail filtered = filtered[:start] @@ -157,6 +161,20 @@ def test_incomplete_sequence_buffered(): assert b"after" in filtered2 +def test_invalid_data_not_buffered(): + """Output with ESC]52; followed by non-base64 data is NOT held in the buffer. + + If a help text or binary file contains the literal bytes ESC]52; but the + following data is not valid base64, the partial-regex guard must reject it + so valid terminal output is not accidentally withheld from VTE. + """ + data = b"info: \x1b]52;c;!!invalid!!" # no terminator; invalid chars in data field + filtered, buf, _ = _filter(data) + # Invalid chars → partial regex does not match → sequence passes through unchanged + assert buf == b"" + assert b"\x1b]52;c;!!invalid!!" in filtered + + def test_utf8_content(): """OSC 52 payload containing multibyte UTF-8 characters is decoded correctly.""" text = "こんにちは" @@ -194,3 +212,87 @@ def test_osc52_both_targets(): use_primary = "p" in params assert use_clipboard is True assert use_primary is True + + +# --------------------------------------------------------------------------- +# Integration tests: verify the correct GTK clipboard targets are populated +# --------------------------------------------------------------------------- + + +def _make_clipboard_mock(): + """Return a simple mock object that records set_text calls.""" + + class _Clipboard: + def __init__(self): + self.text = None + + def set_text(self, text, _length): + self.text = text + + return _Clipboard() + + +def _set_clipboard(text, params, clip_mock=None, primary_mock=None): + """Inline re-implementation of GuakeTerminal._osc52_set_clipboard for testing.""" + if not text: + return + use_clipboard = "c" in params or not any(ch in params for ch in "ps") + use_primary = "p" in params + if use_clipboard and clip_mock is not None: + clip_mock.set_text(text, -1) + if use_primary and primary_mock is not None: + primary_mock.set_text(text, -1) + + +def test_clipboard_mock_default_params(): + """Empty params → only CLIPBOARD is set.""" + clip = _make_clipboard_mock() + primary = _make_clipboard_mock() + _set_clipboard("hello", "", clip, primary) + assert clip.text == "hello" + assert primary.text is None + + +def test_clipboard_mock_c_param(): + """Params 'c' → CLIPBOARD is set, PRIMARY is not.""" + clip = _make_clipboard_mock() + primary = _make_clipboard_mock() + _set_clipboard("world", "c", clip, primary) + assert clip.text == "world" + assert primary.text is None + + +def test_clipboard_mock_p_param(): + """Params 'p' → PRIMARY is set, CLIPBOARD is not.""" + clip = _make_clipboard_mock() + primary = _make_clipboard_mock() + _set_clipboard("primary text", "p", clip, primary) + assert clip.text is None + assert primary.text == "primary text" + + +def test_clipboard_mock_cp_params(): + """Params 'cp' → both CLIPBOARD and PRIMARY are set.""" + clip = _make_clipboard_mock() + primary = _make_clipboard_mock() + _set_clipboard("both", "cp", clip, primary) + assert clip.text == "both" + assert primary.text == "both" + + +def test_clipboard_mock_unknown_params(): + """Unknown params that contain neither 'p' nor 's' → CLIPBOARD.""" + clip = _make_clipboard_mock() + primary = _make_clipboard_mock() + _set_clipboard("data", "q", clip, primary) # 'q' is not 'p' or 's' + assert clip.text == "data" + assert primary.text is None + + +def test_clipboard_mock_empty_text(): + """Empty text → neither clipboard target is touched.""" + clip = _make_clipboard_mock() + primary = _make_clipboard_mock() + _set_clipboard("", "c", clip, primary) + assert clip.text is None + assert primary.text is None