utils.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345
  1. #
  2. # The Qubes OS Project, https://www.qubes-os.org
  3. #
  4. # Copyright (C) 2012 Agnieszka Kostrzewa <agnieszka.kostrzewa@gmail.com>
  5. # Copyright (C) 2012 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. # Copyright (C) 2017 Wojtek Porczyk <woju@invisiblethingslab.com>
  8. #
  9. # This program is free software: you can redistribute it and/or modify
  10. # it under the terms of the GNU General Public License as published by
  11. # the Free Software Foundation, either version 2 of the License, or
  12. # (at your option) any later version.
  13. #
  14. # This program is distributed in the hope that it will be useful,
  15. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17. # GNU General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU General Public License
  20. # along with this program. If not, see <http://www.gnu.org/licenses/>.
  21. #
  22. import itertools
  23. import os
  24. import re
  25. import qubesadmin
  26. import traceback
  27. import asyncio
  28. from contextlib import suppress
  29. import sys
  30. import quamash
  31. from qubesadmin import events
  32. from PyQt5 import QtWidgets, QtCore # pylint: disable=import-error
  33. from PyQt5.QtGui import QIcon # pylint: disable=import-error
  34. def _filter_internal(vm):
  35. return (not vm.klass == 'AdminVM'
  36. and not vm.features.get('internal', False))
  37. def prepare_choice(widget, holder, propname, choice, default,
  38. filter_function=None, *,
  39. icon_getter=None, allow_internal=None, allow_default=False,
  40. allow_none=False, transform=None):
  41. # for newly created vms, set propname to None
  42. debug(
  43. 'prepare_choice(widget={widget!r}, '
  44. 'holder={holder!r}, '
  45. 'propname={propname!r}, '
  46. 'choice={choice!r}, '
  47. 'default={default!r}, '
  48. 'filter_function={filter_function!r}, '
  49. 'icon_getter={icon_getter!r}, '
  50. 'allow_internal={allow_internal!r}, '
  51. 'allow_default={allow_default!r}, '
  52. 'allow_none={allow_none!r})'.format(**locals()))
  53. if propname is not None and allow_default:
  54. default = holder.property_get_default(propname)
  55. if allow_internal is None:
  56. allow_internal = propname is None or not propname.endswith('vm')
  57. if propname is not None:
  58. if holder.property_is_default(propname):
  59. oldvalue = qubesadmin.DEFAULT
  60. else:
  61. oldvalue = getattr(holder, propname)
  62. if oldvalue == '':
  63. oldvalue = None
  64. if transform is not None and oldvalue is not None:
  65. oldvalue = transform(oldvalue)
  66. else:
  67. oldvalue = object() # won't match for identity
  68. idx = 0
  69. choice_list = list(choice)[:]
  70. if not allow_internal:
  71. choice_list = filter(_filter_internal, choice_list)
  72. if filter_function is not None:
  73. choice_list = filter(filter_function, choice_list)
  74. choice_list = list(choice_list)
  75. if allow_default:
  76. choice_list.insert(0, qubesadmin.DEFAULT)
  77. if allow_none:
  78. choice_list.append(None)
  79. for i, item in enumerate(choice_list):
  80. debug('i={} item={}'.format(i, item))
  81. # 0: default (unset)
  82. if item is qubesadmin.DEFAULT:
  83. default_string = str(default) if default is not None else 'none'
  84. if transform is not None:
  85. default_string = transform(default_string)
  86. text = QtCore.QCoreApplication.translate(
  87. "ManagerUtils", 'default ({})').format(default_string)
  88. # N+1: explicit None
  89. elif item is None:
  90. text = QtCore.QCoreApplication.translate("ManagerUtils", '(none)')
  91. # 1..N: choices
  92. else:
  93. text = str(item)
  94. if transform is not None:
  95. text = transform(text)
  96. if item == oldvalue:
  97. text += QtCore.QCoreApplication.translate(
  98. "ManagerUtils", ' (current)')
  99. idx = i
  100. widget.insertItem(i, text)
  101. if icon_getter is not None:
  102. icon = icon_getter(item)
  103. if icon is not None:
  104. widget.setItemIcon(i, icon)
  105. widget.setCurrentIndex(idx)
  106. return choice_list, idx
  107. class KernelVersion: # pylint: disable=too-few-public-methods
  108. # Cannot use distutils.version.LooseVersion, because it fails at handling
  109. # versions that have no numbers in them
  110. def __init__(self, string):
  111. self.string = string
  112. self.groups = re.compile(r'(\d+)').split(self.string)
  113. def __lt__(self, other):
  114. for (self_content, other_content) in itertools.zip_longest(
  115. self.groups, other.groups):
  116. if self_content == other_content:
  117. continue
  118. if self_content is None:
  119. return True
  120. if other_content is None:
  121. return False
  122. if self_content.isdigit() and other_content.isdigit():
  123. return int(self_content) < int(other_content)
  124. return self_content < other_content
  125. def prepare_kernel_choice(widget, holder, propname, default, *args, **kwargs):
  126. try:
  127. app = holder.app
  128. except AttributeError:
  129. app = holder
  130. kernels = [kernel.vid for kernel in app.pools['linux-kernel'].volumes]
  131. kernels = sorted(kernels, key=KernelVersion)
  132. return prepare_choice(
  133. widget, holder, propname, kernels, default, *args, **kwargs)
  134. def prepare_label_choice(widget, holder, propname, default, *args, **kwargs):
  135. try:
  136. app = holder.app
  137. except AttributeError:
  138. app = holder
  139. return prepare_choice(widget, holder, propname,
  140. sorted(app.labels.values(), key=lambda l: l.index),
  141. default, *args,
  142. icon_getter=(lambda label:
  143. QIcon.fromTheme(label.icon)),
  144. **kwargs)
  145. def prepare_vm_choice(widget, holder, propname, default, *args, **kwargs):
  146. try:
  147. app = holder.app
  148. except AttributeError:
  149. app = holder
  150. return prepare_choice(widget, holder, propname, app.domains, default,
  151. *args, **kwargs)
  152. def is_debug():
  153. return os.getenv('QUBES_MANAGER_DEBUG', '') not in ('', '0')
  154. def debug(*args, **kwargs):
  155. if not is_debug():
  156. return
  157. print(*args, **kwargs)
  158. def get_path_from_vm(vm, service_name):
  159. """
  160. Displays a file/directory selection window for the given VM.
  161. :param vm: vm from which to select path
  162. :param service_name: qubes.SelectFile or qubes.SelectDirectory
  163. :return: path to file, checked for validity
  164. """
  165. path_re = re.compile(r"[a-zA-Z0-9/:.,_+=() -]*")
  166. path_max_len = 512
  167. if not vm:
  168. return None
  169. stdout, _stderr = vm.run_service_for_stdio(service_name)
  170. stdout = stdout.strip()
  171. untrusted_path = stdout.decode(encoding='ascii')[:path_max_len]
  172. if not untrusted_path:
  173. return None
  174. if path_re.fullmatch(untrusted_path):
  175. assert '../' not in untrusted_path
  176. assert '\0' not in untrusted_path
  177. return untrusted_path.strip()
  178. raise ValueError(QtCore.QCoreApplication.translate(
  179. "ManagerUtils", 'Unexpected characters in path.'))
  180. def format_dependencies_list(dependencies):
  181. """Given a list of tuples representing properties, formats them in
  182. a readable list."""
  183. list_text = ""
  184. for (holder, prop) in dependencies:
  185. if holder is None:
  186. list_text += QtCore.QCoreApplication.translate(
  187. "ManagerUtils", "- Global property <b>{}</b> <br>").format(prop)
  188. else:
  189. list_text += QtCore.QCoreApplication.translate(
  190. "ManagerUtils", "- <b>{0}</b> for qube <b>{1}</b> <br>").format(
  191. prop, holder.name)
  192. return list_text
  193. def loop_shutdown():
  194. pending = asyncio.Task.all_tasks()
  195. for task in pending:
  196. with suppress(asyncio.CancelledError):
  197. task.cancel()
  198. # Bases on the original code by:
  199. # Copyright (c) 2002-2007 Pascal Varet <p.varet@gmail.com>
  200. def handle_exception(exc_type, exc_value, exc_traceback):
  201. filename, line, _, _ = traceback.extract_tb(exc_traceback).pop()
  202. filename = os.path.basename(filename)
  203. error = "%s: %s" % (exc_type.__name__, exc_value)
  204. strace = ""
  205. stacktrace = traceback.extract_tb(exc_traceback)
  206. while stacktrace:
  207. (filename, line, func, txt) = stacktrace.pop()
  208. strace += "----\n"
  209. strace += "line: %s\n" % txt
  210. strace += "func: %s\n" % func
  211. strace += "line no.: %d\n" % line
  212. strace += "file: %s\n" % filename
  213. msg_box = QtWidgets.QMessageBox()
  214. msg_box.setDetailedText(strace)
  215. msg_box.setIcon(QtWidgets.QMessageBox.Critical)
  216. msg_box.setWindowTitle(QtCore.QCoreApplication.translate(
  217. "ManagerUtils", "Houston, we have a problem..."))
  218. msg_box.setText(QtCore.QCoreApplication.translate(
  219. "ManagerUtils", "Whoops. A critical error has occured. "
  220. "This is most likely a bug in Qubes Manager.<br><br>"
  221. "<b><i>{0}</i></b><br/>at line <b>{1}</b><br/>of file "
  222. "{2}.<br/><br/>").format(error, line, filename))
  223. msg_box.exec_()
  224. def run_asynchronous(window_class):
  225. qt_app = QtWidgets.QApplication(sys.argv)
  226. translator = QtCore.QTranslator(qt_app)
  227. locale = QtCore.QLocale.system().name()
  228. i18n_dir = os.path.join(
  229. os.path.dirname(os.path.realpath(__file__)),
  230. 'i18n')
  231. translator.load("qubesmanager_{!s}.qm".format(locale), i18n_dir)
  232. qt_app.installTranslator(translator)
  233. QtCore.QCoreApplication.installTranslator(translator)
  234. qt_app.setOrganizationName("The Qubes Project")
  235. qt_app.setOrganizationDomain("http://qubes-os.org")
  236. qt_app.lastWindowClosed.connect(loop_shutdown)
  237. qubes_app = qubesadmin.Qubes()
  238. loop = quamash.QEventLoop(qt_app)
  239. asyncio.set_event_loop(loop)
  240. dispatcher = events.EventsDispatcher(qubes_app)
  241. window = window_class(qt_app, qubes_app, dispatcher)
  242. if hasattr(window, "setup_application"):
  243. window.setup_application()
  244. window.show()
  245. try:
  246. loop.run_until_complete(
  247. asyncio.ensure_future(dispatcher.listen_for_events()))
  248. except asyncio.CancelledError:
  249. pass
  250. except Exception: # pylint: disable=broad-except
  251. loop_shutdown()
  252. exc_type, exc_value, exc_traceback = sys.exc_info()[:3]
  253. handle_exception(exc_type, exc_value, exc_traceback)
  254. def run_synchronous(window_class):
  255. qt_app = QtWidgets.QApplication(sys.argv)
  256. translator = QtCore.QTranslator(qt_app)
  257. locale = QtCore.QLocale.system().name()
  258. i18n_dir = os.path.join(
  259. os.path.dirname(os.path.realpath(__file__)),
  260. 'i18n')
  261. translator.load("qubesmanager_{!s}.qm".format(locale), i18n_dir)
  262. qt_app.installTranslator(translator)
  263. QtCore.QCoreApplication.installTranslator(translator)
  264. qt_app.setOrganizationName("The Qubes Project")
  265. qt_app.setOrganizationDomain("http://qubes-os.org")
  266. sys.excepthook = handle_exception
  267. qubes_app = qubesadmin.Qubes()
  268. window = window_class(qt_app, qubes_app)
  269. if hasattr(window, "setup_application"):
  270. window.setup_application()
  271. window.show()
  272. qt_app.exec_()
  273. qt_app.exit()