manager/qubesmanager/utils.py

#
# The Qubes OS Project, https://www.qubes-os.org
#
# Copyright (C) 2012  Agnieszka Kostrzewa <agnieszka.kostrzewa@gmail.com>
# Copyright (C) 2012  Marek Marczykowski-Górecki
#                       <marmarek@invisiblethingslab.com>
# Copyright (C) 2017  Wojtek Porczyk <woju@invisiblethingslab.com>
# Copyright (C) 2020  Marta Marczykowska-Górecka
#                       <marmarta@invisiblethingslab.com>
#
# 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 <http://www.gnu.org/licenses/>.
#
import itertools
import os
import re
import qubesadmin
import traceback
import asyncio
from contextlib import suppress
import sys
import qasync
from qubesadmin import events

from PyQt5 import QtWidgets, QtCore, QtGui  # pylint: disable=import-error


# important usage note: which initialize_widget should I use?
# - if you want a list of VMs, use initialize_widget_with_vms, optionally
#   adding a property if you want to handle qubesadmin.DEFAULT and the
#   current (potentially default) value
# - if you want a list of labels or kernals, use
#       initialize_widget_with_kernels/labels
# - list of some things, but associated with a definite property (optionally
#       with qubesadmin.DEFAULT) - initialize_widget_for_property
# - list of some things, not associated with a property, but still having a
#       default - initialize_widget_with_default
# - just a list, no properties or defaults, just a nice list with a "current"
#       value - initialize_widget

def is_internal(vm):
    """checks if the VM is either an AdminVM or has the 'internal' features"""
    return (vm.klass == 'AdminVM'
            or vm.features.get('internal', False))


def translate(string):
    """helper function for translations"""
    return QtCore.QCoreApplication.translate(
        "ManagerUtils", string)


class SizeSpinBox(QtWidgets.QSpinBox):
    """A SpinBox subclass with extended handling for sizes in MB and GB"""
    # pylint: disable=invalid-name, no-self-use
    def __init__(self, *args, **kwargs):
        super(SizeSpinBox, self).__init__(*args, **kwargs)

        self.pattern = r'(\d+\.?\d?) ?(GB|MB)'
        self.regex = re.compile(self.pattern)
        self.validator = QtGui.QRegExpValidator(QtCore.QRegExp(
            self.pattern), self)

    def textFromValue(self, v: int) -> str:
        if v > 1024:
            return '{:.1f} GB'.format(v / 1024)

        return '{} MB'.format(v)

    def validate(self, text: str, pos: int):
        return self.validator.validate(text, pos)

    def valueFromText(self, text: str) -> int:
        value, unit = self.regex.fullmatch(text.strip()).groups()

        if unit == 'GB':
            multiplier = 1024
        else:
            multiplier = 1

        return int(float(value) * multiplier)


def get_boolean_feature(vm, feature_name):
    """heper function to get a feature converted to a Bool if it does exist.
    Necessary because of the true/false in features being coded as 1/empty
    string."""
    result = vm.features.get(feature_name, None)
    if result is not None:
        result = bool(result)
    return result


def did_widget_selection_change(widget):
    """a simple heuristic to check if the widget text contains appropriately
    translated 'current'"""
    return not translate(" (current)") in widget.currentText()


def initialize_widget(widget, choices, selected_value=None,
                      icon_getter=None, add_current_label=True):
    """
    populates widget (ListBox or ComboBox) with items. Previous widget contents
    are erased.
    :param widget: QListBox or QComboBox; must support addItem and findText
    :param choices: list of tuples (text, value) to use to populate widget.
        text should be a string, value can be of any type, including None
    :param selected_value: initial widget value
    :param icon_getter: function of value that returns desired icon
    :param add_current_label: if initial value should be labelled as (current)
    :return:
    """

    widget.clear()
    selected_item = None

    for (name, value) in choices:
        if value == selected_value:
            selected_item = name
        if icon_getter is not None:
            widget.addItem(icon_getter(value), name, userData=value)
        else:
            widget.addItem(name, userData=value)

    if selected_item is not None:
        widget.setCurrentIndex(widget.findText(selected_item))
    else:
        widget.addItem(str(selected_value), selected_value)
        widget.setCurrentIndex(widget.findText(str(selected_value)))

    if add_current_label:
        widget.setItemText(widget.currentIndex(),
                           widget.currentText() + translate(" (current)"))


def initialize_widget_for_property(
        widget, choices, holder, property_name, allow_default=False,
        icon_getter=None, add_current_label=True):
    """
    populates widget (ListBox or ComboBox) with items, based on a listed
    property. Supports discovering the system default for the given property
    and handling qubesadmin.DEFAULT special value. Value of holder.property
    will be set as current item. Previous widget contents are erased.
    :param widget: QListBox or QComboBox; must support addItem and findText
    :param choices: list of tuples (text, value) to use to populate widget.
        text should be a string, value can be of any type, including None
    :param holder: object to use as property_name's holder
    :param property_name: name of the property
    :param allow_default: boolean, should a position with qubesadmin.DEFAULT
        be added; default False
    :param icon_getter: a function applied to values (from choices) that
        returns a QIcon to be used as a item icon; default None
    :param add_current_label: if initial value should be labelled as (current)
    :return:
    """
    if allow_default:
        default_property = holder.property_get_default(property_name)
        if default_property is None:
            default_property = "none"
        choices.append(
            (translate("default ({})").format(default_property),
             qubesadmin.DEFAULT))

    # calculate current (can be default)
    if holder.property_is_default(property_name):
        current_value = qubesadmin.DEFAULT
    else:
        current_value = getattr(holder, property_name)

    initialize_widget(widget,
                      choices,
                      selected_value=current_value,
                      icon_getter=icon_getter,
                      add_current_label=add_current_label)


# TODO: improvement: add optional icon support
def initialize_widget_with_vms(
        widget, qubes_app, filter_function=(lambda x: True),
        allow_none=False, holder=None, property_name=None,
        allow_default=False, allow_internal=False):
    """
    populates widget (ListBox or ComboBox) with vm items, optionally based on
    a given property. Supports discovering the system default for the property
    and handling qubesadmin.DEFAULT special value. Value of holder.property
    will be set as current item. Previous widget contents are erased.
    :param widget: QListBox or QComboBox; must support addItem and findText
    :param qubes_app: Qubes() object
    :param filter_function: function used to filter vms; optional
    :param allow_none: should a None option be added; default False
    :param holder: object to use as property_name's holder
    :param property_name: name of the property
    :param allow_default: should a position with qubesadmin.DEFAULT be added;
        default False
    :param allow_internal: should AdminVMs and vms with feature 'internal' be
        used
    :return:
    """
    choices = []

    for vm in qubes_app.domains:
        if not allow_internal and is_internal(vm):
            continue
        if not filter_function(vm):
            continue
        choices.append((vm.name, vm))

    if allow_none:
        choices.append((translate("(none)"), None))

    if holder is None:
        initialize_widget(widget,
                          choices,
                          selected_value=choices[0][1],
                          add_current_label=False)
    else:
        initialize_widget_for_property(
            widget=widget, choices=choices, holder=holder,
            property_name=property_name, allow_default=allow_default)


def initialize_widget_with_default(
        widget, choices, add_none=False, add_qubes_default=False,
        mark_existing_as_default=False, default_value=None):
    """
    populates widget (ListBox or ComboBox) with items. Used when there is no
    corresponding property, but support for special qubesadmin.DEFAULT value
    is still needed.
    :param widget: QListBox or QComboBox; must support addItem and findText
    :param choices: list of tuples (text, value) to use to populate widget.
        text should be a string, value can be of any type, including None
    :param add_none: should a 'None' position be added
    :param add_qubes_default: should a qubesadmin.DEFAULT position be added
        (requires default_value to be set to something meaningful)
    :param mark_existing_as_default: should an existing value be marked
        as default. If used with conjuction with add_qubes_default, the
        default_value listed will be replaced by qubesadmin.DEFAULT
    :param default_value: what value should be used as the default
    :return:
    """
    added_existing = False

    if mark_existing_as_default:
        existing_default = [item for item in choices
                            if item[1] == default_value]
        if existing_default:
            choices = [item for item in choices if item not in existing_default]

            if add_qubes_default:
                # if for some reason (e.g. storage pools) we want to mark an
                # actual value as default and replace it with qubesadmin.DEFAULT
                default_value = qubesadmin.DEFAULT

            choices.insert(
                0, (translate("default ({})").format(existing_default[0][0]),
                    default_value))
            added_existing = True

    elif add_qubes_default:
        choices.insert(0, (translate("default ({})").format(default_value),
                           qubesadmin.DEFAULT))

    if add_none:
        if mark_existing_as_default and default_value is None and \
                not added_existing:
            choices.append((translate("default (none)"), None))
        else:
            choices.append((translate("(none)"), None))

    if add_qubes_default:
        selected_value = qubesadmin.DEFAULT
    elif mark_existing_as_default:
        selected_value = default_value
    else:
        selected_value = choices[0][1]

    initialize_widget(
        widget=widget, choices=choices, selected_value=selected_value,
        add_current_label=False)


def initialize_widget_with_kernels(
        widget, qubes_app, allow_none=False, holder=None,
        property_name=None, allow_default=False):
    """
    populates widget (ListBox or ComboBox) with kernel items, based on a given
    property. Supports discovering the system default for the property
    and handling qubesadmin.DEFAULT special value. Value of holder.property
    will be set as current item. Previous widget contents are erased.
    :param widget: QListBox or QComboBox; must support addItem and findText
    :param qubes_app: Qubes() object
    :param allow_none: should a None item be added
    :param holder: object to use as property_name's holder
    :param property_name: name of the property
    :param allow_default: should a qubesadmin.DEFAULT item be added
    :return:
    """
    kernels = [kernel.vid for kernel in qubes_app.pools['linux-kernel'].volumes]
    kernels = sorted(kernels, key=KernelVersion)

    choices = [(kernel, kernel) for kernel in kernels]

    if allow_none:
        choices.append((translate("(none)"), None))

    initialize_widget_for_property(
        widget=widget, choices=choices, holder=holder,
        property_name=property_name, allow_default=allow_default)


def initialize_widget_with_labels(widget, qubes_app,
                                  holder=None, property_name='label'):
    """
    populates widget (ListBox or ComboBox) with label items, optionally based
    on a given property. Value of holder.property will be set as current item.
    Previous widget contents are erased.
    :param widget: QListBox or QComboBox; must support addItem and findText
    :param qubes_app: Qubes() object
    :param holder: object to use as property_name's holder; can be None
    :param property_name: name of the property
    :return:
    """
    labels = sorted(qubes_app.labels.values(), key=lambda l: l.index)
    choices = [(label.name, label) for label in labels]

    icon_getter = (lambda label:
                   QtGui.QIcon.fromTheme(label.icon))

    if holder:
        initialize_widget_for_property(widget=widget,
                                       choices=choices,
                                       holder=holder,
                                       property_name=property_name,
                                       icon_getter=icon_getter)
    else:
        initialize_widget(widget=widget,
                          choices=choices,
                          selected_value=labels[0],
                          icon_getter=icon_getter,
                          add_current_label=False)


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 is None:
                return True
            if other_content is None:
                return False
            if self_content.isdigit() and other_content.isdigit():
                return int(self_content) < int(other_content)
            return self_content < other_content


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(QtCore.QCoreApplication.translate(
        "ManagerUtils", '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 += QtCore.QCoreApplication.translate(
                "ManagerUtils", "- Global property <b>{}</b> <br>").format(prop)
        else:
            list_text += QtCore.QCoreApplication.translate(
                "ManagerUtils", "- <b>{0}</b> for qube <b>{1}</b> <br>").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 <p.varet@gmail.com>
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(QtCore.QCoreApplication.translate(
        "ManagerUtils", "Houston, we have a problem..."))
    msg_box.setText(QtCore.QCoreApplication.translate(
        "ManagerUtils", "Whoops. A critical error has occured. "
                        "This is most likely a bug in Qubes Manager.<br><br>"
                        "<b><i>{0}</i></b><br/>at line <b>{1}</b><br/>of file "
                        "{2}.<br/><br/>").format(error, line, filename))

    msg_box.exec_()


def run_asynchronous(window_class):
    qt_app = QtWidgets.QApplication(sys.argv)

    translator = QtCore.QTranslator(qt_app)
    locale = QtCore.QLocale.system().name()
    i18n_dir = os.path.join(
        os.path.dirname(os.path.realpath(__file__)),
        'i18n')
    translator.load("qubesmanager_{!s}.qm".format(locale), i18n_dir)
    qt_app.installTranslator(translator)
    QtCore.QCoreApplication.installTranslator(translator)

    qt_app.setOrganizationName("The Qubes Project")
    qt_app.setOrganizationDomain("http://qubes-os.org")
    qt_app.lastWindowClosed.connect(loop_shutdown)

    qubes_app = qubesadmin.Qubes()

    loop = qasync.QEventLoop(qt_app)
    asyncio.set_event_loop(loop)
    dispatcher = events.EventsDispatcher(qubes_app)

    window = window_class(qt_app, qubes_app, dispatcher)

    if hasattr(window, "setup_application"):
        window.setup_application()

    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(window_class):
    qt_app = QtWidgets.QApplication(sys.argv)

    translator = QtCore.QTranslator(qt_app)
    locale = QtCore.QLocale.system().name()
    i18n_dir = os.path.join(
        os.path.dirname(os.path.realpath(__file__)),
        'i18n')
    translator.load("qubesmanager_{!s}.qm".format(locale), i18n_dir)
    qt_app.installTranslator(translator)
    QtCore.QCoreApplication.installTranslator(translator)

    qt_app.setOrganizationName("The Qubes Project")
    qt_app.setOrganizationDomain("http://qubes-os.org")

    sys.excepthook = handle_exception

    qubes_app = qubesadmin.Qubes()

    window = window_class(qt_app, qubes_app)

    if hasattr(window, "setup_application"):
        window.setup_application()

    window.show()

    qt_app.exec_()
    qt_app.exit()