Skip to content

Commit

Permalink
Clipboard notification improvements with .metadata
Browse files Browse the repository at this point in the history 
Improving clipboard message using the new GUI protocol 1.8

fixes: QubesOS/qubes-issues#9296
  • Loading branch information
@alimirjamali
alimirjamali committed Oct 26, 2024
1 parent bab3289 commit 8126510
Showing 1 changed file with 160 additions and 32 deletions.
192 changes: 160 additions & 32 deletions qui/clipboard.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@

import asyncio
import contextlib
import json
import math
import os
import fcntl
Expand All @@ -50,13 +51,50 @@
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"
APPVIEWER_LOCK = "/var/run/qubes/appviewer.lock"
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: <b>{vmname}</b>")
ERROR_ON_COPY = _( \
"Failed to fetch clipboard data from qube: <b>{vmname}</b>")
ERROR_ON_PASTE = _( \
"Failed to paste global clipboard contents to qube: " \
"<b>{vmname}</b>")
ERROR_OVERSIZED_DATA = _( \
"Global clipboard size exceeded.\n" \
"qube: <b>{vmname}</b> attempted to send {size} bytes to global clipboard."\
"\nCurrent global clipboard limit is {limit}, increase limit or use " \
"<i>qvm-copy</i> to transfer large amounts of data between qubes.")
WARNING_POSSIBLE_TRUNCATION = _( \
"Global clipboard size limit exceed.\n" \
"qube: <b>{vmname}</b> attempted to send {size} bytes to global clipboard."\
"\nGlobal clipboard might have been truncated.\n" \
"<small>Use <i>qvm-copy</i> to transfer large amounts of data between " \
"qubes.</small>")
WARNING_EMPTY_CLIPBOARD = _( \
"Empty source qube clipboard.\n" \
"qube: <b>{vmname}</b> attempted to send <b>0</b> bytes to global " \
"clipboard.")
MSG_COPY_SUCCESS = _( \
"Clipboard contents fetched from qube: <b>'{vmname}'</b>\n" \
"Copied <b>{size}</b> to the global clipboard.\n" \
"<small>Press {shortcut} in qube to paste to local clipboard.</small>")
MSG_WIPED = _("\n<small>Global clipboard has been wiped</small>")
MSG_PASTE_SUCCESS_METADATA = _( \
"Global clipboard copied <b>{size}</b> to <b>{vmname}</b>.\n" \
"Global clipboard has been wiped.\n" \
"<small>Paste normally in qube (e.g. Ctrl+V).</small>")
MSG_PASTE_SUCCESS_LEGACY = _( \
"Global clipboard copied to qube and wiped.<i/>\n" \
"<small>Paste normally in qube (e.g. Ctrl+V).</small>")

@contextlib.contextmanager
def appviewer_lock():
Expand All @@ -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: <b>'{vmname}'</b>\n"
"Copied <b>{size}</b> to the global clipboard.\n"
"<small>Press {shortcut} in qube "
"to paste to local clipboard.</small>".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.<i/>\n"
"<small>Paste normally in qube (e.g. Ctrl+V).</small>")
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 '''
Expand All @@ -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:
Expand Down Expand Up @@ -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(_(
"<i>Global clipboard is empty</i>"))
Expand All @@ -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(
Expand Down Expand Up @@ -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:
Expand All @@ -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):
Expand Down

0 comments on commit 8126510

Please sign in to comment.