xmpphandler.py (13000B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\xmpp\xmpphandler.py 2 from __future__ import absolute_import 3 import base64 4 import logging 5 import uuid 6 from xml.sax import make_parser 7 from xml.sax.saxutils import escape 8 import unicodedata 9 from .xmppcontenthandler import XmppContentHandler 10 import monolithmetrics 11 logger = logging.getLogger(__name__) 12 13 def remove_control_characters(unicode_str): 14 return u''.join((ch for ch in unicode_str if unicodedata.category(ch)[0] != 'C')) 15 16 17 class XmppHandler(object): 18 19 def __init__(self, owner_xmpp_connection, connection_id): 20 self.xmppConnection = owner_xmpp_connection 21 self.connection_id = connection_id 22 self.id = uuid.uuid4() 23 self.state = 'initial' 24 self.parser = make_parser() 25 self.contentHandler = None 26 self.host = None 27 self.jid = None 28 self.nick = None 29 self.username = None 30 self.password = None 31 self.requests = {} 32 self.next_request_id = 1 33 self.timed_out_requests_ids = set() 34 self.handle_response = self.handle_response_state_not_ready 35 self.stream_element = None 36 self.stream_features_element = None 37 self.verbose = False 38 39 def connect(self, host, username, password): 40 self.host = host 41 self.username = username 42 self.nick = username.split('@')[0] 43 self.jid = self.username 44 self.password = password 45 self.state = 'waiting_for_stream' 46 self.start_stream() 47 48 def start_stream(self): 49 self.parser.reset() 50 self.contentHandler = XmppContentHandler() 51 self.parser.setContentHandler(self.contentHandler) 52 template = "<?xml version='1.0'?><stream:stream to='{0}' version='1.0' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'>" 53 self.send(template.format(self.host)) 54 monolithmetrics.increment('xmpp.handler.send', tags=['type:start_stream']) 55 56 def authenticate(self): 57 key = base64.b64encode('\x00{0}\x00{1}'.format(self.nick, self.password).encode('ascii')).decode() 58 if self.verbose: 59 logger.info('Auth Key: %s', key) 60 package = "<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='PLAIN'>{0}</auth>".format(key) 61 self.state = 'authenticating' 62 self.send(package) 63 monolithmetrics.increment('xmpp.handler.send', tags=['type:authenticate']) 64 65 def bind(self): 66 request = "<bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'><resource>{0}</resource></bind>".format(self.connection_id) 67 self.issue_request(request, 'set', self.handle_bind) 68 69 def handle_bind(self, response): 70 if self.verbose: 71 logger.info('Got response to bind') 72 if response.children[0].tag == 'bind': 73 bind = response.children[0] 74 if bind.children[0].tag == 'jid': 75 self.jid = bind.children[0].text 76 if self.verbose: 77 logger.info('JID is %s', self.jid) 78 self.start_session() 79 80 def start_session(self): 81 request = "<session xmlns='urn:ietf:params:xml:ns:xmpp-session'/>".format(self.host) 82 self.issue_request(request, 'set', self.handle_session) 83 84 def handle_session(self, _response): 85 self.send_initial_presence() 86 self.state = 'ready' 87 88 def send_initial_presence(self): 89 package = '<presence><show/></presence>' 90 self.send(package) 91 monolithmetrics.increment('xmpp.handler.send', tags=['type:send_initial_presence']) 92 93 def message(self, receiver, text): 94 template = '<message from="{0}" to="{1}" xml:lang="en"><body>{2}</body></message>' 95 package = template.format(self.username, receiver, self.sanitize_text(text)) 96 self.send(package) 97 monolithmetrics.increment('xmpp.handler.send', tags=['type:message']) 98 99 def groupchat(self, receiver, text): 100 msg_id = self.get_request_id() 101 template = "<message from='{0}' to='{1}' type='groupchat' xml:lang='en' id='{2}'><body>{3}</body></message>" 102 package = template.format(self.username, receiver, msg_id, self.sanitize_text(text)) 103 self.send(package) 104 monolithmetrics.increment('xmpp.handler.send', tags=['type:groupchat']) 105 return msg_id 106 107 def sanitize_text(self, text): 108 sanitized = escape(remove_control_characters(text)).encode('utf8') 109 return sanitized 110 111 def send_xmpp_subject(self, receiver, text): 112 template = '<message from="{0}" to="{1}" type="groupchat" xml:lang="en"><subject>{2}</subject></message>' 113 package = template.format(self.username, receiver, self.sanitize_text(text)) 114 self.send(package) 115 monolithmetrics.increment('xmpp.handler.send', tags=['type:send_xmpp_subject']) 116 117 def feed_parser(self, response): 118 if self.verbose: 119 logger.info('Recv: %s', response) 120 self.parser.feed(response) 121 122 def is_queue_empty(self): 123 return len(self.contentHandler.queue) == 0 124 125 def pop_queue(self): 126 item = self.contentHandler.queue.pop() 127 monolithmetrics.gauge('xmpp.content_handler.queue', value=len(self.contentHandler.queue)) 128 return item 129 130 def flush_queue(self): 131 if self.contentHandler and self.contentHandler.queue: 132 self.contentHandler.queue = [] 133 134 def process_single_queue_item(self, element): 135 if self.verbose: 136 logger.info('XmppHandler:processing State: %s Username: %s - Received element:\n%s', self.state, self.username, element.toXml()) 137 if self.xmppConnection.disconnectRequested: 138 return 139 if element.tag == 'iq': 140 iq_type = element.attributes.getValue('type') 141 if iq_type == 'get' and element.children[0].tag == 'ping': 142 self._handle_ping(element) 143 else: 144 self._handle_iq_result(element) 145 else: 146 result = self.handle_response(element) 147 if not result: 148 if element.tag == 'stream:error': 149 result = self.handle_stream_error(self.state, element) 150 if not result: 151 logger.warning('Unhandled response - state %s\n%s', self.state, element.toXml()) 152 return True 153 154 def _handle_ping(self, element): 155 request_id = element.attributes.getValue('id') 156 ping_to = element.attributes['to'] 157 ping_from = element.attributes['from'] 158 package = "<iq id='{0}' type='result' from='{1}' to='{2}'/>".format(request_id, ping_to, ping_from) 159 self.send(package) 160 monolithmetrics.increment('xmpp.handler.send', tags=['type:handle_ping']) 161 162 def _handle_iq_result(self, element): 163 request_id = element.attributes.getValue('id') 164 try: 165 callback = self.requests.pop(request_id) 166 if callback: 167 callback(element) 168 monolithmetrics.gauge('xmpp.handler.requests', value=len(self.requests)) 169 except KeyError: 170 try: 171 self.timed_out_requests_ids.remove(request_id) 172 logger.warning('LATE RESPONSE to timed-out request_id=%s with response=%s', request_id, element) 173 except KeyError: 174 logger.error('NO CALLBACK FOUND found for RESPONSE to request %s with response=%s', request_id, element) 175 176 def notify_of_timeout(self, request_id, id, duration_ms, request_type, request, to): 177 if self.id != id: 178 return 179 try: 180 _callback = self.requests.pop(request_id) 181 monolithmetrics.gauge('xmpp.handler.requests', value=len(self.requests)) 182 monolithmetrics.increment('xmpp.handler.timeouts') 183 logger.warning('TIME-OUT NOTIFIED for request_id=%s after duration_ms=%s, type=%s, request=%s', request_id, duration_ms, request_type, request) 184 self.timed_out_requests_ids.add(request_id) 185 except KeyError: 186 logger.error('NO CALLBACK FOUND for TIMED-OUT request %s', request_id) 187 188 def handle_response_state_not_ready(self, response): 189 if self.state == 'waiting_for_stream': 190 if response.tag == 'stream:stream': 191 self.stream_element = response 192 self.state = 'waiting_for_features' 193 return True 194 else: 195 logger.warning('Expected stream:stream tag') 196 return True 197 elif self.state == 'waiting_for_features': 198 if response.tag == 'stream:features': 199 self.stream_features_element = response 200 self.authenticate() 201 return True 202 elif self.state == 'authenticating': 203 if response.tag == 'success': 204 self.start_stream() 205 self.state = 'authenticated_waiting_for_stream' 206 return True 207 if response.tag == 'failure': 208 self.handle_login_failed() 209 return True 210 elif self.state == 'authenticated_waiting_for_stream': 211 if response.tag == 'stream:stream': 212 self.stream_element = response 213 if self.verbose: 214 logger.info('Logged in with id: %s', response.attributes.getValue('id')) 215 self.state = 'authenticated_waiting_for_features' 216 return True 217 elif self.state == 'authenticated_waiting_for_features': 218 if response.tag == 'stream:features': 219 self.stream_features_element = response 220 self.bind() 221 return True 222 elif self.state == 'ready': 223 self.handle_response = self.handle_response_state_ready 224 self.handle_logged_in() 225 return self.handle_response_state_ready(response) 226 227 def handle_response_state_ready(self, response): 228 if response.tag == 'message': 229 self.handle_message(response) 230 return True 231 if response.tag == 'presence': 232 self.handle_presence(response) 233 return True 234 if response.tag == 'stream:closed': 235 self.state = 'closed' 236 self.handle_response = self.handle_response_state_not_ready 237 self.handle_closed() 238 return True 239 240 def handle_logged_in(self): 241 pass 242 243 def handle_login_failed(self): 244 pass 245 246 def handle_message(self, response): 247 pass 248 249 def handle_presence(self, response): 250 pass 251 252 def handle_closed(self): 253 pass 254 255 def handle_stream_error(self, state, response): 256 return None 257 258 def send(self, package): 259 pass 260 261 def issue_request(self, request, request_type, callback, to = None): 262 request_id = self.get_request_id() 263 if to is not None: 264 to_clause = "to='{0}'".format(to) 265 else: 266 to_clause = '' 267 package = "<iq id='{0}' type='{1}' from='{2}' {3}>{4}</iq>".format(request_id, request_type, self.jid, to_clause, request) 268 self.requests[request_id] = callback 269 monolithmetrics.gauge('xmpp.handler.requests', value=len(self.requests)) 270 self.send(package) 271 monolithmetrics.increment('xmpp.handler.send', tags=['type:issue_request']) 272 return request_id 273 274 def join_room(self, room, password = '', historySeconds = 0): 275 request_id = self.get_request_id() 276 if password: 277 password_element = '<password>{0}</password>'.format(self.sanitize_text(password)) 278 else: 279 password_element = '' 280 if historySeconds > 0: 281 history_element = "<history seconds='{0}'/>".format(int(historySeconds)) 282 else: 283 history_element = '' 284 template = "<presence from='{0}' id='{1}' to='{2}/{3}'><x xmlns='http://jabber.org/protocol/muc'>{4}{5}</x></presence>" 285 package = template.format(self.jid, request_id, room, self.nick, password_element, history_element) 286 self.send(package) 287 monolithmetrics.increment('xmpp.handler.send', tags=['type:join_room']) 288 return request_id 289 290 def leave_room(self, room): 291 request_id = self.get_request_id() 292 template = "<presence from='{0}' id='{1}' to='{2}/{3}' type='unavailable'/>" 293 package = template.format(self.jid, request_id, room, self.nick) 294 self.send(package) 295 monolithmetrics.increment('xmpp.handler.send', tags=['type:leave_room']) 296 return request_id 297 298 def get_request_id(self): 299 request_id = 'id{0}'.format(self.next_request_id) 300 self.next_request_id += 1 301 return request_id 302 303 def create_room(self, room): 304 request_id = self.get_request_id() 305 template = "<presence from='{0}' id='{1}' to='{2}/{3}'><x xmlns='http://jabber.org/protocol/muc'/></presence>" 306 package = template.format(self.jid, request_id, room, self.nick) 307 self.send(package) 308 monolithmetrics.increment('xmpp.handler.send', tags=['type:create_room']) 309 return request_id 310 311 def get_id(self): 312 return self.id