diff --git a/qui/clipboard.py b/qui/clipboard.py index c2a4c2dd..aa1da449 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" @@ -57,6 +59,42 @@ COPY_FEATURE = 'gui-default-secure-copy-sequence' PASTE_FEATURE = 'gui-default-secure-paste-sequence' +# Defining all messages in one place for easy modification +ERROR_MALFORMED_DATA = _( \ + "Malformed clipboard data received from " \ + "qube: {vmname}") +ERROR_ON_COPY = _( \ + "Failed to fetch clipboard data from qube: {vmname}") +ERROR_ON_PASTE = _( \ + "Failed to paste global clipboard contents to qube: " \ + "{vmname}") +ERROR_OVERSIZED_DATA = _( \ + "Global clipboard size exceeded.\n" \ + "qube: {vmname} attempted to send {size} bytes to global clipboard."\ + "\nCurrent global clipboard limit is {limit}, increase limit or use " \ + "qvm-copy to transfer large amounts of data between qubes.") +WARNING_POSSIBLE_TRUNCATION = _( \ + "Global clipboard size limit exceed.\n" \ + "qube: {vmname} attempted to send {size} bytes to global clipboard."\ + "\nGlobal clipboard might have been truncated.\n" \ + "Use qvm-copy to transfer large amounts of data between " \ + "qubes.") +WARNING_EMPTY_CLIPBOARD = _( \ + "Empty source qube clipboard.\n" \ + "qube: {vmname} attempted to send 0 bytes to global " \ + "clipboard.") +MSG_COPY_SUCCESS = _( \ + "Clipboard contents fetched from qube: '{vmname}'\n" \ + "Copied {size} to the global clipboard.\n" \ + "Press {shortcut} in qube to paste to local clipboard.") +MSG_WIPED = _("\nGlobal clipboard has been wiped") +MSG_PASTE_SUCCESS_METADATA = _( \ + "Global clipboard copied {size} to {vmname}.\n" \ + "Global clipboard has been wiped.\n" \ + "Paste normally in qube (e.g. Ctrl+V).") +MSG_PASTE_SUCCESS_LEGACY = _( \ + "Global clipboard copied to qube and wiped.\n" \ + "Paste normally in qube (e.g. Ctrl+V).") @contextlib.contextmanager def appviewer_lock(): @@ -75,40 +113,103 @@ 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() + size = clipboard_formatted_size(metadata["sent_size"]) + + if metadata["malformed_request"]: + body = ERROR_MALFORMED_DATA.format(vmname=metadata["vmname"]) + icon = "dialog-error" + elif metadata["qrexec_clipboard"] and \ + metadata["sent_size"] >= metadata["buffer_size"]: + # Microsoft Windows clipboard case + body = WARNING_POSSIBLE_TRUNCATION.format( + vmname=metadata["vmname"], size=size) + icon = "dialog-warning" + elif metadata["oversized_request"]: + body = ERROR_OVERSIZED_DATA.format(vmname=metadata["vmname"], \ + size=size, \ + limit=clipboard_formatted_size(metadata["buffer_size"])) + icon = "dialog-error" + elif metadata["successful"] and metadata["cleared"] and \ + metadata["sent_size"] == 0: + body = WARNING_EMPTY_CLIPBOARD.format(vmname=metadata["vmname"]) + icon = "dialog-warning" + elif not metadata["successful"]: + body = ERROR_ON_COPY.format(vmname=metadata["vmname"]) + icon = "dialog-error" + else: + body = MSG_COPY_SUCCESS.format(vmname=metadata["vmname"], \ + size=size, shortcut=self.gtk_app.paste_shortcut) + icon = "dialog-information" - 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)) + if metadata["cleared"]: + body += MSG_WIPED - self.gtk_app.update_clipboard_contents(vmname, size, message=body) + self.gtk_app.update_clipboard_contents(metadata["vmname"], size, + message=body, icon=icon) - 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).") - self.gtk_app.update_clipboard_contents(message=body) + if not metadata["successful"] or metadata["malformed_request"]: + body = ERROR_ON_PASTE.format(vmname=metadata["vmname"]) + body += MSG_WIPED + icon = "dialog-error" + elif "protocol_version_xside" in metadata.keys() and \ + metadata["protocol_version_xside"] >= 0x00010008: + body = MSG_PASTE_SUCCESS_METADATA.format( \ + size=clipboard_formatted_size(metadata["sent_size"]), \ + vmname=metadata["vmname"]) + icon = "dialog-information" + else: + body = MSG_PASTE_SUCCESS_LEGACY + icon = "dialog-information" + self.gtk_app.update_clipboard_contents(message=body, icon=icon) - 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 +221,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: @@ -204,7 +308,8 @@ def show_menu(self, _unused, event): event.button, # button Gtk.get_current_event_time()) # activate_time - def update_clipboard_contents(self, vm=None, size=0, message=None): + def update_clipboard_contents(self, vm=None, size=0, message=None, \ + icon=None): if not vm or not size: self.clipboard_label.set_markup(_( "Global clipboard is empty")) @@ -218,7 +323,7 @@ def update_clipboard_contents(self, vm=None, size=0, message=None): self.icon.set_from_icon_name("edit-copy") if message: - self.send_notify(message) + self.send_notify(message, icon=icon) def setup_ui(self, *_args, **_kwargs): self.copy_shortcut = self._prettify_shortcut(self.vm.features.get( @@ -265,7 +370,8 @@ def copy_dom0_clipboard(self, *_args, **_kwargs): text = clipboard.wait_for_text() if not text: - self.send_notify(_("Dom0 clipboard is empty!")) + self.send_notify(_("Dom0 clipboard is empty!"), \ + icon="dialog-information") return try: @@ -276,14 +382,36 @@ def copy_dom0_clipboard(self, *_args, **_kwargs): source.write("dom0") with open(XEVENT, "w", encoding='ascii') as timestamp: timestamp.write(str(Gtk.get_current_event_time())) + with open(METADATA, "w", encoding='ascii') as metadata: + metadata.write( + "{{\n" \ + '"vmname":"dom0",\n' \ + '"xevent_timestamp":{xevent_timestamp},\n' \ + '"successful":1,\n' \ + '"copy_action":1,\n' \ + '"paste_action":0,\n' \ + '"malformed_request":0,\n' \ + '"cleared":0,\n' \ + '"qrexec_clipboard":0,\n' \ + '"sent_size":{sent_size},\n' \ + '"buffer_size":{buffer_size},\n' \ + '"protocol_version_xside":65544,\n' \ + '"protocol_version_vmside":65544,\n' \ + '}}\n'.format(xevent_timestamp= \ + str(Gtk.get_current_event_time()), \ + sent_size=os.path.getsize(DATA), \ + buffer_size="256000")) except Exception: # pylint: disable=broad-except - self.send_notify(_("Error while accessing global clipboard!")) + self.send_notify(_("Error while accessing global clipboard!"), \ + icon = "dialog-error") - def send_notify(self, body): + def send_notify(self, body, icon=None): # pylint: disable=attribute-defined-outside-init notification = Gio.Notification.new(_("Global Clipboard")) notification.set_body(body) notification.set_priority(Gio.NotificationPriority.NORMAL) + if icon is not None: + notification.set_icon(Gio.ThemedIcon.new(icon)) self.send_notification(self.get_application_id(), notification) def _prettify_shortcut(self, shortcut: str):