summaryrefslogtreecommitdiff
path: root/src/leap/bitmask/backend/signaler.py
blob: aec2f606240f2db329067b9a4c940776975cda71 (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
# -*- coding: utf-8 -*-
# signaler.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/>.
"""
Signaler client.
Receives signals from the backend and sends to the signaling server.
"""
import Queue
import threading
import time

import zmq

from leap.bitmask.backend.api import SIGNALS
from leap.bitmask.backend.utils import get_frontend_certificates
from leap.bitmask.config import flags

import logging
logger = logging.getLogger(__name__)


class Signaler(object):
    """
    Signaler client.
    Receives signals from the backend and sends to the signaling server.
    """
    if flags.ZMQ_HAS_CURVE:
        PORT = "5667"
        SERVER = "tcp://localhost:%s" % PORT
    else:
        SERVER = "ipc:///tmp/bitmask.socket.1"
    POLL_TIMEOUT = 2000  # ms
    POLL_TRIES = 500

    def __init__(self):
        """
        Initialize the ZMQ socket to talk to the signaling server.
        """
        context = zmq.Context()
        logger.debug("Connecting to signaling server...")
        socket = context.socket(zmq.REQ)

        if flags.ZMQ_HAS_CURVE:
            # 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_frontend_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._signal_queue = Queue.Queue()

        self._do_work = threading.Event()  # used to stop the worker thread.
        self._worker_signaler = threading.Thread(target=self._worker)

    def __getattribute__(self, name):
        """
        This allows the user to do:
            S = Signaler()
            S.SOME_SIGNAL

        Just by having defined 'some_signal' in _SIGNALS

        :param name: the attribute name that is requested.
        :type name: str
        """
        if name in SIGNALS:
            return name
        else:
            return object.__getattribute__(self, name)

    def signal(self, signal, data=None):
        """
        Sends a signal to the signaling server.

        :param signal: the signal to send.
        :type signal: str
        """
        if signal not in SIGNALS:
            raise Exception("Unknown signal: '{0}'".format(signal))

        request = {
            'signal': signal,
            'data': data,
        }

        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._signal_queue.put(request_json)

    def _worker(self):
        """
        Worker loop that processes the Queue of pending requests to do.
        """
        while self._do_work.is_set():
            try:
                request = self._signal_queue.get(block=False)
                self._send_request(request)
            except Queue.Empty:
                pass
            time.sleep(0.01)

        logger.debug("Signaler thread stopped.")

    def start(self):
        """
        Start the Signaler worker.
        """
        self._do_work.set()
        self._worker_signaler.start()

    def stop(self):
        """
        Stop the Signaler worker.
        """
        self._do_work.clear()

    def _send_request(self, request):
        """
        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
        """
        # logger.debug("Signaling '{0}'".format(request))
        self._socket.send(request)

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

        reply = None
        tries = 0

        while True:
            socks = dict(poll.poll(self.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)
        # else:
        #     msg = "Received reply for '{0}' -> '{1}'".format(request, reply)
        #     logger.debug(msg)