← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/server.py

  • Committer: Patch Queue Manager
  • Date: 2013-05-20 17:46:29 UTC
  • mfrom: (6573.1.1 bzr)
  • Revision ID: pqm@pqm.ubuntu.com-20130520174629-dp7zujtuclvomuzd
(jameinel) Fix CVE 2013-2009. Avoid allowing multiple wildcards in a single
 SSL cert hostname segment. (Andrew Starr-Bochicchio)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/server.py
16
16
 
17
17
"""Server for smart-server protocol."""
18
18
 
 
19
from __future__ import absolute_import
 
20
 
19
21
import errno
20
22
import os.path
21
23
import socket
22
24
import sys
 
25
import time
23
26
import threading
24
27
 
25
28
from bzrlib.hooks import Hooks
31
34
from bzrlib.i18n import gettext
32
35
from bzrlib.lazy_import import lazy_import
33
36
lazy_import(globals(), """
34
 
from bzrlib.smart import medium
 
37
from bzrlib.smart import (
 
38
    medium,
 
39
    signals,
 
40
    )
35
41
from bzrlib.transport import (
36
42
    chroot,
37
43
    pathfilter,
38
44
    )
39
45
from bzrlib import (
 
46
    config,
40
47
    urlutils,
41
48
    )
42
49
""")
51
58
    hooks: An instance of SmartServerHooks.
52
59
    """
53
60
 
54
 
    def __init__(self, backing_transport, root_client_path='/'):
 
61
    # This is the timeout on the socket we use .accept() on. It is exposed here
 
62
    # so the test suite can set it faster. (It thread.interrupt_main() will not
 
63
    # fire a KeyboardInterrupt during socket.accept)
 
64
    _ACCEPT_TIMEOUT = 1.0
 
65
    _SHUTDOWN_POLL_TIMEOUT = 1.0
 
66
    _LOG_WAITING_TIMEOUT = 10.0
 
67
 
 
68
    _timer = time.time
 
69
 
 
70
    def __init__(self, backing_transport, root_client_path='/',
 
71
                 client_timeout=None):
55
72
        """Construct a new server.
56
73
 
57
74
        To actually start it running, call either start_background_thread or
60
77
        :param backing_transport: The transport to serve.
61
78
        :param root_client_path: The client path that will correspond to root
62
79
            of backing_transport.
 
80
        :param client_timeout: See SmartServerSocketStreamMedium's timeout
 
81
            parameter.
63
82
        """
64
83
        self.backing_transport = backing_transport
65
84
        self.root_client_path = root_client_path
 
85
        self._client_timeout = client_timeout
 
86
        self._active_connections = []
 
87
        # This is set to indicate we want to wait for clients to finish before
 
88
        # we disconnect.
 
89
        self._gracefully_stopping = False
66
90
 
67
91
    def start_server(self, host, port):
68
92
        """Create the server listening socket.
94
118
        self._sockname = self._server_socket.getsockname()
95
119
        self.port = self._sockname[1]
96
120
        self._server_socket.listen(1)
97
 
        self._server_socket.settimeout(1)
 
121
        self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
 
122
        # Once we start accept()ing connections, we set started.
98
123
        self._started = threading.Event()
 
124
        # Once we stop accept()ing connections (and are closing the socket) we
 
125
        # set _stopped
99
126
        self._stopped = threading.Event()
 
127
        # Once we have finished waiting for all clients, etc. We set
 
128
        # _fully_stopped
 
129
        self._fully_stopped = threading.Event()
100
130
 
101
131
    def _backing_urls(self):
102
132
        # There are three interesting urls:
135
165
        for hook in SmartTCPServer.hooks['server_stopped']:
136
166
            hook(backing_urls, self.get_url())
137
167
 
 
168
    def _stop_gracefully(self):
 
169
        trace.note(gettext('Requested to stop gracefully'))
 
170
        self._should_terminate = True
 
171
        self._gracefully_stopping = True
 
172
        for handler, _ in self._active_connections:
 
173
            handler._stop_gracefully()
 
174
 
 
175
    def _wait_for_clients_to_disconnect(self):
 
176
        self._poll_active_connections()
 
177
        if not self._active_connections:
 
178
            return
 
179
        trace.note(gettext('Waiting for %d client(s) to finish')
 
180
                   % (len(self._active_connections),))
 
181
        t_next_log = self._timer() + self._LOG_WAITING_TIMEOUT
 
182
        while self._active_connections:
 
183
            now = self._timer()
 
184
            if now >= t_next_log:
 
185
                trace.note(gettext('Still waiting for %d client(s) to finish')
 
186
                           % (len(self._active_connections),))
 
187
                t_next_log = now + self._LOG_WAITING_TIMEOUT
 
188
            self._poll_active_connections(self._SHUTDOWN_POLL_TIMEOUT)
 
189
 
138
190
    def serve(self, thread_name_suffix=''):
 
191
        # Note: There is a temptation to do
 
192
        #       signals.register_on_hangup(id(self), self._stop_gracefully)
 
193
        #       However, that creates a temporary object which is a bound
 
194
        #       method. signals._on_sighup is a WeakKeyDictionary so it
 
195
        #       immediately gets garbage collected, because nothing else
 
196
        #       references it. Instead, we need to keep a real reference to the
 
197
        #       bound method for the lifetime of the serve() function.
 
198
        stop_gracefully = self._stop_gracefully
 
199
        signals.register_on_hangup(id(self), stop_gracefully)
139
200
        self._should_terminate = False
140
201
        # for hooks we are letting code know that a server has started (and
141
202
        # later stopped).
151
212
                        pass
152
213
                    except self._socket_error, e:
153
214
                        # if the socket is closed by stop_background_thread
154
 
                        # we might get a EBADF here, any other socket errors
155
 
                        # should get logged.
156
 
                        if e.args[0] != errno.EBADF:
157
 
                            trace.warning("listening socket error: %s", e)
 
215
                        # we might get a EBADF here, or if we get a signal we
 
216
                        # can get EINTR, any other socket errors should get
 
217
                        # logged.
 
218
                        if e.args[0] not in (errno.EBADF, errno.EINTR):
 
219
                            trace.warning(gettext("listening socket error: %s")
 
220
                                          % (e,))
158
221
                    else:
159
222
                        if self._should_terminate:
 
223
                            conn.close()
160
224
                            break
161
225
                        self.serve_conn(conn, thread_name_suffix)
 
226
                    # Cleanout any threads that have finished processing.
 
227
                    self._poll_active_connections()
162
228
            except KeyboardInterrupt:
163
229
                # dont log when CTRL-C'd.
164
230
                raise
166
232
                trace.report_exception(sys.exc_info(), sys.stderr)
167
233
                raise
168
234
        finally:
169
 
            self._stopped.set()
170
235
            try:
171
236
                # ensure the server socket is closed.
172
237
                self._server_socket.close()
173
238
            except self._socket_error:
174
239
                # ignore errors on close
175
240
                pass
 
241
            self._stopped.set()
 
242
            signals.unregister_on_hangup(id(self))
176
243
            self.run_server_stopped_hooks()
 
244
        if self._gracefully_stopping:
 
245
            self._wait_for_clients_to_disconnect()
 
246
        self._fully_stopped.set()
177
247
 
178
248
    def get_url(self):
179
249
        """Return the url of the server"""
180
250
        return "bzr://%s:%s/" % (self._sockname[0], self._sockname[1])
181
251
 
 
252
    def _make_handler(self, conn):
 
253
        return medium.SmartServerSocketStreamMedium(
 
254
            conn, self.backing_transport, self.root_client_path,
 
255
            timeout=self._client_timeout)
 
256
 
 
257
    def _poll_active_connections(self, timeout=0.0):
 
258
        """Check to see if any active connections have finished.
 
259
 
 
260
        This will iterate through self._active_connections, and update any
 
261
        connections that are finished.
 
262
 
 
263
        :param timeout: The timeout to pass to thread.join(). By default, we
 
264
            set it to 0, so that we don't hang if threads are not done yet.
 
265
        :return: None
 
266
        """
 
267
        still_active = []
 
268
        for handler, thread in self._active_connections:
 
269
            thread.join(timeout)
 
270
            if thread.isAlive():
 
271
                still_active.append((handler, thread))
 
272
        self._active_connections = still_active
 
273
 
182
274
    def serve_conn(self, conn, thread_name_suffix):
183
275
        # For WIN32, where the timeout value from the listening socket
184
276
        # propagates to the newly accepted socket.
185
277
        conn.setblocking(True)
186
278
        conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
187
 
        handler = medium.SmartServerSocketStreamMedium(
188
 
            conn, self.backing_transport, self.root_client_path)
189
279
        thread_name = 'smart-server-child' + thread_name_suffix
 
280
        handler = self._make_handler(conn)
190
281
        connection_thread = threading.Thread(
191
282
            None, handler.serve, name=thread_name)
192
 
        # FIXME: This thread is never joined, it should at least be collected
193
 
        # somewhere so that tests that want to check for leaked threads can get
194
 
        # rid of them -- vila 20100531
 
283
        self._active_connections.append((handler, connection_thread))
195
284
        connection_thread.setDaemon(True)
196
285
        connection_thread.start()
197
286
        return connection_thread
341
430
            transport = _mod_transport.get_transport_from_url(expand_userdirs.get_url())
342
431
        self.transport = transport
343
432
 
344
 
    def _make_smart_server(self, host, port, inet):
 
433
    def _get_stdin_stdout(self):
 
434
        return sys.stdin, sys.stdout
 
435
 
 
436
    def _make_smart_server(self, host, port, inet, timeout):
 
437
        if timeout is None:
 
438
            c = config.GlobalStack()
 
439
            timeout = c.get('serve.client_timeout')
345
440
        if inet:
 
441
            stdin, stdout = self._get_stdin_stdout()
346
442
            smart_server = medium.SmartServerPipeStreamMedium(
347
 
                sys.stdin, sys.stdout, self.transport)
 
443
                stdin, stdout, self.transport, timeout=timeout)
348
444
        else:
349
445
            if host is None:
350
446
                host = medium.BZR_DEFAULT_INTERFACE
351
447
            if port is None:
352
448
                port = medium.BZR_DEFAULT_PORT
353
 
            smart_server = SmartTCPServer(self.transport)
 
449
            smart_server = SmartTCPServer(self.transport,
 
450
                                          client_timeout=timeout)
354
451
            smart_server.start_server(host, port)
355
452
            trace.note(gettext('listening on port: %s') % smart_server.port)
356
453
        self.smart_server = smart_server
369
466
        self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
370
467
        ui.ui_factory = ui.SilentUIFactory()
371
468
        lockdir._DEFAULT_TIMEOUT_SECONDS = 0
 
469
        orig = signals.install_sighup_handler()
 
470
        def restore_signals():
 
471
            signals.restore_sighup_handler(orig)
 
472
        self.cleanups.append(restore_signals)
372
473
 
373
 
    def set_up(self, transport, host, port, inet):
 
474
    def set_up(self, transport, host, port, inet, timeout):
374
475
        self._make_backing_transport(transport)
375
 
        self._make_smart_server(host, port, inet)
 
476
        self._make_smart_server(host, port, inet, timeout)
376
477
        self._change_globals()
377
478
 
378
479
    def tear_down(self):
379
480
        for cleanup in reversed(self.cleanups):
380
481
            cleanup()
381
482
 
382
 
def serve_bzr(transport, host=None, port=None, inet=False):
 
483
 
 
484
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
383
485
    """This is the default implementation of 'bzr serve'.
384
 
    
 
486
 
385
487
    It creates a TCP or pipe smart server on 'transport, and runs it.  The
386
488
    transport will be decorated with a chroot and pathfilter (using
387
489
    os.path.expanduser).
388
490
    """
389
491
    bzr_server = BzrServerFactory()
390
492
    try:
391
 
        bzr_server.set_up(transport, host, port, inet)
 
493
        bzr_server.set_up(transport, host, port, inet, timeout)
392
494
        bzr_server.smart_server.serve()
393
495
    except:
394
496
        hook_caught_exception = False