Skip to content

Commit

Permalink
Cleanup Admin API denial reporting
Browse files Browse the repository at this point in the history
Rename QubesDaemonNoResponseError to more intuitive
QubesDaemonAccessError (keep legacy name still working).
Use QubesPropertyAccessError whenever the access is about @Property -
this makes it easy to use `getattr` to use default value instead.

QubesOS/qubes-issues#5811
  • Loading branch information
marmarek committed Aug 11, 2020
1 parent b04a146 commit 7425a53
Show file tree
Hide file tree
Showing 4 changed files with 81 additions and 31 deletions.
2 changes: 1 addition & 1 deletion qubesadmin/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -838,7 +838,7 @@ def qubesd_call(self, dest, method, arg=None, payload=None,
stderr=subprocess.PIPE)
(stdout, stderr) = p.communicate(payload)
if p.returncode != 0:
raise qubesadmin.exc.QubesDaemonNoResponseError(
raise qubesadmin.exc.QubesDaemonAccessError(
'Service call error: %s', stderr.decode())

return self._parse_qubesd_response(stdout)
Expand Down
28 changes: 17 additions & 11 deletions qubesadmin/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ def _parse_qubesd_response(response_data):
'''

if response_data == b'':
raise qubesadmin.exc.QubesDaemonNoResponseError(
raise qubesadmin.exc.QubesDaemonAccessError(
'Got empty response from qubesd. See journalctl in dom0 for '
'details.')

Expand Down Expand Up @@ -151,11 +151,14 @@ def property_is_default(self, item):
# cached properties list
if self._properties is not None and item not in self._properties:
raise AttributeError(item)
property_str = self.qubesd_call(
self._method_dest,
self._method_prefix + 'Get',
item,
None)
try:
property_str = self.qubesd_call(
self._method_dest,
self._method_prefix + 'Get',
item,
None)
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError(item)
is_default, value = self._deserialize_property(property_str)
if self.app.cache_enabled:
self._properties_cache[item] = (is_default, value)
Expand All @@ -170,11 +173,14 @@ def property_get_default(self, item):
'''
if item.startswith('_'):
raise AttributeError(item)
property_str = self.qubesd_call(
self._method_dest,
self._method_prefix + 'GetDefault',
item,
None)
try:
property_str = self.qubesd_call(
self._method_dest,
self._method_prefix + 'GetDefault',
item,
None)
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError(item)
if not property_str:
raise AttributeError(item + ' has no default')
(prop_type, value) = property_str.split(b' ', 1)
Expand Down
10 changes: 7 additions & 3 deletions qubesadmin/exc.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,13 +161,17 @@ def __init__(self, msg, backup_log=None):
self.backup_log = backup_log

# pylint: disable=too-many-ancestors
class QubesDaemonNoResponseError(QubesDaemonCommunicationError):
'''Got empty response from qubesd'''
class QubesDaemonAccessError(QubesDaemonCommunicationError):
'''Got empty response from qubesd. This can be lack of permission,
or some server-side issue.'''


class QubesPropertyAccessError(QubesException, AttributeError):
class QubesPropertyAccessError(QubesDaemonAccessError, AttributeError):
'''Failed to read/write property value, cause is unknown (insufficient
permissions, no such property, invalid value, other)'''
def __init__(self, prop):
super(QubesPropertyAccessError, self).__init__(
'Failed to access \'%s\' property' % prop)

# legacy name
QubesDaemonNoResponseError = QubesDaemonAccessError
72 changes: 56 additions & 16 deletions qubesadmin/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
# with this program; if not, see <http://www.gnu.org/licenses/>.

'''Storage subsystem.'''
import qubesadmin.exc

class Volume(object):
'''Storage volume.'''
Expand Down Expand Up @@ -112,33 +113,48 @@ def pool(self):
'''Storage volume pool name.'''
if self._pool is not None:
return self._pool
self._fetch_info()
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('pool')
return str(self._info['pool'])

@property
def vid(self):
'''Storage volume id, unique within given pool.'''
if self._vid is not None:
return self._vid
self._fetch_info()
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('vid')
return str(self._info['vid'])

@property
def size(self):
'''Size of volume, in bytes.'''
self._fetch_info(True)
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('size')
return int(self._info['size'])

@property
def usage(self):
'''Used volume space, in bytes.'''
self._fetch_info(True)
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('usage')
return int(self._info['usage'])

@property
def rw(self):
'''True if volume is read-write.'''
self._fetch_info()
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('rw')
return self._info['rw'] == 'True'

@rw.setter
Expand All @@ -150,13 +166,19 @@ def rw(self, value):
@property
def snap_on_start(self):
'''Create a snapshot from source on VM start.'''
self._fetch_info()
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('snap_on_start')
return self._info['snap_on_start'] == 'True'

@property
def save_on_stop(self):
'''Commit changes to original volume on VM stop.'''
self._fetch_info()
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('save_on_stop')
return self._info['save_on_stop'] == 'True'

@property
Expand All @@ -165,15 +187,21 @@ def source(self):
If None, this volume itself will be used.
'''
self._fetch_info()
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('source')
if self._info['source']:
return self._info['source']
return None

@property
def revisions_to_keep(self):
'''Number of revisions to keep around'''
self._fetch_info()
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('revisions_to_keep')
return int(self._info['revisions_to_keep'])

@revisions_to_keep.setter
Expand All @@ -186,7 +214,10 @@ def is_outdated(self):
'''Returns `True` if this snapshot of a source volume (for
`snap_on_start`=True) is outdated.
'''
self._fetch_info(True)
try:
self._fetch_info()
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('is_outdated')
return self._info.get('is_outdated', False) == 'True'

def resize(self, size):
Expand Down Expand Up @@ -290,8 +321,11 @@ def __lt__(self, other):
@property
def usage_details(self):
''' Storage pool usage details (current - not cached) '''
pool_usage_data = self.app.qubesd_call(
'dom0', 'admin.pool.UsageDetails', self.name, None)
try:
pool_usage_data = self.app.qubesd_call(
'dom0', 'admin.pool.UsageDetails', self.name, None)
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('usage_details')
pool_usage_data = pool_usage_data.decode('utf-8')
assert pool_usage_data.endswith('\n') or pool_usage_data == ''
pool_usage_data = pool_usage_data[:-1]
Expand All @@ -306,8 +340,11 @@ def _int_split(text): # pylint: disable=missing-docstring
def config(self):
''' Storage pool config '''
if self._config is None:
pool_info_data = self.app.qubesd_call(
'dom0', 'admin.pool.Info', self.name, None)
try:
pool_info_data = self.app.qubesd_call(
'dom0', 'admin.pool.Info', self.name, None)
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('config')
pool_info_data = pool_info_data.decode('utf-8')
assert pool_info_data.endswith('\n')
pool_info_data = pool_info_data[:-1]
Expand Down Expand Up @@ -355,8 +392,11 @@ def revisions_to_keep(self, value):
@property
def volumes(self):
''' Volumes managed by this pool '''
volumes_data = self.app.qubesd_call(
'dom0', 'admin.pool.volume.List', self.name, None)
try:
volumes_data = self.app.qubesd_call(
'dom0', 'admin.pool.volume.List', self.name, None)
except qubesadmin.exc.QubesDaemonAccessError:
raise qubesadmin.exc.QubesPropertyAccessError('volumes')
assert volumes_data.endswith(b'\n')
volumes_data = volumes_data[:-1].decode('ascii')
for vid in volumes_data.splitlines():
Expand Down

0 comments on commit 7425a53

Please sign in to comment.