#!/usr/bin/python3 # # The Qubes OS Project, http://www.qubes-os.org # # Copyright (C) 2012 Agnieszka Kostrzewa # Copyright (C) 2012 Marek Marczykowski # # 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 Lesser General Public License along # with this program; if not, see . # # import os import subprocess from PyQt5 import QtWidgets, QtCore, QtGui # pylint: disable=import-error from qubesadmin.utils import parse_size from . import ui_globalsettingsdlg # pylint: disable=no-name-in-module from . import utils from configparser import ConfigParser qmemman_config_path = '/etc/qubes/qmemman.conf' def _run_qrexec_repo(service, arg=''): # Set default locale to C in order to prevent error msg # in subprocess call related to falling back to C locale env = os.environ.copy() env['LC_ALL'] = 'C' # Fake up a "qrexec call" to dom0 because dom0 can't qrexec to itself yet cmd = '/etc/qubes-rpc/' + service p = subprocess.run( ['sudo', cmd, arg], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, env=env ) if p.stderr: raise RuntimeError( QtCore.QCoreApplication.translate( "GlobalSettings", 'qrexec call stderr was not empty'), {'stderr': p.stderr.decode('utf-8')}) if p.returncode != 0: raise RuntimeError( QtCore.QCoreApplication.translate( "GlobalSettings", 'qrexec call exited with non-zero return code'), {'returncode': p.returncode}) return p.stdout.decode('utf-8') # pylint: disable=too-many-instance-attributes class GlobalSettingsWindow(ui_globalsettingsdlg.Ui_GlobalSettings, QtWidgets.QDialog): def __init__(self, app, qvm_collection, parent=None): super(GlobalSettingsWindow, self).__init__(parent) self.app = app self.qvm_collection = qvm_collection self.setupUi(self) self.buttonBox.accepted.connect(self.save_and_apply) self.buttonBox.rejected.connect(self.reject) self.__init_system_defaults__() self.__init_kernel_defaults__() self.__init_mem_defaults__() self.__init_updates__() def setup_application(self): self.app.setApplicationName(self.tr("Qubes Global Settings")) self.app.setWindowIcon(QtGui.QIcon.fromTheme("qubes-manager")) def __init_system_defaults__(self): # set up updatevm choice self.update_vm_vmlist, self.update_vm_idx = utils.prepare_vm_choice( self.update_vm_combo, self.qvm_collection, 'updatevm', None, allow_none=True, filter_function=(lambda vm: vm.klass != 'TemplateVM') ) # set up clockvm choice self.clock_vm_vmlist, self.clock_vm_idx = utils.prepare_vm_choice( self.clock_vm_combo, self.qvm_collection, 'clockvm', None, allow_none=True, filter_function=(lambda vm: vm.klass != 'TemplateVM') ) # set up default netvm self.default_netvm_vmlist, self.default_netvm_idx = \ utils.prepare_vm_choice( self.default_netvm_combo, self.qvm_collection, 'default_netvm', None, filter_function=(lambda vm: vm.provides_network), allow_none=True) # default template self.default_template_vmlist, self.default_template_idx = \ utils.prepare_vm_choice( self.default_template_combo, self.qvm_collection, 'default_template', None, filter_function=(lambda vm: vm.klass == 'TemplateVM'), allow_none=True ) # default dispvm self.default_dispvm_vmlist, self.default_dispvm_idx = \ utils.prepare_vm_choice( self.default_dispvm_combo, self.qvm_collection, 'default_dispvm', None, (lambda vm: getattr(vm, 'template_for_dispvms', False)), allow_none=True ) def __apply_system_defaults__(self): # updatevm if self.qvm_collection.updatevm != \ self.update_vm_vmlist[self.update_vm_combo.currentIndex()]: self.qvm_collection.updatevm = \ self.update_vm_vmlist[self.update_vm_combo.currentIndex()] # clockvm if self.qvm_collection.clockvm != \ self.clock_vm_vmlist[self.clock_vm_combo.currentIndex()]: self.qvm_collection.clockvm = \ self.clock_vm_vmlist[self.clock_vm_combo.currentIndex()] # default netvm if self.qvm_collection.default_netvm != \ self.default_netvm_vmlist[ self.default_netvm_combo.currentIndex()]: self.qvm_collection.default_netvm = \ self.default_netvm_vmlist[ self.default_netvm_combo.currentIndex()] # default template if self.qvm_collection.default_template != \ self.default_template_vmlist[ self.default_template_combo.currentIndex()]: self.qvm_collection.default_template = \ self.default_template_vmlist[ self.default_template_combo.currentIndex()] # default_dispvm if self.qvm_collection.default_dispvm != \ self.default_dispvm_vmlist[ self.default_dispvm_combo.currentIndex()]: self.qvm_collection.default_dispvm = \ self.default_dispvm_vmlist[ self.default_dispvm_combo.currentIndex()] def __init_kernel_defaults__(self): self.kernels_list, self.kernels_idx = utils.prepare_kernel_choice( self.default_kernel_combo, self.qvm_collection, 'default_kernel', None, allow_none=True ) def __apply_kernel_defaults__(self): if self.qvm_collection.default_kernel != \ self.kernels_list[self.default_kernel_combo.currentIndex()]: self.qvm_collection.default_kernel = \ self.kernels_list[self.default_kernel_combo.currentIndex()] def __init_mem_defaults__(self): # qmemman settings self.qmemman_config = ConfigParser() self.vm_min_mem_val = '200MiB' # str(qmemman_algo.MIN_PREFMEM) self.dom0_mem_boost_val = '350MiB' # str(qmemman_algo.DOM0_MEM_BOOST) self.qmemman_config.read(qmemman_config_path) if self.qmemman_config.has_section('global'): self.vm_min_mem_val = \ self.qmemman_config.get('global', 'vm-min-mem') self.dom0_mem_boost_val = \ self.qmemman_config.get('global', 'dom0-mem-boost') self.vm_min_mem_val = parse_size(self.vm_min_mem_val) self.dom0_mem_boost_val = parse_size(self.dom0_mem_boost_val) self.min_vm_mem.setValue(self.vm_min_mem_val / 1024 / 1024) self.dom0_mem_boost.setValue(self.dom0_mem_boost_val / 1024 / 1024) def __apply_mem_defaults__(self): # qmemman settings current_min_vm_mem = self.min_vm_mem.value() current_dom0_mem_boost = self.dom0_mem_boost.value() if current_min_vm_mem * 1024 * 1024 != self.vm_min_mem_val or \ current_dom0_mem_boost * 1024 * 1024 != self.dom0_mem_boost_val: current_min_vm_mem = str(current_min_vm_mem) + 'MiB' current_dom0_mem_boost = str(current_dom0_mem_boost) + 'MiB' if not self.qmemman_config.has_section('global'): # add the whole section self.qmemman_config.add_section('global') self.qmemman_config.set( 'global', 'vm-min-mem', current_min_vm_mem) self.qmemman_config.set( 'global', 'dom0-mem-boost', current_dom0_mem_boost) self.qmemman_config.set( 'global', 'cache-margin-factor', str(1.3)) # removed qmemman_algo.CACHE_FACTOR qmemman_config_file = open(qmemman_config_path, 'a') self.qmemman_config.write(qmemman_config_file) qmemman_config_file.close() else: # If there already is a 'global' section, we don't use # SafeConfigParser.write() - it would get rid of # all the comments... lines_to_add = {} lines_to_add['vm-min-mem'] = \ "vm-min-mem = " + current_min_vm_mem + "\n" lines_to_add['dom0-mem-boost'] = \ "dom0-mem-boost = " + current_dom0_mem_boost + "\n" config_lines = [] qmemman_config_file = open(qmemman_config_path, 'r') for line in qmemman_config_file: if line.strip().startswith('vm-min-mem'): config_lines.append(lines_to_add['vm-min-mem']) del lines_to_add['vm-min-mem'] elif line.strip().startswith('dom0-mem-boost'): config_lines.append(lines_to_add['dom0-mem-boost']) del lines_to_add['dom0-mem-boost'] else: config_lines.append(line) qmemman_config_file.close() for line in lines_to_add: config_lines.append(line) qmemman_config_file = open(qmemman_config_path, 'w') qmemman_config_file.writelines(config_lines) qmemman_config_file.close() def __init_updates__(self): try: self.updates_dom0_val = bool(self.qvm_collection.domains[ 'dom0'].features[ 'service.qubes-update-check']) except KeyError: self.updates_dom0_val = True self.updates_dom0.setChecked(self.updates_dom0_val) self.updates_vm.setChecked(self.qvm_collection.check_updates_vm) self.enable_updates_all.clicked.connect(self.__enable_updates_all) self.disable_updates_all.clicked.connect(self.__disable_updates_all) self.repos = repos = dict() for i in _run_qrexec_repo('qubes.repos.List').split('\n'): lst = i.split('\0') # Keyed by repo name dct = repos[lst[0]] = dict() dct['prettyname'] = lst[1] dct['enabled'] = lst[2] == 'enabled' if repos['qubes-dom0-unstable']['enabled']: self.dom0_updates_repo.setCurrentIndex(3) elif repos['qubes-dom0-current-testing']['enabled']: self.dom0_updates_repo.setCurrentIndex(2) elif repos['qubes-dom0-security-testing']['enabled']: self.dom0_updates_repo.setCurrentIndex(1) elif repos['qubes-dom0-current']['enabled']: self.dom0_updates_repo.setCurrentIndex(0) else: raise Exception( self.tr('Cannot detect enabled dom0 update repositories')) if repos['qubes-templates-itl-testing']['enabled']: self.itl_tmpl_updates_repo.setCurrentIndex(1) elif repos['qubes-templates-itl']['enabled']: self.itl_tmpl_updates_repo.setCurrentIndex(0) else: raise Exception(self.tr('Cannot detect enabled ITL template update ' 'repositories')) if repos['qubes-templates-community-testing']['enabled']: self.comm_tmpl_updates_repo.setCurrentIndex(2) elif repos['qubes-templates-community']['enabled']: self.comm_tmpl_updates_repo.setCurrentIndex(1) else: self.comm_tmpl_updates_repo.setCurrentIndex(0) def __enable_updates_all(self): reply = QtWidgets.QMessageBox.question( self, self.tr("Change state of all qubes"), self.tr("Are you sure you want to set all qubes to check " "for updates?"), QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.Cancel) if reply == QtWidgets.QMessageBox.Cancel: return self.__set_updates_all(True) def __disable_updates_all(self): reply = QtWidgets.QMessageBox.question( self, self.tr("Change state of all qubes"), self.tr("Are you sure you want to set all qubes to not check " "for updates?"), QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.Cancel) if reply == QtWidgets.QMessageBox.Cancel: return self.__set_updates_all(False) def __set_updates_all(self, state): for vm in self.qvm_collection.domains: if vm.klass != "AdminVM": vm.features['service.qubes-update-check'] = state def __apply_updates__(self): if self.updates_dom0.isChecked() != self.updates_dom0_val: self.qvm_collection.domains['dom0'].features[ 'service.qubes-update-check'] = \ self.updates_dom0.isChecked() if self.qvm_collection.check_updates_vm != self.updates_vm.isChecked(): self.qvm_collection.check_updates_vm = self.updates_vm.isChecked() def _manage_repos(self, repolist, action): for name in repolist: if self.repos[name]['enabled'] and action == 'Enable' or \ not self.repos[name]['enabled'] and action == 'Disable': continue try: result = _run_qrexec_repo('qubes.repos.' + action, name) if result != 'ok\n': raise RuntimeError( self.tr('qrexec call stdout did not contain "ok"' ' as expected'), {'stdout': result}) except RuntimeError as ex: msg = '{desc}; {args}'.format(desc=ex.args[0], args=', '.join( # This is kind of hard to mentally parse but really all # it does is pretty-print args[1], which is a dictionary ['{key}: {val}'.format(key=i[0], val=i[1]) for i in ex.args[1].items()] )) QtWidgets.QMessageBox.warning( None, self.tr("ERROR!"), self.tr("Error managing {repo} repository settings:" " {msg}".format(repo=name, msg=msg))) def _handle_dom0_updates_combobox(self, idx): idx += 1 repolist = ['qubes-dom0-current', 'qubes-dom0-security-testing', 'qubes-dom0-current-testing', 'qubes-dom0-unstable'] enable = repolist[:idx] disable = repolist[idx:] self._manage_repos(enable, 'Enable') self._manage_repos(disable, 'Disable') # pylint: disable=invalid-name def _handle_itl_tmpl_updates_combobox(self, idx): idx += 1 repolist = ['qubes-templates-itl', 'qubes-templates-itl-testing'] enable = repolist[:idx] disable = repolist[idx:] self._manage_repos(enable, 'Enable') self._manage_repos(disable, 'Disable') # pylint: disable=invalid-name def _handle_comm_tmpl_updates_combobox(self, idx): # We don't increment idx by 1 because this is the only combobox that # has an explicit "disable this repository entirely" option repolist = ['qubes-templates-community', 'qubes-templates-community-testing'] enable = repolist[:idx] disable = repolist[idx:] self._manage_repos(enable, 'Enable') self._manage_repos(disable, 'Disable') def __apply_repos__(self): self._handle_dom0_updates_combobox( self.dom0_updates_repo.currentIndex()) self._handle_itl_tmpl_updates_combobox( self.itl_tmpl_updates_repo.currentIndex()) self._handle_comm_tmpl_updates_combobox( self.comm_tmpl_updates_repo.currentIndex()) def reject(self): self.done(0) def save_and_apply(self): self.__apply_system_defaults__() self.__apply_kernel_defaults__() self.__apply_mem_defaults__() self.__apply_updates__() self.__apply_repos__() def main(): utils.run_synchronous(GlobalSettingsWindow) if __name__ == "__main__": main()