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

qubesadmin: make PEP8 happy

Browse Source
This commit is contained in:
Frédéric Pierret (fepitre) 2019-08-10 19:03:17 +02:00
parent 769f8a5ee8
commit a982e1e538
No known key found for this signature in database
GPG Key ID: 484010B5CDC576E2
1 changed files with 76 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
qubesadmin/app.py
View File

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