Add base communication framework.

1 files changed, 196 insertions, 0 deletions

diff --git a/src/leap/bitmask/backend/backend.py b/src/leap/bitmask/backend/backend.py
new file mode 100644
index 00000000..26c547b6
--- /dev/null
+++ b/src/leap/bitmask/backend/backend.py
@@ -0,0 +1,196 @@
+#!/usr/bin/env python
+# encoding: utf-8
+import json
+import threading
+import time
+
+from twisted.internet import defer, reactor, threads
+
+import zmq
+from zmq.auth.thread import ThreadAuthenticator
+
+from leap.bitmask.backend.api import API
+from leap.bitmask.backend.utils import get_backend_certificates
+from leap.bitmask.backend.signaler import Signaler
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+class Backend(object):
+    """
+    Backend server.
+    Receives signals from backend_proxy and emit signals if needed.
+    """
+    PORT = '5556'
+    BIND_ADDR = "tcp://127.0.0.1:%s" % PORT
+
+    def __init__(self):
+        """
+        Backend constructor, create needed instances.
+        """
+        self._signaler = Signaler()
+
+        self._do_work = threading.Event()  # used to stop the worker thread.
+        self._zmq_socket = None
+
+        self._ongoing_defers = []
+        self._init_zmq()
+
+    def _init_zmq(self):
+        """
+        Configure the zmq components and connection.
+        """
+        context = zmq.Context()
+        socket = context.socket(zmq.REP)
+
+        # Start an authenticator for this context.
+        auth = ThreadAuthenticator(context)
+        auth.start()
+        auth.allow('127.0.0.1')
+
+        # Tell authenticator to use the certificate in a directory
+        auth.configure_curve(domain='*', location=zmq.auth.CURVE_ALLOW_ANY)
+        public, secret = get_backend_certificates()
+        socket.curve_publickey = public
+        socket.curve_secretkey = secret
+        socket.curve_server = True  # must come before bind
+
+        socket.bind(self.BIND_ADDR)
+
+        self._zmq_socket = socket
+
+    def _worker(self):
+        """
+        Receive requests and send it to process.
+
+        Note: we use a simple while since is less resource consuming than a
+        Twisted's LoopingCall.
+        """
+        while self._do_work.is_set():
+            # Wait for next request from client
+            try:
+                request = self._zmq_socket.recv(zmq.NOBLOCK)
+                self._zmq_socket.send("OK")
+                logger.debug("Received request: '{0}'".format(request))
+                self._process_request(request)
+            except zmq.ZMQError as e:
+                if e.errno != zmq.EAGAIN:
+                    raise
+            time.sleep(0.01)
+
+    def _stop_reactor(self):
+        """
+        Stop the Twisted reactor, but first wait a little for some threads to
+        complete their work.
+
+        Note: this method needs to be run in a different thread so the
+        time.sleep() does not block and other threads can finish.
+        i.e.:
+            use threads.deferToThread(this_method) instead of this_method()
+        """
+        wait_max = 5  # seconds
+        wait_step = 0.5
+        wait = 0
+        while self._ongoing_defers and wait < wait_max:
+            time.sleep(wait_step)
+            wait += wait_step
+            msg = "Waiting for running threads to finish... {0}/{1}"
+            msg = msg.format(wait, wait_max)
+            logger.debug(msg)
+
+        # after a timeout we shut down the existing threads.
+        for d in self._ongoing_defers:
+            d.cancel()
+
+        reactor.stop()
+        logger.debug("Twisted reactor stopped.")
+
+    def run(self):
+        """
+        Start the ZMQ server and run the loop to handle requests.
+        """
+        self._signaler.start()
+        self._do_work.set()
+        threads.deferToThread(self._worker)
+        reactor.run()
+
+    def stop(self):
+        """
+        Stop the server and the zmq request parse loop.
+        """
+        logger.debug("STOP received.")
+        self._signaler.stop()
+        self._do_work.clear()
+        threads.deferToThread(self._stop_reactor)
+
+    def _process_request(self, request_json):
+        """
+        Process a request and call the according method with the given
+        parameters.
+
+        :param request_json: a json specification of a request.
+        :type request_json: str
+        """
+        try:
+            # request = zmq.utils.jsonapi.loads(request_json)
+            # We use stdlib's json to ensure that we get unicode strings
+            request = json.loads(request_json)
+            api_method = request['api_method']
+            kwargs = request['arguments'] or None
+        except Exception as e:
+            msg = "Malformed JSON data in Backend request '{0}'. Exc: {1!r}"
+            msg = msg.format(request_json, e)
+            msg = msg.format(request_json)
+            logger.critical(msg)
+            raise
+
+        if api_method not in API:
+            logger.error("Invalid API call '{0}'".format(api_method))
+            return
+
+        self._run_in_thread(api_method, kwargs)
+
+    def _run_in_thread(self, api_method, kwargs):
+        """
+        Run the method name in a thread with the given arguments.
+
+        :param api_method: the callable name to run in a thread.
+        :type api_method: str
+        :param kwargs: the arguments dict that will be sent to the callable.
+        :type kwargs: tuple
+        """
+        func = getattr(self, api_method)
+
+        method = func
+        if kwargs is not None:
+            method = lambda: func(**kwargs)
+
+        logger.debug("Running method: '{0}' "
+                     "with args: '{1}' in a thread".format(api_method, kwargs))
+
+        # run the action in a thread and keep track of it
+        d = threads.deferToThread(method)
+        d.addCallback(self._done_action, d)
+        d.addErrback(self._done_action, d)
+        self._ongoing_defers.append(d)
+
+    def _done_action(self, failure, d):
+        """
+        Remove the defer from the ongoing list.
+
+        :param failure: the failure that triggered the errback.
+                        None if no error.
+        :type failure: twisted.python.failure.Failure
+        :param d: defer to remove
+        :type d: twisted.internet.defer.Deferred
+        """
+        if failure is not None:
+            if failure.check(defer.CancelledError):
+                logger.debug("A defer was cancelled.")
+            else:
+                logger.error("There was a failure - {0!r}".format(failure))
+                logger.error(failure.getTraceback())
+
+        if d in self._ongoing_defers:
+            self._ongoing_defers.remove(d)