utils.py 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308
  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 # 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 = 'default ({})'.format(default_string)
  87. # N+1: explicit None
  88. elif item is None:
  89. text = '(none)'
  90. # 1..N: choices
  91. else:
  92. text = str(item)
  93. if transform is not None:
  94. text = transform(text)
  95. if item == oldvalue:
  96. text += ' (current)'
  97. idx = i
  98. widget.insertItem(i, text)
  99. if icon_getter is not None:
  100. icon = icon_getter(item)
  101. if icon is not None:
  102. widget.setItemIcon(i, icon)
  103. widget.setCurrentIndex(idx)
  104. return choice_list, idx
  105. class KernelVersion: # pylint: disable=too-few-public-methods
  106. # Cannot use distutils.version.LooseVersion, because it fails at handling
  107. # versions that have no numbers in them
  108. def __init__(self, string):
  109. self.string = string
  110. self.groups = re.compile(r'(\d+)').split(self.string)
  111. def __lt__(self, other):
  112. for (self_content, other_content) in itertools.zip_longest(
  113. self.groups, other.groups):
  114. if self_content == other_content:
  115. continue
  116. if self_content.isdigit() and other_content.isdigit():
  117. return int(self_content) < int(other_content)
  118. return self_content < other_content
  119. def prepare_kernel_choice(widget, holder, propname, default, *args, **kwargs):
  120. # TODO get from storage API (pool 'linux-kernel') (suggested by @marmarta)
  121. kernels = sorted(os.listdir('/var/lib/qubes/vm-kernels'),
  122. key=KernelVersion)
  123. return prepare_choice(
  124. widget, holder, propname, kernels, default, *args, **kwargs)
  125. def prepare_label_choice(widget, holder, propname, default, *args, **kwargs):
  126. try:
  127. app = holder.app
  128. except AttributeError:
  129. app = holder
  130. return prepare_choice(widget, holder, propname,
  131. sorted(app.labels.values(), key=lambda l: l.index),
  132. default, *args,
  133. icon_getter=(lambda label:
  134. QIcon.fromTheme(label.icon)),
  135. **kwargs)
  136. def prepare_vm_choice(widget, holder, propname, default, *args, **kwargs):
  137. try:
  138. app = holder.app
  139. except AttributeError:
  140. app = holder
  141. return prepare_choice(widget, holder, propname, app.domains, default,
  142. *args, **kwargs)
  143. def is_debug():
  144. return os.getenv('QUBES_MANAGER_DEBUG', '') not in ('', '0')
  145. def debug(*args, **kwargs):
  146. if not is_debug():
  147. return
  148. print(*args, **kwargs)
  149. def get_path_from_vm(vm, service_name):
  150. """
  151. Displays a file/directory selection window for the given VM.
  152. :param vm: vm from which to select path
  153. :param service_name: qubes.SelectFile or qubes.SelectDirectory
  154. :return: path to file, checked for validity
  155. """
  156. path_re = re.compile(r"[a-zA-Z0-9/:.,_+=() -]*")
  157. path_max_len = 512
  158. if not vm:
  159. return None
  160. stdout, _stderr = vm.run_service_for_stdio(service_name)
  161. stdout = stdout.strip()
  162. untrusted_path = stdout.decode(encoding='ascii')[:path_max_len]
  163. if not untrusted_path:
  164. return None
  165. if path_re.fullmatch(untrusted_path):
  166. assert '../' not in untrusted_path
  167. assert '\0' not in untrusted_path
  168. return untrusted_path.strip()
  169. raise ValueError('Unexpected characters in path.')
  170. def format_dependencies_list(dependencies):
  171. """Given a list of tuples representing properties, formats them in
  172. a readable list."""
  173. list_text = ""
  174. for (holder, prop) in dependencies:
  175. if holder is None:
  176. list_text += "- Global property <b>{}</b> <br>".format(prop)
  177. else:
  178. list_text += "- <b>{}</b> for qube <b>{}</b> <br>".format(
  179. prop, holder.name)
  180. return list_text
  181. def loop_shutdown():
  182. pending = asyncio.Task.all_tasks()
  183. for task in pending:
  184. with suppress(asyncio.CancelledError):
  185. task.cancel()
  186. # Bases on the original code by:
  187. # Copyright (c) 2002-2007 Pascal Varet <p.varet@gmail.com>
  188. def handle_exception(exc_type, exc_value, exc_traceback):
  189. filename, line, _, _ = traceback.extract_tb(exc_traceback).pop()
  190. filename = os.path.basename(filename)
  191. error = "%s: %s" % (exc_type.__name__, exc_value)
  192. strace = ""
  193. stacktrace = traceback.extract_tb(exc_traceback)
  194. while stacktrace:
  195. (filename, line, func, txt) = stacktrace.pop()
  196. strace += "----\n"
  197. strace += "line: %s\n" % txt
  198. strace += "func: %s\n" % func
  199. strace += "line no.: %d\n" % line
  200. strace += "file: %s\n" % filename
  201. msg_box = QtWidgets.QMessageBox()
  202. msg_box.setDetailedText(strace)
  203. msg_box.setIcon(QtWidgets.QMessageBox.Critical)
  204. msg_box.setWindowTitle("Houston, we have a problem...")
  205. msg_box.setText("Whoops. A critical error has occured. "
  206. "This is most likely a bug in Qubes Manager.<br><br>"
  207. "<b><i>%s</i></b>" % error +
  208. "<br/>at line <b>%d</b><br/>of file %s.<br/><br/>"
  209. % (line, filename))
  210. msg_box.exec_()
  211. def run_asynchronous(app_name, icon, window_class):
  212. qt_app = QtWidgets.QApplication(sys.argv)
  213. qt_app.setOrganizationName("The Qubes Project")
  214. qt_app.setOrganizationDomain("http://qubes-os.org")
  215. qt_app.setApplicationName(app_name)
  216. qt_app.setWindowIcon(icon)
  217. qt_app.lastWindowClosed.connect(loop_shutdown)
  218. qubes_app = qubesadmin.Qubes()
  219. loop = quamash.QEventLoop(qt_app)
  220. asyncio.set_event_loop(loop)
  221. dispatcher = events.EventsDispatcher(qubes_app)
  222. window = window_class(qt_app, qubes_app, dispatcher)
  223. window.show()
  224. try:
  225. loop.run_until_complete(
  226. asyncio.ensure_future(dispatcher.listen_for_events()))
  227. except asyncio.CancelledError:
  228. pass
  229. except Exception: # pylint: disable=broad-except
  230. loop_shutdown()
  231. exc_type, exc_value, exc_traceback = sys.exc_info()[:3]
  232. handle_exception(exc_type, exc_value, exc_traceback)
  233. def run_synchronous(app_name, window_class):
  234. qt_app = QtWidgets.QApplication(sys.argv)
  235. qt_app.setOrganizationName("The Qubes Project")
  236. qt_app.setOrganizationDomain("http://qubes-os.org")
  237. qt_app.setApplicationName(app_name)
  238. sys.excepthook = handle_exception
  239. qubes_app = qubesadmin.Qubes()
  240. window = window_class(qt_app, qubes_app)
  241. window.show()
  242. qt_app.exec_()
  243. qt_app.exit()