base.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Marek Marczykowski-Górecki
  6. # <marmarek@invisiblethingslab.com>
  7. #
  8. # This program is free software; you can redistribute it and/or modify
  9. # it under the terms of the GNU Lesser General Public License as published by
  10. # the Free Software Foundation; either version 2 of the License, or
  11. # (at your option) any later version.
  12. #
  13. # This program is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16. # GNU Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public License along
  19. # with this program; if not, see <http://www.gnu.org/licenses/>.
  20. '''Base classes for managed objects'''
  21. import ast
  22. import qubesadmin.exc
  23. DEFAULT = object()
  24. class PropertyHolder(object):
  25. '''A base class for object having properties retrievable using mgmt API.
  26. Warning: each (non-private) local attribute needs to be defined at class
  27. level, even if initialized in __init__; otherwise will be treated as
  28. property retrievable using mgmt call.
  29. '''
  30. #: a place for appropriate Qubes() object (QubesLocal or QubesRemote),
  31. # use None for self
  32. app = None
  33. def __init__(self, app, method_prefix, method_dest):
  34. #: appropriate Qubes() object (QubesLocal or QubesRemote), use None
  35. # for self
  36. self.app = app
  37. self._method_prefix = method_prefix
  38. self._method_dest = method_dest
  39. self._properties = None
  40. self._properties_help = None
  41. def qubesd_call(self, dest, method, arg=None, payload=None,
  42. payload_stream=None):
  43. '''
  44. Call into qubesd using appropriate mechanism. This method should be
  45. defined by a subclass.
  46. Only one of `payload` and `payload_stream` can be specified.
  47. :param dest: Destination VM name
  48. :param method: Full API method name ('admin...')
  49. :param arg: Method argument (if any)
  50. :param payload: Payload send to the method
  51. :param payload_stream: file-like object to read payload from
  52. :return: Data returned by qubesd (string)
  53. '''
  54. if dest is None:
  55. dest = self._method_dest
  56. # have the actual implementation at Qubes() instance
  57. if self.app:
  58. return self.app.qubesd_call(dest, method, arg, payload,
  59. payload_stream)
  60. raise NotImplementedError
  61. @staticmethod
  62. def _parse_qubesd_response(response_data):
  63. '''Parse response from qubesd.
  64. In case of success, return actual data. In case of error,
  65. raise appropriate exception.
  66. '''
  67. if response_data == b'':
  68. raise qubesadmin.exc.QubesDaemonNoResponseError(
  69. 'Got empty response from qubesd. See journalctl in dom0 for '
  70. 'details.')
  71. if response_data[0:2] == b'\x30\x00':
  72. return response_data[2:]
  73. elif response_data[0:2] == b'\x32\x00':
  74. (_, exc_type, _traceback, format_string, args) = \
  75. response_data.split(b'\x00', 4)
  76. # drop last field because of terminating '\x00'
  77. args = [arg.decode() for arg in args.split(b'\x00')[:-1]]
  78. format_string = format_string.decode('utf-8')
  79. exc_type = exc_type.decode('ascii')
  80. try:
  81. exc_class = getattr(qubesadmin.exc, exc_type)
  82. except AttributeError:
  83. if exc_type.endswith('Error'):
  84. exc_class = __builtins__.get(exc_type,
  85. qubesadmin.exc.QubesException)
  86. else:
  87. exc_class = qubesadmin.exc.QubesException
  88. # TODO: handle traceback if given
  89. raise exc_class(format_string, *args)
  90. else:
  91. raise qubesadmin.exc.QubesDaemonCommunicationError(
  92. 'Invalid response format')
  93. def property_list(self):
  94. '''
  95. List available properties (their names).
  96. :return: list of strings
  97. '''
  98. if self._properties is None:
  99. properties_str = self.qubesd_call(
  100. self._method_dest,
  101. self._method_prefix + 'List',
  102. None,
  103. None)
  104. self._properties = properties_str.decode('ascii').splitlines()
  105. # TODO: make it somehow immutable
  106. return self._properties
  107. def property_help(self, name):
  108. '''
  109. Get description of a property.
  110. :return: property help text
  111. '''
  112. help_text = self.qubesd_call(
  113. self._method_dest,
  114. self._method_prefix + 'Help',
  115. name,
  116. None)
  117. return help_text.decode('ascii')
  118. def property_is_default(self, item):
  119. '''
  120. Check if given property have default value
  121. :param str item: name of property
  122. :return: bool
  123. '''
  124. if item.startswith('_'):
  125. raise AttributeError(item)
  126. property_str = self.qubesd_call(
  127. self._method_dest,
  128. self._method_prefix + 'Get',
  129. item,
  130. None)
  131. (default, _value) = property_str.split(b' ', 1)
  132. assert default.startswith(b'default=')
  133. is_default_str = default.split(b'=')[1]
  134. is_default = ast.literal_eval(is_default_str.decode('ascii'))
  135. assert isinstance(is_default, bool)
  136. return is_default
  137. def clone_properties(self, src, proplist=None):
  138. '''Clone properties from other object.
  139. :param PropertyHolder src: source object
  140. :param list proplist: list of properties \
  141. (:py:obj:`None` or omit for all properties)
  142. '''
  143. if proplist is None:
  144. proplist = self.property_list()
  145. for prop in proplist:
  146. try:
  147. setattr(self, prop, getattr(src, prop))
  148. except AttributeError:
  149. continue
  150. def __getattr__(self, item):
  151. # pylint: disable=too-many-return-statements
  152. if item.startswith('_'):
  153. raise AttributeError(item)
  154. try:
  155. property_str = self.qubesd_call(
  156. self._method_dest,
  157. self._method_prefix + 'Get',
  158. item,
  159. None)
  160. except qubesadmin.exc.QubesDaemonNoResponseError:
  161. raise qubesadmin.exc.QubesPropertyAccessError(item)
  162. (_default, prop_type, value) = property_str.split(b' ', 2)
  163. prop_type = prop_type.decode('ascii')
  164. if not prop_type.startswith('type='):
  165. raise qubesadmin.exc.QubesDaemonCommunicationError(
  166. 'Invalid type prefix received: {}'.format(prop_type))
  167. (_, prop_type) = prop_type.split('=', 1)
  168. value = value.decode()
  169. if prop_type == 'str':
  170. return str(value)
  171. elif prop_type == 'bool':
  172. if value == '':
  173. raise AttributeError
  174. return ast.literal_eval(value)
  175. elif prop_type == 'int':
  176. if value == '':
  177. raise AttributeError
  178. return ast.literal_eval(value)
  179. elif prop_type == 'vm':
  180. if value == '':
  181. return None
  182. return self.app.domains[value]
  183. elif prop_type == 'label':
  184. if value == '':
  185. return None
  186. # TODO
  187. return self.app.labels[value]
  188. else:
  189. raise qubesadmin.exc.QubesDaemonCommunicationError(
  190. 'Received invalid value type: {}'.format(prop_type))
  191. def __setattr__(self, key, value):
  192. if key.startswith('_') or key in dir(self):
  193. return super(PropertyHolder, self).__setattr__(key, value)
  194. if value is qubesadmin.DEFAULT:
  195. try:
  196. self.qubesd_call(
  197. self._method_dest,
  198. self._method_prefix + 'Reset',
  199. key,
  200. None)
  201. except qubesadmin.exc.QubesDaemonNoResponseError:
  202. raise qubesadmin.exc.QubesPropertyAccessError(key)
  203. else:
  204. if isinstance(value, qubesadmin.vm.QubesVM):
  205. value = value.name
  206. if value is None:
  207. value = ''
  208. try:
  209. self.qubesd_call(
  210. self._method_dest,
  211. self._method_prefix + 'Set',
  212. key,
  213. str(value).encode('utf-8'))
  214. except qubesadmin.exc.QubesDaemonNoResponseError:
  215. raise qubesadmin.exc.QubesPropertyAccessError(key)
  216. def __delattr__(self, name):
  217. if name.startswith('_') or name in dir(self):
  218. return super(PropertyHolder, self).__delattr__(name)
  219. try:
  220. self.qubesd_call(
  221. self._method_dest,
  222. self._method_prefix + 'Reset',
  223. name
  224. )
  225. except qubesadmin.exc.QubesDaemonNoResponseError:
  226. raise qubesadmin.exc.QubesPropertyAccessError(name)
  227. class WrapperObjectsCollection(object):
  228. '''Collection of simple named objects'''
  229. def __init__(self, app, list_method, object_class):
  230. '''
  231. Construct manager of named wrapper objects.
  232. :param app: Qubes() object
  233. :param list_method: name of API method used to list objects,
  234. must return simple "one name per line" list
  235. :param object_class: object class (callable) for wrapper objects,
  236. will be called with just two arguments: app and a name
  237. '''
  238. self.app = app
  239. self._list_method = list_method
  240. self._object_class = object_class
  241. #: names cache
  242. self._names_list = None
  243. #: returned objects cache
  244. self._objects = {}
  245. def clear_cache(self):
  246. '''Clear cached list of names'''
  247. self._names_list = None
  248. def refresh_cache(self, force=False):
  249. '''Refresh cached list of names'''
  250. if not force and self._names_list is not None:
  251. return
  252. list_data = self.app.qubesd_call('dom0', self._list_method)
  253. list_data = list_data.decode('ascii')
  254. assert list_data[-1] == '\n'
  255. self._names_list = [str(name) for name in list_data[:-1].splitlines()]
  256. for name, obj in list(self._objects.items()):
  257. if obj.name not in self._names_list:
  258. # Object no longer exists
  259. del self._objects[name]
  260. def __getitem__(self, item):
  261. if not self.app.blind_mode and item not in self:
  262. raise KeyError(item)
  263. if item not in self._objects:
  264. self._objects[item] = self._object_class(self.app, item)
  265. return self._objects[item]
  266. def __contains__(self, item):
  267. self.refresh_cache()
  268. return item in self._names_list
  269. def __iter__(self):
  270. self.refresh_cache()
  271. for obj in self._names_list:
  272. yield self[obj]
  273. def keys(self):
  274. '''Get list of names.'''
  275. self.refresh_cache()
  276. return self._names_list.copy()