From e6ab2a3e5603b90e30862b3bffdec272ce4f3a2c Mon Sep 17 00:00:00 2001 From: Ali Mirjamali Date: Mon, 7 Oct 2024 00:58:03 +0330 Subject: [PATCH] Notify if qube tried to send large clipboard fixes: https://github.com/QubesOS/qubes-issues/issues/9296 --- qui/clipboard.py | 119 +++++++++++++++++++++++++++++++++++++---------- 1 file changed, 95 insertions(+), 24 deletions(-) diff --git a/qui/clipboard.py b/qui/clipboard.py index c2a4c2dd..35e76c83 100644 --- a/qui/clipboard.py +++ b/qui/clipboard.py @@ -28,6 +28,7 @@ import asyncio import contextlib +import json import math import os import fcntl @@ -50,6 +51,7 @@ gbulb.install() DATA = "/var/run/qubes/qubes-clipboard.bin" +METADATA = "/var/run/qubes/qubes-clipboard.bin.metadata" FROM = "/var/run/qubes/qubes-clipboard.bin.source" FROM_DIR = "/var/run/qubes/" XEVENT = "/var/run/qubes/qubes-clipboard.bin.xevent" @@ -75,40 +77,106 @@ def my_init(self, loop=None, gtk_app=None): self.gtk_app = gtk_app self.loop = loop if loop else asyncio.get_event_loop() - def _copy(self, vmname: str = None): + def _copy(self, metadata: dict) -> None: ''' Sends Copy notification via Gio.Notification ''' - if vmname is None: - with appviewer_lock(): - with open(FROM, 'r', encoding='ascii') as vm_from_file: - vmname = vm_from_file.readline().strip('\n') - - size = clipboard_formatted_size() - - body = _("Clipboard contents fetched from qube: '{vmname}'\n" + size = clipboard_formatted_size(metadata["sent_size"]) + + if metadata["malformed_request"]: + body = _("Malformed clipboard request received from qube: " + "{vmname}!").format(vmname=metadata["vmname"]) + elif metadata["qrexec_clipboard"] and \ + metadata["sent_size"] >= metadata["buffer_size"]: + # Microsoft Windows clipboard case + body = _("Qube: {vmname} sent {size} bytes to global " + "clipboard which is over its set limit!\n" + "Global clipboard might be truncated").format( + vmname=metadata["vmname"], + size=size) + elif metadata["oversized_request"]: + body = _("Qube: {vmname} clipboard is over allowed size:\n " + "Size: {size} - Limit: {limit}\n" + "Increase limit or use qvm-copy instead.\n" + ).format(vmname=metadata["vmname"], size=size,limit= \ + clipboard_formatted_size(metadata["buffer_size"])) + elif not metadata["successful"]: + body = _("Failed clipboard copy request received from qube: " + "{vmname}!").format(vmname=metadata["vmname"]) + else: + body = _("Clipboard contents fetched from qube: '{vmname}'\n" "Copied {size} to the global clipboard.\n" "Press {shortcut} in qube " "to paste to local clipboard.".format( - vmname=vmname, size=size, shortcut=self.gtk_app.paste_shortcut)) + vmname=metadata["vmname"], size=size, + shortcut=self.gtk_app.paste_shortcut)) + + if metadata["cleared"]: + body += _("\nGlobal clipboard is wiped") - self.gtk_app.update_clipboard_contents(vmname, size, message=body) + self.gtk_app.update_clipboard_contents(metadata["vmname"], size, + message=body) - def _paste(self): + def _paste(self, metadata: dict) -> None: ''' Sends Paste notification via Gio.Notification. ''' - body = _("Global clipboard contents copied to qube and wiped.\n" - "Paste normally in qube (e.g. Ctrl+V).") + if not metadata["successful"] or metadata["malformed_request"]: + body = _("Failed to paste global clipboard contents to qube: " + "{vmname}".format(vmname=metadata["vmname"])) + body += _("\nGlobal clipboard is wiped") + elif "protocol_version_xside" in metadata.keys() and \ + metadata["protocol_version_xside"] >= 0x00010008: + body = _("Global clipboard contents of {size} copied to " + "{vmname} and wiped.\n" + "Paste normally in qube (e.g. Ctrl+V)." + "".format(size=clipboard_formatted_size( + metadata["sent_size"]), vmname=metadata["vmname"])) + else: + body = _("Global clipboard contents copied to qube and wiped.\n" + "Paste normally in qube (e.g. Ctrl+V).") self.gtk_app.update_clipboard_contents(message=body) - def process_IN_CLOSE_WRITE(self, _unused): + def process_IN_CLOSE_WRITE(self, _unused=None): ''' Reacts to modifications of the FROM file ''' + metadata = {} with appviewer_lock(): - with open(FROM, 'r', encoding='ascii') as vm_from_file: - vmname = vm_from_file.readline().strip('\n') - if vmname == "": - self._paste() - else: - self._copy(vmname=vmname) + if os.path.isfile(METADATA): + # parse JSON .metadata file if qubes-guid protocol 1.8 or newer + try: + with open(METADATA, 'r', encoding='ascii') as metadata_file: + metadata = json.loads(metadata_file.read()) + except OSError: + return + except json.decoder.JSONDecodeError: + return + else: + # revert to .source file on qubes-guid protocol 1.7 or older + # synthesize metadata based on limited available information + with open(FROM, 'r', encoding='ascii') as vm_from_file: + metadata["vmname"] = vm_from_file.readline().strip('\n') + + metadata["copy_action"] = metadata["vmname"] != "" + metadata["paste_action"] = metadata["vmname"] == "" + + try: + metadata["sent_size"] = os.path.getsize(DATA) + except OSError: + metadata["sent_size"] = 0 + + metadata["cleared"] = metadata["sent_size"] == 0 + metadata["qrexec_request"] = False + metadata["malformed_request"] = False + metadata["oversized_request"] = metadata["sent_size"] >= 65000 + metadata["buffer_size"] = 65000 + + if metadata["copy_action"] and metadata["sent_size"] == 0: + metadata["successful"] = False + else: + metadata["successful"] = True + + if metadata["copy_action"]: + self._copy(metadata=metadata) + elif metadata["paste_action"]: + self._paste(metadata=metadata) def process_IN_MOVE_SELF(self, _unused): ''' Stop loop if file is moved ''' @@ -120,15 +188,18 @@ def process_IN_DELETE(self, _unused): def process_IN_CREATE(self, event): if event.pathname == FROM: - self._copy() + self.process_IN_CLOSE_WRITE() self.gtk_app.setup_watcher() -def clipboard_formatted_size() -> str: +def clipboard_formatted_size(size: int = None) -> str: units = ['B', 'KiB', 'MiB', 'GiB'] try: - file_size = os.path.getsize(DATA) + if size: + file_size = size + else: + file_size = os.path.getsize(DATA) except OSError: return _('? bytes') if file_size == 1: