From c298eddd161525b84d7f50ad9f1690b0fbca8b8b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Fri, 24 Feb 2017 01:21:02 +0100 Subject: [PATCH] Avoid cyclic imports QubesOS/qubes-issues#853 --- ci/pylintrc | 1 - qubesmgmt/__init__.py | 134 +-------------------------------- qubesmgmt/app.py | 4 +- qubesmgmt/base.py | 156 +++++++++++++++++++++++++++++++++++++++ qubesmgmt/vm/__init__.py | 4 +- 5 files changed, 163 insertions(+), 136 deletions(-) create mode 100644 qubesmgmt/base.py diff --git a/ci/pylintrc b/ci/pylintrc index 973e99b..0ac8194 100644 --- a/ci/pylintrc +++ b/ci/pylintrc @@ -6,7 +6,6 @@ ignore=tests # abstract-class-little-used: see http://www.logilab.org/ticket/111138 disable= bad-continuation, - cyclic-import, fixme, locally-disabled, missing-docstring diff --git a/qubesmgmt/__init__.py b/qubesmgmt/__init__.py index 4b834bc..ffab4ce 100644 --- a/qubesmgmt/__init__.py +++ b/qubesmgmt/__init__.py @@ -19,140 +19,12 @@ # with this program; if not, see . -import ast import os - -import qubesmgmt.exc - -DEFAULT = object() - - -class PropertyHolder(object): - '''A base class for object having properties retrievable using mgmt API. - - Warning: each (non-private) local attribute needs to be defined at class - level, even if initialized in __init__; otherwise will be treated as - property retrievable using mgmt call. - ''' - #: a place for appropriate Qubes() object (QubesLocal or QubesRemote), - # use None for self - app = None - - def __init__(self, app, method_prefix, method_dest): - #: appropriate Qubes() object (QubesLocal or QubesRemote), use None - # for self - self.app = app - self._method_prefix = method_prefix - self._method_dest = method_dest - self._properties = None - self._properties_help = None - - def qubesd_call(self, dest, method, arg=None, payload=None): - ''' - Call into qubesd using appropriate mechanism. This method should be - defined by a subclass. - - :param dest: Destination VM name - :param method: Full API method name ('mgmt...') - :param arg: Method argument (if any) - :param payload: Payload send to the method - :return: Data returned by qubesd (string) - ''' - # have the actual implementation at Qubes() instance - if self.app: - return self.app.qubesd_call(dest, method, arg, payload) - raise NotImplementedError - - @staticmethod - def _parse_qubesd_response(response_data): - if response_data[0:2] == b'\x30\x00': - return response_data[2:] - elif response_data[0:2] == b'\x32\x00': - (_, exc_type, _traceback, format_string, args) = \ - response_data.split(b'\x00', 4) - # drop last field because of terminating '\x00' - args = [arg.decode() for arg in args.split(b'\x00')[:-1]] - format_string = format_string.decode('utf-8') - exc_type = exc_type.decode('ascii') - exc_class = getattr(qubesmgmt.exc, exc_type, 'QubesException') - # TODO: handle traceback if given - raise exc_class(format_string, *args) - else: - raise qubesmgmt.exc.QubesException('Invalid response format') - - - def property_list(self): - if self._properties is None: - properties_str = self.qubesd_call( - self._method_dest, - self._method_prefix + 'List', - None, - None) - self._properties = properties_str.decode('ascii').splitlines() - # TODO: make it somehow immutable - return self._properties - - def property_is_default(self, item): - if item.startswith('_'): - raise AttributeError(item) - property_str = self.qubesd_call( - self._method_dest, - self._method_prefix + 'Get', - item, - None) - (default, _value) = property_str.split(b' ', 1) - assert default.startswith(b'default=') - is_default_str = default.split(b'=')[1] - is_default = ast.literal_eval(is_default_str.decode('ascii')) - assert isinstance(is_default, bool) - return is_default - - def __getattr__(self, item): - if item.startswith('_'): - raise AttributeError(item) - property_str = self.qubesd_call( - self._method_dest, - self._method_prefix + 'Get', - item, - None) - (_default, value) = property_str.split(b' ', 1) - value = value.decode() - if value[0] == '\'': - return ast.literal_eval('u' + value) - else: - return ast.literal_eval(value) - - def __setattr__(self, key, value): - if key.startswith('_') or key in dir(self): - return super(PropertyHolder, self).__setattr__(key, value) - if value is DEFAULT: - self.qubesd_call( - self._method_dest, - self._method_prefix + 'Reset', - key, - None) - else: - if isinstance(value, qubesmgmt.vm.QubesVM): - # pylint: disable=protected-access - value = value._name - self.qubesd_call( - self._method_dest, - self._method_prefix + 'Set', - key, - str(value).encode('utf-8')) - - def __delattr__(self, name): - if name.startswith('_') or name in dir(self): - return super(PropertyHolder, self).__delattr__(name) - self.qubesd_call( - self._method_dest, - self._method_prefix + 'Reset', - name - ) - -# pylint: disable=wrong-import-position +import qubesmgmt.base import qubesmgmt.app +DEFAULT = qubesmgmt.base.DEFAULT + if os.path.exists(qubesmgmt.app.QUBESD_SOCK): Qubes = qubesmgmt.app.QubesLocal else: diff --git a/qubesmgmt/app.py b/qubesmgmt/app.py index b56d07f..f3b2ebc 100644 --- a/qubesmgmt/app.py +++ b/qubesmgmt/app.py @@ -22,7 +22,7 @@ import socket import subprocess -import qubesmgmt +import qubesmgmt.base import qubesmgmt.vm import qubesmgmt.exc @@ -71,7 +71,7 @@ class VMCollection(object): return self._vm_list.keys() -class QubesBase(qubesmgmt.PropertyHolder): +class QubesBase(qubesmgmt.base.PropertyHolder): '''Main Qubes application''' #: domains (VMs) collection diff --git a/qubesmgmt/base.py b/qubesmgmt/base.py new file mode 100644 index 0000000..3506a1d --- /dev/null +++ b/qubesmgmt/base.py @@ -0,0 +1,156 @@ +# -*- encoding: utf8 -*- +# +# The Qubes OS Project, http://www.qubes-os.org +# +# Copyright (C) 2017 Marek Marczykowski-Górecki +# +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License along +# with this program; if not, see . +import ast +import qubesmgmt.exc + +DEFAULT = object() + +class PropertyHolder(object): + '''A base class for object having properties retrievable using mgmt API. + + Warning: each (non-private) local attribute needs to be defined at class + level, even if initialized in __init__; otherwise will be treated as + property retrievable using mgmt call. + ''' + #: a place for appropriate Qubes() object (QubesLocal or QubesRemote), + # use None for self + app = None + + def __init__(self, app, method_prefix, method_dest): + #: appropriate Qubes() object (QubesLocal or QubesRemote), use None + # for self + self.app = app + self._method_prefix = method_prefix + self._method_dest = method_dest + self._properties = None + self._properties_help = None + + def qubesd_call(self, dest, method, arg=None, payload=None): + ''' + Call into qubesd using appropriate mechanism. This method should be + defined by a subclass. + + :param dest: Destination VM name + :param method: Full API method name ('mgmt...') + :param arg: Method argument (if any) + :param payload: Payload send to the method + :return: Data returned by qubesd (string) + ''' + # have the actual implementation at Qubes() instance + if self.app: + return self.app.qubesd_call(dest, method, arg, payload) + raise NotImplementedError + + @staticmethod + def _parse_qubesd_response(response_data): + if response_data[0:2] == b'\x30\x00': + return response_data[2:] + elif response_data[0:2] == b'\x32\x00': + (_, exc_type, _traceback, format_string, args) = \ + response_data.split(b'\x00', 4) + # drop last field because of terminating '\x00' + args = [arg.decode() for arg in args.split(b'\x00')[:-1]] + format_string = format_string.decode('utf-8') + exc_type = exc_type.decode('ascii') + exc_class = getattr(qubesmgmt.exc, exc_type, 'QubesException') + # TODO: handle traceback if given + raise exc_class(format_string, *args) + else: + raise qubesmgmt.exc.QubesException('Invalid response format') + + def property_list(self): + ''' + List available properties (their names). + + :return: list of strings + ''' + if self._properties is None: + properties_str = self.qubesd_call( + self._method_dest, + self._method_prefix + 'List', + None, + None) + self._properties = properties_str.decode('ascii').splitlines() + # TODO: make it somehow immutable + return self._properties + + def property_is_default(self, item): + ''' + Check if given property have default value + + :param str item: name of property + :return: bool + ''' + if item.startswith('_'): + raise AttributeError(item) + property_str = self.qubesd_call( + self._method_dest, + self._method_prefix + 'Get', + item, + None) + (default, _value) = property_str.split(b' ', 1) + assert default.startswith(b'default=') + is_default_str = default.split(b'=')[1] + is_default = ast.literal_eval(is_default_str.decode('ascii')) + assert isinstance(is_default, bool) + return is_default + + def __getattr__(self, item): + if item.startswith('_'): + raise AttributeError(item) + property_str = self.qubesd_call( + self._method_dest, + self._method_prefix + 'Get', + item, + None) + (_default, value) = property_str.split(b' ', 1) + value = value.decode() + if value[0] == '\'': + return ast.literal_eval('u' + value) + else: + return ast.literal_eval(value) + + def __setattr__(self, key, value): + if key.startswith('_') or key in dir(self): + return super(PropertyHolder, self).__setattr__(key, value) + if value is qubesmgmt.DEFAULT: + self.qubesd_call( + self._method_dest, + self._method_prefix + 'Reset', + key, + None) + else: + if isinstance(value, qubesmgmt.vm.QubesVM): + # pylint: disable=protected-access + value = value._name + self.qubesd_call( + self._method_dest, + self._method_prefix + 'Set', + key, + str(value).encode('utf-8')) + + def __delattr__(self, name): + if name.startswith('_') or name in dir(self): + return super(PropertyHolder, self).__delattr__(name) + self.qubesd_call( + self._method_dest, + self._method_prefix + 'Reset', + name + ) diff --git a/qubesmgmt/vm/__init__.py b/qubesmgmt/vm/__init__.py index a6b9d09..521cfdb 100644 --- a/qubesmgmt/vm/__init__.py +++ b/qubesmgmt/vm/__init__.py @@ -18,10 +18,10 @@ # You should have received a copy of the GNU General Public License along # with this program; if not, see . -import qubesmgmt +import qubesmgmt.base -class QubesVM(qubesmgmt.PropertyHolder): +class QubesVM(qubesmgmt.base.PropertyHolder): def __init__(self, app, name, vm_class): self._name = name self._class = vm_class