__init__.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429
  1. # -*- encoding: utf8 -*-
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2017 Wojtek Porczyk <woju@invisiblethingslab.com>
  6. # Copyright (C) 2017 Marek Marczykowski-Górecki
  7. # <marmarek@invisiblethingslab.com>
  8. #
  9. # This library is free software; you can redistribute it and/or
  10. # modify it under the terms of the GNU Lesser General Public
  11. # License as published by the Free Software Foundation; either
  12. # version 2.1 of the License, or (at your option) any later version.
  13. #
  14. # This library is distributed in the hope that it will be useful,
  15. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  17. # Lesser General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU Lesser General Public
  20. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  21. import asyncio
  22. import errno
  23. import functools
  24. import io
  25. import os
  26. import shutil
  27. import socket
  28. import struct
  29. import traceback
  30. import qubes.exc
  31. class ProtocolError(AssertionError):
  32. '''Raised when something is wrong with data received'''
  33. pass
  34. class PermissionDenied(Exception):
  35. '''Raised deliberately by handlers when we decide not to cooperate'''
  36. pass
  37. def method(name, *, no_payload=False, endpoints=None, **classifiers):
  38. '''Decorator factory for methods intended to appear in API.
  39. The decorated method can be called from public API using a child of
  40. :py:class:`AbstractQubesMgmt` class. The method becomes "public", and can be
  41. called using remote management interface.
  42. :param str name: qrexec rpc method name
  43. :param bool no_payload: if :py:obj:`True`, will barf on non-empty payload; \
  44. also will not pass payload at all to the method
  45. :param iterable endpoints: if specified, method serve multiple API calls
  46. generated by replacing `{endpoint}` with each value in this iterable
  47. The expected function method should have one argument (other than usual
  48. *self*), ``untrusted_payload``, which will contain the payload.
  49. .. warning::
  50. This argument has to be named such, to remind the programmer that the
  51. content of this variable is indeed untrusted.
  52. If *no_payload* is true, then the method is called with no arguments.
  53. '''
  54. def decorator(func):
  55. if no_payload:
  56. # the following assignment is needed for how closures work in Python
  57. _func = func
  58. @functools.wraps(_func)
  59. def wrapper(self, untrusted_payload, **kwargs):
  60. if untrusted_payload != b'':
  61. raise ProtocolError('unexpected payload')
  62. return _func(self, **kwargs)
  63. func = wrapper
  64. # pylint: disable=protected-access
  65. if endpoints is None:
  66. func.rpcnames = ((name, None),)
  67. else:
  68. func.rpcnames = tuple(
  69. (name.format(endpoint=endpoint), endpoint)
  70. for endpoint in endpoints)
  71. func.classifiers = classifiers
  72. return func
  73. return decorator
  74. def apply_filters(iterable, filters):
  75. '''Apply filters returned by admin-permission:... event'''
  76. for selector in filters:
  77. iterable = filter(selector, iterable)
  78. return iterable
  79. class AbstractQubesAPI:
  80. '''Common code for Qubes Management Protocol handling
  81. Different interfaces can expose different API call sets, however they share
  82. common protocol and common implementation framework. This class is the
  83. latter.
  84. To implement a new interface, inherit from this class and write at least one
  85. method and decorate it with :py:func:`api` decorator. It will have access to
  86. pre-defined attributes: :py:attr:`app`, :py:attr:`src`, :py:attr:`dest`,
  87. :py:attr:`arg` and :py:attr:`method`.
  88. There are also two helper functions for firing events associated with API
  89. calls.
  90. '''
  91. #: the preferred socket location (to be overridden in child's class)
  92. SOCKNAME = None
  93. def __init__(self, app, src, method_name, dest, arg, send_event=None):
  94. #: :py:class:`qubes.Qubes` object
  95. self.app = app
  96. #: source qube
  97. self.src = self.app.domains[src.decode('ascii')]
  98. #: destination qube
  99. self.dest = self.app.domains[dest.decode('ascii')]
  100. #: argument
  101. self.arg = arg.decode('ascii')
  102. #: name of the method
  103. self.method = method_name.decode('ascii')
  104. #: callback for sending events if applicable
  105. self.send_event = send_event
  106. #: is this operation cancellable?
  107. self.cancellable = False
  108. candidates = list(self.list_methods(self.method))
  109. if not candidates:
  110. raise ProtocolError('no such method: {!r}'.format(self.method))
  111. assert len(candidates) == 1, \
  112. 'multiple candidates for method {!r}'.format(self.method)
  113. #: the method to execute
  114. self._handler = candidates[0]
  115. self._running_handler = None
  116. @classmethod
  117. def list_methods(cls, select_method=None):
  118. for attr in dir(cls):
  119. func = getattr(cls, attr)
  120. if not callable(func):
  121. continue
  122. try:
  123. # pylint: disable=protected-access
  124. rpcnames = func.rpcnames
  125. except AttributeError:
  126. continue
  127. for mname, endpoint in rpcnames:
  128. if select_method is None or mname == select_method:
  129. yield (func, mname, endpoint)
  130. def execute(self, *, untrusted_payload):
  131. '''Execute management operation.
  132. This method is a coroutine.
  133. '''
  134. handler, _, endpoint = self._handler
  135. kwargs = {}
  136. if endpoint is not None:
  137. kwargs['endpoint'] = endpoint
  138. self._running_handler = asyncio.ensure_future(handler(self,
  139. untrusted_payload=untrusted_payload, **kwargs))
  140. return self._running_handler
  141. def cancel(self):
  142. '''If operation is cancellable, interrupt it'''
  143. if self.cancellable and self._running_handler is not None:
  144. self._running_handler.cancel()
  145. def fire_event_for_permission(self, **kwargs):
  146. '''Fire an event on the source qube to check for permission'''
  147. return self.src.fire_event('admin-permission:' + self.method,
  148. pre_event=True, dest=self.dest, arg=self.arg, **kwargs)
  149. def fire_event_for_filter(self, iterable, **kwargs):
  150. '''Fire an event on the source qube to filter for permission'''
  151. return apply_filters(iterable,
  152. self.fire_event_for_permission(**kwargs))
  153. class QubesDaemonProtocol(asyncio.Protocol):
  154. buffer_size = 65536
  155. header = struct.Struct('Bx')
  156. # keep track of connections, to gracefully close them at server exit
  157. # (including cleanup of integration test)
  158. connections = set()
  159. def __init__(self, handler, *args, app, debug=False, **kwargs):
  160. super().__init__(*args, **kwargs)
  161. self.handler = handler
  162. self.app = app
  163. self.untrusted_buffer = io.BytesIO()
  164. self.len_untrusted_buffer = 0
  165. self.transport = None
  166. self.debug = debug
  167. self.event_sent = False
  168. self.mgmt = None
  169. def connection_made(self, transport):
  170. self.transport = transport
  171. self.connections.add(self)
  172. def connection_lost(self, exc):
  173. self.untrusted_buffer.close()
  174. # for cancellable operation, interrupt it, otherwise it will do nothing
  175. if self.mgmt is not None:
  176. self.mgmt.cancel()
  177. self.transport = None
  178. self.connections.remove(self)
  179. def data_received(self, untrusted_data): # pylint: disable=arguments-differ
  180. if self.len_untrusted_buffer + len(untrusted_data) > self.buffer_size:
  181. self.app.log.warning('request too long')
  182. self.transport.abort()
  183. self.untrusted_buffer.close()
  184. return
  185. self.len_untrusted_buffer += \
  186. self.untrusted_buffer.write(untrusted_data)
  187. def eof_received(self):
  188. try:
  189. src, meth, dest, arg, untrusted_payload = \
  190. self.untrusted_buffer.getvalue().split(b'\0', 4)
  191. except ValueError:
  192. self.app.log.warning('framing error')
  193. self.transport.abort()
  194. return None
  195. finally:
  196. self.untrusted_buffer.close()
  197. asyncio.ensure_future(self.respond(
  198. src, meth, dest, arg, untrusted_payload=untrusted_payload))
  199. return True
  200. @asyncio.coroutine
  201. def respond(self, src, meth, dest, arg, *, untrusted_payload):
  202. try:
  203. self.mgmt = self.handler(self.app, src, meth, dest, arg,
  204. self.send_event)
  205. response = yield from self.mgmt.execute(
  206. untrusted_payload=untrusted_payload)
  207. assert not (self.event_sent and response)
  208. if self.transport is None:
  209. return
  210. # except clauses will fall through to transport.abort() below
  211. except PermissionDenied:
  212. self.app.log.warning(
  213. 'permission denied for call %s+%s (%s → %s) '
  214. 'with payload of %d bytes',
  215. meth, arg, src, dest, len(untrusted_payload))
  216. except ProtocolError:
  217. self.app.log.warning(
  218. 'protocol error for call %s+%s (%s → %s) '
  219. 'with payload of %d bytes',
  220. meth, arg, src, dest, len(untrusted_payload))
  221. except qubes.exc.QubesException as err:
  222. msg = ('%r while calling '
  223. 'src=%r meth=%r dest=%r arg=%r len(untrusted_payload)=%d')
  224. if self.debug:
  225. self.app.log.debug(msg,
  226. err, src, meth, dest, arg, len(untrusted_payload),
  227. exc_info=1)
  228. if self.transport is not None:
  229. self.send_exception(err)
  230. self.transport.write_eof()
  231. self.transport.close()
  232. return
  233. except Exception: # pylint: disable=broad-except
  234. self.app.log.exception(
  235. 'unhandled exception while calling '
  236. 'src=%r meth=%r dest=%r arg=%r len(untrusted_payload)=%d',
  237. src, meth, dest, arg, len(untrusted_payload))
  238. else:
  239. if not self.event_sent:
  240. self.send_response(response)
  241. try:
  242. self.transport.write_eof()
  243. except NotImplementedError:
  244. pass
  245. self.transport.close()
  246. return
  247. # this is reached if from except: blocks; do not put it in finally:,
  248. # because this will prevent the good case from sending the reply
  249. if self.transport:
  250. self.transport.abort()
  251. def send_header(self, *args):
  252. self.transport.write(self.header.pack(*args))
  253. def send_response(self, content):
  254. assert not self.event_sent
  255. self.send_header(0x30)
  256. if content is not None:
  257. self.transport.write(content.encode('utf-8'))
  258. def send_event(self, subject, event, **kwargs):
  259. if self.transport is None:
  260. return
  261. self.event_sent = True
  262. self.send_header(0x31)
  263. if subject is not self.app:
  264. self.transport.write(str(subject).encode('ascii'))
  265. self.transport.write(b'\0')
  266. self.transport.write(event.encode('ascii') + b'\0')
  267. for k, v in kwargs.items():
  268. self.transport.write('{}\0{}\0'.format(k, str(v)).encode('ascii'))
  269. self.transport.write(b'\0')
  270. def send_exception(self, exc):
  271. self.send_header(0x32)
  272. self.transport.write(type(exc).__name__.encode() + b'\0')
  273. if self.debug:
  274. self.transport.write(''.join(traceback.format_exception(
  275. type(exc), exc, exc.__traceback__)).encode('utf-8'))
  276. self.transport.write(b'\0')
  277. self.transport.write(str(exc).encode('utf-8') + b'\0')
  278. def cleanup_socket(sockpath, force):
  279. '''Remove socket if stale, or force=True
  280. :param sockpath: path to a socket
  281. :param force: should remove even if still used
  282. '''
  283. if force:
  284. os.unlink(sockpath)
  285. else:
  286. sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  287. try:
  288. sock.connect(sockpath)
  289. except ConnectionRefusedError:
  290. # dead socket, remove it anyway
  291. os.unlink(sockpath)
  292. else:
  293. # woops, someone is listening
  294. sock.close()
  295. raise FileExistsError(errno.EEXIST,
  296. 'socket already exists: {!r}'.format(sockpath))
  297. @asyncio.coroutine
  298. def create_servers(*args, force=False, loop=None, **kwargs):
  299. '''Create multiple Qubes API servers
  300. :param qubes.Qubes app: the app that is a backend of the servers
  301. :param bool force: if :py:obj:`True`, unconditionally remove existing \
  302. sockets; if :py:obj:`False`, raise an error if there is some process \
  303. listening to such socket
  304. :param asyncio.Loop loop: loop
  305. *args* are supposed to be classes inheriting from
  306. :py:class:`AbstractQubesAPI`
  307. *kwargs* (like *app* or *debug* for example) are passed to
  308. :py:class:`QubesDaemonProtocol` constructor
  309. '''
  310. loop = loop or asyncio.get_event_loop()
  311. servers = []
  312. old_umask = os.umask(0o007)
  313. try:
  314. # XXX this can be optimised with asyncio.wait() to start servers in
  315. # parallel, but I currently don't see the need
  316. for handler in args:
  317. sockpath = handler.SOCKNAME
  318. assert sockpath is not None, \
  319. 'SOCKNAME needs to be overloaded in {}'.format(
  320. type(handler).__name__)
  321. if os.path.exists(sockpath):
  322. cleanup_socket(sockpath, force)
  323. server = yield from loop.create_unix_server(
  324. functools.partial(QubesDaemonProtocol, handler, **kwargs),
  325. sockpath)
  326. for sock in server.sockets:
  327. shutil.chown(sock.getsockname(), group='qubes')
  328. servers.append(server)
  329. except:
  330. for server in servers:
  331. for sock in server.sockets:
  332. try:
  333. os.unlink(sock.getsockname())
  334. except FileNotFoundError:
  335. pass
  336. server.close()
  337. if servers:
  338. yield from asyncio.wait([
  339. server.wait_closed() for server in servers])
  340. raise
  341. finally:
  342. os.umask(old_umask)
  343. return servers