qvm_device.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301
  1. # encoding=utf-8
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2016 Bahtiar `kalkin-` Gadimov <bahtiar@gadimov.de>
  6. # Copyright (C) 2016 Marek Marczykowski-Górecki
  7. # <marmarek@invisiblethingslab.com>
  8. #
  9. # This program is free software; you can redistribute it and/or modify
  10. # it under the terms of the GNU Lesser General Public License as published by
  11. # the Free Software Foundation; either version 2.1 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 Lesser General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU Lesser 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. """Qubes volume and block device managment"""
  23. import argparse
  24. import os
  25. import sys
  26. import qubesadmin
  27. import qubesadmin.exc
  28. import qubesadmin.tools
  29. import qubesadmin.devices
  30. def prepare_table(dev_list):
  31. """ Converts a list of :py:class:`qubes.devices.DeviceInfo` objects to a
  32. list of tupples for the :py:func:`qubes.tools.print_table`.
  33. If :program:`qvm-devices` is running in a TTY, it will ommit duplicate
  34. data.
  35. :param iterable dev_list: List of :py:class:`qubes.devices.DeviceInfo`
  36. objects.
  37. :returns: list of tupples
  38. """
  39. output = []
  40. header = []
  41. if sys.stdout.isatty():
  42. header += [('BACKEND:DEVID', 'DESCRIPTION', 'USED BY')] # NOQA
  43. for line in dev_list:
  44. output += [(
  45. line.ident,
  46. line.description,
  47. str(line.assignments),
  48. )]
  49. return header + sorted(output)
  50. class Line(object):
  51. """Helper class to hold single device info for listing"""
  52. # pylint: disable=too-few-public-methods
  53. def __init__(self, device: qubesadmin.devices.DeviceInfo, attached_to=None):
  54. self.ident = "{!s}:{!s}".format(device.backend_domain, device.ident)
  55. self.description = device.description
  56. self.attached_to = attached_to if attached_to else ""
  57. self.frontends = []
  58. @property
  59. def assignments(self):
  60. """list of frontends the device is assigned to"""
  61. return ', '.join(self.frontends)
  62. def list_devices(args):
  63. """ Called by the parser to execute the qubes-devices list
  64. subcommand. """
  65. app = args.app
  66. devices = set()
  67. if hasattr(args, 'domains') and args.domains:
  68. for domain in args.domains:
  69. for dev in domain.devices[args.devclass].attached():
  70. devices.add(dev)
  71. for dev in domain.devices[args.devclass].available():
  72. devices.add(dev)
  73. else:
  74. for domain in app.domains:
  75. for dev in domain.devices[args.devclass].available():
  76. devices.add(dev)
  77. result = {dev: Line(dev) for dev in devices}
  78. for dev in result:
  79. for domain in app.domains:
  80. if domain == dev.backend_domain:
  81. continue
  82. for assignment in domain.devices[args.devclass].assignments():
  83. if dev != assignment:
  84. continue
  85. if assignment.options:
  86. result[dev].frontends.append('{!s} ({})'.format(
  87. domain, ', '.join('{}={}'.format(key, value)
  88. for key, value in
  89. assignment.options.items())))
  90. else:
  91. result[dev].frontends.append(str(domain))
  92. qubesadmin.tools.print_table(prepare_table(result.values()))
  93. def attach_device(args):
  94. """ Called by the parser to execute the :program:`qvm-devices attach`
  95. subcommand.
  96. """
  97. device_assignment = args.device_assignment
  98. vm = args.domains[0]
  99. options = dict(opt.split('=', 1) for opt in args.option or [])
  100. if args.ro:
  101. options['read-only'] = 'yes'
  102. device_assignment.persistent = args.persistent
  103. device_assignment.options = options
  104. vm.devices[args.devclass].attach(device_assignment)
  105. def detach_device(args):
  106. """ Called by the parser to execute the :program:`qvm-devices detach`
  107. subcommand.
  108. """
  109. vm = args.domains[0]
  110. if args.device_assignment:
  111. vm.devices[args.devclass].detach(args.device_assignment)
  112. else:
  113. for device_assignment in vm.devices[args.devclass].assignments():
  114. vm.devices[args.devclass].detach(device_assignment)
  115. def init_list_parser(sub_parsers):
  116. """ Configures the parser for the :program:`qvm-devices list` subcommand """
  117. # pylint: disable=protected-access
  118. list_parser = sub_parsers.add_parser('list', aliases=('ls', 'l'),
  119. help='list devices')
  120. vm_name_group = qubesadmin.tools.VmNameGroup(
  121. list_parser, required=False, vm_action=qubesadmin.tools.VmNameAction,
  122. help='list devices assigned to specific domain(s)')
  123. list_parser._mutually_exclusive_groups.append(vm_name_group)
  124. list_parser.set_defaults(func=list_devices)
  125. class DeviceAction(qubesadmin.tools.QubesAction):
  126. """ Action for argument parser that gets the
  127. :py:class:``qubesadmin.device.DeviceAssignment`` from a
  128. BACKEND:DEVICE_ID string.
  129. """ # pylint: disable=too-few-public-methods
  130. def __init__(self, help='A backend & device id combination',
  131. required=True, allow_unknown=False, **kwargs):
  132. # pylint: disable=redefined-builtin
  133. self.allow_unknown = allow_unknown
  134. super(DeviceAction, self).__init__(help=help, required=required,
  135. **kwargs)
  136. def __call__(self, parser, namespace, values, option_string=None):
  137. """ Set ``namespace.device_assignment`` to ``values`` """
  138. setattr(namespace, self.dest, values)
  139. def parse_qubes_app(self, parser, namespace):
  140. app = namespace.app
  141. backend_device_id = getattr(namespace, self.dest)
  142. devclass = namespace.devclass
  143. if backend_device_id is None:
  144. return
  145. try:
  146. vmname, device_id = backend_device_id.split(':', 1)
  147. vm = None
  148. try:
  149. vm = app.domains[vmname]
  150. except KeyError:
  151. parser.error_runtime("no backend vm {!r}".format(vmname))
  152. try:
  153. dev = vm.devices[devclass][device_id]
  154. if not self.allow_unknown and \
  155. isinstance(dev, qubesadmin.devices.UnknownDevice):
  156. raise KeyError(device_id)
  157. except KeyError:
  158. parser.error_runtime(
  159. "backend vm {!r} doesn't expose device {!r}".format(
  160. vmname, device_id))
  161. device_assignment = qubesadmin.devices.DeviceAssignment(
  162. vm, device_id)
  163. setattr(namespace, self.dest, device_assignment)
  164. except ValueError:
  165. parser.error(
  166. 'expected a backend vm & device id combination like foo:bar '
  167. 'got %s' % backend_device_id)
  168. def get_parser(device_class=None):
  169. """Create :py:class:`argparse.ArgumentParser` suitable for
  170. :program:`qvm-block`.
  171. """
  172. parser = qubesadmin.tools.QubesArgumentParser(description=__doc__,
  173. want_app=True)
  174. parser.register('action', 'parsers',
  175. qubesadmin.tools.AliasedSubParsersAction)
  176. parser.allow_abbrev = False
  177. if device_class:
  178. parser.add_argument('devclass', const=device_class,
  179. action='store_const',
  180. help=argparse.SUPPRESS)
  181. else:
  182. parser.add_argument('devclass', metavar='DEVICE_CLASS', action='store',
  183. help="Device class to manage ('pci', 'usb', etc)")
  184. # default action
  185. parser.set_defaults(func=list_devices)
  186. sub_parsers = parser.add_subparsers(
  187. title='commands',
  188. description="For more information see qvm-device command -h",
  189. dest='command')
  190. init_list_parser(sub_parsers)
  191. attach_parser = sub_parsers.add_parser(
  192. 'attach', help="Attach device to domain", aliases=('at', 'a'))
  193. detach_parser = sub_parsers.add_parser(
  194. "detach", help="Detach device from domain", aliases=('d', 'dt'))
  195. attach_parser.add_argument('VMNAME', nargs=1,
  196. action=qubesadmin.tools.VmNameAction)
  197. detach_parser.add_argument('VMNAME', nargs=1,
  198. action=qubesadmin.tools.VmNameAction)
  199. attach_parser.add_argument(metavar='BACKEND:DEVICE_ID',
  200. dest='device_assignment',
  201. action=DeviceAction)
  202. detach_parser.add_argument(metavar='BACKEND:DEVICE_ID',
  203. dest='device_assignment',
  204. nargs=argparse.OPTIONAL,
  205. action=DeviceAction, allow_unknown=True)
  206. attach_parser.add_argument('--option', '-o', action='append',
  207. help="Set option for the device in opt=value "
  208. "form (can be specified "
  209. "multiple times), see man qvm-device for "
  210. "details")
  211. attach_parser.add_argument('--ro', action='store_true', default=False,
  212. help="Attach device read-only (alias for "
  213. "read-only=yes option, "
  214. "takes precedence)")
  215. attach_parser.add_argument('--persistent', '-p', action='store_true',
  216. default=False,
  217. help="Attach device persistently (so it will "
  218. "be automatically "
  219. "attached at qube startup)")
  220. attach_parser.set_defaults(func=attach_device)
  221. detach_parser.set_defaults(func=detach_device)
  222. parser.add_argument('--list-device-classes', action='store_true',
  223. default=False)
  224. return parser
  225. def main(args=None, app=None):
  226. """Main routine of :program:`qvm-block`."""
  227. basename = os.path.basename(sys.argv[0])
  228. devclass = None
  229. if basename.startswith('qvm-') and basename != 'qvm-device':
  230. devclass = basename[4:]
  231. args = get_parser(devclass).parse_args(args, app=app)
  232. if args.list_device_classes:
  233. print('\n'.join(qubesadmin.Qubes().list_deviceclass()))
  234. return 0
  235. try:
  236. args.func(args)
  237. except qubesadmin.exc.QubesException as e:
  238. print(str(e), file=sys.stderr)
  239. return 1
  240. return 0
  241. if __name__ == '__main__':
  242. # Special treatment for '--list-device-classes' (alias --list-classes)
  243. curr_action = sys.argv[1:]
  244. if set(curr_action).intersection(
  245. {'--list-device-classes', '--list-classes'}):
  246. sys.exit(main(args=['', '--list-device-classes']))
  247. sys.exit(main())