__init__.py 15 KB

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