base.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343
  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 qubesadmin.exc
  22. DEFAULT = object()
  23. class PropertyHolder(object):
  24. '''A base class for object having properties retrievable using mgmt API.
  25. Warning: each (non-private) local attribute needs to be defined at class
  26. level, even if initialized in __init__; otherwise will be treated as
  27. property retrievable using mgmt call.
  28. '''
  29. #: a place for appropriate Qubes() object (QubesLocal or QubesRemote),
  30. # use None for self
  31. app = None
  32. def __init__(self, app, method_prefix, method_dest):
  33. #: appropriate Qubes() object (QubesLocal or QubesRemote), use None
  34. # for self
  35. self.app = app
  36. self._method_prefix = method_prefix
  37. self._method_dest = method_dest
  38. self._properties = None
  39. self._properties_help = None
  40. def qubesd_call(self, dest, method, arg=None, payload=None,
  41. payload_stream=None):
  42. '''
  43. Call into qubesd using appropriate mechanism. This method should be
  44. defined by a subclass.
  45. Only one of `payload` and `payload_stream` can be specified.
  46. :param dest: Destination VM name
  47. :param method: Full API method name ('admin...')
  48. :param arg: Method argument (if any)
  49. :param payload: Payload send to the method
  50. :param payload_stream: file-like object to read payload from
  51. :return: Data returned by qubesd (string)
  52. '''
  53. if dest is None:
  54. dest = self._method_dest
  55. # have the actual implementation at Qubes() instance
  56. if self.app:
  57. return self.app.qubesd_call(dest, method, arg, payload,
  58. payload_stream)
  59. raise NotImplementedError
  60. @staticmethod
  61. def _parse_qubesd_response(response_data):
  62. '''Parse response from qubesd.
  63. In case of success, return actual data. In case of error,
  64. raise appropriate exception.
  65. '''
  66. if response_data == b'':
  67. raise qubesadmin.exc.QubesDaemonNoResponseError(
  68. 'Got empty response from qubesd. See journalctl in dom0 for '
  69. 'details.')
  70. if response_data[0:2] == b'\x30\x00':
  71. return response_data[2:]
  72. elif response_data[0:2] == b'\x32\x00':
  73. (_, exc_type, _traceback, format_string, args) = \
  74. response_data.split(b'\x00', 4)
  75. # drop last field because of terminating '\x00'
  76. args = [arg.decode() for arg in args.split(b'\x00')[:-1]]
  77. format_string = format_string.decode('utf-8')
  78. exc_type = exc_type.decode('ascii')
  79. try:
  80. exc_class = getattr(qubesadmin.exc, exc_type)
  81. except AttributeError:
  82. if exc_type.endswith('Error'):
  83. exc_class = __builtins__.get(exc_type,
  84. qubesadmin.exc.QubesException)
  85. else:
  86. exc_class = qubesadmin.exc.QubesException
  87. # TODO: handle traceback if given
  88. raise exc_class(format_string, *args)
  89. else:
  90. raise qubesadmin.exc.QubesDaemonCommunicationError(
  91. 'Invalid response format')
  92. def property_list(self):
  93. '''
  94. List available properties (their names).
  95. :return: list of strings
  96. '''
  97. if self._properties is None:
  98. properties_str = self.qubesd_call(
  99. self._method_dest,
  100. self._method_prefix + 'List',
  101. None,
  102. None)
  103. self._properties = properties_str.decode('ascii').splitlines()
  104. # TODO: make it somehow immutable
  105. return self._properties
  106. def property_help(self, name):
  107. '''
  108. Get description of a property.
  109. :return: property help text
  110. '''
  111. help_text = self.qubesd_call(
  112. self._method_dest,
  113. self._method_prefix + 'Help',
  114. name,
  115. None)
  116. return help_text.decode('ascii')
  117. def property_is_default(self, item):
  118. '''
  119. Check if given property have default value
  120. :param str item: name of property
  121. :return: bool
  122. '''
  123. if item.startswith('_'):
  124. raise AttributeError(item)
  125. property_str = self.qubesd_call(
  126. self._method_dest,
  127. self._method_prefix + 'Get',
  128. item,
  129. None)
  130. (default, _value) = property_str.split(b' ', 1)
  131. assert default.startswith(b'default=')
  132. is_default_str = default.split(b'=')[1]
  133. is_default = is_default_str.decode('ascii') == "True"
  134. assert isinstance(is_default, bool)
  135. return is_default
  136. def clone_properties(self, src, proplist=None):
  137. '''Clone properties from other object.
  138. :param PropertyHolder src: source object
  139. :param list proplist: list of properties \
  140. (:py:obj:`None` or omit for all properties)
  141. '''
  142. if proplist is None:
  143. proplist = self.property_list()
  144. for prop in proplist:
  145. try:
  146. setattr(self, prop, getattr(src, prop))
  147. except AttributeError:
  148. continue
  149. def __getattr__(self, item):
  150. # pylint: disable=too-many-return-statements
  151. if item.startswith('_'):
  152. raise AttributeError(item)
  153. try:
  154. property_str = self.qubesd_call(
  155. self._method_dest,
  156. self._method_prefix + 'Get',
  157. item,
  158. None)
  159. except qubesadmin.exc.QubesDaemonNoResponseError:
  160. raise qubesadmin.exc.QubesPropertyAccessError(item)
  161. (_default, prop_type, value) = property_str.split(b' ', 2)
  162. prop_type = prop_type.decode('ascii')
  163. if not prop_type.startswith('type='):
  164. raise qubesadmin.exc.QubesDaemonCommunicationError(
  165. 'Invalid type prefix received: {}'.format(prop_type))
  166. (_, prop_type) = prop_type.split('=', 1)
  167. value = value.decode()
  168. if prop_type == 'str':
  169. return str(value)
  170. elif prop_type == 'bool':
  171. if value == '':
  172. raise AttributeError
  173. return value == "True"
  174. elif prop_type == 'int':
  175. if value == '':
  176. raise AttributeError
  177. return int(value)
  178. elif prop_type == 'vm':
  179. if value == '':
  180. return None
  181. return self.app.domains[value]
  182. elif prop_type == 'label':
  183. if value == '':
  184. return None
  185. return self.app.labels.get_blind(value)
  186. else:
  187. raise qubesadmin.exc.QubesDaemonCommunicationError(
  188. 'Received invalid value type: {}'.format(prop_type))
  189. @classmethod
  190. def _local_properties(cls):
  191. '''
  192. Get set of property names that are properties on the Python object,
  193. and must not be set on the remote object
  194. '''
  195. if "_local_properties_set" not in cls.__dict__:
  196. props = set()
  197. for class_ in cls.__mro__:
  198. for key in class_.__dict__:
  199. props.add(key)
  200. cls._local_properties_set = props
  201. return cls._local_properties_set
  202. def __setattr__(self, key, value):
  203. if key.startswith('_') or key in self._local_properties():
  204. return super(PropertyHolder, self).__setattr__(key, value)
  205. if value is qubesadmin.DEFAULT:
  206. try:
  207. self.qubesd_call(
  208. self._method_dest,
  209. self._method_prefix + 'Reset',
  210. key,
  211. None)
  212. except qubesadmin.exc.QubesDaemonNoResponseError:
  213. raise qubesadmin.exc.QubesPropertyAccessError(key)
  214. else:
  215. if isinstance(value, qubesadmin.vm.QubesVM):
  216. value = value.name
  217. if value is None:
  218. value = ''
  219. try:
  220. self.qubesd_call(
  221. self._method_dest,
  222. self._method_prefix + 'Set',
  223. key,
  224. str(value).encode('utf-8'))
  225. except qubesadmin.exc.QubesDaemonNoResponseError:
  226. raise qubesadmin.exc.QubesPropertyAccessError(key)
  227. def __delattr__(self, name):
  228. if name.startswith('_') or name in self._local_properties():
  229. return super(PropertyHolder, self).__delattr__(name)
  230. try:
  231. self.qubesd_call(
  232. self._method_dest,
  233. self._method_prefix + 'Reset',
  234. name
  235. )
  236. except qubesadmin.exc.QubesDaemonNoResponseError:
  237. raise qubesadmin.exc.QubesPropertyAccessError(name)
  238. class WrapperObjectsCollection(object):
  239. '''Collection of simple named objects'''
  240. def __init__(self, app, list_method, object_class):
  241. '''
  242. Construct manager of named wrapper objects.
  243. :param app: Qubes() object
  244. :param list_method: name of API method used to list objects,
  245. must return simple "one name per line" list
  246. :param object_class: object class (callable) for wrapper objects,
  247. will be called with just two arguments: app and a name
  248. '''
  249. self.app = app
  250. self._list_method = list_method
  251. self._object_class = object_class
  252. #: names cache
  253. self._names_list = None
  254. #: returned objects cache
  255. self._objects = {}
  256. def clear_cache(self):
  257. '''Clear cached list of names'''
  258. self._names_list = None
  259. def refresh_cache(self, force=False):
  260. '''Refresh cached list of names'''
  261. if not force and self._names_list is not None:
  262. return
  263. list_data = self.app.qubesd_call('dom0', self._list_method)
  264. list_data = list_data.decode('ascii')
  265. assert list_data[-1] == '\n'
  266. self._names_list = [str(name) for name in list_data[:-1].splitlines()]
  267. for name, obj in list(self._objects.items()):
  268. if obj.name not in self._names_list:
  269. # Object no longer exists
  270. del self._objects[name]
  271. def __getitem__(self, item):
  272. if not self.app.blind_mode and item not in self:
  273. raise KeyError(item)
  274. return self.get_blind(item)
  275. def get_blind(self, item):
  276. '''
  277. Get a property without downloading the list
  278. and checking if it's present
  279. '''
  280. if item not in self._objects:
  281. self._objects[item] = self._object_class(self.app, item)
  282. return self._objects[item]
  283. def __contains__(self, item):
  284. self.refresh_cache()
  285. return item in self._names_list
  286. def __iter__(self):
  287. self.refresh_cache()
  288. for key in self._names_list:
  289. yield key
  290. def keys(self):
  291. '''Get list of names.'''
  292. self.refresh_cache()
  293. return [key for key in self._names_list]
  294. def items(self):
  295. '''Get list of (key, value) pairs'''
  296. self.refresh_cache()
  297. return [(key, self.get_blind(key)) for key in self._names_list]
  298. def values(self):
  299. '''Get list of objects'''
  300. self.refresh_cache()
  301. return [self.get_blind(key) for key in self._names_list]