Mercurial > libervia-backend
view sat/plugins/plugin_xep_0047.py @ 2744:e6716d90c2fe
plugin XEP-0384: various bug fixes:
- fixed trust management on message sending, when trusts are missing
- retrieve identiy keys when trust has not been set yet
- added our own device when it is found missing on PEP notification
- mark message as trusted when suitable
author | Goffi <goffi@goffi.org> |
---|---|
date | Thu, 03 Jan 2019 21:04:55 +0100 |
parents | 56f94936df1e |
children | 003b8b4b56a7 |
line wrap: on
line source
#!/usr/bin/env python2 # -*- coding: utf-8 -*- # SAT plugin for managing gateways (xep-0047) # Copyright (C) 2009-2018 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero 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 Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from sat.core.i18n import _ from sat.core.log import getLogger log = getLogger(__name__) from sat.core.constants import Const as C from sat.core import exceptions from twisted.words.protocols.jabber import jid from twisted.words.protocols.jabber import xmlstream from twisted.words.protocols.jabber import error from twisted.internet import reactor from twisted.internet import defer from twisted.python import failure from wokkel import disco, iwokkel from zope.interface import implements import base64 try: from twisted.words.protocols.xmlstream import XMPPHandler except ImportError: from wokkel.subprotocols import XMPPHandler MESSAGE = "/message" IQ_SET = '/iq[@type="set"]' NS_IBB = "http://jabber.org/protocol/ibb" IBB_OPEN = IQ_SET + '/open[@xmlns="' + NS_IBB + '"]' IBB_CLOSE = IQ_SET + '/close[@xmlns="' + NS_IBB + '" and @sid="{}"]' IBB_IQ_DATA = IQ_SET + '/data[@xmlns="' + NS_IBB + '" and @sid="{}"]' IBB_MESSAGE_DATA = MESSAGE + '/data[@xmlns="' + NS_IBB + '" and @sid="{}"]' TIMEOUT = 120 # timeout for workflow DEFER_KEY = "finished" # key of the deferred used to track session end PLUGIN_INFO = { C.PI_NAME: "In-Band Bytestream Plugin", C.PI_IMPORT_NAME: "XEP-0047", C.PI_TYPE: "XEP", C.PI_MODES: C.PLUG_MODE_BOTH, C.PI_PROTOCOLS: ["XEP-0047"], C.PI_MAIN: "XEP_0047", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("""Implementation of In-Band Bytestreams"""), } class XEP_0047(object): NAMESPACE = NS_IBB BLOCK_SIZE = 4096 def __init__(self, host): log.info(_("In-Band Bytestreams plugin initialization")) self.host = host def getHandler(self, client): return XEP_0047_handler(self) def profileConnected(self, client): client.xep_0047_current_stream = {} # key: stream_id, value: data(dict) def _timeOut(self, sid, client): """Delete current_stream id, called after timeout @param sid(unicode): session id of client.xep_0047_current_stream @param client: %(doc_client)s """ log.info( u"In-Band Bytestream: TimeOut reached for id {sid} [{profile}]".format( sid=sid, profile=client.profile ) ) self._killSession(sid, client, "TIMEOUT") def _killSession(self, sid, client, failure_reason=None): """Delete a current_stream id, clean up associated observers @param sid(unicode): session id @param client: %(doc_client)s @param failure_reason(None, unicode): if None the session is successful else, will be used to call failure_cb """ try: session = client.xep_0047_current_stream[sid] except KeyError: log.warning(u"kill id called on a non existant id") return try: observer_cb = session["observer_cb"] except KeyError: pass else: client.xmlstream.removeObserver(session["event_data"], observer_cb) if session["timer"].active(): session["timer"].cancel() del client.xep_0047_current_stream[sid] success = failure_reason is None stream_d = session[DEFER_KEY] if success: stream_d.callback(None) else: stream_d.errback(failure.Failure(exceptions.DataError(failure_reason))) def createSession(self, *args, **kwargs): """like [_createSession] but return the session deferred instead of the whole session session deferred is fired when transfer is finished """ return self._createSession(*args, **kwargs)[DEFER_KEY] def _createSession(self, client, stream_object, to_jid, sid): """Called when a bytestream is imminent @param stream_object(IConsumer): stream object where data will be written @param to_jid(jid.JId): jid of the other peer @param sid(unicode): session id @return (dict): session data """ if sid in client.xep_0047_current_stream: raise exceptions.ConflictError(u"A session with this id already exists !") session_data = client.xep_0047_current_stream[sid] = { "id": sid, DEFER_KEY: defer.Deferred(), "to": to_jid, "stream_object": stream_object, "seq": -1, "timer": reactor.callLater(TIMEOUT, self._timeOut, sid, client), } return session_data def _onIBBOpen(self, iq_elt, client): """"Called when an IBB <open> element is received @param iq_elt(domish.Element): the whole <iq> stanza """ log.debug(_(u"IBB stream opening")) iq_elt.handled = True open_elt = iq_elt.elements(NS_IBB, "open").next() block_size = open_elt.getAttribute("block-size") sid = open_elt.getAttribute("sid") stanza = open_elt.getAttribute("stanza", "iq") if not sid or not block_size or int(block_size) > 65535: return self._sendError("not-acceptable", sid or None, iq_elt, client) if not sid in client.xep_0047_current_stream: log.warning(_(u"Ignoring unexpected IBB transfer: %s" % sid)) return self._sendError("not-acceptable", sid or None, iq_elt, client) session_data = client.xep_0047_current_stream[sid] if session_data["to"] != jid.JID(iq_elt["from"]): log.warning( _("sended jid inconsistency (man in the middle attack attempt ?)") ) return self._sendError("not-acceptable", sid, iq_elt, client) # at this stage, the session looks ok and will be accepted # we reset the timeout: session_data["timer"].reset(TIMEOUT) # we save the xmlstream, events and observer data to allow observer removal session_data["event_data"] = event_data = ( IBB_MESSAGE_DATA if stanza == "message" else IBB_IQ_DATA ).format(sid) session_data["observer_cb"] = observer_cb = self._onIBBData event_close = IBB_CLOSE.format(sid) # we now set the stream observer to look after data packet # FIXME: if we never get the events, the observers stay. # would be better to have generic observer and check id once triggered client.xmlstream.addObserver(event_data, observer_cb, client=client) client.xmlstream.addOnetimeObserver(event_close, self._onIBBClose, client=client) # finally, we send the accept stanza iq_result_elt = xmlstream.toResponse(iq_elt, "result") client.send(iq_result_elt) def _onIBBClose(self, iq_elt, client): """"Called when an IBB <close> element is received @param iq_elt(domish.Element): the whole <iq> stanza """ iq_elt.handled = True log.debug(_("IBB stream closing")) close_elt = iq_elt.elements(NS_IBB, "close").next() # XXX: this observer is only triggered on valid sid, so we don't need to check it sid = close_elt["sid"] iq_result_elt = xmlstream.toResponse(iq_elt, "result") client.send(iq_result_elt) self._killSession(sid, client) def _onIBBData(self, element, client): """Observer called on <iq> or <message> stanzas with data element Manage the data elelement (check validity and write to the stream_object) @param element(domish.Element): <iq> or <message> stanza """ element.handled = True data_elt = element.elements(NS_IBB, "data").next() sid = data_elt["sid"] try: session_data = client.xep_0047_current_stream[sid] except KeyError: log.warning(_(u"Received data for an unknown session id")) return self._sendError("item-not-found", None, element, client) from_jid = session_data["to"] stream_object = session_data["stream_object"] if from_jid.full() != element["from"]: log.warning( _( u"sended jid inconsistency (man in the middle attack attempt ?)\ninitial={initial}\ngiven={given}" ).format(initial=from_jid, given=element["from"]) ) if element.name == "iq": self._sendError("not-acceptable", sid, element, client) return session_data["seq"] = (session_data["seq"] + 1) % 65535 if int(data_elt.getAttribute("seq", -1)) != session_data["seq"]: log.warning(_(u"Sequence error")) if element.name == "iq": reason = "not-acceptable" self._sendError(reason, sid, element, client) self.terminateStream(session_data, client, reason) return # we reset the timeout: session_data["timer"].reset(TIMEOUT) # we can now decode the data try: stream_object.write(base64.b64decode(str(data_elt))) except TypeError: # The base64 data is invalid log.warning(_(u"Invalid base64 data")) if element.name == "iq": self._sendError("not-acceptable", sid, element, client) self.terminateStream(session_data, client, reason) return # we can now ack success if element.name == "iq": iq_result_elt = xmlstream.toResponse(element, "result") client.send(iq_result_elt) def _sendError(self, error_condition, sid, iq_elt, client): """Send error stanza @param error_condition: one of twisted.words.protocols.jabber.error.STANZA_CONDITIONS keys @param sid(unicode,None): jingle session id, or None, if session must not be destroyed @param iq_elt(domish.Element): full <iq> stanza @param client: %(doc_client)s """ iq_elt = error.StanzaError(error_condition).toResponse(iq_elt) log.warning( u"Error while managing in-band bytestream session, cancelling: {}".format( error_condition ) ) if sid is not None: self._killSession(sid, client, error_condition) client.send(iq_elt) def startStream(self, client, stream_object, to_jid, sid, block_size=None): """Launch the stream workflow @param stream_object(ifaces.IStreamProducer): stream object to send @param to_jid(jid.JID): JID of the recipient @param sid(unicode): Stream session id @param block_size(int, None): size of the block (or None for default) """ session_data = self._createSession(client, stream_object, to_jid, sid) if block_size is None: block_size = XEP_0047.BLOCK_SIZE assert block_size <= 65535 session_data["block_size"] = block_size iq_elt = client.IQ() iq_elt["to"] = to_jid.full() open_elt = iq_elt.addElement((NS_IBB, "open")) open_elt["block-size"] = str(block_size) open_elt["sid"] = sid open_elt["stanza"] = "iq" # TODO: manage <message> stanza ? args = [session_data, client] d = iq_elt.send() d.addCallbacks(self._IQDataStreamCb, self._IQDataStreamEb, args, None, args) return session_data[DEFER_KEY] def _IQDataStreamCb(self, iq_elt, session_data, client): """Called during the whole data streaming @param iq_elt(domish.Element): iq result @param session_data(dict): data of this streaming session @param client: %(doc_client)s """ session_data["timer"].reset(TIMEOUT) buffer_ = session_data["stream_object"].read(session_data["block_size"]) if buffer_: next_iq_elt = client.IQ() next_iq_elt["to"] = session_data["to"].full() data_elt = next_iq_elt.addElement((NS_IBB, "data")) seq = session_data["seq"] = (session_data["seq"] + 1) % 65535 data_elt["seq"] = unicode(seq) data_elt["sid"] = session_data["id"] data_elt.addContent(base64.b64encode(buffer_)) args = [session_data, client] d = next_iq_elt.send() d.addCallbacks(self._IQDataStreamCb, self._IQDataStreamEb, args, None, args) else: self.terminateStream(session_data, client) def _IQDataStreamEb(self, failure, session_data, client): if failure.check(error.StanzaError): log.warning(u"IBB transfer failed: {}".format(failure.value)) else: log.error(u"IBB transfer failed: {}".format(failure.value)) self.terminateStream(session_data, client, "IQ_ERROR") def terminateStream(self, session_data, client, failure_reason=None): """Terminate the stream session @param session_data(dict): data of this streaming session @param client: %(doc_client)s @param failure_reason(unicode, None): reason of the failure, or None if steam was successful """ iq_elt = client.IQ() iq_elt["to"] = session_data["to"].full() close_elt = iq_elt.addElement((NS_IBB, "close")) close_elt["sid"] = session_data["id"] iq_elt.send() self._killSession(session_data["id"], client, failure_reason) class XEP_0047_handler(XMPPHandler): implements(iwokkel.IDisco) def __init__(self, parent): self.plugin_parent = parent def connectionInitialized(self): self.xmlstream.addObserver( IBB_OPEN, self.plugin_parent._onIBBOpen, client=self.parent ) def getDiscoInfo(self, requestor, target, nodeIdentifier=""): return [disco.DiscoFeature(NS_IBB)] def getDiscoItems(self, requestor, target, nodeIdentifier=""): return []