Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Notify if qube tried to send large clipboard #223

Merged
merged 1 commit into from
Nov 4, 2024
Merged

Notify if qube tried to send large clipboard #223

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
192 changes: 160 additions & 32 deletions qui/clipboard.py
View file
Open in desktop
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"]:
alimirjamali marked this conversation as resolved.
Show resolved Hide resolved
# 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