client.py 28.7 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# Copyright (C) 2020 Philipp Hörist <philipp AT hoerist.com>
#
# This file is part of nbxmpp.
#
# 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/>.

import logging

from gi.repository import GLib

Philipp Hörist's avatar
Philipp Hörist committed
22
from nbxmpp.namespaces import Namespace
23
24
25
26
27
28
29
30
31
from nbxmpp.protocol import Features
from nbxmpp.protocol import StanzaMalformed
from nbxmpp.protocol import SessionRequest
from nbxmpp.protocol import BindRequest
from nbxmpp.protocol import TLSRequest
from nbxmpp.protocol import isResultNode
from nbxmpp.protocol import JID
from nbxmpp.protocol import Protocol
from nbxmpp.protocol import WebsocketCloseHeader
Philipp Hörist's avatar
Philipp Hörist committed
32
33
34
from nbxmpp.errors import TimeoutStanzaError
from nbxmpp.errors import StanzaError
from nbxmpp.errors import CancelledError
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from nbxmpp.addresses import ServerAddresses
from nbxmpp.addresses import NoMoreAddresses
from nbxmpp.tcp import TCPConnection
from nbxmpp.websocket import WebsocketConnection
from nbxmpp.smacks import Smacks
from nbxmpp.auth import SASL
from nbxmpp.const import StreamState
from nbxmpp.const import StreamError
from nbxmpp.const import ConnectionType
from nbxmpp.const import ConnectionProtocol
from nbxmpp.const import Mode
from nbxmpp.dispatcher import StanzaDispatcher
from nbxmpp.util import get_stream_header
from nbxmpp.util import get_stanza_id
from nbxmpp.util import Observable
from nbxmpp.util import validate_stream_header
51
from nbxmpp.util import LogAdapter
52
53
54
55
56

log = logging.getLogger('nbxmpp.stream')


class Client(Observable):
57
    def __init__(self, log_context=None):
58
59
60
61
62
        '''
        Signals:
            resume-failed
            resume-successful
            login-successful
63
            anonymous-supported
64
65
66
67
68
69
            disconnected
            connected
            connection-failed
            stanza-sent
            stanza-received
        '''
70
71
72
73
74
75
76
77
78

        self._log_context = log_context
        if log_context is None:
            self._log_context = str(id(self))

        self._log = LogAdapter(log, {'context': self._log_context})

        Observable.__init__(self, self._log)

79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
        self._jid = None
        self._lang = 'en'
        self._domain = None
        self._username = None
        self._resource = None

        self._custom_host = None

        self._addresses = None
        self._current_address = None
        self._address_generator = None

        self._client_cert = None
        self._client_cert_pass = None
        self._proxy = None

        self._allowed_con_types = None
        self._allowed_protocols = None
        self._allowed_mechs = None

99
100
        self._sm_disabled = False

101
102
103
104
105
106
107
        self._stream_id = None
        self._stream_secure = False
        self._stream_authenticated = False
        self._stream_features = None
        self._session_required = False
        self._connect_successful = False
        self._stream_close_initiated = False
Philipp Hörist's avatar
Philipp Hörist committed
108
        self._ping_task = None
109
110
        self._error = None, None, None

111
        self._ignored_tls_errors = set()
112
113
114
115
116
117
        self._ignore_tls_errors = False
        self._accepted_certificates = []
        self._peer_certificate = None
        self._peer_certificate_errors = None

        self._con = None
Philipp Hörist's avatar
Philipp Hörist committed
118
        self._local_address = None
119
        self._remote_address = None
120
121
122
        self._mode = Mode.CLIENT

        self._ping_source_id = None
Philipp Hörist's avatar
Philipp Hörist committed
123
        self._tasks = []
124

125
        self._dispatcher = StanzaDispatcher(self)
126
127
128
129
        self._dispatcher.subscribe('before-dispatch', self._on_before_dispatch)
        self._dispatcher.subscribe('parsing-error', self._on_parsing_error)
        self._dispatcher.subscribe('stream-end', self._on_stream_end)

130
131
        self._smacks = Smacks(self)
        self._sasl = SASL(self)
132
133
134

        self._state = StreamState.DISCONNECTED

Philipp Hörist's avatar
Philipp Hörist committed
135
136
137
138
139
140
141
142
143
    def add_task(self, task):
        self._tasks.append(task)

    def remove_task(self, task, _context):
        try:
            self._tasks.remove(task)
        except Exception:
            pass

144
145
146
147
    @property
    def log_context(self):
        return self._log_context

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
    @property
    def features(self):
        return self._stream_features

    @property
    def sm_supported(self):
        return self._smacks.sm_supported

    @property
    def lang(self):
        return self._lang

    @property
    def username(self):
        return self._username

    @property
    def domain(self):
        return self._domain

    @property
    def resource(self):
        return self._resource

    def set_username(self, username):
        self._username = username

    def set_domain(self, domain):
        self._domain = domain

    def set_resource(self, resource):
        self._resource = resource

    def set_mode(self, mode):
        self._mode = mode

184
185
186
187
    @property
    def custom_host(self):
        return self._custom_host

188
    def set_custom_host(self, host_or_uri, protocol, type_):
189
190
        if self._domain is None:
            raise ValueError('Call set_domain() first before set_custom_host()')
191
        self._custom_host = (host_or_uri, protocol, type_)
192
193
194
195
196
197
198
199
200

    def set_accepted_certificates(self, certificates):
        self._accepted_certificates = certificates

    @property
    def ignored_tls_errors(self):
        return self._ignored_tls_errors

    def set_ignored_tls_errors(self, errors):
201
202
        if errors is None:
            errors = set()
203
204
205
206
207
208
209
210
211
212
213
214
        self._ignored_tls_errors = errors

    @property
    def ignore_tls_errors(self):
        return self._ignore_tls_errors

    def set_ignore_tls_errors(self, ignore):
        self._ignore_tls_errors = ignore

    def set_password(self, password):
        self._sasl.set_password(password)

215
216
217
218
    @property
    def password(self):
        return self._sasl.password

219
220
221
222
    @property
    def peer_certificate(self):
        return self._peer_certificate, self._peer_certificate_errors

223
224
225
226
    @property
    def current_address(self):
        return self._current_address

227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
    @property
    def current_connection_type(self):
        return self._current_address.type

    @property
    def is_websocket(self):
        return self._current_address.protocol == ConnectionProtocol.WEBSOCKET

    @property
    def stream_id(self):
        return self._stream_id

    @property
    def is_stream_secure(self):
        direct_tls = self.current_connection_type == ConnectionType.DIRECT_TLS
        return self._stream_secure or direct_tls

    @property
    def is_stream_authenticated(self):
        return self._stream_authenticated

    @property
    def state(self):
        return self._state

    @state.setter
    def state(self, value):
        self._state = value
255
        self._log.info('Set state: %s', value)
256
257
258
259
260

    def set_state(self, state):
        self.state = state
        self._xmpp_state_machine()

Philipp Hörist's avatar
Philipp Hörist committed
261
262
263
264
    @property
    def local_address(self):
        return self._local_address

265
266
267
268
    @property
    def remote_address(self):
        return self._remote_address

269
270
    @property
    def connection_types(self):
271
272
        if self._custom_host is not None:
            return [self._custom_host[2]]
273
274
275
276
277
278
279
280
281
282
        return list(self._allowed_con_types or [ConnectionType.DIRECT_TLS,
                                                ConnectionType.START_TLS])

    def set_connection_types(self, con_types):
        self._allowed_con_types = con_types

    @property
    def mechs(self):
        return set(self._allowed_mechs or set(['SCRAM-SHA-256',
                                               'SCRAM-SHA-1',
283
                                               'GSSAPI',
284
285
286
287
288
289
290
291
292
                                               # 'SCRAM-SHA-256-PLUS',
                                               # 'SCRAM-SHA-1-PLUS',
                                               'PLAIN']))

    def set_mechs(self, mechs):
        self._allowed_mechs = mechs

    @property
    def protocols(self):
293
294
        if self._custom_host is not None:
            return [self._custom_host[1]]
295
296
297
298
299
300
        return list(self._allowed_protocols or [ConnectionProtocol.TCP,
                                                ConnectionProtocol.WEBSOCKET])

    def set_protocols(self, protocols):
        self._allowed_protocols = protocols

301
302
303
304
305
306
307
    def set_sm_disabled(self, value):
        self._sm_disabled = value

    @property
    def sm_disabled(self):
        return self._sm_disabled

308
309
310
311
312
313
314
315
316
317
318
319
    @property
    def client_cert(self):
        return self._client_cert, self._client_cert_pass

    def set_client_cert(self, client_cert, client_cert_pass):
        self._client_cert = client_cert
        self._client_cert_pass = client_cert_pass

    def set_proxy(self, proxy):
        self._proxy = proxy
        self._dispatcher.get_module('Muclumbus').set_proxy(proxy)

320
321
322
323
    @property
    def proxy(self):
        return self._proxy

324
325
326
327
    def get_bound_jid(self):
        return self._jid

    def _set_bound_jid(self, jid):
328
        self._jid = JID.from_string(jid)
329
330
331
332
333
334
335
336
337
338
339
340

    @property
    def has_error(self):
        return self._error[0] is not None

    def get_error(self):
        return self._error

    def _reset_error(self):
        self._error = None, None, None

    def _set_error(self, domain, error, text=None):
341
        self._log.info('Set error: %s, %s, %s', domain, error, text)
342
343
344
345
        self._error = domain, error, text

    def _connect(self):
        if self._state not in (StreamState.DISCONNECTED, StreamState.RESOLVED):
346
347
            self._log.error('Stream can\'t connect, stream state: %s',
                            self._state)
348
349
350
351
352
            return

        self.state = StreamState.CONNECTING
        self._reset_error()

353
354
        self._con = self._get_connection(self._log_context,
                                         self._current_address,
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
                                         self._accepted_certificates,
                                         self._ignore_tls_errors,
                                         self._ignored_tls_errors,
                                         self.client_cert)

        self._con.subscribe('connected', self._on_connected)
        self._con.subscribe('connection-failed', self._on_connection_failed)
        self._con.subscribe('disconnected', self._on_disconnected)
        self._con.subscribe('data-sent', self._on_data_sent)
        self._con.subscribe('data-received', self._on_data_received)
        self._con.subscribe('bad-certificate', self._on_bad_certificate)
        self._con.subscribe('certificate-set', self._on_certificate_set)
        self._con.connect()

    def _get_connection(self, *args):
        if self.is_websocket:
            return WebsocketConnection(*args)
        return TCPConnection(*args)

    def connect(self):
        if self._state != StreamState.DISCONNECTED:
376
377
            self._log.error('Stream can\'t reconnect, stream state: %s',
                            self._state)
Philipp Hörist's avatar
Philipp Hörist committed
378
            return
379
380

        if self._connect_successful:
381
            self._log.info('Reconnect')
382
383
384
            self._connect()
            return

385
        self._log.info('Connect')
386
387
388
389
390
391
392
393
394
395
        self._reset_error()
        self.state = StreamState.RESOLVE

        self._addresses = ServerAddresses(self._domain)
        self._addresses.set_custom_host(self._custom_host)
        self._addresses.set_proxy(self._proxy)
        self._addresses.subscribe('resolved', self._on_addresses_resolved)
        self._addresses.resolve()

    def _on_addresses_resolved(self, _addresses, _signal_name):
396
397
        self._log.info('Domain resolved')
        self._log.info(self._addresses)
398
399
400
401
402
403
404
405
406
407
408
409
410
        self.state = StreamState.RESOLVED
        self._address_generator = self._addresses.get_next_address(
            self.connection_types,
            self.protocols)

        self._try_next_ip()

    def _try_next_ip(self, *args):
        try:
            self._current_address = next(self._address_generator)
        except NoMoreAddresses:
            self._current_address = None
            self.state = StreamState.DISCONNECTED
411
            self._log.error('Unable to connect to %s', self._addresses.domain)
412
413
414
415
416
417
            self._set_error(StreamError.CONNECTION_FAILED,
                            'connection-failed',
                            'Unable to connect to %s' % self._addresses.domain)
            self.notify('connection-failed')
            return

418
        self._log.info('Current address: %s', self._current_address)
419
420
421
422
423
424
425
426
427
428
429
430
431
432
        self._connect()

    def disconnect(self, immediate=False):
        if self._state == StreamState.RESOLVE:
            self._addresses.cancel_resolve()
            self.state = StreamState.DISCONNECTED
            return

        if self._state == StreamState.CONNECTING:
            self._disconnect()
            return

        if self._state in (StreamState.DISCONNECTED,
                           StreamState.DISCONNECTING):
433
434
            self._log.warning('Stream can\'t disconnect, stream state: %s',
                              self._state)
435
436
437
438
439
440
441
            return

        self._disconnect(immediate=immediate)

    def _disconnect(self, immediate=True):
        self.state = StreamState.DISCONNECTING
        self._remove_ping_timer()
Philipp Hörist's avatar
Philipp Hörist committed
442
443
        self._cancel_ping_task()

444
445
446
447
448
449
450
451
452
453
454
455
        if not immediate:
            self._stream_close_initiated = True
            self._smacks.close_session()
            self._end_stream()
            self._con.shutdown_output()
        else:
            self._con.disconnect()

    def send(self, stanza, *args, **kwargs):
        # Alias for backwards compat
        return self.send_stanza(stanza)

Philipp Hörist's avatar
Philipp Hörist committed
456
    def _on_connected(self, connection, _signal_name):
457
        self.set_state(StreamState.CONNECTED)
Philipp Hörist's avatar
Philipp Hörist committed
458
        self._local_address = connection.local_address
459
        self._remote_address = connection.remote_address
460
461
462

    def _on_disconnected(self, _connection, _signal_name):
        self.state = StreamState.DISCONNECTED
Philipp Hörist's avatar
Philipp Hörist committed
463
464
        for task in self._tasks:
            task.cancel()
465
        self._remove_ping_timer()
Philipp Hörist's avatar
Philipp Hörist committed
466
        self._cancel_ping_task()
467
468
469
470
471
472
473
474
475
476
477
        self._reset_stream()
        self.notify('disconnected')

    def _on_connection_failed(self, _connection, _signal_name):
        self.state = StreamState.DISCONNECTED
        self._reset_stream()
        if not self._connect_successful:
            self._try_next_ip()
        else:
            self._set_error(StreamError.CONNECTION_FAILED,
                            'connection-failed',
Philipp Hörist's avatar
Philipp Hörist committed
478
479
                            (f'Unable to connect to last '
                             'successful address: {self._current_address}'))
480
481
482
483
484
485
486
            self.notify('connection-failed')

    def _disconnect_with_error(self, error_domain, error, text=None):
        self._set_error(error_domain, error, text)
        self.disconnect()

    def _on_parsing_error(self, _dispatcher, _signal_name, error):
487
488
489
        if self._state == StreamState.DISCONNECTING:
            # Don't notify about parsing errors if we already ended the stream
            return
490
491
492
493
        self._disconnect_with_error(StreamError.PARSING, 'parsing-error', error)

    def _on_stream_end(self, _dispatcher, _signal_name, error):
        if not self.has_error:
494
            self._set_error(StreamError.STREAM, error or 'stream-end')
495
496
497
498

        self._con.shutdown_input()
        if not self._stream_close_initiated:
            self.state = StreamState.DISCONNECTING
499
            self._remove_ping_timer()
Philipp Hörist's avatar
Philipp Hörist committed
500
            self._cancel_ping_task()
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
            self._smacks.close_session()
            self._end_stream()
            self._con.shutdown_output()

    def _reset_stream(self):
        self._stream_id = None
        self._stream_secure = False
        self._stream_authenticated = False
        self._stream_features = None
        self._session_required = False
        self._con = None

    def _end_stream(self):
        if self.is_websocket:
            nonza = WebsocketCloseHeader()
        else:
            nonza = '</stream:stream>'
        self.send_nonza(nonza)

    def get_module(self, name):
        return self._dispatcher.get_module(name)

    def _on_bad_certificate(self, connection, _signal_name):
Philipp Hörist's avatar
Philipp Hörist committed
524
525
        self._peer_certificate, self._peer_certificate_errors = \
            connection.peer_certificate
526
527
528
        self._set_error(StreamError.BAD_CERTIFICATE, 'bad certificate')

    def _on_certificate_set(self, connection, _signal_name):
Philipp Hörist's avatar
Philipp Hörist committed
529
530
        self._peer_certificate, self._peer_certificate_errors = \
            connection.peer_certificate
531
532

    def accept_certificate(self):
533
        self._log.info('Certificate accepted')
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
        self._accepted_certificates.append(self._peer_certificate)
        self._connect()

    def _on_data_sent(self, _connection, _signal_name, data):
        self.notify('stanza-sent', data)

    def _on_before_dispatch(self, _dispatcher, _signal_name, data):
        self.notify('stanza-received', data)

    def _on_data_received(self, _connection, _signal_name, data):
        self._dispatcher.process_data(data)
        self._reset_ping_timer()

    def _reset_ping_timer(self):
        if self.is_websocket:
            return

        if not self._mode.is_client:
            return

        if self.state != StreamState.ACTIVE:
            return

        if self._ping_source_id is not None:
558
            self._log.info('Remove ping timer')
559
560
561
            GLib.source_remove(self._ping_source_id)
            self._ping_source_id = None

562
        self._log.info('Start ping timer')
Philipp Hörist's avatar
Philipp Hörist committed
563
        self._ping_source_id = GLib.timeout_add_seconds(180, self._ping)
564
565
566
567

    def _remove_ping_timer(self):
        if self._ping_source_id is None:
            return
568
        self._log.info('Remove ping timer')
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
        GLib.source_remove(self._ping_source_id)
        self._ping_source_id = None

    def send_stanza(self, stanza, now=False, callback=None,
                    timeout=None, user_data=None):
        if user_data is not None and not isinstance(user_data, dict):
            raise ValueError('arg user_data must be of dict type')

        if not isinstance(stanza, Protocol):
            raise ValueError('Nonzas not allowed, use send_nonza()')

        id_ = stanza.getID()
        if id_ is None:
            id_ = get_stanza_id()
            stanza.setID(id_)

        if callback is not None:
            self._dispatcher.add_callback_for_id(
                id_, callback, timeout, user_data)
        self._con.send(stanza, now)
        self._smacks.save_in_queue(stanza)
        return id_

    def SendAndCallForResponse(self, stanza, callback, user_data=None):
        self.send_stanza(stanza, callback=callback, user_data=user_data)

    def send_nonza(self, nonza, now=False):
        self._con.send(nonza, now)

    def _xmpp_state_machine(self, stanza=None):
599
        self._log.info('Execute state machine')
600
601
        if stanza is not None:
            if stanza.getName() == 'error':
602
                self._log.info('Stream error')
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
                # TODO:
                # self._disconnect_with_error(StreamError.SASL,
                #                             stanza.get_condition())
                return

        if self.state == StreamState.CONNECTED:
            self._dispatcher.set_dispatch_callback(self._xmpp_state_machine)
            if (self.current_connection_type == ConnectionType.DIRECT_TLS and
                    not self.is_websocket):
                self._con.start_tls_negotiation()
                self._stream_secure = True
                self._start_stream()
                return

            self._start_stream()

        elif self.state == StreamState.WAIT_FOR_STREAM_START:
            try:
                self._stream_id = validate_stream_header(stanza,
                                                         self._domain,
                                                         self.is_websocket)
            except StanzaMalformed as error:
625
                self._log.error(error)
626
627
628
629
630
                self._disconnect_with_error(StreamError.STREAM,
                                            'stanza-malformed',
                                            'Invalid stream header')
                return

631
632
633
634
635
636
637
638
            if (self._stream_secure or
                    self.current_connection_type == ConnectionType.PLAIN):
                # TLS Negotiation succeeded or we are connected PLAIN
                # We received the stream header and consider this as
                # successfully connected, this means we will not try
                # other connection methods if an error happensafterwards
                self._connect_successful = True

639
640
641
642
            self.state = StreamState.WAIT_FOR_FEATURES

        elif self.state == StreamState.WAIT_FOR_FEATURES:
            if stanza.getName() != 'features':
643
                self._log.error('Invalid response: %s', stanza)
644
645
646
647
648
649
650
651
                self._disconnect_with_error(
                    StreamError.STREAM,
                    'stanza-malformed',
                    'Invalid response, expected features')
                return
            self._on_stream_features(Features(stanza))

        elif self.state == StreamState.WAIT_FOR_TLS_PROCEED:
Philipp Hörist's avatar
Philipp Hörist committed
652
            if stanza.getNamespace() != Namespace.TLS:
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
                self._disconnect_with_error(
                    StreamError.TLS,
                    'stanza-malformed',
                    'Invalid namespace for TLS response')
                return

            if stanza.getName() == 'failure':
                self._disconnect_with_error(StreamError.TLS,
                                            'negotiation-failed')
                return

            if stanza.getName() == 'proceed':
                self._con.start_tls_negotiation()
                self._stream_secure = True
                self._start_stream()
                return

670
            self._log.error('Invalid response')
671
672
673
674
675
676
677
678
679
680
681
682
            self._disconnect_with_error(StreamError.TLS,
                                        'stanza-malformed',
                                        'Invalid TLS response')
            return

        elif self.state == StreamState.PROCEED_WITH_AUTH:
            self._sasl.delegate(stanza)

        elif self.state == StreamState.AUTH_SUCCESSFUL:
            self._stream_authenticated = True
            if self._mode.is_login_test:
                self.notify('login-successful')
683
684
685
                # Reset parser because we will receive a new stream header
                # which will otherwise lead to a parsing error
                self._dispatcher.reset_parser()
686
687
688
689
690
691
692
693
694
695
696
697
698
                self.disconnect()
                return

            self._start_stream()

        elif self.state == StreamState.AUTH_FAILED:
            self._disconnect_with_error(StreamError.SASL,
                                        *self._sasl.error)

        elif self.state == StreamState.WAIT_FOR_BIND:
            self._on_bind(stanza)

        elif self.state == StreamState.BIND_SUCCESSFUL:
699
            self._dispatcher.clear_iq_callbacks()
700
            self._dispatcher.set_dispatch_callback(None)
701
            self._smacks.send_enable()
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
            self.state = StreamState.ACTIVE
            self.notify('connected')

        elif self.state == StreamState.WAIT_FOR_SESSION:
            self._on_session(stanza)

        elif self.state == StreamState.WAIT_FOR_RESUMED:
            self._smacks.delegate(stanza)

        elif self.state == StreamState.RESUME_FAILED:
            self.notify('resume-failed')
            self._start_bind()

        elif self.state == StreamState.RESUME_SUCCESSFUL:
            self._dispatcher.set_dispatch_callback(None)
            self.state = StreamState.ACTIVE
            self.notify('resume-successful')

    def _on_stream_features(self, features):
        if self.is_stream_authenticated:
            self._stream_features = features
            self._smacks.sm_supported = features.has_sm()
            self._session_required = features.session_required()
            if self._smacks.resume_supported:
                self._smacks.resume_request()
                self.state = StreamState.WAIT_FOR_RESUMED
            else:
                self._start_bind()

        elif self.is_stream_secure:
            if self._mode.is_register:
Philipp Hörist's avatar
Philipp Hörist committed
733
                if features.has_register():
734
735
736
737
738
739
740
741
                    self.state = StreamState.ACTIVE
                    self._dispatcher.set_dispatch_callback(None)
                    self.notify('connected')
                else:
                    self._disconnect_with_error(StreamError.REGISTER,
                                                'register-not-supported')
                return

742
743
744
745
746
747
748
749
750
            if self._mode.is_anonymous_test:
                if features.has_anonymous():
                    self.notify('anonymous-supported')
                    self.disconnect()
                else:
                    self._disconnect_with_error(StreamError.SASL,
                                                'anonymous-not-supported')
                return

751
752
753
754
755
756
            self._start_auth(features)

        else:
            tls_supported, required = features.has_starttls()
            if self._current_address.type == ConnectionType.PLAIN:
                if tls_supported and required:
757
                    self._log.error('Server requires TLS')
758
759
760
761
762
763
                    self._disconnect_with_error(StreamError.TLS, 'tls-required')
                    return
                self._start_auth(features)
                return

            if not tls_supported:
764
                self._log.error('Server does not support TLS')
765
766
767
768
769
770
                self._disconnect_with_error(StreamError.TLS,
                                            'tls-not-supported')
                return
            self._start_tls()

    def _start_stream(self):
771
        self._log.info('Start stream')
772
773
774
775
776
777
778
779
780
781
782
783
        self._stream_id = None
        self._dispatcher.reset_parser()
        header = get_stream_header(self._domain, self._lang, self.is_websocket)
        self.send_nonza(header)
        self.state = StreamState.WAIT_FOR_STREAM_START

    def _start_tls(self):
        self.send_nonza(TLSRequest())
        self.state = StreamState.WAIT_FOR_TLS_PROCEED

    def _start_auth(self, features):
        if not features.has_sasl():
784
            self._log.error('Server does not support SASL')
785
786
787
788
789
790
791
            self._disconnect_with_error(StreamError.SASL,
                                        'sasl-not-supported')
            return
        self.state = StreamState.PROCEED_WITH_AUTH
        self._sasl.start_auth(features)

    def _start_bind(self):
792
        self._log.info('Send bind')
793
794
795
796
797
798
        bind_request = BindRequest(self.resource)
        self.send_stanza(bind_request)
        self.state = StreamState.WAIT_FOR_BIND

    def _on_bind(self, stanza):
        if not isResultNode(stanza):
799
800
801
            self._disconnect_with_error(StreamError.BIND,
                                        stanza.getError(),
                                        stanza.getErrorMsg())
802
803
804
            return

        jid = stanza.getTag('bind').getTagData('jid')
805
        self._log.info('Successfully bound %s', jid)
806
807
808
809
        self._set_bound_jid(jid)

        if not self._session_required:
            # Server don't want us to initialize a session
810
            self._log.info('No session required')
811
812
813
814
815
816
817
818
            self.set_state(StreamState.BIND_SUCCESSFUL)
        else:
            session_request = SessionRequest()
            self.send_stanza(session_request)
            self.state = StreamState.WAIT_FOR_SESSION

    def _on_session(self, stanza):
        if isResultNode(stanza):
819
            self._log.info('Successfully started session')
820
821
            self.set_state(StreamState.BIND_SUCCESSFUL)
        else:
822
            self._log.error('Session open failed')
823
824
825
            self._disconnect_with_error(StreamError.SESSION,
                                        stanza.getError(),
                                        stanza.getErrorMsg())
826
827

    def _ping(self):
828
        self._ping_source_id = None
Philipp Hörist's avatar
Philipp Hörist committed
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
        self._ping_task = self.get_module('Ping').ping(
            self.domain,
            timeout=10,
            callback=self._on_pong)

    def _on_pong(self, task):
        self._ping_task = None

        try:
            task.finish()
        except TimeoutStanzaError:
            self._log.info('Ping timeout')
            self._disconnect(immediate=True)
            return

        except CancelledError:
            return

        except StanzaError:
            pass

850
        self._log.info('Pong')
851

Philipp Hörist's avatar
Philipp Hörist committed
852
853
854
855
    def _cancel_ping_task(self):
        if self._ping_task is not None:
            self._ping_task.cancel()

856
857
858
859
860
861
862
    def register_handler(self, *args, **kwargs):
        self._dispatcher.register_handler(*args, **kwargs)

    def unregister_handler(self, *args, **kwargs):
        self._dispatcher.unregister_handler(*args, **kwargs)

    def destroy(self):
Philipp Hörist's avatar
Philipp Hörist committed
863
864
        for task in self._tasks:
            task.cancel()
865
866
867
868
869
870
        self._remove_ping_timer()
        self._smacks = None
        self._sasl = None
        self._dispatcher.cleanup()
        self._dispatcher = None
        self.remove_subscriptions()