__init__.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436
  1. #!/usr/bin/python2 -O
  2. # vim: fileencoding=utf-8
  3. #
  4. # The Qubes OS Project, https://www.qubes-os.org/
  5. #
  6. # Copyright (C) 2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
  7. # Copyright (C) 2015 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 along
  20. # with this program; if not, write to the Free Software Foundation, Inc.,
  21. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  22. #
  23. '''Qubes' command line tools
  24. '''
  25. from __future__ import print_function
  26. import argparse
  27. import importlib
  28. import logging
  29. import os
  30. import subprocess
  31. import sys
  32. import textwrap
  33. import qubes.log
  34. #: constant returned when some action should be performed on all qubes
  35. VM_ALL = object()
  36. class QubesAction(argparse.Action):
  37. ''' Interface providing a convinience method to be called, after
  38. `namespace.app` is instantiated.
  39. '''
  40. # pylint: disable=too-few-public-methods
  41. def parse_qubes_app(self, parser, namespace):
  42. ''' This method is called by :py:class:`qubes.tools.QubesArgumentParser`
  43. after the `namespace.app` is instantiated. Oerwrite this method when
  44. extending :py:class:`qubes.tools.QubesAction` to initialized values
  45. based on the `namespace.app`
  46. '''
  47. raise NotImplementedError
  48. class PropertyAction(argparse.Action):
  49. '''Action for argument parser that stores a property.'''
  50. # pylint: disable=redefined-builtin,too-few-public-methods
  51. def __init__(self,
  52. option_strings,
  53. dest,
  54. metavar='NAME=VALUE',
  55. required=False,
  56. help='set property to a value'):
  57. super(PropertyAction, self).__init__(option_strings, 'properties',
  58. metavar=metavar, default={}, help=help)
  59. def __call__(self, parser, namespace, values, option_string=None):
  60. try:
  61. prop, value = values.split('=', 1)
  62. except ValueError:
  63. parser.error('invalid property token: {!r}'.format(values))
  64. getattr(namespace, self.dest)[prop] = value
  65. class SinglePropertyAction(argparse.Action):
  66. '''Action for argument parser that stores a property.'''
  67. # pylint: disable=redefined-builtin,too-few-public-methods
  68. def __init__(self,
  69. option_strings,
  70. dest,
  71. metavar='VALUE',
  72. const=None,
  73. nargs=None,
  74. required=False,
  75. help=None):
  76. if help is None:
  77. help = 'set {!r} property to a value'.format(dest)
  78. if const is not None:
  79. help += ' {!r}'.format(const)
  80. if const is not None:
  81. nargs = 0
  82. super(SinglePropertyAction, self).__init__(option_strings, 'properties',
  83. metavar=metavar, help=help, default={}, const=const,
  84. nargs=nargs)
  85. self.name = dest
  86. def __call__(self, parser, namespace, values, option_string=None):
  87. getattr(namespace, self.dest)[self.name] = values \
  88. if self.const is None else self.const
  89. class HelpPropertiesAction(argparse.Action):
  90. '''Action for argument parser that displays all properties and exits.'''
  91. # pylint: disable=redefined-builtin,too-few-public-methods
  92. def __init__(self,
  93. option_strings,
  94. klass=None,
  95. dest=argparse.SUPPRESS,
  96. default=argparse.SUPPRESS,
  97. help='list all available properties with short descriptions'
  98. ' and exit'):
  99. super(HelpPropertiesAction, self).__init__(
  100. option_strings=option_strings,
  101. dest=dest,
  102. default=default,
  103. nargs=0,
  104. help=help)
  105. # late import because of circular dependency
  106. import qubes # pylint: disable=redefined-outer-name
  107. self._klass = klass if klass is not None else qubes.Qubes
  108. def __call__(self, parser, namespace, values, option_string=None):
  109. # pylint: disable=redefined-outer-name
  110. properties = self._klass.property_list()
  111. width = max(len(prop.__name__) for prop in properties)
  112. wrapper = textwrap.TextWrapper(width=80,
  113. initial_indent=' ', subsequent_indent=' ' * (width + 6))
  114. text = 'Common properties:\n' + '\n'.join(
  115. wrapper.fill('{name:{width}s} {doc}'.format(
  116. name=prop.__name__,
  117. doc=qubes.utils.format_doc(prop.__doc__) if prop.__doc__ else'',
  118. width=width))
  119. for prop in sorted(properties))
  120. if self._klass is not qubes.Qubes:
  121. text += '\n\n' \
  122. 'There may be more properties in specific domain classes.\n'
  123. parser.exit(message=text)
  124. class VmNameAction(QubesAction):
  125. ''' Action for parsing one ore multiple domains from provided VMNAMEs '''
  126. # pylint: disable=too-few-public-methods,redefined-builtin
  127. def __init__(self, option_strings, nargs=1, dest='vmnames', help=None,
  128. **kwargs):
  129. if help is None:
  130. if nargs == argparse.OPTIONAL:
  131. help = 'at most one domain name'
  132. elif nargs == 1:
  133. help = 'a domain name'
  134. elif nargs == argparse.ZERO_OR_MORE:
  135. help = 'zero or more domain names'
  136. elif nargs == argparse.ONE_OR_MORE:
  137. help = 'one or more domain names'
  138. elif nargs > 1:
  139. help = '%s domain names' % nargs
  140. else:
  141. raise argparse.ArgumentError(
  142. nargs, "Passed unexpected value {!s} as {!s} nargs ".format(
  143. nargs, dest))
  144. super(VmNameAction, self).__init__(option_strings, dest=dest, help=help,
  145. nargs=nargs, **kwargs)
  146. def __call__(self, parser, namespace, values, option_string=None):
  147. ''' Set ``namespace.vmname`` to ``values`` '''
  148. setattr(namespace, self.dest, values)
  149. def parse_qubes_app(self, parser, namespace):
  150. assert hasattr(namespace, 'app')
  151. setattr(namespace, 'domains', [])
  152. app = namespace.app
  153. if hasattr(namespace, 'all_domains') and namespace.all_domains:
  154. namespace.domains = [
  155. vm
  156. for vm in app.domains
  157. if vm.qid != 0 and vm.name not in namespace.exclude
  158. ]
  159. else:
  160. if hasattr(namespace, 'exclude') and namespace.exclude:
  161. parser.error('--exclude can only be used with --all')
  162. for vm_name in getattr(namespace, self.dest):
  163. try:
  164. namespace.domains += [app.domains[vm_name]]
  165. except KeyError:
  166. parser.error('no such domain: {!r}'.format(vm_name))
  167. class PoolsAction(QubesAction):
  168. ''' Action for argument parser to gather multiple pools '''
  169. # pylint: disable=too-few-public-methods
  170. def __call__(self, parser, namespace, values, option_string=None):
  171. ''' Set ``namespace.vmname`` to ``values`` '''
  172. setattr(namespace, self.dest, values)
  173. def parse_qubes_app(self, parser, namespace):
  174. app = namespace.app
  175. name = getattr(namespace, self.dest)
  176. if not name:
  177. return
  178. try:
  179. setattr(namespace, self.dest, app.get_pool(name))
  180. except qubes.exc.QubesException as e:
  181. parser.error(e.message)
  182. sys.exit(2)
  183. class QubesArgumentParser(argparse.ArgumentParser):
  184. '''Parser preconfigured for use in most of the Qubes command-line tools.
  185. :param bool want_app: instantiate :py:class:`qubes.Qubes` object
  186. :param bool want_app_no_instance: don't actually instantiate \
  187. :py:class:`qubes.Qubes` object, just add argument for custom xml file
  188. :param bool want_force_root: add ``--force-root`` option
  189. :param mixed vmname_nargs: The number of ``VMNAME`` arguments that should be
  190. consumed. Values include:
  191. - N (an integer) consumes N arguments (and produces a list)
  192. - '?' consumes zero or one arguments
  193. - '*' consumes zero or more arguments (and produces a list)
  194. - '+' consumes one or more arguments (and produces a list)
  195. *kwargs* are passed to :py:class:`argparser.ArgumentParser`.
  196. Currenty supported options:
  197. ``--force-root`` (optional)
  198. ``--qubesxml`` location of :file:`qubes.xml` (help is suppressed)
  199. ``--verbose`` and ``--quiet``
  200. '''
  201. def __init__(self, want_app=True, want_app_no_instance=False,
  202. want_force_root=False, vmname_nargs=None, **kwargs):
  203. super(QubesArgumentParser, self).__init__(**kwargs)
  204. self._want_app = want_app
  205. self._want_app_no_instance = want_app_no_instance
  206. self._want_force_root = want_force_root
  207. self._vmname_nargs = vmname_nargs
  208. if self._want_app:
  209. self.add_argument('--qubesxml', metavar='FILE', action='store',
  210. dest='app', help=argparse.SUPPRESS)
  211. self.add_argument('--verbose', '-v', action='count',
  212. help='increase verbosity')
  213. self.add_argument('--quiet', '-q', action='count',
  214. help='decrease verbosity')
  215. if self._want_force_root:
  216. self.add_argument('--force-root', action='store_true',
  217. default=False, help='force to run as root')
  218. if self._vmname_nargs in [argparse.ZERO_OR_MORE, argparse.ONE_OR_MORE]:
  219. vm_name_group = VmNameGroup(self, self._vmname_nargs)
  220. self._mutually_exclusive_groups.append(vm_name_group)
  221. elif self._vmname_nargs is not None:
  222. self.add_argument('VMNAME', nargs=self._vmname_nargs,
  223. action=VmNameAction)
  224. self.set_defaults(verbose=1, quiet=0)
  225. def parse_args(self, *args, **kwargs):
  226. namespace = super(QubesArgumentParser, self).parse_args(*args, **kwargs)
  227. if self._want_app and not self._want_app_no_instance:
  228. self.set_qubes_verbosity(namespace)
  229. namespace.app = qubes.Qubes(namespace.app)
  230. if self._want_force_root:
  231. self.dont_run_as_root(namespace)
  232. for action in self._actions:
  233. if issubclass(action.__class__, QubesAction):
  234. action.parse_qubes_app(self, namespace)
  235. return namespace
  236. def error_runtime(self, message):
  237. '''Runtime error, without showing usage.
  238. :param str message: message to show
  239. '''
  240. self.exit(1, '{}: error: {}\n'.format(self.prog, message))
  241. def dont_run_as_root(self, namespace):
  242. '''Prevent running as root.
  243. :param argparse.Namespace args: if there is ``.force_root`` attribute \
  244. set to true, run anyway
  245. '''
  246. try:
  247. euid = os.geteuid()
  248. except AttributeError: # no geteuid(), probably NT
  249. return
  250. if euid == 0 and not namespace.force_root:
  251. self.error_runtime(
  252. 'refusing to run as root; add --force-root to override')
  253. @staticmethod
  254. def get_loglevel_from_verbosity(namespace):
  255. ''' Return loglevel calculated from quiet and verbose arguments '''
  256. return (namespace.quiet - namespace.verbose) * 10 + logging.WARNING
  257. @staticmethod
  258. def set_qubes_verbosity(namespace):
  259. '''Apply a verbosity setting.
  260. This is done by configuring global logging.
  261. :param argparse.Namespace args: args as parsed by parser
  262. '''
  263. verbose = namespace.verbose - namespace.quiet
  264. if verbose >= 2:
  265. qubes.log.enable_debug()
  266. elif verbose >= 1:
  267. qubes.log.enable()
  268. # pylint: disable=no-self-use
  269. def print_error(self, *args, **kwargs):
  270. ''' Print to ``sys.stderr``'''
  271. print(*args, file=sys.stderr, **kwargs)
  272. class AliasedSubParsersAction(argparse._SubParsersAction):
  273. # source https://gist.github.com/sampsyo/471779
  274. # pylint: disable=protected-access,too-few-public-methods
  275. class _AliasedPseudoAction(argparse.Action):
  276. # pylint: disable=redefined-builtin
  277. def __init__(self, name, aliases, help):
  278. dest = name
  279. if aliases:
  280. dest += ' (%s)' % ','.join(aliases)
  281. sup = super(AliasedSubParsersAction._AliasedPseudoAction, self)
  282. sup.__init__(option_strings=[], dest=dest, help=help)
  283. def __call__(self, **kwargs):
  284. super(AliasedSubParsersAction._AliasedPseudoAction, self).__call__(
  285. **kwargs)
  286. def add_parser(self, name, **kwargs):
  287. if 'aliases' in kwargs:
  288. aliases = kwargs['aliases']
  289. del kwargs['aliases']
  290. else:
  291. aliases = []
  292. local_parser = super(AliasedSubParsersAction, self).add_parser(
  293. name, **kwargs)
  294. # Make the aliases work.
  295. for alias in aliases:
  296. self._name_parser_map[alias] = local_parser
  297. # Make the help text reflect them, first removing old help entry.
  298. if 'help' in kwargs:
  299. self._choices_actions.pop()
  300. pseudo_action = self._AliasedPseudoAction(name, aliases,
  301. kwargs.pop('help'))
  302. self._choices_actions.append(pseudo_action)
  303. return local_parser
  304. def get_parser_for_command(command):
  305. '''Get parser for given qvm-tool.
  306. :param str command: command name
  307. :rtype: argparse.ArgumentParser
  308. :raises ImportError: when command's module is not found
  309. :raises AttributeError: when parser was not found
  310. '''
  311. module = importlib.import_module(
  312. '.' + command.replace('-', '_'), 'qubes.tools')
  313. try:
  314. parser = module.parser
  315. except AttributeError:
  316. try:
  317. parser = module.get_parser()
  318. except AttributeError:
  319. raise AttributeError('cannot find parser in module')
  320. return parser
  321. # pylint: disable=protected-access
  322. class VmNameGroup(argparse._MutuallyExclusiveGroup):
  323. ''' Adds an a VMNAME, --all & --exclude parameters to a
  324. :py:class:``argparse.ArgumentParser```.
  325. '''
  326. def __init__(self, container, required, vm_action=VmNameAction):
  327. super(VmNameGroup, self).__init__(container, required=required)
  328. self.add_argument('--all', action='store_true',
  329. dest='all_domains',
  330. help='perform the action on all qubes')
  331. container.add_argument('--exclude', action='append', default=[],
  332. help='exclude the qube from --all')
  333. self.add_argument('VMNAME', action=vm_action, nargs='*',
  334. default=[]) # the default parameter is important! see
  335. # https://stackoverflow.com/questions/35044288
  336. # and `argparse.ArgumentParser.parse_args()`
  337. def print_table(table):
  338. ''' Uses the unix column command to print pretty table.
  339. :param str text: list of lists/sets
  340. '''
  341. unit_separator = chr(31)
  342. cmd = ['column', '-t', '-s', unit_separator]
  343. text_table = '\n'.join([unit_separator.join(row) for row in table])
  344. p = subprocess.Popen(cmd, stdin=subprocess.PIPE)
  345. p.stdin.write(text_table)
  346. p.communicate()