summaryrefslogtreecommitdiff
path: root/src/leap/bitmask/backend/backend_proxy.py
blob: e9ad9b4dd6404b0634e4ba5bb8a593f3e515c333 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
# -*- coding: utf-8 -*-
# backend_proxy.py
# Copyright (C) 2013, 2014 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""
The BackendProxy handles calls from the GUI and forwards (through ZMQ)
to the backend.
"""
# XXX should document the relationship to the API here.

import functools
import Queue
import threading
import time

import zmq

from leap.bitmask.backend.api import API, STOP_REQUEST, PING_REQUEST
from leap.bitmask.backend.settings import Settings
from leap.bitmask.backend.utils import generate_zmq_certificates_if_needed
from leap.bitmask.backend.utils import get_backend_certificates

import logging
logger = logging.getLogger(__name__)


class BackendProxy(object):
    """
    The BackendProxy handles calls from the GUI and forwards (through ZMQ)
    to the backend.
    """

    PORT = '5556'
    SERVER = "tcp://localhost:%s" % PORT

    POLL_TIMEOUT = 4000  # ms
    POLL_TRIES = 3

    PING_INTERVAL = 2  # secs

    def __init__(self):
        generate_zmq_certificates_if_needed()

        self._socket = None

        self.settings = Settings()

        # initialize ZMQ stuff:
        context = zmq.Context()
        logger.debug("Connecting to server...")
        socket = context.socket(zmq.REQ)

        # public, secret = zmq.curve_keypair()
        client_keys = zmq.curve_keypair()
        socket.curve_publickey = client_keys[0]
        socket.curve_secretkey = client_keys[1]

        # The client must know the server's public key to make a CURVE
        # connection.
        public, _ = get_backend_certificates()
        socket.curve_serverkey = public

        socket.setsockopt(zmq.RCVTIMEO, 1000)
        socket.setsockopt(zmq.LINGER, 0)  # Terminate early
        socket.connect(self.SERVER)
        self._socket = socket

        self._ping_at = 0
        self.online = False

        self._call_queue = Queue.Queue()
        self._worker_caller = threading.Thread(target=self._worker)

    def start(self):
        self._worker_caller.start()

    def check_online(self):
        """
        Return whether the backend is accessible or not.
        You don't need to do `run` in order to use this.

        :rtype: bool
        """
        # we use a small timeout in order to response quickly if the backend is
        # offline
        self._send_request(PING_REQUEST, retry=False, timeout=500)
        self._socket.close()
        return self.online

    def _worker(self):
        """
        Worker loop that processes the Queue of pending requests to do.
        """
        while True:
            try:
                request = self._call_queue.get(block=False)
                # break the loop after sending the 'stop' action to the
                # backend.
                if request == STOP_REQUEST:
                    break

                self._send_request(request)
            except Queue.Empty:
                pass
            time.sleep(0.01)
            self._ping()

        logger.debug("BackendProxy worker stopped.")

    def _reset_ping(self):
        """
        Reset the ping timeout counter.
        This is called for every ping and request.
        """
        self._ping_at = time.time() + self.PING_INTERVAL

    def _ping(self):
        """
        Heartbeat helper.
        Sends a PING request just to know that the server is alive.
        """
        if time.time() > self._ping_at:
            self._send_request(PING_REQUEST)
            self._reset_ping()

    def _api_call(self, *args, **kwargs):
        """
        Call the `api_method` method in backend (through zmq).

        :param kwargs: named arguments to forward to the backend api method.
        :type kwargs: dict

        Note: is mandatory to have the kwarg 'api_method' defined.
        """
        if args:
            # Use a custom message to be more clear about using kwargs *only*
            raise Exception("All arguments need to be kwargs!")

        api_method = kwargs.pop('api_method', None)
        if api_method is None:
            raise Exception("Missing argument, no method name specified.")

        request = {
            'api_method': api_method,
            'arguments': kwargs,
        }

        try:
            request_json = zmq.utils.jsonapi.dumps(request)
        except Exception as e:
            msg = ("Error serializing request into JSON.\n"
                   "Exception: {0} Data: {1}")
            msg = msg.format(e, request)
            logger.critical(msg)
            raise

        # queue the call in order to handle the request in a thread safe way.
        self._call_queue.put(request_json)

        if api_method == STOP_REQUEST:
            self._call_queue.put(STOP_REQUEST)

    def _send_request(self, request, retry=True, timeout=None):
        """
        Send the given request to the server.
        This is used from a thread safe loop in order to avoid sending a
        request without receiving a response from a previous one.

        :param request: the request to send.
        :type request: str
        :param retry: whether we should retry or not in case of timeout.
        :type retry: bool
        :param timeout: a custom timeout (milliseconds) to wait for a response.
        :type timeout: int
        """
        # logger.debug("Sending request to backend: {0}".format(request))
        self._socket.send(request)

        poll = zmq.Poller()
        poll.register(self._socket, zmq.POLLIN)

        reply = None

        tries = 0
        if not retry:
            tries = self.POLL_TRIES + 1  # this means: no retries left

        if timeout is None:
            timeout = self.POLL_TIMEOUT

        while True:
            socks = dict(poll.poll(timeout))
            if socks.get(self._socket) == zmq.POLLIN:
                reply = self._socket.recv()
                break

            tries += 1
            if tries < self.POLL_TRIES:
                logger.warning('Retrying receive... {0}/{1}'.format(
                    tries, self.POLL_TRIES))
            else:
                break

        if reply is None:
            msg = "Timeout error contacting backend."
            logger.critical(msg)
            self.online = False
        else:
            # msg = "Received reply for '{0}' -> '{1}'".format(request, reply)
            # logger.debug(msg)
            self.online = True
            # request received, no ping needed for other interval.
            self._reset_ping()

    def __getattribute__(self, name):
        """
        This allows the user to do:
            bp = BackendProxy()
            bp.some_method()

        Just by having defined 'some_method' in the API

        :param name: the attribute name that is requested.
        :type name: str
        """
        if name in API:
            return functools.partial(self._api_call, api_method=name)
        else:
            return object.__getattribute__(self, name)