499 lines
16 KiB
Python
499 lines
16 KiB
Python
#
|
|
# 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>
|
|
#
|
|
# 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 quamash
|
|
from qubesadmin import events
|
|
|
|
from PyQt5 import QtWidgets, QtCore, QtGui # pylint: disable=import-error
|
|
|
|
|
|
#TODO: remove
|
|
def _filter_internal(vm):
|
|
return (not vm.klass == 'AdminVM'
|
|
and not vm.features.get('internal', False))
|
|
|
|
|
|
def is_internal(vm):
|
|
return (vm.klass == 'AdminVM'
|
|
or vm.features.get('internal', False))
|
|
|
|
|
|
def translate(string):
|
|
return QtCore.QCoreApplication.translate(
|
|
"ManagerUtils", string)
|
|
|
|
|
|
class SizeSpinBox(QtWidgets.QSpinBox):
|
|
# 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):
|
|
result = vm.features.get(feature_name, None)
|
|
if result is not None:
|
|
result = bool(result)
|
|
return result
|
|
|
|
# TODO: doublecheck translation
|
|
|
|
|
|
def did_widget_selection_change(widget):
|
|
return not translate(" (current)") in widget.currentText()
|
|
|
|
|
|
def initialize_widget(widget, choices,
|
|
selected_value=None):
|
|
"""
|
|
populates widget (ListBox or ComboBox) with items. Previous widget contents
|
|
are erased.
|
|
:param widget: widget to populate
|
|
:param choices: list of tuples (text, value) to use to populate widget
|
|
:param selected_value: value to populate widget with
|
|
:return:
|
|
"""
|
|
|
|
widget.clear()
|
|
selected_item = None
|
|
|
|
for (name, value) in choices:
|
|
if value == selected_value:
|
|
selected_item = name
|
|
widget.addItem(name, 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)))
|
|
|
|
widget.setItemText(widget.currentIndex(),
|
|
widget.currentText() + translate(" (current)"))
|
|
|
|
|
|
def initialize_widget_for_property(
|
|
widget, choices, holder, property_name, allow_default=False):
|
|
# potentially add default
|
|
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)
|
|
|
|
|
|
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):
|
|
choices = []
|
|
|
|
for vm in qubes_app.domains:
|
|
if not allow_internal and is_internal(vm):
|
|
continue
|
|
if not filter_function(vm):
|
|
continue
|
|
else:
|
|
choices.append((vm.name, vm))
|
|
|
|
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_kernels(widget,
|
|
qubes_app,
|
|
allow_none=False,
|
|
holder=None,
|
|
property_name=None,
|
|
allow_default=False
|
|
):
|
|
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 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
|
|
|
|
# clear the widget, so that prepare_choice functions can be used
|
|
# to refresh widget values
|
|
while widget.count() > 0:
|
|
widget.removeItem(0)
|
|
|
|
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 = QtCore.QCoreApplication.translate(
|
|
"ManagerUtils", 'default ({})').format(default_string)
|
|
# N+1: explicit None
|
|
elif item is None:
|
|
text = QtCore.QCoreApplication.translate("ManagerUtils", '(none)')
|
|
# 1..N: choices
|
|
else:
|
|
text = str(item)
|
|
if transform is not None:
|
|
text = transform(text)
|
|
|
|
if item == oldvalue:
|
|
text += QtCore.QCoreApplication.translate(
|
|
"ManagerUtils", ' (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 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 prepare_kernel_choice(widget, holder, propname, default, *args, **kwargs):
|
|
try:
|
|
app = holder.app
|
|
except AttributeError:
|
|
app = holder
|
|
kernels = [kernel.vid for kernel in app.pools['linux-kernel'].volumes]
|
|
kernels = sorted(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:
|
|
QtGui.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(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 = quamash.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()
|