1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 import locale, random
22 from hashlib import sha1
23 from transports_nb import NonBlockingTransport, NonBlockingHTTPBOSH,\
24 CONNECTED, CONNECTING, DISCONNECTED, DISCONNECTING,\
25 urisplit, DISCONNECT_TIMEOUT_SECONDS
26 from protocol import BOSHBody, Protocol, NS_CLIENT
27 from simplexml import Node
28
29 import logging
30 log = logging.getLogger('nbxmpp.bosh')
31
32 KEY_COUNT = 10
33
34
35
36 FAKE_DESCRIPTOR = -1337
37
38
40 - def __init__(self, raise_event, on_disconnect, idlequeue, estabilish_tls, certs,
41 xmpp_server, domain, bosh_dict, proxy_creds):
42 NonBlockingTransport.__init__(self, raise_event, on_disconnect, idlequeue,
43 estabilish_tls, certs)
44
45 self.bosh_sid = None
46 if locale.getdefaultlocale()[0]:
47 self.bosh_xml_lang = locale.getdefaultlocale()[0].split('_')[0]
48 else:
49 self.bosh_xml_lang = 'en'
50
51 self.http_version = 'HTTP/1.1'
52 self.http_persistent = True
53 self.http_pipelining = bosh_dict['bosh_http_pipelining']
54 self.bosh_to = domain
55
56 self.route_host, self.route_port = xmpp_server
57
58 self.bosh_wait = bosh_dict['bosh_wait']
59 if not self.http_pipelining:
60 self.bosh_hold = 1
61 else:
62 self.bosh_hold = bosh_dict['bosh_hold']
63 self.bosh_requests = self.bosh_hold
64 self.bosh_uri = bosh_dict['bosh_uri']
65 self.bosh_content = bosh_dict['bosh_content']
66 self.over_proxy = bosh_dict['bosh_useproxy']
67 if estabilish_tls:
68 self.bosh_secure = 'true'
69 else:
70 self.bosh_secure = 'false'
71 self.use_proxy_auth = bosh_dict['useauth']
72 self.proxy_creds = proxy_creds
73 self.wait_cb_time = None
74 self.http_socks = []
75 self.stanza_buffer = []
76 self.prio_bosh_stanzas = []
77 self.current_recv_handler = None
78 self.current_recv_socket = None
79 self.key_stack = None
80 self.ack_checker = None
81 self.after_init = False
82 self.proxy_dict = {}
83 if self.over_proxy and self.estabilish_tls:
84 self.proxy_dict['type'] = 'http'
85
86
87 host, port = urisplit(self.bosh_uri)[1:3]
88 self.proxy_dict['xmpp_server'] = (host, port)
89 self.proxy_dict['credentials'] = self.proxy_creds
90
91
92 - def connect(self, conn_5tuple, on_connect, on_connect_failure):
113
118
119
120
126
128 """
129 Called when HTTP request it's possible to send a HTTP request. It can be when
130 socket is connected or when HTTP response arrived
131
132 There should be always one pending request to BOSH CM.
133 """
134 log.debug('on_http_req possible, state:\n%s' % self.get_current_state())
135 if self.get_state()==DISCONNECTED: return
136
137
138 if self._owner.got_features:
139 if (hasattr(self._owner, 'NonBlockingNonSASL') or hasattr(self._owner, 'SASL')):
140 self.send_BOSH(None)
141 else:
142
143
144
145
146 return
147 else:
148 self.send_BOSH(None)
149
150
151
153 """
154 Get sockets in desired state
155 """
156 for s in self.http_socks:
157 if s.get_state()==state: return s
158 return None
159
160
162 """
163 Select and returns socket eligible for sending a data to
164 """
165 if self.http_pipelining:
166 return self.get_socket_in(CONNECTED)
167 else:
168 last_recv_time, tmpsock = 0, None
169 for s in self.http_socks:
170
171 if s.get_state()==CONNECTED and s.pending_requests==0:
172
173
174 if (last_recv_time==0) or (s.last_recv_time < last_recv_time):
175 last_recv_time = s.last_recv_time
176 tmpsock = s
177 if tmpsock:
178 return tmpsock
179 else:
180 return None
181
182
239
241 stanza = None
242 s = self.get_free_socket()
243 if s:
244 s._plug_idle(writable=True, readable=True)
245 else:
246 log.error('=====!!!!!!!!====> Couldn\'t get free socket in plug_socket())')
247
249 """
250 Build a BOSH body tag from data in buffers and adds key, rid and ack
251 attributes to it
252
253 This method is called from _do_send() of underlying transport. This is to
254 ensure rid and keys will be processed in correct order. If I generate
255 them before plugging a socket for write (and did it for two sockets/HTTP
256 connections) in parallel, they might be sent in wrong order, which
257 results in violating the BOSH session and server-side disconnect.
258 """
259 if self.prio_bosh_stanzas:
260 stanza, add_payload = self.prio_bosh_stanzas.pop(0)
261 if add_payload:
262 stanza.setPayload(self.stanza_buffer)
263 self.stanza_buffer = []
264 else:
265 stanza = self.boshify_stanzas(self.stanza_buffer)
266 self.stanza_buffer = []
267
268 stanza = self.ack_checker.backup_stanza(stanza, socket)
269
270 key, newkey = self.key_stack.get()
271 if key:
272 stanza.setAttr('key', key)
273 if newkey:
274 stanza.setAttr('newkey', newkey)
275
276
277 log.info('sending msg with rid=%s to sock %s' % (stanza.getAttr('rid'), id(socket)))
278 self.renew_bosh_wait_timeout(self.bosh_wait + 3)
279 return stanza
280
281
283 log.error('Connection Manager didn\'t respond within %s + 3 seconds --> forcing disconnect' % self.bosh_wait)
284 self.disconnect()
285
286
292
297
299 """
300 Called from underlying transport when server closes TCP connection
301
302 :param socket: disconnected transport object
303 """
304 if socket.http_persistent:
305 log.warn('Fallback to nonpersistent HTTP (no pipelining as well)')
306 socket.http_persistent = False
307 self.http_persistent = False
308 self.http_pipelining = False
309 socket.disconnect(do_callback=False)
310 self.connect_and_flush(socket)
311 else:
312 socket.disconnect()
313
314
315
316 - def handle_body_attrs(self, stanza_attrs):
317 """
318 Called for each incoming body stanza from dispatcher. Checks body
319 attributes.
320 """
321 self.remove_bosh_wait_timeout()
322
323 if self.after_init:
324 if stanza_attrs.has_key('sid'):
325
326 self.bosh_sid = stanza_attrs['sid']
327
328 if stanza_attrs.has_key('requests'):
329 self.bosh_requests = int(stanza_attrs['requests'])
330
331 if stanza_attrs.has_key('wait'):
332 self.bosh_wait = int(stanza_attrs['wait'])
333 self.after_init = False
334
335 ack = None
336 if stanza_attrs.has_key('ack'):
337 ack = stanza_attrs['ack']
338 self.ack_checker.process_incoming_ack(ack=ack,
339 socket=self.current_recv_socket)
340
341 if stanza_attrs.has_key('type'):
342 if stanza_attrs['type'] in ['terminate', 'terminal']:
343 condition = 'n/a'
344 if stanza_attrs.has_key('condition'):
345 condition = stanza_attrs['condition']
346 if condition == 'n/a':
347 log.info('Received sesion-ending terminating stanza')
348 else:
349 log.error('Received terminating stanza: %s - %s' % (condition,
350 bosh_errors[condition]))
351 self.disconnect()
352 return
353
354 if stanza_attrs['type'] == 'error':
355
356 pass
357 return
358
359
361 """
362 Append stanza to a buffer to send
363 """
364 if stanza:
365 if isinstance(stanza, tuple):
366
367 self.prio_bosh_stanzas.append(stanza)
368 else:
369
370 self.stanza_buffer.append(stanza)
371
372
373 - def send(self, stanza, now=False):
375
376
377
379 t = '------ SOCKET_ID\tSOCKET_STATE\tPENDING_REQS\n'
380 for s in self.http_socks:
381 t = '%s------ %s\t%s\t%s\n' % (t, id(s), s.get_state(), s.pending_requests)
382 t = '%s------ prio stanzas: %s, queued XMPP stanzas: %s, not_acked stanzas: %s' \
383 % (t, self.prio_bosh_stanzas, self.stanza_buffer,
384 self.ack_checker.get_not_acked_rids())
385 return t
386
387
393
394
396 """
397 Wraps zero to many stanzas by body tag with xmlns and sid
398 """
399 log.debug('boshify_staza - type is: %s, stanza is %s' % (type(stanzas), stanzas))
400 tag = BOSHBody(attrs={'sid': self.bosh_sid})
401 tag.setPayload(stanzas)
402 return tag
403
404
406 if after_SASL:
407 t = BOSHBody(
408 attrs={ 'to': self.bosh_to,
409 'sid': self.bosh_sid,
410 'xml:lang': self.bosh_xml_lang,
411 'xmpp:restart': 'true',
412 'secure': self.bosh_secure,
413 'xmlns:xmpp': 'urn:xmpp:xbosh'})
414 else:
415 t = BOSHBody(
416 attrs={ 'content': self.bosh_content,
417 'hold': str(self.bosh_hold),
418 'route': 'xmpp:%s:%s' % (self.route_host, self.route_port),
419 'to': self.bosh_to,
420 'wait': str(self.bosh_wait),
421 'xml:lang': self.bosh_xml_lang,
422 'xmpp:version': '1.0',
423 'ver': '1.6',
424 'xmlns:xmpp': 'urn:xmpp:xbosh'})
425 self.send_BOSH((t, True))
426
432
433
435 http_dict = {'http_uri': self.bosh_uri,
436 'http_version': self.http_version,
437 'http_persistent': self.http_persistent,
438 'add_proxy_headers': self.over_proxy and not self.estabilish_tls}
439 if self.use_proxy_auth:
440 http_dict['proxy_user'], http_dict['proxy_pass'] = self.proxy_creds
441
442 s = NonBlockingHTTPBOSH(
443 raise_event=self.raise_event,
444 on_disconnect=self.disconnect,
445 idlequeue = self.idlequeue,
446 estabilish_tls = self.estabilish_tls,
447 certs = self.certs,
448 on_http_request_possible = self.on_http_request_possible,
449 http_dict = http_dict,
450 proxy_dict = self.proxy_dict,
451 on_persistent_fallback = self.on_persistent_fallback)
452
453 s.onreceive(self.on_received_http)
454 s.set_stanza_build_cb(self.build_stanza)
455 return s
456
457
462
463
465 self.current_recv_socket = socket
466 self.current_recv_handler(data)
467
468
476
477
479
480
481
482
483 r = random.Random()
484 r.seed()
485 return r.getrandbits(50)
486
487
488
490 """
491 Class for generating rids and generating and checking acknowledgements in
492 BOSH messages
493 """
495 self.rid = get_rand_number()
496 self.ack = 1
497 self.last_rids = {}
498 self.not_acked = []
499
500
502
504 socket.pending_requests += 1
505 rid = self.get_rid()
506 self.not_acked.append((rid, stanza))
507 stanza.setAttr('rid', str(rid))
508 self.last_rids[socket]=rid
509
510 if self.rid != self.ack + 1:
511 stanza.setAttr('ack', str(self.ack))
512 return stanza
513
515 socket.pending_requests -= 1
516 if ack:
517 ack = int(ack)
518 else:
519 ack = self.last_rids[socket]
520
521 i = len([rid for rid, st in self.not_acked if ack >= rid])
522 self.not_acked = self.not_acked[i:]
523
524 self.ack = ack
525
526
528 self.rid = self.rid + 1
529 return self.rid
530
531
532
533
534
536 """
537 Class implementing key sequences for BOSH messages
538 """
540 self.count = count
541 self.keys = []
542 self.reset()
543 self.first_call = True
544
546 seed = str(get_rand_number())
547 self.keys = [sha1(seed).hexdigest()]
548 for i in range(self.count-1):
549 curr_seed = self.keys[i]
550 self.keys.append(sha1(curr_seed).hexdigest())
551
553 if self.first_call:
554 self.first_call = False
555 return (None, self.keys.pop())
556
557 if len(self.keys)>1:
558 return (self.keys.pop(), None)
559 else:
560 last_key = self.keys.pop()
561 self.reset()
562 new_key = self.keys.pop()
563 return (last_key, new_key)
564
565
566 bosh_errors = {
567 'n/a': 'none or unknown condition in terminating body stanza',
568 'bad-request': 'The format of an HTTP header or binding element received from the client is unacceptable (e.g., syntax error), or Script Syntax is not supported.',
569 'host-gone': 'The target domain specified in the "to" attribute or the target host or port specified in the "route" attribute is no longer serviced by the connection manager.',
570 'host-unknown': 'The target domain specified in the "to" attribute or the target host or port specified in the "route" attribute is unknown to the connection manager.',
571 'improper-addressing': 'The initialization element lacks a "to" or "route" attribute (or the attribute has no value) but the connection manager requires one.',
572 'internal-server-error': 'The connection manager has experienced an internal error that prevents it from servicing the request.',
573 'item-not-found': '(1) "sid" is not valid, (2) "stream" is not valid, (3) "rid" is larger than the upper limit of the expected window, (4) connection manager is unable to resend response, (5) "key" sequence is invalid',
574 'other-request': 'Another request being processed at the same time as this request caused the session to terminate.',
575 'policy-violation': 'The client has broken the session rules (polling too frequently, requesting too frequently, too many simultaneous requests).',
576 'remote-connection-failed': 'The connection manager was unable to connect to, or unable to connect securely to, or has lost its connection to, the server.',
577 'remote-stream-error': 'Encapsulates an error in the protocol being transported.',
578 'see-other-uri': 'The connection manager does not operate at this URI (e.g., the connection manager accepts only SSL or TLS connections at some https: URI rather than the http: URI requested by the client). The client may try POSTing to the URI in the content of the <uri/> child element.',
579 'system-shutdown': 'The connection manager is being shut down. All active HTTP sessions are being terminated. No new sessions can be created.',
580 'undefined-condition': 'The error is not one of those defined herein; the connection manager SHOULD include application-specific information in the content of the <body/> wrapper.'
581 }
582