# # The Qubes OS Project, https://www.qubes-os.org # # Copyright (C) 2012 Agnieszka Kostrzewa # Copyright (C) 2012 Marek Marczykowski-Górecki # # Copyright (C) 2017 Wojtek Porczyk # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # import itertools import os import re import qubesadmin import traceback import asyncio from contextlib import suppress import sys import quamash from qubesadmin import events from PyQt5 import QtWidgets # pylint: disable=import-error from PyQt5.QtGui import QIcon # pylint: disable=import-error def _filter_internal(vm): return (not vm.klass == 'AdminVM' and not vm.features.get('internal', False)) def prepare_choice(widget, holder, propname, choice, default, filter_function=None, *, icon_getter=None, allow_internal=None, allow_default=False, allow_none=False, transform=None): # for newly created vms, set propname to None debug( 'prepare_choice(widget={widget!r}, ' 'holder={holder!r}, ' 'propname={propname!r}, ' 'choice={choice!r}, ' 'default={default!r}, ' 'filter_function={filter_function!r}, ' 'icon_getter={icon_getter!r}, ' 'allow_internal={allow_internal!r}, ' 'allow_default={allow_default!r}, ' 'allow_none={allow_none!r})'.format(**locals())) if propname is not None and allow_default: default = holder.property_get_default(propname) if allow_internal is None: allow_internal = propname is None or not propname.endswith('vm') if propname is not None: if holder.property_is_default(propname): oldvalue = qubesadmin.DEFAULT else: oldvalue = getattr(holder, propname) if oldvalue == '': oldvalue = None if transform is not None and oldvalue is not None: oldvalue = transform(oldvalue) else: oldvalue = object() # won't match for identity idx = 0 choice_list = list(choice)[:] if not allow_internal: choice_list = filter(_filter_internal, choice_list) if filter_function is not None: choice_list = filter(filter_function, choice_list) choice_list = list(choice_list) if allow_default: choice_list.insert(0, qubesadmin.DEFAULT) if allow_none: choice_list.append(None) for i, item in enumerate(choice_list): debug('i={} item={}'.format(i, item)) # 0: default (unset) if item is qubesadmin.DEFAULT: default_string = str(default) if default is not None else 'none' if transform is not None: default_string = transform(default_string) text = 'default ({})'.format(default_string) # N+1: explicit None elif item is None: text = '(none)' # 1..N: choices else: text = str(item) if transform is not None: text = transform(text) if item == oldvalue: text += ' (current)' idx = i widget.insertItem(i, text) if icon_getter is not None: icon = icon_getter(item) if icon is not None: widget.setItemIcon(i, icon) widget.setCurrentIndex(idx) return choice_list, idx class KernelVersion: # pylint: disable=too-few-public-methods # Cannot use distutils.version.LooseVersion, because it fails at handling # versions that have no numbers in them def __init__(self, string): self.string = string self.groups = re.compile(r'(\d+)').split(self.string) def __lt__(self, other): for (self_content, other_content) in itertools.zip_longest( self.groups, other.groups): if self_content == other_content: continue if self_content.isdigit() and other_content.isdigit(): return int(self_content) < int(other_content) return self_content < other_content def prepare_kernel_choice(widget, holder, propname, default, *args, **kwargs): # TODO get from storage API (pool 'linux-kernel') (suggested by @marmarta) kernels = sorted(os.listdir('/var/lib/qubes/vm-kernels'), key=KernelVersion) return prepare_choice( widget, holder, propname, kernels, default, *args, **kwargs) def prepare_label_choice(widget, holder, propname, default, *args, **kwargs): try: app = holder.app except AttributeError: app = holder return prepare_choice(widget, holder, propname, sorted(app.labels.values(), key=lambda l: l.index), default, *args, icon_getter=(lambda label: QIcon.fromTheme(label.icon)), **kwargs) def prepare_vm_choice(widget, holder, propname, default, *args, **kwargs): try: app = holder.app except AttributeError: app = holder return prepare_choice(widget, holder, propname, app.domains, default, *args, **kwargs) def is_debug(): return os.getenv('QUBES_MANAGER_DEBUG', '') not in ('', '0') def debug(*args, **kwargs): if not is_debug(): return print(*args, **kwargs) def get_path_from_vm(vm, service_name): """ Displays a file/directory selection window for the given VM. :param vm: vm from which to select path :param service_name: qubes.SelectFile or qubes.SelectDirectory :return: path to file, checked for validity """ path_re = re.compile(r"[a-zA-Z0-9/:.,_+=() -]*") path_max_len = 512 if not vm: return None stdout, _stderr = vm.run_service_for_stdio(service_name) stdout = stdout.strip() untrusted_path = stdout.decode(encoding='ascii')[:path_max_len] if not untrusted_path: return None if path_re.fullmatch(untrusted_path): assert '../' not in untrusted_path assert '\0' not in untrusted_path return untrusted_path.strip() raise ValueError('Unexpected characters in path.') def format_dependencies_list(dependencies): """Given a list of tuples representing properties, formats them in a readable list.""" list_text = "" for (holder, prop) in dependencies: if holder is None: list_text += "- Global property {}
".format(prop) else: list_text += "- {} for qube {}
".format( prop, holder.name) return list_text def loop_shutdown(): pending = asyncio.Task.all_tasks() for task in pending: with suppress(asyncio.CancelledError): task.cancel() # Bases on the original code by: # Copyright (c) 2002-2007 Pascal Varet def handle_exception(exc_type, exc_value, exc_traceback): filename, line, _, _ = traceback.extract_tb(exc_traceback).pop() filename = os.path.basename(filename) error = "%s: %s" % (exc_type.__name__, exc_value) strace = "" stacktrace = traceback.extract_tb(exc_traceback) while stacktrace: (filename, line, func, txt) = stacktrace.pop() strace += "----\n" strace += "line: %s\n" % txt strace += "func: %s\n" % func strace += "line no.: %d\n" % line strace += "file: %s\n" % filename msg_box = QtWidgets.QMessageBox() msg_box.setDetailedText(strace) msg_box.setIcon(QtWidgets.QMessageBox.Critical) msg_box.setWindowTitle("Houston, we have a problem...") msg_box.setText("Whoops. A critical error has occured. " "This is most likely a bug in Qubes Manager.

" "%s" % error + "
at line %d
of file %s.

" % (line, filename)) msg_box.exec_() def run_asynchronous(app_name, icon, window_class): qt_app = QtWidgets.QApplication(sys.argv) qt_app.setOrganizationName("The Qubes Project") qt_app.setOrganizationDomain("http://qubes-os.org") qt_app.setApplicationName(app_name) qt_app.setWindowIcon(icon) qt_app.lastWindowClosed.connect(loop_shutdown) qubes_app = qubesadmin.Qubes() loop = quamash.QEventLoop(qt_app) asyncio.set_event_loop(loop) dispatcher = events.EventsDispatcher(qubes_app) window = window_class(qt_app, qubes_app, dispatcher) window.show() try: loop.run_until_complete( asyncio.ensure_future(dispatcher.listen_for_events())) except asyncio.CancelledError: pass except Exception: # pylint: disable=broad-except loop_shutdown() exc_type, exc_value, exc_traceback = sys.exc_info()[:3] handle_exception(exc_type, exc_value, exc_traceback) def run_synchronous(app_name, window_class): qt_app = QtWidgets.QApplication(sys.argv) qt_app.setOrganizationName("The Qubes Project") qt_app.setOrganizationDomain("http://qubes-os.org") qt_app.setApplicationName(app_name) sys.excepthook = handle_exception qubes_app = qubesadmin.Qubes() window = window_class(qt_app, qubes_app) window.show() qt_app.exec_() qt_app.exit()