qvm_device.py 9.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266
  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 assignment.options.items())))
  89. else:
  90. result[dev].frontends.append(str(domain))
  91. qubesadmin.tools.print_table(prepare_table(result.values()))
  92. def attach_device(args):
  93. ''' Called by the parser to execute the :program:`qvm-devices attach`
  94. subcommand.
  95. '''
  96. device_assignment = args.device_assignment
  97. vm = args.domains[0]
  98. options = dict(opt.split('=', 1) for opt in args.option or [])
  99. device_assignment.persistent = args.persistent
  100. device_assignment.options = options
  101. vm.devices[args.devclass].attach(device_assignment)
  102. def detach_device(args):
  103. ''' Called by the parser to execute the :program:`qvm-devices detach`
  104. subcommand.
  105. '''
  106. device_assignment = args.device_assignment
  107. vm = args.domains[0]
  108. vm.devices[args.devclass].detach(device_assignment)
  109. def init_list_parser(sub_parsers):
  110. ''' Configures the parser for the :program:`qvm-devices list` subcommand '''
  111. # pylint: disable=protected-access
  112. list_parser = sub_parsers.add_parser('list', aliases=('ls', 'l'),
  113. help='list devices')
  114. vm_name_group = qubesadmin.tools.VmNameGroup(
  115. list_parser, required=False, vm_action=qubesadmin.tools.VmNameAction,
  116. help='list devices assigned to specific domain(s)')
  117. list_parser._mutually_exclusive_groups.append(vm_name_group)
  118. list_parser.set_defaults(func=list_devices)
  119. class DeviceAction(qubesadmin.tools.QubesAction):
  120. ''' Action for argument parser that gets the
  121. :py:class:``qubesadmin.device.DeviceAssignment`` from a
  122. BACKEND:DEVICE_ID string.
  123. ''' # pylint: disable=too-few-public-methods
  124. def __init__(self, help='A backend & device id combination',
  125. required=True, allow_unknown=False, **kwargs):
  126. # pylint: disable=redefined-builtin
  127. self.allow_unknown = allow_unknown
  128. super(DeviceAction, self).__init__(help=help, required=required,
  129. **kwargs)
  130. def __call__(self, parser, namespace, values, option_string=None):
  131. ''' Set ``namespace.device_assignment`` to ``values`` '''
  132. setattr(namespace, self.dest, values)
  133. def parse_qubes_app(self, parser, namespace):
  134. app = namespace.app
  135. backend_device_id = getattr(namespace, self.dest)
  136. devclass = namespace.devclass
  137. try:
  138. vmname, device_id = backend_device_id.split(':', 1)
  139. try:
  140. vm = app.domains[vmname]
  141. except KeyError:
  142. parser.error_runtime("no backend vm {!r}".format(vmname))
  143. try:
  144. dev = vm.devices[devclass][device_id]
  145. if not self.allow_unknown and isinstance(dev,
  146. qubesadmin.devices.UnknownDevice):
  147. raise KeyError(device_id)
  148. except KeyError:
  149. parser.error_runtime(
  150. "backend vm {!r} doesn't expose device {!r}"
  151. .format(vmname, device_id))
  152. device_assignment = qubesadmin.devices.DeviceAssignment(vm,
  153. device_id)
  154. setattr(namespace, self.dest, device_assignment)
  155. except ValueError:
  156. parser.error('expected a backend vm & device id combination ' \
  157. 'like foo:bar got %s' % backend_device_id)
  158. def get_parser(device_class=None):
  159. '''Create :py:class:`argparse.ArgumentParser` suitable for
  160. :program:`qvm-block`.
  161. '''
  162. parser = qubesadmin.tools.QubesArgumentParser(description=__doc__,
  163. want_app=True)
  164. parser.register('action', 'parsers',
  165. qubesadmin.tools.AliasedSubParsersAction)
  166. if device_class:
  167. parser.add_argument('devclass', const=device_class,
  168. action='store_const',
  169. help=argparse.SUPPRESS)
  170. else:
  171. parser.add_argument('devclass', metavar='DEVICE_CLASS', action='store',
  172. help="Device class to manage ('pci', 'usb', etc)")
  173. # default action
  174. parser.set_defaults(func=list_devices)
  175. sub_parsers = parser.add_subparsers(
  176. title='commands',
  177. description="For more information see qvm-device command -h",
  178. dest='command')
  179. init_list_parser(sub_parsers)
  180. attach_parser = sub_parsers.add_parser(
  181. 'attach', help="Attach device to domain", aliases=('at', 'a'))
  182. detach_parser = sub_parsers.add_parser(
  183. "detach", help="Detach device from domain", aliases=('d', 'dt'))
  184. attach_parser.add_argument('VMNAME', nargs=1,
  185. action=qubesadmin.tools.VmNameAction)
  186. detach_parser.add_argument('VMNAME', nargs=1,
  187. action=qubesadmin.tools.VmNameAction)
  188. attach_parser.add_argument(metavar='BACKEND:DEVICE_ID',
  189. dest='device_assignment',
  190. action=DeviceAction)
  191. detach_parser.add_argument(metavar='BACKEND:DEVICE_ID',
  192. dest='device_assignment',
  193. action=DeviceAction, allow_unknown=True)
  194. attach_parser.add_argument('--option', '-o', action='append',
  195. help="Set option for the device in opt=value form (can be specified "
  196. "multiple times)")
  197. attach_parser.add_argument('--persistent', '-p', action='store_true',
  198. default=False,
  199. help="Attach device persistently (so it will be automatically "
  200. "attached at qube startup)")
  201. attach_parser.set_defaults(func=attach_device)
  202. detach_parser.set_defaults(func=detach_device)
  203. return parser
  204. def main(args=None, app=None):
  205. '''Main routine of :program:`qvm-block`.'''
  206. basename = os.path.basename(sys.argv[0])
  207. devclass = None
  208. if basename.startswith('qvm-') and basename != 'qvm-device':
  209. devclass = basename[4:]
  210. args = get_parser(devclass).parse_args(args, app=app)
  211. try:
  212. args.func(args)
  213. except qubesadmin.exc.QubesException as e:
  214. print(str(e), file=sys.stderr)
  215. return 1
  216. return 0
  217. if __name__ == '__main__':
  218. sys.exit(main())