__init__.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534
  1. #
  2. # The Qubes OS Project, https://www.qubes-os.org/
  3. #
  4. # Copyright (C) 2015 Joanna Rutkowska <joanna@invisiblethingslab.com>
  5. # Copyright (C) 2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  6. #
  7. # This program is free software; you can redistribute it and/or modify
  8. # it under the terms of the GNU General Public License as published by
  9. # the Free Software Foundation; either version 2 of the License, or
  10. # (at your option) any later version.
  11. #
  12. # This program is distributed in the hope that it will be useful,
  13. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15. # GNU General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU General Public License along
  18. # with this program; if not, write to the Free Software Foundation, Inc.,
  19. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  20. #
  21. '''Qubes' command line tools
  22. '''
  23. from __future__ import print_function
  24. import argparse
  25. import importlib
  26. import logging
  27. import os
  28. import subprocess
  29. import sys
  30. import textwrap
  31. import qubes.log
  32. #: constant returned when some action should be performed on all qubes
  33. VM_ALL = object()
  34. class QubesAction(argparse.Action):
  35. ''' Interface providing a convinience method to be called, after
  36. `namespace.app` is instantiated.
  37. '''
  38. # pylint: disable=too-few-public-methods
  39. def parse_qubes_app(self, parser, namespace):
  40. ''' This method is called by :py:class:`qubes.tools.QubesArgumentParser`
  41. after the `namespace.app` is instantiated. Oerwrite this method when
  42. extending :py:class:`qubes.tools.QubesAction` to initialized values
  43. based on the `namespace.app`
  44. '''
  45. raise NotImplementedError
  46. class PropertyAction(argparse.Action):
  47. '''Action for argument parser that stores a property.'''
  48. # pylint: disable=redefined-builtin,too-few-public-methods
  49. def __init__(self,
  50. option_strings,
  51. dest,
  52. metavar='NAME=VALUE',
  53. required=False,
  54. help='set property to a value'):
  55. super(PropertyAction, self).__init__(option_strings, 'properties',
  56. metavar=metavar, default={}, help=help)
  57. def __call__(self, parser, namespace, values, option_string=None):
  58. try:
  59. prop, value = values.split('=', 1)
  60. except ValueError:
  61. parser.error('invalid property token: {!r}'.format(values))
  62. getattr(namespace, self.dest)[prop] = value
  63. class SinglePropertyAction(argparse.Action):
  64. '''Action for argument parser that stores a property.'''
  65. # pylint: disable=redefined-builtin,too-few-public-methods
  66. def __init__(self,
  67. option_strings,
  68. dest,
  69. metavar='VALUE',
  70. const=None,
  71. nargs=None,
  72. required=False,
  73. help=None):
  74. if help is None:
  75. help = 'set {!r} property to a value'.format(dest)
  76. if const is not None:
  77. help += ' {!r}'.format(const)
  78. if const is not None:
  79. nargs = 0
  80. super(SinglePropertyAction, self).__init__(option_strings, 'properties',
  81. metavar=metavar, help=help, default={}, const=const,
  82. nargs=nargs)
  83. self.name = dest
  84. def __call__(self, parser, namespace, values, option_string=None):
  85. getattr(namespace, self.dest)[self.name] = values \
  86. if self.const is None else self.const
  87. class HelpPropertiesAction(argparse.Action):
  88. '''Action for argument parser that displays all properties and exits.'''
  89. # pylint: disable=redefined-builtin,too-few-public-methods
  90. def __init__(self,
  91. option_strings,
  92. klass=None,
  93. dest=argparse.SUPPRESS,
  94. default=argparse.SUPPRESS,
  95. help='list all available properties with short descriptions'
  96. ' and exit'):
  97. super(HelpPropertiesAction, self).__init__(
  98. option_strings=option_strings,
  99. dest=dest,
  100. default=default,
  101. nargs=0,
  102. help=help)
  103. # late import because of circular dependency
  104. import qubes # pylint: disable=redefined-outer-name
  105. self._klass = klass if klass is not None else qubes.Qubes
  106. def __call__(self, parser, namespace, values, option_string=None):
  107. # pylint: disable=redefined-outer-name
  108. properties = self._klass.property_list()
  109. width = max(len(prop.__name__) for prop in properties)
  110. wrapper = textwrap.TextWrapper(width=80,
  111. initial_indent=' ', subsequent_indent=' ' * (width + 6))
  112. text = 'Common properties:\n' + '\n'.join(
  113. wrapper.fill('{name:{width}s} {doc}'.format(
  114. name=prop.__name__,
  115. doc=qubes.utils.format_doc(prop.__doc__) if prop.__doc__ else'',
  116. width=width))
  117. for prop in sorted(properties))
  118. if self._klass is not qubes.Qubes:
  119. text += '\n\n' \
  120. 'There may be more properties in specific domain classes.\n'
  121. parser.exit(message=text)
  122. class VmNameAction(QubesAction):
  123. ''' Action for parsing one ore multiple domains from provided VMNAMEs '''
  124. # pylint: disable=too-few-public-methods,redefined-builtin
  125. def __init__(self, option_strings, nargs=1, dest='vmnames', help=None,
  126. **kwargs):
  127. if help is None:
  128. if nargs == argparse.OPTIONAL:
  129. help = 'at most one domain name'
  130. elif nargs == 1:
  131. help = 'a domain name'
  132. elif nargs == argparse.ZERO_OR_MORE:
  133. help = 'zero or more domain names'
  134. elif nargs == argparse.ONE_OR_MORE:
  135. help = 'one or more domain names'
  136. elif nargs > 1:
  137. help = '%s domain names' % nargs
  138. else:
  139. raise argparse.ArgumentError(
  140. nargs, "Passed unexpected value {!s} as {!s} nargs ".format(
  141. nargs, dest))
  142. super(VmNameAction, self).__init__(option_strings, dest=dest, help=help,
  143. nargs=nargs, **kwargs)
  144. def __call__(self, parser, namespace, values, option_string=None):
  145. ''' Set ``namespace.vmname`` to ``values`` '''
  146. setattr(namespace, self.dest, values)
  147. def parse_qubes_app(self, parser, namespace):
  148. assert hasattr(namespace, 'app')
  149. setattr(namespace, 'domains', [])
  150. app = namespace.app
  151. if hasattr(namespace, 'all_domains') and namespace.all_domains:
  152. namespace.domains = [
  153. vm
  154. for vm in app.domains
  155. if vm.qid != 0 and vm.name not in namespace.exclude
  156. ]
  157. else:
  158. if hasattr(namespace, 'exclude') and namespace.exclude:
  159. parser.error('--exclude can only be used with --all')
  160. for vm_name in getattr(namespace, self.dest):
  161. try:
  162. namespace.domains += [app.domains[vm_name]]
  163. except KeyError:
  164. parser.error('no such domain: {!r}'.format(vm_name))
  165. class RunningVmNameAction(VmNameAction):
  166. ''' Action for argument parser that gets a running domain from VMNAME '''
  167. # pylint: disable=too-few-public-methods
  168. def __init__(self, option_strings, nargs=1, dest='vmnames', help=None,
  169. **kwargs):
  170. # pylint: disable=redefined-builtin
  171. if help is None:
  172. if nargs == argparse.OPTIONAL:
  173. help = 'at most one running domain'
  174. elif nargs == 1:
  175. help = 'running domain name'
  176. elif nargs == argparse.ZERO_OR_MORE:
  177. help = 'zero or more running domains'
  178. elif nargs == argparse.ONE_OR_MORE:
  179. help = 'one or more running domains'
  180. elif nargs > 1:
  181. help = '%s running domains' % nargs
  182. else:
  183. raise argparse.ArgumentError(
  184. nargs, "Passed unexpected value {!s} as {!s} nargs ".format(
  185. nargs, dest))
  186. super(RunningVmNameAction, self).__init__(
  187. option_strings, dest=dest, help=help, nargs=nargs, **kwargs)
  188. def parse_qubes_app(self, parser, namespace):
  189. super(RunningVmNameAction, self).parse_qubes_app(parser, namespace)
  190. for vm in namespace.domains:
  191. if not vm.is_running():
  192. parser.error_runtime("domain {!r} is not running".format(
  193. vm.name))
  194. class VolumeAction(QubesAction):
  195. ''' Action for argument parser that gets the
  196. :py:class:``qubes.storage.Volume`` from a POOL_NAME:VOLUME_ID string.
  197. '''
  198. # pylint: disable=too-few-public-methods
  199. def __init__(self, help='A pool & volume id combination',
  200. required=True, **kwargs):
  201. # pylint: disable=redefined-builtin
  202. super(VolumeAction, self).__init__(help=help, required=required,
  203. **kwargs)
  204. def __call__(self, parser, namespace, values, option_string=None):
  205. ''' Set ``namespace.vmname`` to ``values`` '''
  206. setattr(namespace, self.dest, values)
  207. def parse_qubes_app(self, parser, namespace):
  208. ''' Acquire the :py:class:``qubes.storage.Volume`` object from
  209. ``namespace.app``.
  210. '''
  211. assert hasattr(namespace, 'app')
  212. app = namespace.app
  213. try:
  214. pool_name, vid = getattr(namespace, self.dest).split(':')
  215. try:
  216. pool = app.pools[pool_name]
  217. volume = [v for v in pool.volumes if v.vid == vid]
  218. assert len(volume) == 1, 'Duplicate vids in pool %s' % pool_name
  219. if not volume:
  220. parser.error_runtime(
  221. 'no volume with id {!r} pool: {!r}'.format(vid,
  222. pool_name))
  223. else:
  224. setattr(namespace, self.dest, volume[0])
  225. except KeyError:
  226. parser.error_runtime('no pool {!r}'.format(pool_name))
  227. except ValueError:
  228. parser.error('expected a pool & volume id combination like foo:bar')
  229. class PoolsAction(QubesAction):
  230. ''' Action for argument parser to gather multiple pools '''
  231. # pylint: disable=too-few-public-methods
  232. def __call__(self, parser, namespace, values, option_string=None):
  233. ''' Set ``namespace.vmname`` to ``values`` '''
  234. if hasattr(namespace, self.dest) and getattr(namespace, self.dest):
  235. names = getattr(namespace, self.dest)
  236. else:
  237. names = []
  238. names += [values]
  239. setattr(namespace, self.dest, names)
  240. def parse_qubes_app(self, parser, namespace):
  241. app = namespace.app
  242. pool_names = getattr(namespace, self.dest)
  243. if pool_names:
  244. try:
  245. pools = [app.get_pool(name) for name in pool_names]
  246. setattr(namespace, self.dest, pools)
  247. except qubes.exc.QubesException as e:
  248. parser.error(str(e))
  249. sys.exit(2)
  250. class QubesArgumentParser(argparse.ArgumentParser):
  251. '''Parser preconfigured for use in most of the Qubes command-line tools.
  252. :param bool want_app: instantiate :py:class:`qubes.Qubes` object
  253. :param bool want_app_no_instance: don't actually instantiate \
  254. :py:class:`qubes.Qubes` object, just add argument for custom xml file
  255. :param bool want_force_root: add ``--force-root`` option
  256. :param mixed vmname_nargs: The number of ``VMNAME`` arguments that should be
  257. consumed. Values include:
  258. - N (an integer) consumes N arguments (and produces a list)
  259. - '?' consumes zero or one arguments
  260. - '*' consumes zero or more arguments (and produces a list)
  261. - '+' consumes one or more arguments (and produces a list)
  262. *kwargs* are passed to :py:class:`argparser.ArgumentParser`.
  263. Currenty supported options:
  264. ``--force-root`` (optional)
  265. ``--qubesxml`` location of :file:`qubes.xml` (help is suppressed)
  266. ``--offline-mode`` do not talk to hypervisor (help is suppressed)
  267. ``--verbose`` and ``--quiet``
  268. '''
  269. def __init__(self, want_app=True, want_app_no_instance=False,
  270. want_force_root=False, vmname_nargs=None, **kwargs):
  271. super(QubesArgumentParser, self).__init__(**kwargs)
  272. self._want_app = want_app
  273. self._want_app_no_instance = want_app_no_instance
  274. self._want_force_root = want_force_root
  275. self._vmname_nargs = vmname_nargs
  276. if self._want_app:
  277. self.add_argument('--qubesxml', metavar='FILE', action='store',
  278. dest='app', help=argparse.SUPPRESS)
  279. self.add_argument('--offline-mode', action='store_true',
  280. default=None, dest='offline_mode', help=argparse.SUPPRESS)
  281. self.add_argument('--verbose', '-v', action='count',
  282. help='increase verbosity')
  283. self.add_argument('--quiet', '-q', action='count',
  284. help='decrease verbosity')
  285. if self._want_force_root:
  286. self.add_argument('--force-root', action='store_true',
  287. default=False, help='force to run as root')
  288. if self._vmname_nargs in [argparse.ZERO_OR_MORE, argparse.ONE_OR_MORE]:
  289. vm_name_group = VmNameGroup(self, self._vmname_nargs)
  290. self._mutually_exclusive_groups.append(vm_name_group)
  291. elif self._vmname_nargs is not None:
  292. self.add_argument('VMNAME', nargs=self._vmname_nargs,
  293. action=VmNameAction)
  294. self.set_defaults(verbose=1, quiet=0)
  295. def parse_args(self, args=None, namespace=None):
  296. namespace = super(QubesArgumentParser, self).parse_args(args, namespace)
  297. if self._want_app and not self._want_app_no_instance:
  298. self.set_qubes_verbosity(namespace)
  299. namespace.app = qubes.Qubes(namespace.app,
  300. offline_mode=namespace.offline_mode)
  301. if self._want_force_root:
  302. self.dont_run_as_root(namespace)
  303. for action in self._actions:
  304. # pylint: disable=protected-access
  305. if issubclass(action.__class__, QubesAction):
  306. action.parse_qubes_app(self, namespace)
  307. elif issubclass(action.__class__,
  308. argparse._SubParsersAction): # pylint: disable=no-member
  309. assert hasattr(namespace, 'command')
  310. command = namespace.command
  311. subparser = action._name_parser_map[command]
  312. for subaction in subparser._actions:
  313. if issubclass(subaction.__class__, QubesAction):
  314. subaction.parse_qubes_app(self, namespace)
  315. return namespace
  316. def error_runtime(self, message):
  317. '''Runtime error, without showing usage.
  318. :param str message: message to show
  319. '''
  320. self.exit(1, '{}: error: {}\n'.format(self.prog, message))
  321. def dont_run_as_root(self, namespace):
  322. '''Prevent running as root.
  323. :param argparse.Namespace args: if there is ``.force_root`` attribute \
  324. set to true, run anyway
  325. '''
  326. try:
  327. euid = os.geteuid()
  328. except AttributeError: # no geteuid(), probably NT
  329. return
  330. if euid == 0 and not namespace.force_root:
  331. self.error_runtime(
  332. 'refusing to run as root; add --force-root to override')
  333. @staticmethod
  334. def get_loglevel_from_verbosity(namespace):
  335. ''' Return loglevel calculated from quiet and verbose arguments '''
  336. return (namespace.quiet - namespace.verbose) * 10 + logging.WARNING
  337. @staticmethod
  338. def set_qubes_verbosity(namespace):
  339. '''Apply a verbosity setting.
  340. This is done by configuring global logging.
  341. :param argparse.Namespace args: args as parsed by parser
  342. '''
  343. verbose = namespace.verbose - namespace.quiet
  344. if verbose >= 2:
  345. qubes.log.enable_debug()
  346. elif verbose >= 1:
  347. qubes.log.enable()
  348. # pylint: disable=no-self-use
  349. def print_error(self, *args, **kwargs):
  350. ''' Print to ``sys.stderr``'''
  351. print(*args, file=sys.stderr, **kwargs)
  352. class AliasedSubParsersAction(argparse._SubParsersAction):
  353. # source https://gist.github.com/sampsyo/471779
  354. # pylint: disable=protected-access,too-few-public-methods
  355. class _AliasedPseudoAction(argparse.Action):
  356. # pylint: disable=redefined-builtin,arguments-differ
  357. def __init__(self, name, aliases, help):
  358. dest = name
  359. if aliases:
  360. dest += ' (%s)' % ','.join(aliases)
  361. sup = super(AliasedSubParsersAction._AliasedPseudoAction, self)
  362. sup.__init__(option_strings=[], dest=dest, help=help)
  363. def __call__(self, parser, namespace, values, option_string=None):
  364. raise NotImplementedError
  365. def add_parser(self, name, **kwargs):
  366. if 'aliases' in kwargs:
  367. aliases = kwargs['aliases']
  368. del kwargs['aliases']
  369. else:
  370. aliases = []
  371. local_parser = super(AliasedSubParsersAction, self).add_parser(
  372. name, **kwargs)
  373. # Make the aliases work.
  374. for alias in aliases:
  375. self._name_parser_map[alias] = local_parser
  376. # Make the help text reflect them, first removing old help entry.
  377. if 'help' in kwargs:
  378. self._choices_actions.pop()
  379. pseudo_action = self._AliasedPseudoAction(name, aliases,
  380. kwargs.pop('help'))
  381. self._choices_actions.append(pseudo_action)
  382. return local_parser
  383. def get_parser_for_command(command):
  384. '''Get parser for given qvm-tool.
  385. :param str command: command name
  386. :rtype: argparse.ArgumentParser
  387. :raises ImportError: when command's module is not found
  388. :raises AttributeError: when parser was not found
  389. '''
  390. module = importlib.import_module(
  391. '.' + command.replace('-', '_'), 'qubes.tools')
  392. try:
  393. parser = module.parser
  394. except AttributeError:
  395. try:
  396. parser = module.get_parser()
  397. except AttributeError:
  398. raise AttributeError('cannot find parser in module')
  399. return parser
  400. # pylint: disable=protected-access
  401. class VmNameGroup(argparse._MutuallyExclusiveGroup):
  402. ''' Adds an a VMNAME, --all & --exclude parameters to a
  403. :py:class:``argparse.ArgumentParser```.
  404. '''
  405. def __init__(self, container, required, vm_action=VmNameAction, help=None):
  406. # pylint: disable=redefined-builtin
  407. super(VmNameGroup, self).__init__(container, required=required)
  408. if not help:
  409. help = 'perform the action on all qubes'
  410. self.add_argument('--all', action='store_true', dest='all_domains',
  411. help=help)
  412. container.add_argument('--exclude', action='append', default=[],
  413. help='exclude the qube from --all')
  414. # ⚠ the default parameter below is important! ⚠
  415. # See https://stackoverflow.com/questions/35044288 and
  416. # `argparse.ArgumentParser.parse_args()` implementation
  417. self.add_argument('VMNAME', action=vm_action, nargs='*', default=[])
  418. def print_table(table):
  419. ''' Uses the unix column command to print pretty table.
  420. :param str text: list of lists/sets
  421. '''
  422. unit_separator = chr(31)
  423. cmd = ['column', '-t', '-s', unit_separator]
  424. text_table = '\n'.join([unit_separator.join(row) for row in table])
  425. text_table += '\n'
  426. # for tests...
  427. if sys.stdout != sys.__stdout__:
  428. p = subprocess.Popen(cmd + ['-c', '80'], stdin=subprocess.PIPE,
  429. stdout=subprocess.PIPE)
  430. p.stdin.write(text_table.encode())
  431. (out, _) = p.communicate()
  432. sys.stdout.write(out.decode())
  433. else:
  434. p = subprocess.Popen(cmd, stdin=subprocess.PIPE)
  435. p.communicate(text_table.encode())