xmppConnection.py (22320B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\xmpp\xmppConnection.py 2 from weakref import WeakSet 3 import blue 4 import logging 5 import socket 6 from eveprefs import prefs 7 from stackless_tracing.attributes import SERVICE_NAME, SERVICE_NAME_PREFIX 8 from xml.sax import SAXParseException 9 from xml.sax.saxutils import escape, quoteattr 10 from stackless_tracing.attributes import HOST_NAME 11 import uthread2 12 import ssl 13 import certifi 14 import monolithmetrics 15 import monolithhoneycomb_events as mhe 16 from monolithhoneycomb_events.events import CHAT_CONNECTION_CREATED, CHAT_CONNECTION_DELETED, CHAT_CONNECTION_CONNECTED, CHAT_CONNECTION_DISCONNECTED 17 from xmpp import XmppHandler 18 logger = logging.getLogger(__name__) 19 ca_certs_file = certifi.where() 20 21 def channel_timeout_helper(waitingChannel, timeInSecs, msg): 22 uthread2.sleep(timeInSecs) 23 if waitingChannel.balance < 0: 24 logger.warning(msg) 25 while waitingChannel.balance < 0: 26 waitingChannel.send(None) 27 28 29 def channel_timeout(channel, timeoutInSecs, msg): 30 uthread2.start_tasklet(channel_timeout_helper, channel, timeoutInSecs, msg) 31 32 33 CONNECTION_OBJECT_CREATED = 'xmpp.connection_object.created' 34 CONNECTION_OBJECT_DESTROYED = 'xmpp.connection_object.destroyed' 35 CONNECTION_ATTEMPTED = 'xmpp.connection.attempted' 36 CONNECTION_LOGIN_FAILED = 'xmpp.connection.login_failed' 37 CONNECTION_FAILED = 'xmpp.connection.failed' 38 CONNECTION_SUCCEEDED = 'xmpp.connection.succeeded' 39 CONNECTION_DISCONNECTED = 'xmpp.connection.disconnected' 40 CONNECTION_RECONNECTED = 'xmpp.connection.reconnected' 41 42 class XmppConnection(object): 43 44 def __init__(self, hostname = '', ejabberd_address = 'localhost'): 45 self.hostname = hostname 46 self.ejabberd_address = ejabberd_address 47 monolithmetrics.increment(CONNECTION_OBJECT_CREATED, value=1, tags=self._get_tags()) 48 self.port = 5222 49 self.username = None 50 self.xmppHandler = None 51 self.socket = None 52 self.connection_id = 'default' 53 self.verbose = False 54 self.disconnectRequested = False 55 self.hasConnectedOnce = False 56 self.myTasklets = WeakSet() 57 self.receive_t = None 58 self._SetupStatsGathering() 59 self.send_honey_event(CHAT_CONNECTION_CREATED) 60 61 def __del__(self): 62 monolithmetrics.increment(CONNECTION_OBJECT_DESTROYED, value=1, tags=self._get_tags()) 63 self.send_honey_event(CHAT_CONNECTION_DELETED) 64 65 def close_socket(self): 66 if not self.socket: 67 return 68 self.socket.close() 69 self.socket = None 70 monolithmetrics.increment('xmpp.sockets.closed') 71 72 def _SetupStatsGathering(self): 73 self.lastSocketActivityTime = blue.os.GetWallclockTime() 74 self.bytesReceived = 0 75 self.bytesReceivedTotal = 0 76 self.bytesReceivedPerSec = 0 77 self.bytesReceivedPerSecAvg = 0 78 self.bytesSent = 0 79 self.bytesSentTotal = 0 80 self.bytesSentPerSec = 0 81 self.bytesSentPerSecAvg = 0 82 self.bytesReceivedStat = blue.statistics.Find('xmpp/bytesReceived') 83 if not self.bytesReceivedStat: 84 self.bytesReceivedStat = blue.CcpStatisticsEntry() 85 self.bytesReceivedStat.name = 'xmpp/bytesReceived' 86 self.bytesReceivedStat.type = 1 87 self.bytesReceivedStat.resetPerFrame = True 88 self.bytesReceivedStat.description = 'Number of bytes received from Xmpp chat' 89 blue.statistics.Register(self.bytesReceivedStat) 90 self.bytesSentStat = blue.statistics.Find('xmpp/bytesSent') 91 if not self.bytesSentStat: 92 self.bytesSentStat = blue.CcpStatisticsEntry() 93 self.bytesSentStat.name = 'xmpp/bytesSent' 94 self.bytesSentStat.type = 1 95 self.bytesSentStat.resetPerFrame = True 96 self.bytesSentStat.description = 'Number of bytes sent to Xmpp chat' 97 blue.statistics.Register(self.bytesSentStat) 98 self.bytesReceivedPerSecStat = blue.statistics.Find('xmpp/bytesReceivedPerSec') 99 if not self.bytesReceivedPerSecStat: 100 self.bytesReceivedPerSecStat = blue.CcpStatisticsEntry() 101 self.bytesReceivedPerSecStat.name = 'xmpp/bytesReceivedPerSec' 102 self.bytesReceivedPerSecStat.type = 1 103 self.bytesReceivedPerSecStat.resetPerFrame = False 104 self.bytesReceivedPerSecStat.description = 'Number of bytes received per second from Xmpp chat' 105 blue.statistics.Register(self.bytesReceivedStat) 106 self.bytesSentPerSecStat = blue.statistics.Find('xmpp/bytesSentPerSec') 107 if not self.bytesSentPerSecStat: 108 self.bytesSentPerSecStat = blue.CcpStatisticsEntry() 109 self.bytesSentPerSecStat.name = 'xmpp/bytesSentPerSec' 110 self.bytesSentPerSecStat.type = 1 111 self.bytesSentPerSecStat.resetPerFrame = False 112 self.bytesSentPerSecStat.description = 'Number of bytes sent per second to Xmpp chat' 113 blue.statistics.Register(self.bytesSentPerSecStat) 114 self.bytesReceivedPerSecAvgStat = blue.statistics.Find('xmpp/bytesReceivedPerSecAvg') 115 if not self.bytesReceivedPerSecAvgStat: 116 self.bytesReceivedPerSecAvgStat = blue.CcpStatisticsEntry() 117 self.bytesReceivedPerSecAvgStat.name = 'xmpp/bytesReceivedAvgPerSec' 118 self.bytesReceivedPerSecAvgStat.type = 1 119 self.bytesReceivedPerSecAvgStat.resetPerFrame = False 120 self.bytesReceivedPerSecAvgStat.description = 'Number of bytes received per second from Xmpp chat' 121 blue.statistics.Register(self.bytesReceivedStat) 122 self.bytesSentPerSecAvgStat = blue.statistics.Find('xmpp/bytesSentPerSecAvg') 123 if not self.bytesSentPerSecAvgStat: 124 self.bytesSentPerSecAvgStat = blue.CcpStatisticsEntry() 125 self.bytesSentPerSecAvgStat.name = 'xmpp/bytesSentPerSecAvg' 126 self.bytesSentPerSecAvgStat.type = 1 127 self.bytesSentPerSecAvgStat.resetPerFrame = False 128 self.bytesSentPerSecAvgStat.description = 'Number of bytes sent per second to Xmpp chat' 129 blue.statistics.Register(self.bytesSentPerSecAvgStat) 130 self.requestsStat = blue.statistics.Find('xmpp/requests') 131 if not self.requestsStat: 132 self.requestsStat = blue.CcpStatisticsEntry() 133 self.requestsStat.name = 'xmpp/requests' 134 self.requestsStat.type = 1 135 self.requestsStat.resetPerFrame = False 136 self.requestsStat.description = 'Number of requests to Xmpp chat' 137 blue.statistics.Register(self.requestsStat) 138 uthread2.start_tasklet(self._GatherStats) 139 140 def _GatherStats(self): 141 bytesReceivedSamples = [] 142 bytesSentSamples = [] 143 NUM_SAMPLES_FOR_AVERAGE = 10 144 while True: 145 uthread2.sleep(1) 146 bytesReceivedSamples.append(self.bytesReceived) 147 bytesReceivedSamples = bytesReceivedSamples[-NUM_SAMPLES_FOR_AVERAGE:] 148 self.bytesReceivedTotal += self.bytesReceived 149 self.bytesReceivedPerSec = self.bytesReceived 150 self.bytesReceived = 0 151 self.bytesReceivedPerSecAvg = sum(bytesReceivedSamples) / len(bytesReceivedSamples) 152 self.bytesReceivedPerSecStat.Set(self.bytesReceivedPerSec) 153 self.bytesReceivedPerSecAvgStat.Set(self.bytesReceivedPerSecAvg) 154 bytesSentSamples.append(self.bytesSent) 155 bytesSentSamples = bytesSentSamples[-NUM_SAMPLES_FOR_AVERAGE:] 156 self.bytesSentTotal += self.bytesSent 157 self.bytesSentPerSec = self.bytesSent 158 self.bytesSent = 0 159 self.bytesSentPerSecAvg = sum(bytesSentSamples) / len(bytesSentSamples) 160 self.bytesSentPerSecStat.Set(self.bytesSentPerSec) 161 self.bytesSentPerSecAvgStat.Set(self.bytesSentPerSecAvg) 162 163 def Connect(self, username, connection_id = None): 164 self.disconnectRequested = False 165 self.username = username 166 if connection_id: 167 self.connection_id = connection_id 168 self._connectToXmppServer() 169 self.send_honey_event(CHAT_CONNECTION_CONNECTED, {'connection_id': self.connection_id}) 170 171 def Disconnect(self): 172 is_connected = self.IsConnected() 173 if is_connected: 174 self.xmppHandler.send('</stream:stream>') 175 self.close_socket() 176 self.disconnectRequested = True 177 if self.receive_t: 178 self.receive_t.kill() 179 self.receive_t = None 180 tasklets = self.myTasklets.copy() 181 self.myTasklets.clear() 182 for tasklet in tasklets: 183 tasklet.kill() 184 185 self.xmppHandler = None 186 self.send_honey_event(CHAT_CONNECTION_DISCONNECTED, {'was_connected': is_connected}) 187 188 def IsConnected(self): 189 if self.xmppHandler and self.xmppHandler.state == 'ready': 190 return True 191 return False 192 193 def IssueRequest(self, request, request_type, to = None, method_tag = 'unknown'): 194 self.requestsStat.Inc() 195 if not self.IsConnected(): 196 logger.warning("CAN'T HANDLE REQUEST while NOT CONNECTED: iq %s: %s, %s", request_type, request, to) 197 return 198 TIMEOUT_DELAY_SECS = 30 199 startTime = blue.os.GetWallclockTime() 200 import stackless 201 callback_channel = stackless.channel() 202 203 def callback(response): 204 if callback_channel.balance < 0: 205 callback_channel.send(response) 206 else: 207 logger.warning('NO LISTENERS for callback channel!') 208 209 request_id = self.xmppHandler.issue_request(request, request_type, callback, to=to) 210 handler_id = self.xmppHandler.get_id() 211 channel_timeout(callback_channel, TIMEOUT_DELAY_SECS, msg='TIMED-OUT awaiting response for request: \n%s' % request) 212 result = callback_channel.receive() 213 endTime = blue.os.GetWallclockTime() 214 duration_ms = blue.os.TimeDiffInMs(startTime, endTime) 215 monolithmetrics.histogram('xmpp.issue_request.duration', value=duration_ms, tags=['method:' + method_tag], buckets=(100, 500, 1000)) 216 if result is None and self.xmppHandler is not None: 217 self.xmppHandler.notify_of_timeout(request_id, handler_id, duration_ms, request_type, request, to) 218 if duration_ms > 3000: 219 logger.warning('SLOW RESPONSE time of %d ms, type:%s, request:%s, to:%s, result:%s', duration_ms, request_type, request, to, result) 220 return result 221 222 def Message(self, receiver, text): 223 return self.xmppHandler.message(receiver, text) 224 225 def GroupChat(self, receiver, text): 226 return self.xmppHandler.groupchat(receiver, text) 227 228 def SendXmppSubject(self, receiver, text): 229 self.xmppHandler.send_xmpp_subject(receiver, text) 230 231 def JoinRoom(self, room, password = '', historySeconds = 0): 232 return self.xmppHandler.join_room(room, password=password, historySeconds=historySeconds) 233 234 def LeaveRoom(self, room): 235 return self.xmppHandler.leave_room(room) 236 237 def SetRole(self, room, user, role, reason): 238 template = "<query xmlns='http://jabber.org/protocol/muc#admin'><item nick='{0}' role='{1}'>{2}</item></query>" 239 if reason is not None: 240 reason_element = '<reason>{0}</reason>'.format(self.xmppHandler.sanitize_text(reason)) 241 else: 242 reason_element = '' 243 query = template.format(user, role, reason_element) 244 response = self.IssueRequest(query, 'set', to=room, method_tag='set_role') 245 if not response.attributes.getValue('type') == 'result': 246 logger.warning('set_role: Incorrect response') 247 248 def RegisterTemporaryRestriction(self, category, room, user, durationInSecs): 249 template = "<query xmlns='urn:xmpp:expiring_record' jid='{0}' room='{1}' duration='{2}' action='set' category='{3}'/>" 250 query = template.format(user, room, durationInSecs, category) 251 self.IssueRequest(query, 'set', to=self.hostname, method_tag='register_temporary_restriction') 252 253 def UnregisterTemporaryRestriction(self, category, room, user): 254 template = "<query xmlns='urn:xmpp:expiring_record' jid='{0}' room='{1}' action='delete' category='{2}'/>" 255 query = template.format(user, room, category) 256 self.IssueRequest(query, 'set', to=self.hostname, method_tag='unregister_temporary_restriction') 257 258 def SetEveUserData(self, user, data): 259 attrs = '' 260 for k, v in data.iteritems(): 261 try: 262 attribute = quoteattr(escape(v).encode('utf-8')) 263 except Exception: 264 attribute = "'{0}'".format(v) 265 266 attrs += '{0}={1} '.format(k, attribute) 267 268 template = "<query xmlns='urn:xmpp:eve_user_data' jid='{0}'><eve_user_data {1}/></query>" 269 query = template.format(user, attrs) 270 self.IssueRequest(query, 'set', to=self.hostname, method_tag='set_eve_user_data') 271 272 def GetEveUserData(self, user): 273 template = "<query xmlns='urn:xmpp:eve_user_data' jid='{0}'/>" 274 query = template.format(user) 275 result = self.IssueRequest(query, 'get', to=self.hostname, method_tag='get_eve_user_data') 276 logger.debug(result) 277 return result 278 279 def Invite(self, user, roomName, reason): 280 template = "<message from='{sender}' to='{room}'><x xmlns='http://jabber.org/protocol/muc#user'><invite to='{to}'><reason>{reason}</reason></invite></x></message>" 281 package = template.format(sender=self.username, room=roomName, reason=reason, to=user) 282 self.Send(package) 283 284 def RejectInvite(self, user, roomName, reason): 285 template = "<message from='{sender}' to='{room}'><x xmlns='http://jabber.org/protocol/muc#user'><decline to='{to}'><reason>{reason}</reason></decline></x></message>" 286 package = template.format(sender=self.username, room=roomName, reason=reason, to=user) 287 self.Send(package) 288 289 def Send(self, package): 290 return self.xmppHandler.send(package) 291 292 def GetFullUsername(self, user): 293 return '{0}@{1}'.format(user, self.hostname) 294 295 def _receive_t(self): 296 while not self.disconnectRequested and self.socket: 297 wasConnected = self.IsConnected() 298 try: 299 startTime = blue.os.GetWallclockTime() 300 response = self.socket.recv(4096) 301 endTime = blue.os.GetWallclockTime() 302 self.lastSocketActivityTime = endTime 303 duration = blue.os.TimeDiffInMs(startTime, endTime) 304 monolithmetrics.histogram('xmpp.bytes.rx.duration', value=duration, buckets=(100, 500, 1000)) 305 except socket.error: 306 idleTime = self.GetIdleTimeInSeconds() 307 logger.warning('Socket error in receive tasklet after %ds idle time', idleTime) 308 response = '' 309 310 bytesReceived = len(response) 311 if bytesReceived == 0: 312 self.close_socket() 313 self.xmppHandler = None 314 if self.disconnectRequested: 315 monolithmetrics.increment(CONNECTION_DISCONNECTED, value=1, tags=self._get_tags()) 316 else: 317 uthread2.start_tasklet(self._handleXmppConnectionLost, wasConnected) 318 break 319 monolithmetrics.increment('xmpp.bytes.rx', value=bytesReceived) 320 self.bytesReceivedStat.Add(bytesReceived) 321 self.bytesReceived += bytesReceived 322 try: 323 self.xmppHandler.feed_parser(response) 324 while self.xmppHandler and not self.xmppHandler.is_queue_empty(): 325 element = self.xmppHandler.pop_queue() 326 if self.disconnectRequested and self.xmppHandler: 327 self.xmppHandler.flush_queue() 328 elif self.IsConnected(): 329 self.myTasklets.add(uthread2.start_tasklet(self.xmppHandler.process_single_queue_item, element)) 330 monolithmetrics.gauge('xmpp.connection.tasklets', value=len(self.myTasklets)) 331 else: 332 self.xmppHandler.process_single_queue_item(element) 333 334 except SAXParseException: 335 logger.exception('Error parsing XMPP stream') 336 self.close_socket() 337 self.xmppHandler = None 338 uthread2.start_tasklet(self._handleXmppConnectionLost, wasConnected) 339 break 340 except Exception: 341 logger.exception('Exception in receive tasklet') 342 343 if self.disconnectRequested: 344 self.close_socket() 345 self.disconnectRequested = False 346 347 def GetIdleTimeInSeconds(self): 348 idleTime = blue.os.TimeDiffInMs(self.lastSocketActivityTime, blue.os.GetWallclockTime()) / 1000 349 return idleTime 350 351 def _send(self, data): 352 if self.verbose: 353 logger.info('Send: %s', data) 354 try: 355 startTime = blue.os.GetWallclockTime() 356 self.socket.sendall(data) 357 endTime = blue.os.GetWallclockTime() 358 duration = blue.os.TimeDiffInMs(startTime, endTime) 359 bytesSent = len(data) 360 if duration > 300: 361 logger.warning('Slow: Sending %d bytes took %d ms', bytesSent, duration) 362 monolithmetrics.increment('xmpp.bytes.tx', value=bytesSent) 363 monolithmetrics.histogram('xmpp.bytes.tx.duration', value=duration, buckets=(100, 500, 1000)) 364 self.bytesSentStat.Add(bytesSent) 365 self.bytesSent += bytesSent 366 self.lastSocketActivityTime = startTime 367 except socket.error: 368 idleTime = self.GetIdleTimeInSeconds() 369 logger.warning('Socket error when sending after %ds idle time', idleTime) 370 logger.exception('Socket error in send') 371 372 def _connectToXmppServer(self): 373 self.xmppHandler = XmppHandler(self, self.connection_id) 374 self.xmppHandler.handle_message = self.handle_message 375 self.xmppHandler.handle_presence = self.handle_presence 376 self.xmppHandler.handle_logged_in = self.handle_logged_in 377 self.xmppHandler.handle_login_failed = self._handleXmppLoginFailed 378 self.xmppHandler.handle_closed = self._handleXmppConnectionLost 379 self.xmppHandler.handle_stream_error = self.handle_stream_error 380 self.xmppHandler.send = self._send 381 self.xmppHandler.verbose = self.verbose 382 s = socket.socket() 383 monolithmetrics.increment('xmpp.sockets.opened') 384 try: 385 monolithmetrics.increment(CONNECTION_ATTEMPTED, value=1, tags=self._get_tags()) 386 s.connect((self.ejabberd_address, self.port)) 387 if not prefs.GetValue('chatserver_disable_tls', False): 388 s = ssl.wrap_socket(s, cert_reqs=ssl.CERT_REQUIRED, ca_certs=ca_certs_file) 389 self.socket = s 390 monolithmetrics.increment(CONNECTION_SUCCEEDED, value=1, tags=self._get_tags()) 391 except Exception as e: 392 monolithmetrics.increment(CONNECTION_FAILED, value=1, tags=self._get_tags()) 393 raise e 394 395 self.receive_t = uthread2.start_tasklet(self._receive_t) 396 self.xmppHandler.connect(self.hostname, self.username, self.get_password()) 397 398 def _reconnectToXmppServer(self): 399 isFirst = True 400 while True: 401 try: 402 self._connectToXmppServer() 403 monolithmetrics.increment(CONNECTION_RECONNECTED, value=1, tags=self._get_tags()) 404 break 405 except socket.error: 406 if isFirst: 407 logger.exception('Socket error when reconnecting to chat server') 408 isFirst = False 409 else: 410 logger.warning('Error reconnecting to chat server') 411 except ssl.SSLError: 412 if isFirst: 413 logger.exception('TLS error when reconnecting to chat server') 414 isFirst = False 415 else: 416 logger.warning('Error reconnecting to chat server') 417 418 uthread2.sleep(5) 419 420 def _handleXmppConnectionLost(self, wasConnected): 421 self.xmppHandler = None 422 if wasConnected: 423 monolithmetrics.increment(CONNECTION_DISCONNECTED, value=1, tags=self._get_tags()) 424 logger.warning('Connection to chat server lost') 425 uthread2.sleep(5) 426 try: 427 self.handle_connection_lost() 428 except Exception: 429 logger.exception('Exception when handling lost connection') 430 431 else: 432 logger.warning("Couldn't connect to %s", self.ejabberd_address) 433 uthread2.sleep(5) 434 uthread2.start_tasklet(self._reconnectToXmppServer) 435 436 def _handleXmppLoginFailed(self): 437 self.close_socket() 438 self.xmppHandler = None 439 monolithmetrics.increment(CONNECTION_LOGIN_FAILED, value=1, tags=self._get_tags()) 440 logger.warning('Failed to login to chat server') 441 try: 442 self.handle_login_failed() 443 except Exception: 444 logger.exception('Exception when handling failed login') 445 446 uthread2.sleep(5) 447 uthread2.start_tasklet(self._reconnectToXmppServer) 448 449 def handle_logged_in(self): 450 if self.verbose: 451 logger.info('Logged in to xmpp server') 452 self.hasConnectedOnce = True 453 454 def _get_tags(self): 455 return ['address:{}'.format(self.ejabberd_address), 'host:{}'.format(self.hostname)] 456 457 def handle_login_failed(self): 458 pass 459 460 def handle_message(self, response): 461 pass 462 463 def handle_presence(self, response): 464 pass 465 466 def handle_connection_lost(self): 467 pass 468 469 def handle_stream_error(self, state, response): 470 pass 471 472 def get_password(self): 473 pass 474 475 def send_honey_event(self, event_name, extra = None): 476 fields = {'eve.xmpp.connection.id': id(self), 477 HOST_NAME: self.hostname, 478 'eve.xmpp.connection.address': self.ejabberd_address, 479 'eve.xmpp.connection.has_connected': self.hasConnectedOnce, 480 SERVICE_NAME: SERVICE_NAME_PREFIX + 'xmppConnection'} 481 if extra is not None: 482 fields.update(extra) 483 try: 484 mhe.send(event_name, fields) 485 except Exception as e: 486 logger.warning('Failed to send honey event: %s' % e)