libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0198.py @ 3028:ab2696e34d29

Python 3 port: /!\ this is a huge commit /!\ starting from this commit, SàT is needs Python 3.6+ /!\ SàT maybe be instable or some feature may not work anymore, this will improve with time This patch port backend, bridge and frontends to Python 3. Roughly this has been done this way: - 2to3 tools has been applied (with python 3.7) - all references to python2 have been replaced with python3 (notably shebangs) - fixed files not handled by 2to3 (notably the shell script) - several manual fixes - fixed issues reported by Python 3 that where not handled in Python 2 - replaced "async" with "async_" when needed (it's a reserved word from Python 3.7) - replaced zope's "implements" with @implementer decorator - temporary hack to handle data pickled in database, as str or bytes may be returned, to be checked later - fixed hash comparison for password - removed some code which is not needed anymore with Python 3 - deactivated some code which needs to be checked (notably certificate validation) - tested with jp, fixed reported issues until some basic commands worked - ported Primitivus (after porting dependencies like urwid satext) - more manual fixes

author	Goffi <goffi@goffi.org>
date	Tue, 13 Aug 2019 19:08:41 +0200
parents	af9d71303605
children	87b8808ac49d

comparison

equal deleted inserted replaced

-:ff5bcb12ae60
+:ab2696e34d29
-#!/usr/bin/env python2
+#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 # SàT plugin for managing raw XML log
 # Copyright (C) 2011  Jérôme Poisson (goffi@goffi.org)
 from twisted.words.xish import domish
 from twisted.internet import defer
 from twisted.internet import task, reactor
 from functools import partial
 from wokkel import disco, iwokkel
-from zope.interface import implements
+from zope.interface import implementer
 import collections
 import time
 log = getLogger(__name__)
 PLUGIN_INFO = {
-C.PI_NAME: u"Stream Management",
+C.PI_NAME: "Stream Management",
-C.PI_IMPORT_NAME: u"XEP-0198",
+C.PI_IMPORT_NAME: "XEP-0198",
-C.PI_TYPE: u"XEP",
+C.PI_TYPE: "XEP",
 C.PI_MODES: C.PLUG_MODE_BOTH,
-C.PI_PROTOCOLS: [u"XEP-0198"],
+C.PI_PROTOCOLS: ["XEP-0198"],
 C.PI_DEPENDENCIES: [],
-C.PI_RECOMMENDATIONS: [u"XEP-0045", u"XEP-0313"],
+C.PI_RECOMMENDATIONS: ["XEP-0045", "XEP-0313"],
-C.PI_MAIN: u"XEP_0198",
+C.PI_MAIN: "XEP_0198",
-C.PI_HANDLER: u"yes",
+C.PI_HANDLER: "yes",
-C.PI_DESCRIPTION: _(u"""Implementation of Stream Management"""),
+C.PI_DESCRIPTION: _("""Implementation of Stream Management"""),
 }
-NS_SM = u"urn:xmpp:sm:3"
+NS_SM = "urn:xmpp:sm:3"
 SM_ENABLED = '/enabled[@xmlns="' + NS_SM + '"]'
 SM_RESUMED = '/resumed[@xmlns="' + NS_SM + '"]'
 SM_FAILED = '/failed[@xmlns="' + NS_SM + '"]'
 SM_R_REQUEST = '/r[@xmlns="' + NS_SM + '"]'
 SM_A_REQUEST = '/a[@xmlns="' + NS_SM + '"]'
 @enabled.setter
 def enabled(self, enabled):
 if enabled:
 if self._enabled:
 raise exceptions.InternalError(
-u"Stream Management can't be enabled twice")
+"Stream Management can't be enabled twice")
 self._enabled = True
 callback, kw = self.callback_data
 self.timer = task.LoopingCall(callback, **kw)
 self.timer.start(MAX_DELAY_ACK_R, now=False)
 else:
 self.session_id = self.location = None
 self.ack_requested = False
 self.last_ack_r = 0
 if self.req_timer is not None:
 if self.req_timer.active():
-log.error(u"req_timer has been called/cancelled but not reset")
+log.error("req_timer has been called/cancelled but not reset")
 else:
 self.req_timer.cancel()
 self.req_timer = None
 def getBufferCopy(self):
 # FIXME: location is not handled yet
 def __init__(self, host):
 log.info(_("Plugin Stream Management initialization"))
 self.host = host
-host.registerNamespace(u'sm', NS_SM)
+host.registerNamespace('sm', NS_SM)
 host.trigger.add("stream_hooks", self.addHooks)
 host.trigger.add("xml_init", self._XMLInitTrigger)
 host.trigger.add("disconnecting", self._disconnectingTrigger)
 host.trigger.add("disconnected", self._disconnectedTrigger)
 try:
 self._ack_timeout = int(host.memory.getConfig("", "ack_timeout", ACK_TIMEOUT))
 except ValueError:
-log.error(_(u"Invalid ack_timeout value, please check your configuration"))
+log.error(_("Invalid ack_timeout value, please check your configuration"))
 self._ack_timeout = ACK_TIMEOUT
 if not self._ack_timeout:
-log.info(_(u"Ack timeout disabled"))
+log.info(_("Ack timeout disabled"))
 else:
-log.info(_(u"Ack timeout set to {timeout}s").format(
+log.info(_("Ack timeout set to {timeout}s").format(
 timeout=self._ack_timeout))
 def profileConnecting(self, client):
 client._xep_0198_session = ProfileSessionData(callback=self.checkAcks,
 client=client)
 send_hooks.append(partial(self.onSend, client=client))
 return True
 def _XMLInitTrigger(self, client):
 """Enable or resume a stream mangement"""
-if not (NS_SM, u'sm') in client.xmlstream.features:
+if not (NS_SM, 'sm') in client.xmlstream.features:
 log.warning(_(
-u"Your server doesn't support stream management ({namespace}), this is "
+"Your server doesn't support stream management ({namespace}), this is "
-u"used to improve connection problems detection (like network outages). "
+"used to improve connection problems detection (like network outages). "
-u"Please ask your server administrator to enable this feature.".format(
+"Please ask your server administrator to enable this feature.".format(
 namespace=NS_SM)))
 return True
 session = client._xep_0198_session
 # a disconnect timer from a previous disconnection may still be active
 del session.disconnect_timer
 if session.resume_enabled:
 # we are resuming a session
 resume_elt = domish.Element((NS_SM, 'resume'))
-resume_elt['h'] = unicode(session.in_counter)
+resume_elt['h'] = str(session.in_counter)
 resume_elt['previd'] = session.session_id
 client.send(resume_elt)
 session.resuming = True
 # session.enabled will be set on <resumed/> reception
 return False
 else:
 # we start a new session
 assert session.out_counter == 0
 enable_elt = domish.Element((NS_SM, 'enable'))
-enable_elt[u'resume'] = u'true'
+enable_elt['resume'] = 'true'
 client.send(enable_elt)
 session.enabled = True
 return True
 def _disconnectingTrigger(self, client):
 def updateBuffer(self, session, server_acked):
 """Update buffer and buffer_index"""
 if server_acked > session.buffer_idx:
 diff = server_acked - session.buffer_idx
 try:
-for i in xrange(diff):
+for i in range(diff):
 session.buffer.pop()
 except IndexError:
 log.error(
-u"error while cleaning buffer, invalid index (buffer is empty):\n"
+"error while cleaning buffer, invalid index (buffer is empty):\n"
-u"diff = {diff}\n"
+"diff = {diff}\n"
-u"server_acked = {server_acked}\n"
+"server_acked = {server_acked}\n"
-u"buffer_idx = {buffer_id}".format(
+"buffer_idx = {buffer_id}".format(
 diff=diff, server_acked=server_acked,
 buffer_id=session.buffer_idx))
 session.buffer_idx += diff
 def replayBuffer(self, client, buffer_, discard_results=False):
 stanza = buffer_.pop()
 except IndexError:
 break
 else:
 if ((discard_results
-and stanza.name == u'iq'
+and stanza.name == 'iq'
-and stanza.getAttribute(u'type') == 'result')):
+and stanza.getAttribute('type') == 'result')):
 continue
 client.send(stanza)
 def sendAck(self, client):
 """Send an answer element with current IN counter"""
 a_elt = domish.Element((NS_SM, 'a'))
-a_elt['h'] = unicode(client._xep_0198_session.in_counter)
+a_elt['h'] = str(client._xep_0198_session.in_counter)
 client.send(a_elt)
 def requestAck(self, client):
 """Send a request element"""
 session = client._xep_0198_session
 def _connectionFailed(self, failure_, connector):
 normal_host, normal_port = connector.normal_location
 del connector.normal_location
 log.warning(_(
-u"Connection failed using location given by server (host: {host}, port: "
+"Connection failed using location given by server (host: {host}, port: "
-u"{port}), switching to normal host and port (host: {normal_host}, port: "
+"{port}), switching to normal host and port (host: {normal_host}, port: "
-u"{normal_port})".format(host=connector.host, port=connector.port,
+"{normal_port})".format(host=connector.host, port=connector.port,
 normal_host=normal_host, normal_port=normal_port)))
 connector.host, connector.port = normal_host, normal_port
 connector.connectionFailed = connector.connectionFailed_ori
 del connector.connectionFailed_ori
 return connector.connectionFailed(failure_)
 def onEnabled(self, enabled_elt, client):
 session = client._xep_0198_session
 session.in_counter = 0
 # we check that resuming is possible and that we have a session id
-resume = C.bool(enabled_elt.getAttribute(u'resume'))
+resume = C.bool(enabled_elt.getAttribute('resume'))
-session_id = enabled_elt.getAttribute(u'id')
+session_id = enabled_elt.getAttribute('id')
 if not session_id:
-log.warning(_(u'Incorrect <enabled/> element received, no "id" attribute'))
+log.warning(_('Incorrect <enabled/> element received, no "id" attribute'))
 if not resume or not session_id:
 log.warning(_(
-u"You're server doesn't support session resuming with stream management, "
+"You're server doesn't support session resuming with stream management, "
-u"please contact your server administrator to enable it"))
+"please contact your server administrator to enable it"))
 return
 session.session_id = session_id
 # XXX: we disable resource binding, which must not be done
 #      when we resume the session.
 client.factory.authenticator.res_binding = False
 # location, in case server want resuming session to be elsewhere
 try:
-location = enabled_elt[u'location']
+location = enabled_elt['location']
 except KeyError:
 pass
 else:
 # TODO: handle IPv6 here (in brackets, cf. XEP)
 try:
 domain, port = location.split(':', 1)
 port = int(port)
 except ValueError:
-log.warning(_(u"Invalid location received: {location}")
+log.warning(_("Invalid location received: {location}")
 .format(location=location))
 else:
 session.location = (domain, port)
 # we monkey patch connector to use the new location
 connector = client.xmlstream.transport.connector
 connector.connectionFailed = partial(self._connectionFailed,
 connector=connector)
 # resuming time
 try:
-max_s = int(enabled_elt[u'max'])
+max_s = int(enabled_elt['max'])
 except (ValueError, KeyError) as e:
 if isinstance(e, ValueError):
-log.warning(_(u'Invalid "max" attribute'))
+log.warning(_('Invalid "max" attribute'))
 max_s = RESUME_MAX
-log.info(_(u"Using default session max value ({max_s} s).".format(
+log.info(_("Using default session max value ({max_s} s).".format(
 max_s=max_s)))
-log.info(_(u"Stream Management enabled"))
+log.info(_("Stream Management enabled"))
 else:
 log.info(_(
-u"Stream Management enabled, with a resumption time of {res_m} min"
+"Stream Management enabled, with a resumption time of {res_m} min"
 .format(res_m = max_s/60)))
 session.session_max = max_s
 def onResumed(self, enabled_elt, client):
 session = client._xep_0198_session
 # we resend all stanza which have not been received properly
 self.replayBuffer(client, session.buffer)
 # now we can continue the session
 session.enabled = True
 d_time = time.time() - session.disconnected_time
-log.info(_(u"Stream session resumed (disconnected for {d_time} s, {count} "
+log.info(_("Stream session resumed (disconnected for {d_time} s, {count} "
-u"stanza(s) resent)").format(d_time=int(d_time), count=resend_count))
+"stanza(s) resent)").format(d_time=int(d_time), count=resend_count))
 def onFailed(self, failed_elt, client):
 session = client._xep_0198_session
 condition_elt = failed_elt.firstChildElement()
 buffer_ = session.getBufferCopy()
 try:
 del session.resuming
 except AttributeError:
 # stream management can't be started at all
-msg = _(u"Can't use stream management")
+msg = _("Can't use stream management")
 if condition_elt is None:
-log.error(msg + u'.')
+log.error(msg + '.')
 else:
-log.error(_(u"{msg}: {reason}").format(
+log.error(_("{msg}: {reason}").format(
 msg=msg, reason=condition_elt.name))
 else:
 # only stream resumption failed, we can try full session init
 # XXX: we try to start full session init from this point, with many
 #      variables/attributes already initialised with a potentially different
 #      jid. This is experimental and may not be safe. It may be more
 #      secured to abord the connection and restart everything with a fresh
 #      client.
-msg = _(u"stream resumption not possible, restarting full session")
+msg = _("stream resumption not possible, restarting full session")
 if condition_elt is None:
-log.warning(u'{msg}.'.format(msg=msg))
+log.warning('{msg}.'.format(msg=msg))
 else:
-log.warning(u"{msg}: {reason}".format(
+log.warning("{msg}: {reason}".format(
 msg=msg, reason=condition_elt.name))
 # stream resumption failed, but we still can do normal stream management
 # we restore attributes as if the session was new, and init stream
 # we keep everything initialized, and only do binding, roster request
 # and initial presence sending.
 if client.conn_deferred.called:
 client.conn_deferred = defer.Deferred()
 else:
-log.error(u"conn_deferred should be called at this point")
+log.error("conn_deferred should be called at this point")
-plg_0045 = self.host.plugins.get(u'XEP-0045')
+plg_0045 = self.host.plugins.get('XEP-0045')
-plg_0313 = self.host.plugins.get(u'XEP-0313')
+plg_0313 = self.host.plugins.get('XEP-0313')
 # FIXME: we should call all loaded plugins with generic callbacks
 #        (e.g. prepareResume and resume), so a hot resuming can be done
 #        properly for all plugins.
 session.req_timer.cancel()
 session.req_timer = None
 try:
 server_acked = int(a_elt['h'])
 except ValueError:
-log.warning(_(u"Server returned invalid ack element, disabling stream "
+log.warning(_("Server returned invalid ack element, disabling stream "
-u"management: {xml}").format(xml=a_elt))
+"management: {xml}").format(xml=a_elt))
 session.enabled = False
 return
 if server_acked > session.out_counter:
-log.error(_(u"Server acked more stanzas than we have sent, disabling stream "
+log.error(_("Server acked more stanzas than we have sent, disabling stream "
-u"management."))
+"management."))
 session.reset()
 return
 self.updateBuffer(session, server_acked)
 self.checkAcks(client)
 def onAckTimeOut(self, client):
 """Called when a requested ACK has not been received in time"""
-log.info(_(u"Ack was not received in time, aborting connection"))
+log.info(_("Ack was not received in time, aborting connection"))
 transport = client.xmlstream.transport
 if transport is None:
-log.warning(u"transport was already removed")
+log.warning("transport was already removed")
 else:
 transport.abortConnection()
 client._xep_0198_session.req_timer = None
+@implementer(iwokkel.IDisco)
 class XEP_0198_handler(xmlstream.XMPPHandler):
-implements(iwokkel.IDisco)
 def __init__(self, plugin_parent):
 self.plugin_parent = plugin_parent
 self.host = plugin_parent.host

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0198.py @ 3028:ab2696e34d29