__init__.py 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217
  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. '''Event handling implementation, require Python >=3.5.2 for asyncio.'''
  21. import asyncio
  22. import fnmatch
  23. import subprocess
  24. import qubesadmin.config
  25. import qubesadmin.exc
  26. class EventsDispatcher(object):
  27. ''' Events dispatcher, responsible for receiving events and calling
  28. appropriate handlers'''
  29. def __init__(self, app, api_method='admin.Events'):
  30. '''Initialize EventsDispatcher'''
  31. #: Qubes() object
  32. self.app = app
  33. self._api_method = api_method
  34. #: event handlers - dict of event -> handlers
  35. self.handlers = {}
  36. def add_handler(self, event, handler):
  37. '''Register handler for event
  38. Use '*' as event to register a handler for all events.
  39. Handler function is called with:
  40. * subject (VM object or None)
  41. * event name (str)
  42. * keyword arguments related to the event, if any - all values as str
  43. :param event Event name, or '*' for all events
  44. :param handler Handler function'''
  45. self.handlers.setdefault(event, set()).add(handler)
  46. def remove_handler(self, event, handler):
  47. '''Remove previously registered event handler
  48. :param event Event name
  49. :param handler Handler function
  50. '''
  51. self.handlers[event].remove(handler)
  52. @asyncio.coroutine
  53. def _get_events_reader(self, vm=None) -> (asyncio.StreamReader, callable):
  54. '''Make connection to qubesd and return stream to read events from
  55. :param vm: Specific VM for which events should be handled, use None
  56. to handle events from all VMs (and non-VM objects)
  57. :return stream to read events from and a cleanup function
  58. (call it to terminate qubesd connection)'''
  59. if vm is not None:
  60. dest = vm.name
  61. else:
  62. dest = 'dom0'
  63. if self.app.qubesd_connection_type == 'socket':
  64. reader, writer = yield from asyncio.open_unix_connection(
  65. qubesadmin.config.QUBESD_SOCKET)
  66. writer.write(b'dom0\0') # source
  67. writer.write(self._api_method.encode() + b'\0') # method
  68. writer.write(dest.encode('ascii') + b'\0') # dest
  69. writer.write(b'\0') # arg
  70. writer.write_eof()
  71. def cleanup_func():
  72. '''Close connection to qubesd'''
  73. writer.close()
  74. elif self.app.qubesd_connection_type == 'qrexec':
  75. proc = yield from asyncio.create_subprocess_exec(
  76. 'qrexec-client-vm', dest, self._api_method,
  77. stdin=subprocess.PIPE, stdout=subprocess.PIPE)
  78. proc.stdin.write_eof()
  79. reader = proc.stdout
  80. def cleanup_func():
  81. '''Close connection to qubesd'''
  82. try:
  83. proc.kill()
  84. except ProcessLookupError:
  85. pass
  86. else:
  87. raise NotImplementedError('Unsupported qubesd connection type: '
  88. + self.app.qubesd_connection_type)
  89. return reader, cleanup_func
  90. @asyncio.coroutine
  91. def listen_for_events(self, vm=None, reconnect=True):
  92. '''
  93. Listen for events and call appropriate handlers.
  94. This function do not exit until manually terminated.
  95. This is coroutine.
  96. :param vm: Listen for events only for this VM, use None to listen for
  97. events about all VMs and not related to any particular VM.
  98. :param reconnect: should reconnect to qubesd if connection is
  99. interrupted?
  100. :rtype: None
  101. '''
  102. while True:
  103. try:
  104. yield from self._listen_for_events(vm)
  105. except (OSError, qubesadmin.exc.QubesDaemonCommunicationError):
  106. pass
  107. if not reconnect:
  108. break
  109. self.app.log.warning(
  110. 'Connection to qubesd terminated, reconnecting in {} '
  111. 'seconds'.format(qubesadmin.config.QUBESD_RECONNECT_DELAY))
  112. # avoid busy-loop if qubesd is dead
  113. yield from asyncio.sleep(qubesadmin.config.QUBESD_RECONNECT_DELAY)
  114. @asyncio.coroutine
  115. def _listen_for_events(self, vm=None):
  116. '''
  117. Listen for events and call appropriate handlers.
  118. This function do not exit until manually terminated.
  119. This is coroutine.
  120. :param vm: Listen for events only for this VM, use None to listen for
  121. events about all VMs and not related to any particular VM.
  122. :return: True if any event was received, otherwise False
  123. :rtype: bool
  124. '''
  125. reader, cleanup_func = yield from self._get_events_reader(vm)
  126. try:
  127. some_event_received = False
  128. while not reader.at_eof():
  129. try:
  130. event_header = yield from reader.readuntil(b'\0')
  131. if event_header != b'1\0':
  132. raise qubesadmin.exc.QubesDaemonCommunicationError(
  133. 'Non-event received on events connection: '
  134. + repr(event_header))
  135. subject = (yield from reader.readuntil(b'\0'))[:-1].decode(
  136. 'utf-8')
  137. event = (yield from reader.readuntil(b'\0'))[:-1].decode(
  138. 'utf-8')
  139. kwargs = {}
  140. while True:
  141. key = (yield from reader.readuntil(b'\0'))[:-1].decode(
  142. 'utf-8')
  143. if not key:
  144. break
  145. value = (yield from reader.readuntil(b'\0'))[:-1].\
  146. decode('utf-8')
  147. kwargs[key] = value
  148. except BrokenPipeError:
  149. break
  150. except asyncio.IncompleteReadError as err:
  151. if err.partial == b'':
  152. break
  153. raise
  154. if not subject:
  155. subject = None
  156. self.handle(subject, event, **kwargs)
  157. some_event_received = True
  158. finally:
  159. cleanup_func()
  160. return some_event_received
  161. def handle(self, subject, event, **kwargs):
  162. '''Call handlers for given event'''
  163. if subject:
  164. if event in ['property-set:name']:
  165. self.app.domains.clear_cache()
  166. try:
  167. subject = self.app.domains.get_blind(subject)
  168. except KeyError:
  169. return
  170. else:
  171. # handle cache refreshing on best-effort basis
  172. if event in ['domain-add', 'domain-delete']:
  173. self.app.domains.clear_cache()
  174. subject = None
  175. # deserialize known attributes
  176. if event.startswith('device-') and 'device' in kwargs:
  177. try:
  178. devclass = event.split(':', 1)[1]
  179. backend_domain, ident = kwargs['device'].split(':', 1)
  180. kwargs['device'] = self.app.domains.get_blind(backend_domain)\
  181. .devices[devclass][ident]
  182. except (KeyError, ValueError):
  183. pass
  184. handlers = [h_func for h_name, h_func_set in self.handlers.items()
  185. for h_func in h_func_set
  186. if fnmatch.fnmatch(event, h_name)]
  187. for handler in handlers:
  188. handler(subject, event, **kwargs)