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 qubesadmin import exc

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"""
    try:
        return (vm.klass == 'AdminVM'
                or vm.features.get('internal', False))
    except exc.QubesDaemonAccessError:
        return False


def is_running(vm, default_state):
    """Checks if the VM is running, returns default_state if we have
    insufficient permissions to deteremine that."""
    try:
        return vm.is_running()
    except exc.QubesDaemonAccessError:
        return default_state


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().__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_feature(vm, feature_name, default_value):
    try:
        return vm.features.get(feature_name, default_value)
    except exc.QubesDaemonAccessError:
        return default_value


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 = get_feature(vm, 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'"""
    if not widget.isEnabled():
        return False
    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:
        try:
            default_property = holder.property_get_default(property_name)
        except exc.QubesDaemonAccessError:
            default_property = "ERROR: unavailable"
        if default_property is None:
            default_property = "none"
        choices.append(
            (translate("default ({})").format(default_property),
             qubesadmin.DEFAULT))

    # calculate current (can be default)
    try:
        is_default = holder.property_is_default(property_name)
    except exc.QubesDaemonAccessError:
        is_default = False

    if is_default:
        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()