Qubes/core-admin-client
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge remote-tracking branch 'origin/pr/100'

Browse Source 
* origin/pr/100:
  qvm-device: add manpage entry
  qvm-device: prevent parser allowing abbreviations
  qvm-device: handle 'list-device-classes' and 'list-classes'
  qubesadmin: define methods list_vmclass and list_devicesclass
  qubesadmin: make PEP8 happy
This commit is contained in:
Marek Marczykowski-Górecki 2019-09-06 13:08:32 +02:00
commit 9158412a24
No known key found for this signature in database
GPG Key ID: 063938BA42CFA724
6 changed files with 114 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
doc/manpages/qvm-device.rst
View File

@ -28,6 +28,10 @@ Options
decrease verbosity
.. option:: --list-device-classes
list device classes
Commands
========
@ -122,3 +126,4 @@ Authors
| Joanna Rutkowska <joanna at invisiblethingslab dot com>
| Rafal Wojtczuk <rafal at invisiblethingslab dot com>
| Marek Marczykowski <marmarek at invisiblethingslab dot com>
| Frédéric Pierret <frederic.pierret at qubes dash os dot org>

116
qubesadmin/app.py
View File

@ -19,9 +19,9 @@
# with this program; if not, see <http://www.gnu.org/licenses/>.
'''
"""
Main Qubes() class and related classes.
'''
"""
import os
import shlex
import socket
@ -40,19 +40,21 @@ import qubesadmin.config
BUF_SIZE = 4096
class VMCollection(object):
'''Collection of VMs objects'''
"""Collection of VMs objects"""
def __init__(self, app):
self.app = app
self._vm_list = None
self._vm_objects = {}
def clear_cache(self):
'''Clear cached list of VMs'''
"""Clear cached list of VMs"""
self._vm_list = None
def refresh_cache(self, force=False):
'''Refresh cached list of VMs'''
"""Refresh cached list of VMs"""
if not force and self._vm_list is not None:
return
vm_list_data = self.app.qubesd_call(
@ -90,10 +92,10 @@ class VMCollection(object):
return self.get_blind(item)
def get_blind(self, item):
'''
"""
Get a vm without downloading the list
and checking if exists
'''
"""
if item not in self._vm_objects:
cls = qubesadmin.vm.QubesVM
# provide class name to constructor, if already cached (which can be
@ -121,23 +123,23 @@ class VMCollection(object):
yield self[vm]
def keys(self):
'''Get list of VM names.'''
"""Get list of VM names."""
self.refresh_cache()
return self._vm_list.keys()
def values(self):
'''Get list of VM objects.'''
"""Get list of VM objects."""
self.refresh_cache()
return [self[name] for name in self._vm_list]
class QubesBase(qubesadmin.base.PropertyHolder):
'''Main Qubes application.
"""Main Qubes application.
This is a base abstract class, don't use it directly. Use specialized
class in py:class:`qubesadmin.Qubes` instead, which points at
:py:class:`QubesLocal` or :py:class:`QubesRemote`.
'''
"""
#: domains (VMs) collection
domains = None
@ -163,12 +165,24 @@ class QubesBase(qubesadmin.base.PropertyHolder):
self._pool_drivers = None
self.log = logging.getLogger('app')
def list_vmclass(self):
"""Call Qubesd in order to obtain the vm classes list"""
vmclass = self.qubesd_call('dom0', 'admin.vmclass.List')\
.decode().splitlines()
return sorted(vmclass)
def list_deviceclass(self):
"""Call Qubesd in order to obtain the device classes list"""
deviceclasses = self.qubesd_call('dom0', 'admin.deviceclass.List')\
.decode().splitlines()
return sorted(deviceclasses)
def _refresh_pool_drivers(self):
'''
"""
Refresh cached storage pool drivers and their parameters.
:return: None
'''
"""
if self._pool_drivers is None:
pool_drivers_data = self.qubesd_call(
'dom0', 'admin.pool.ListDrivers', None, None)
@ -183,24 +197,24 @@ class QubesBase(qubesadmin.base.PropertyHolder):
@property
def pool_drivers(self):
''' Available storage pool drivers '''
""" Available storage pool drivers """
self._refresh_pool_drivers()
return self._pool_drivers.keys()
def pool_driver_parameters(self, driver):
''' Parameters to initialize storage pool using given driver '''
""" Parameters to initialize storage pool using given driver """
self._refresh_pool_drivers()
return self._pool_drivers[driver]
def add_pool(self, name, driver, **kwargs):
''' Add a storage pool to config
""" Add a storage pool to config
:param name: name of storage pool to create
:param driver: driver to use, see :py:meth:`pool_drivers` for
available drivers
:param kwargs: configuration parameters for storage pool,
see :py:meth:`pool_driver_parameters` for a list
'''
"""
# sort parameters only to ease testing, not required by API
payload = 'name={}\n'.format(name) + \
''.join('{}={}\n'.format(key, value)
@ -209,14 +223,14 @@ class QubesBase(qubesadmin.base.PropertyHolder):
payload.encode('utf-8'))
def remove_pool(self, name):
''' Remove a storage pool '''
""" Remove a storage pool """
self.qubesd_call('dom0', 'admin.pool.Remove', name, None)
def get_label(self, label):
'''Get label as identified by index or name
"""Get label as identified by index or name
:throws KeyError: when label is not found
'''
"""
# first search for name, verbatim
try:
@ -233,19 +247,19 @@ class QubesBase(qubesadmin.base.PropertyHolder):
@staticmethod
def get_vm_class(clsname):
'''Find the class for a domain.
"""Find the class for a domain.
Compatibility function, client tools use str to identify domain classes.
:param str clsname: name of the class
:return str: class
'''
"""
return clsname
def add_new_vm(self, cls, name, label, template=None, pool=None,
pools=None):
'''Create new Virtual Machine
"""Create new Virtual Machine
Example usage with custom storage pools:
@ -264,7 +278,7 @@ class QubesBase(qubesadmin.base.PropertyHolder):
:param dict pools: storage pool for specific volumes
:return new VM object
'''
"""
if not isinstance(cls, str):
cls = cls.__name__
@ -293,9 +307,9 @@ class QubesBase(qubesadmin.base.PropertyHolder):
self.domains.clear_cache()
return self.domains[name]
def clone_vm(self, src_vm, new_name, new_cls=None,
pool=None, pools=None, ignore_errors=False, ignore_volumes=None):
'''Clone Virtual Machine
def clone_vm(self, src_vm, new_name, new_cls=None, pool=None, pools=None,
ignore_errors=False, ignore_volumes=None):
"""Clone Virtual Machine
Example usage with custom storage pools:
@ -317,7 +331,7 @@ class QubesBase(qubesadmin.base.PropertyHolder):
like 'private' or 'root'
:return new VM object
'''
"""
if pool and pools:
raise ValueError('only one of pool= and pools= can be used')
@ -454,7 +468,7 @@ class QubesBase(qubesadmin.base.PropertyHolder):
def qubesd_call(self, dest, method, arg=None, payload=None,
payload_stream=None):
'''
"""
Execute Admin API method.
Only one of `payload` and `payload_stream` can be specified.
@ -467,14 +481,14 @@ class QubesBase(qubesadmin.base.PropertyHolder):
:return: Data returned by qubesd (string)
.. warning:: *payload_stream* will get closed by this function
'''
"""
raise NotImplementedError(
'qubesd_call not implemented in QubesBase class; use specialized '
'class: qubesadmin.Qubes()')
def run_service(self, dest, service, filter_esc=False, user=None,
localcmd=None, wait=True, **kwargs):
'''Run qrexec service in a given destination
"""Run qrexec service in a given destination
*kwargs* are passed verbatim to :py:meth:`subprocess.Popen`.
@ -485,24 +499,25 @@ class QubesBase(qubesadmin.base.PropertyHolder):
emulator
:param str user: username to run service as
:param str localcmd: Command to connect stdin/stdout to
:param bool wait: Wait service run
:rtype: subprocess.Popen
'''
"""
raise NotImplementedError(
'run_service not implemented in QubesBase class; use specialized '
'class: qubesadmin.Qubes()')
class QubesLocal(QubesBase):
'''Application object communicating through local socket.
"""Application object communicating through local socket.
Used when running in dom0.
'''
"""
qubesd_connection_type = 'socket'
def qubesd_call(self, dest, method, arg=None, payload=None,
payload_stream=None):
'''
"""
Execute Admin API method.
Only one of `payload` and `payload_stream` can be specified.
@ -515,7 +530,7 @@ class QubesLocal(QubesBase):
:return: Data returned by qubesd (string)
.. warning:: *payload_stream* will get closed by this function
'''
"""
if payload and payload_stream:
raise ValueError(
'Only one of payload and payload_stream can be used')
@ -562,7 +577,7 @@ class QubesLocal(QubesBase):
def run_service(self, dest, service, filter_esc=False, user=None,
localcmd=None, wait=True, **kwargs):
'''Run qrexec service in a given destination
"""Run qrexec service in a given destination
:param str dest: Destination - may be a VM name or empty
string for default (for a given service)
@ -572,9 +587,8 @@ class QubesLocal(QubesBase):
:param str user: username to run service as
:param str localcmd: Command to connect stdin/stdout to
:param bool wait: wait for remote process to finish
:param int connect_timeout: qrexec client connection timeout
:rtype: subprocess.Popen
'''
"""
if not dest:
raise ValueError('Empty destination name allowed only from a VM')
@ -600,23 +614,23 @@ class QubesLocal(QubesBase):
kwargs.setdefault('stdin', subprocess.PIPE)
kwargs.setdefault('stdout', subprocess.PIPE)
kwargs.setdefault('stderr', subprocess.PIPE)
proc = subprocess.Popen([qubesadmin.config.QREXEC_CLIENT] +
qrexec_opts + ['{}:QUBESRPC {} dom0'.format(user, service)],
**kwargs)
proc = subprocess.Popen(
[qubesadmin.config.QREXEC_CLIENT] + qrexec_opts + [
'{}:QUBESRPC {} dom0'.format(user, service)], **kwargs)
return proc
class QubesRemote(QubesBase):
'''Application object communicating through qrexec services.
"""Application object communicating through qrexec services.
Used when running in VM.
'''
"""
qubesd_connection_type = 'qrexec'
def qubesd_call(self, dest, method, arg=None, payload=None,
payload_stream=None):
'''
"""
Execute Admin API method.
Only one of `payload` and `payload_stream` can be specified.
@ -629,7 +643,7 @@ class QubesRemote(QubesBase):
:return: Data returned by qubesd (string)
.. warning:: *payload_stream* will get closed by this function
'''
"""
if payload and payload_stream:
raise ValueError(
'Only one of payload and payload_stream can be used')
@ -652,7 +666,7 @@ class QubesRemote(QubesBase):
def run_service(self, dest, service, filter_esc=False, user=None,
localcmd=None, wait=True, **kwargs):
'''Run qrexec service in a given destination
"""Run qrexec service in a given destination
:param str dest: Destination - may be a VM name or empty
string for default (for a given service)
@ -663,7 +677,7 @@ class QubesRemote(QubesBase):
:param str localcmd: Command to connect stdin/stdout to
:param bool wait: wait for process to finish
:rtype: subprocess.Popen
'''
"""
if filter_esc:
raise NotImplementedError(
'filter_esc not implemented for calls from VM')
@ -685,7 +699,7 @@ class QubesRemote(QubesBase):
kwargs.setdefault('stdin', subprocess.PIPE)
kwargs.setdefault('stdout', subprocess.PIPE)
kwargs.setdefault('stderr', subprocess.PIPE)
proc = subprocess.Popen([qubesadmin.config.QREXEC_CLIENT_VM,
dest or '', service] + (shlex.split(localcmd) if localcmd else []),
**kwargs)
proc = subprocess.Popen(
[qubesadmin.config.QREXEC_CLIENT_VM, dest or '', service] + (
shlex.split(localcmd) if localcmd else []), **kwargs)
return proc

14
qubesadmin/devices.py
View File

@ -303,17 +303,7 @@ class DeviceManager(dict):
return self[key]
def __iter__(self):
return iter(self._get_device_classes())
return iter(self._vm.app.list_deviceclass())
def keys(self):
return self._get_device_classes()
def _get_device_classes(self):
"""Function used to call Qubesd in order to obtain
the device classes list
"""
device_classes = \
self._vm.app.qubesd_call('dom0', 'admin.deviceclass.List').decode()
device_classes = sorted(device_classes.splitlines())
return device_classes
return self._vm.app.list_deviceclass()

1
qubesadmin/qubesadmin Symbolic link
View File

@ -0,0 +1 @@
/home/user/qubes-builder/qubes-src/core-admin-client/qubesadmin

5
qubesadmin/tools/qvm_create.py
View File

@ -103,9 +103,8 @@ def main(args=None, app=None):
args = parser.parse_args(args, app=app)
if args.help_classes:
vm_classes = args.app.qubesd_call('dom0', 'admin.vmclass.List').decode()
vm_classes = vm_classes.splitlines()
print('\n'.join(sorted(vm_classes)))
vm_classes = args.app.list_vmclass()
print('\n'.join(vm_classes))
return 0
pools = {}

16
qubesadmin/tools/qvm_device.py
View File

@ -211,6 +211,7 @@ def get_parser(device_class=None):
want_app=True)
parser.register('action', 'parsers',
qubesadmin.tools.AliasedSubParsersAction)
parser.allow_abbrev = False
if device_class:
parser.add_argument('devclass', const=device_class,
action='store_const',
@ -263,6 +264,9 @@ def get_parser(device_class=None):
attach_parser.set_defaults(func=attach_device)
detach_parser.set_defaults(func=detach_device)
parser.add_argument('--list-device-classes', action='store_true',
default=False)
return parser
@ -272,7 +276,13 @@ def main(args=None, app=None):
devclass = None
if basename.startswith('qvm-') and basename != 'qvm-device':
devclass = basename[4:]
args = get_parser(devclass).parse_args(args, app=app)
if args.list_device_classes:
print('\n'.join(qubesadmin.Qubes().list_deviceclass()))
return 0
try:
args.func(args)
except qubesadmin.exc.QubesException as e:
@ -282,4 +292,10 @@ def main(args=None, app=None):
if __name__ == '__main__':
# Special treatment for '--list-device-classes' (alias --list-classes)
curr_action = sys.argv[1:]
if set(curr_action).intersection(
{'--list-device-classes', '--list-classes'}):
sys.exit(main(args=['', '--list-device-classes']))
sys.exit(main())