Mercurial > libervia-backend
view libervia/backend/plugins/plugin_sec_otr.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # SAT plugin for OTR encryption # Copyright (C) 2009-2021 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/>. # XXX: thanks to Darrik L Mazey for his documentation # (https://blog.darmasoft.net/2013/06/30/using-pure-python-otr.html) # this implentation is based on it import copy import time import uuid from binascii import hexlify, unhexlify from libervia.backend.core.i18n import _, D_ from libervia.backend.core.constants import Const as C from libervia.backend.core.log import getLogger from libervia.backend.core import exceptions from libervia.backend.tools import xml_tools from twisted.words.protocols.jabber import jid from twisted.python import failure from twisted.internet import defer from libervia.backend.memory import persistent import potr log = getLogger(__name__) PLUGIN_INFO = { C.PI_NAME: "OTR", C.PI_IMPORT_NAME: "OTR", C.PI_MODES: [C.PLUG_MODE_CLIENT], C.PI_TYPE: "SEC", C.PI_PROTOCOLS: ["XEP-0364"], C.PI_DEPENDENCIES: ["XEP-0280", "XEP-0334"], C.PI_MAIN: "OTR", C.PI_HANDLER: "no", C.PI_DESCRIPTION: _("""Implementation of OTR"""), } NS_OTR = "urn:xmpp:otr:0" PRIVATE_KEY = "PRIVATE KEY" OTR_MENU = D_("OTR") AUTH_TXT = D_( "To authenticate your correspondent, you need to give your below fingerprint " "*BY AN EXTERNAL CANAL* (i.e. not in this chat), and check that the one he gives " "you is the same as below. If there is a mismatch, there can be a spy between you!" ) DROP_TXT = D_( "You private key is used to encrypt messages for your correspondent, nobody except " "you must know it, if you are in doubt, you should drop it!\n\nAre you sure you " "want to drop your private key?" ) # NO_LOG_AND = D_(u"/!\\Your history is not logged anymore, and") # FIXME: not used at the moment NO_ADV_FEATURES = D_("Some of advanced features are disabled !") DEFAULT_POLICY_FLAGS = {"ALLOW_V1": False, "ALLOW_V2": True, "REQUIRE_ENCRYPTION": True} OTR_STATE_TRUSTED = "trusted" OTR_STATE_UNTRUSTED = "untrusted" OTR_STATE_UNENCRYPTED = "unencrypted" OTR_STATE_ENCRYPTED = "encrypted" class Context(potr.context.Context): def __init__(self, context_manager, other_jid): self.context_manager = context_manager super(Context, self).__init__(context_manager.account, other_jid) @property def host(self): return self.context_manager.host @property def _p_hints(self): return self.context_manager.parent._p_hints @property def _p_carbons(self): return self.context_manager.parent._p_carbons def get_policy(self, key): if key in DEFAULT_POLICY_FLAGS: return DEFAULT_POLICY_FLAGS[key] else: return False def inject(self, msg_str, appdata=None): """Inject encrypted data in the stream if appdata is not None, we are sending a message in sendMessageDataTrigger stanza will be injected directly if appdata is None, else we just update the element and follow normal workflow @param msg_str(str): encrypted message body @param appdata(None, dict): None for signal message, message data when an encrypted message is going to be sent """ assert isinstance(self.peer, jid.JID) msg = msg_str.decode("utf-8") client = self.user.client log.debug("injecting encrypted message to {to}".format(to=self.peer)) if appdata is None: mess_data = { "from": client.jid, "to": self.peer, "uid": str(uuid.uuid4()), "message": {"": msg}, "subject": {}, "type": "chat", "extra": {}, "timestamp": time.time(), } client.generate_message_xml(mess_data) xml = mess_data["xml"] self._p_carbons.set_private(xml) self._p_hints.add_hint_elements( xml, [self._p_hints.HINT_NO_COPY, self._p_hints.HINT_NO_PERMANENT_STORE] ) client.send(mess_data["xml"]) else: message_elt = appdata["xml"] assert message_elt.name == "message" message_elt.addElement("body", content=msg) def stop_cb(self, __, feedback): client = self.user.client self.host.bridge.otr_state( OTR_STATE_UNENCRYPTED, self.peer.full(), client.profile ) client.feedback(self.peer, feedback) def stop_eb(self, failure_): # encryption may be already stopped in case of manual stop if not failure_.check(exceptions.NotFound): log.error("Error while stopping OTR encryption: {msg}".format(msg=failure_)) def is_trusted(self): # we have to check value because potr code says that a 2-tuples should be # returned while in practice it's either None or u"trusted" trusted = self.getCurrentTrust() if trusted is None: return False elif trusted == "trusted": return True else: log.error("Unexpected getCurrentTrust() value: {value}".format(value=trusted)) return False def set_state(self, state): client = self.user.client old_state = self.state super(Context, self).set_state(state) log.debug("set_state: %s (old_state=%s)" % (state, old_state)) if state == potr.context.STATE_PLAINTEXT: feedback = _("/!\\ conversation with %(other_jid)s is now UNENCRYPTED") % { "other_jid": self.peer.full() } d = defer.ensureDeferred(client.encryption.stop(self.peer, NS_OTR)) d.addCallback(self.stop_cb, feedback=feedback) d.addErrback(self.stop_eb) return elif state == potr.context.STATE_ENCRYPTED: defer.ensureDeferred(client.encryption.start(self.peer, NS_OTR)) try: trusted = self.is_trusted() except TypeError: trusted = False trusted_str = _("trusted") if trusted else _("untrusted") if old_state == potr.context.STATE_ENCRYPTED: feedback = D_( "{trusted} OTR conversation with {other_jid} REFRESHED" ).format(trusted=trusted_str, other_jid=self.peer.full()) else: feedback = D_( "{trusted} encrypted OTR conversation started with {other_jid}\n" "{extra_info}" ).format( trusted=trusted_str, other_jid=self.peer.full(), extra_info=NO_ADV_FEATURES, ) self.host.bridge.otr_state( OTR_STATE_ENCRYPTED, self.peer.full(), client.profile ) elif state == potr.context.STATE_FINISHED: feedback = D_("OTR conversation with {other_jid} is FINISHED").format( other_jid=self.peer.full() ) d = defer.ensureDeferred(client.encryption.stop(self.peer, NS_OTR)) d.addCallback(self.stop_cb, feedback=feedback) d.addErrback(self.stop_eb) return else: log.error(D_("Unknown OTR state")) return client.feedback(self.peer, feedback) def disconnect(self): """Disconnect the session.""" if self.state != potr.context.STATE_PLAINTEXT: super(Context, self).disconnect() def finish(self): """Finish the session avoid to send any message but the user still has to end the session himself. """ if self.state == potr.context.STATE_ENCRYPTED: self.processTLVs([potr.proto.DisconnectTLV()]) class Account(potr.context.Account): # TODO: manage trusted keys: if a fingerprint is not used anymore, # we have no way to remove it from database yet (same thing for a # correspondent jid) # TODO: manage explicit message encryption def __init__(self, host, client): log.debug("new account: %s" % client.jid) if not client.jid.resource: log.warning("Account created without resource") super(Account, self).__init__(str(client.jid), "xmpp", 1024) self.host = host self.client = client def load_privkey(self): log.debug("load_privkey") return self.privkey def save_privkey(self): log.debug("save_privkey") if self.privkey is None: raise exceptions.InternalError(_("Save is called but privkey is None !")) priv_key = hexlify(self.privkey.serializePrivateKey()) encrypted_priv_key = self.host.memory.encrypt_value(priv_key, self.client.profile) self.client._otr_data[PRIVATE_KEY] = encrypted_priv_key def load_trusts(self): trust_data = self.client._otr_data.get("trust", {}) for jid_, jid_data in trust_data.items(): for fingerprint, trust_level in jid_data.items(): log.debug( 'setting trust for {jid}: [{fingerprint}] = "{trust_level}"'.format( jid=jid_, fingerprint=fingerprint, trust_level=trust_level ) ) self.trusts.setdefault(jid.JID(jid_), {})[fingerprint] = trust_level def save_trusts(self): log.debug("saving trusts for {profile}".format(profile=self.client.profile)) log.debug("trusts = {}".format(self.client._otr_data["trust"])) self.client._otr_data.force("trust") def set_trust(self, other_jid, fingerprint, trustLevel): try: trust_data = self.client._otr_data["trust"] except KeyError: trust_data = {} self.client._otr_data["trust"] = trust_data jid_data = trust_data.setdefault(other_jid.full(), {}) jid_data[fingerprint] = trustLevel super(Account, self).set_trust(other_jid, fingerprint, trustLevel) class ContextManager(object): def __init__(self, parent, client): self.parent = parent self.account = Account(parent.host, client) self.contexts = {} @property def host(self): return self.parent.host def start_context(self, other_jid): assert isinstance(other_jid, jid.JID) context = self.contexts.setdefault(other_jid, Context(self, other_jid)) return context def get_context_for_user(self, other): log.debug("get_context_for_user [%s]" % other) if not other.resource: log.warning("get_context_for_user called with a bare jid: %s" % other.full()) return self.start_context(other) class OTR(object): def __init__(self, host): log.info(_("OTR plugin initialization")) self.host = host self.context_managers = {} self.skipped_profiles = ( set() ) # FIXME: OTR should not be skipped per profile, this need to be refactored self._p_hints = host.plugins["XEP-0334"] self._p_carbons = host.plugins["XEP-0280"] host.trigger.add( "message_received", self.message_received_trigger, priority=100000 ) host.trigger.add("sendMessage", self.send_message_trigger, priority=100000) host.trigger.add("send_message_data", self._send_message_data_trigger) host.bridge.add_method( "skip_otr", ".plugin", in_sign="s", out_sign="", method=self._skip_otr ) # FIXME: must be removed, must be done on per-message basis host.bridge.add_signal( "otr_state", ".plugin", signature="sss" ) # args: state, destinee_jid, profile # XXX: menus are disabled in favor to the new more generic encryption menu # there are let here commented for a little while as a reference # host.import_menu( # (OTR_MENU, D_(u"Start/Refresh")), # self._otr_start_refresh, # security_limit=0, # help_string=D_(u"Start or refresh an OTR session"), # type_=C.MENU_SINGLE, # ) # host.import_menu( # (OTR_MENU, D_(u"End session")), # self._otr_session_end, # security_limit=0, # help_string=D_(u"Finish an OTR session"), # type_=C.MENU_SINGLE, # ) # host.import_menu( # (OTR_MENU, D_(u"Authenticate")), # self._otr_authenticate, # security_limit=0, # help_string=D_(u"Authenticate user/see your fingerprint"), # type_=C.MENU_SINGLE, # ) # host.import_menu( # (OTR_MENU, D_(u"Drop private key")), # self._drop_priv_key, # security_limit=0, # type_=C.MENU_SINGLE, # ) host.trigger.add("presence_received", self._presence_received_trigger) self.host.register_encryption_plugin(self, "OTR", NS_OTR, directed=True) def _skip_otr(self, profile): """Tell the backend to not handle OTR for this profile. @param profile (str): %(doc_profile)s """ # FIXME: should not be done per profile but per message, using extra data # for message received, profile wide hook may be need, but client # should be used anyway instead of a class attribute self.skipped_profiles.add(profile) @defer.inlineCallbacks def profile_connecting(self, client): if client.profile in self.skipped_profiles: return ctxMng = client._otr_context_manager = ContextManager(self, client) client._otr_data = persistent.PersistentBinaryDict(NS_OTR, client.profile) yield client._otr_data.load() encrypted_priv_key = client._otr_data.get(PRIVATE_KEY, None) if encrypted_priv_key is not None: priv_key = self.host.memory.decrypt_value(encrypted_priv_key, client.profile) ctxMng.account.privkey = potr.crypt.PK.parsePrivateKey( unhexlify(priv_key.encode("utf-8")) )[0] else: ctxMng.account.privkey = None ctxMng.account.load_trusts() def profile_disconnected(self, client): if client.profile in self.skipped_profiles: self.skipped_profiles.remove(client.profile) return for context in list(client._otr_context_manager.contexts.values()): context.disconnect() del client._otr_context_manager # encryption plugin methods def start_encryption(self, client, entity_jid): self.start_refresh(client, entity_jid) def stop_encryption(self, client, entity_jid): self.end_session(client, entity_jid) def get_trust_ui(self, client, entity_jid): if not entity_jid.resource: entity_jid.resource = self.host.memory.main_resource_get( client, entity_jid ) # FIXME: temporary and unsecure, must be changed when frontends # are refactored ctxMng = client._otr_context_manager otrctx = ctxMng.get_context_for_user(entity_jid) priv_key = ctxMng.account.privkey if priv_key is None: # we have no private key yet dialog = xml_tools.XMLUI( C.XMLUI_DIALOG, dialog_opt={ C.XMLUI_DATA_TYPE: C.XMLUI_DIALOG_MESSAGE, C.XMLUI_DATA_MESS: _( "You have no private key yet, start an OTR conversation to " "have one" ), C.XMLUI_DATA_LVL: C.XMLUI_DATA_LVL_WARNING, }, title=_("No private key"), ) return dialog other_fingerprint = otrctx.getCurrentKey() if other_fingerprint is None: # we have a private key, but not the fingerprint of our correspondent dialog = xml_tools.XMLUI( C.XMLUI_DIALOG, dialog_opt={ C.XMLUI_DATA_TYPE: C.XMLUI_DIALOG_MESSAGE, C.XMLUI_DATA_MESS: _( "Your fingerprint is:\n{fingerprint}\n\n" "Start an OTR conversation to have your correspondent one." ).format(fingerprint=priv_key), C.XMLUI_DATA_LVL: C.XMLUI_DATA_LVL_INFO, }, title=_("Fingerprint"), ) return dialog def set_trust(raw_data, profile): if xml_tools.is_xmlui_cancelled(raw_data): return {} # This method is called when authentication form is submited data = xml_tools.xmlui_result_2_data_form_result(raw_data) if data["match"] == "yes": otrctx.setCurrentTrust(OTR_STATE_TRUSTED) note_msg = _("Your correspondent {correspondent} is now TRUSTED") self.host.bridge.otr_state( OTR_STATE_TRUSTED, entity_jid.full(), client.profile ) else: otrctx.setCurrentTrust("") note_msg = _("Your correspondent {correspondent} is now UNTRUSTED") self.host.bridge.otr_state( OTR_STATE_UNTRUSTED, entity_jid.full(), client.profile ) note = xml_tools.XMLUI( C.XMLUI_DIALOG, dialog_opt={ C.XMLUI_DATA_TYPE: C.XMLUI_DIALOG_NOTE, C.XMLUI_DATA_MESS: note_msg.format(correspondent=otrctx.peer), }, ) return {"xmlui": note.toXml()} submit_id = self.host.register_callback(set_trust, with_data=True, one_shot=True) trusted = otrctx.is_trusted() xmlui = xml_tools.XMLUI( C.XMLUI_FORM, title=_("Authentication ({entity_jid})").format(entity_jid=entity_jid.full()), submit_id=submit_id, ) xmlui.addText(_(AUTH_TXT)) xmlui.addDivider() xmlui.addText( D_("Your own fingerprint is:\n{fingerprint}").format(fingerprint=priv_key) ) xmlui.addText( D_("Your correspondent fingerprint should be:\n{fingerprint}").format( fingerprint=other_fingerprint ) ) xmlui.addDivider("blank") xmlui.change_container("pairs") xmlui.addLabel(D_("Is your correspondent fingerprint the same as here ?")) xmlui.addList( "match", [("yes", _("yes")), ("no", _("no"))], ["yes" if trusted else "no"] ) return xmlui def _otr_start_refresh(self, menu_data, profile): """Start or refresh an OTR session @param menu_data: %(menu_data)s @param profile: %(doc_profile)s """ client = self.host.get_client(profile) try: to_jid = jid.JID(menu_data["jid"]) except KeyError: log.error(_("jid key is not present !")) return defer.fail(exceptions.DataError) self.start_refresh(client, to_jid) return {} def start_refresh(self, client, to_jid): """Start or refresh an OTR session @param to_jid(jid.JID): jid to start encrypted session with """ encrypted_session = client.encryption.getSession(to_jid.userhostJID()) if encrypted_session and encrypted_session["plugin"].namespace != NS_OTR: raise exceptions.ConflictError( _( "Can't start an OTR session, there is already an encrypted session " "with {name}" ).format(name=encrypted_session["plugin"].name) ) if not to_jid.resource: to_jid.resource = self.host.memory.main_resource_get( client, to_jid ) # FIXME: temporary and unsecure, must be changed when frontends # are refactored otrctx = client._otr_context_manager.get_context_for_user(to_jid) query = otrctx.sendMessage(0, b"?OTRv?") otrctx.inject(query) def _otr_session_end(self, menu_data, profile): """End an OTR session @param menu_data: %(menu_data)s @param profile: %(doc_profile)s """ client = self.host.get_client(profile) try: to_jid = jid.JID(menu_data["jid"]) except KeyError: log.error(_("jid key is not present !")) return defer.fail(exceptions.DataError) self.end_session(client, to_jid) return {} def end_session(self, client, to_jid): """End an OTR session""" if not to_jid.resource: to_jid.resource = self.host.memory.main_resource_get( client, to_jid ) # FIXME: temporary and unsecure, must be changed when frontends # are refactored otrctx = client._otr_context_manager.get_context_for_user(to_jid) otrctx.disconnect() return {} def _otr_authenticate(self, menu_data, profile): """End an OTR session @param menu_data: %(menu_data)s @param profile: %(doc_profile)s """ client = self.host.get_client(profile) try: to_jid = jid.JID(menu_data["jid"]) except KeyError: log.error(_("jid key is not present !")) return defer.fail(exceptions.DataError) return self.authenticate(client, to_jid) def authenticate(self, client, to_jid): """Authenticate other user and see our own fingerprint""" xmlui = self.get_trust_ui(client, to_jid) return {"xmlui": xmlui.toXml()} def _drop_priv_key(self, menu_data, profile): """Drop our private Key @param menu_data: %(menu_data)s @param profile: %(doc_profile)s """ client = self.host.get_client(profile) try: to_jid = jid.JID(menu_data["jid"]) if not to_jid.resource: to_jid.resource = self.host.memory.main_resource_get( client, to_jid ) # FIXME: temporary and unsecure, must be changed when frontends # are refactored except KeyError: log.error(_("jid key is not present !")) return defer.fail(exceptions.DataError) ctxMng = client._otr_context_manager if ctxMng.account.privkey is None: return { "xmlui": xml_tools.note(_("You don't have a private key yet !")).toXml() } def drop_key(data, profile): if C.bool(data["answer"]): # we end all sessions for context in list(ctxMng.contexts.values()): context.disconnect() ctxMng.account.privkey = None ctxMng.account.getPrivkey() # as account.privkey is None, getPrivkey # will generate a new key, and save it return { "xmlui": xml_tools.note( D_("Your private key has been dropped") ).toXml() } return {} submit_id = self.host.register_callback(drop_key, with_data=True, one_shot=True) confirm = xml_tools.XMLUI( C.XMLUI_DIALOG, title=_("Confirm private key drop"), dialog_opt={"type": C.XMLUI_DIALOG_CONFIRM, "message": _(DROP_TXT)}, submit_id=submit_id, ) return {"xmlui": confirm.toXml()} def _received_treatment(self, data, client): from_jid = data["from"] log.debug("_received_treatment [from_jid = %s]" % from_jid) otrctx = client._otr_context_manager.get_context_for_user(from_jid) try: message = next( iter(data["message"].values()) ) # FIXME: Q&D fix for message refactoring, message is now a dict res = otrctx.receiveMessage(message.encode("utf-8")) except (potr.context.UnencryptedMessage, potr.context.NotOTRMessage): # potr has a bug with Python 3 and test message against str while bytes are # expected, resulting in a NoOTRMessage raised instead of UnencryptedMessage; # so we catch NotOTRMessage as a workaround # TODO: report this upstream encrypted = False if otrctx.state == potr.context.STATE_ENCRYPTED: log.warning( "Received unencrypted message in an encrypted context (from {jid})".format( jid=from_jid.full() ) ) feedback = ( D_( "WARNING: received unencrypted data in a supposedly encrypted " "context" ), ) client.feedback(from_jid, feedback) except potr.context.NotEncryptedError: msg = D_("WARNING: received OTR encrypted data in an unencrypted context") log.warning(msg) feedback = msg client.feedback(from_jid, msg) raise failure.Failure(exceptions.CancelError(msg)) except potr.context.ErrorReceived as e: msg = D_("WARNING: received OTR error message: {msg}".format(msg=e)) log.warning(msg) feedback = msg client.feedback(from_jid, msg) raise failure.Failure(exceptions.CancelError(msg)) except potr.crypt.InvalidParameterError as e: msg = D_("Error while trying de decrypt OTR message: {msg}".format(msg=e)) log.warning(msg) feedback = msg client.feedback(from_jid, msg) raise failure.Failure(exceptions.CancelError(msg)) except StopIteration: return data else: encrypted = True if encrypted: if res[0] != None: # decrypted messages handling. # receiveMessage() will return a tuple, # the first part of which will be the decrypted message data["message"] = { "": res[0] } # FIXME: Q&D fix for message refactoring, message is now a dict try: # we want to keep message in history, even if no store is # requested in message hints del data["history"] except KeyError: pass # TODO: add skip history as an option, but by default we don't skip it # data[u'history'] = C.HISTORY_SKIP # we send the decrypted message to # frontends, but we don't want it in # history else: raise failure.Failure( exceptions.CancelError("Cancelled by OTR") ) # no message at all (no history, no signal) client.encryption.mark_as_encrypted(data, namespace=NS_OTR) trusted = otrctx.is_trusted() if trusted: client.encryption.mark_as_trusted(data) else: client.encryption.mark_as_untrusted(data) return data def _received_treatment_for_skipped_profiles(self, data): """This profile must be skipped because the frontend manages OTR itself, but we still need to check if the message must be stored in history or not """ # XXX: FIXME: this should not be done on a per-profile basis, but per-message try: message = ( iter(data["message"].values()).next().encode("utf-8") ) # FIXME: Q&D fix for message refactoring, message is now a dict except StopIteration: return data if message.startswith(potr.proto.OTRTAG): # FIXME: it may be better to cancel the message and send it direclty to # bridge # this is used by Libervia, but this may send garbage message to # other frontends # if they are used at the same time as Libervia. # Hard to avoid with decryption on Libervia though. data["history"] = C.HISTORY_SKIP return data def message_received_trigger(self, client, message_elt, post_treat): if client.is_component: return True if message_elt.getAttribute("type") == C.MESS_TYPE_GROUPCHAT: # OTR is not possible in group chats return True from_jid = jid.JID(message_elt["from"]) if not from_jid.resource or from_jid.userhostJID() == client.jid.userhostJID(): # OTR is only usable when resources are present return True if client.profile in self.skipped_profiles: post_treat.addCallback(self._received_treatment_for_skipped_profiles) else: post_treat.addCallback(self._received_treatment, client) return True def _send_message_data_trigger(self, client, mess_data): if client.is_component: return True encryption = mess_data.get(C.MESS_KEY_ENCRYPTION) if encryption is None or encryption["plugin"].namespace != NS_OTR: return to_jid = mess_data["to"] if not to_jid.resource: to_jid.resource = self.host.memory.main_resource_get( client, to_jid ) # FIXME: temporary and unsecure, must be changed when frontends otrctx = client._otr_context_manager.get_context_for_user(to_jid) message_elt = mess_data["xml"] if otrctx.state == potr.context.STATE_ENCRYPTED: log.debug("encrypting message") body = None for child in list(message_elt.children): if child.name == "body": # we remove all unencrypted body, # and will only encrypt the first one if body is None: body = child message_elt.children.remove(child) elif child.name == "html": # we don't want any XHTML-IM element message_elt.children.remove(child) if body is None: log.warning("No message found") else: self._p_carbons.set_private(message_elt) self._p_hints.add_hint_elements( message_elt, [self._p_hints.HINT_NO_COPY, self._p_hints.HINT_NO_PERMANENT_STORE], ) otrctx.sendMessage(0, str(body).encode("utf-8"), appdata=mess_data) else: feedback = D_( "Your message was not sent because your correspondent closed the " "encrypted conversation on his/her side. " "Either close your own side, or refresh the session." ) log.warning(_("Message discarded because closed encryption channel")) client.feedback(to_jid, feedback) raise failure.Failure(exceptions.CancelError("Cancelled by OTR plugin")) def send_message_trigger( self, client, mess_data, pre_xml_treatments, post_xml_treatments ): if client.is_component: return True if mess_data["type"] == "groupchat": return True if client.profile in self.skipped_profiles: # FIXME: should not be done on a per-profile basis return True to_jid = copy.copy(mess_data["to"]) if client.encryption.getSession(to_jid.userhostJID()): # there is already an encrypted session with this entity return True if not to_jid.resource: to_jid.resource = self.host.memory.main_resource_get( client, to_jid ) # FIXME: full jid may not be known otrctx = client._otr_context_manager.get_context_for_user(to_jid) if otrctx.state != potr.context.STATE_PLAINTEXT: defer.ensureDeferred(client.encryption.start(to_jid, NS_OTR)) client.encryption.set_encryption_flag(mess_data) if not mess_data["to"].resource: # if not resource was given, we force it here mess_data["to"] = to_jid return True def _presence_received_trigger(self, client, entity, show, priority, statuses): if show != C.PRESENCE_UNAVAILABLE: return True if not entity.resource: try: entity.resource = self.host.memory.main_resource_get( client, entity ) # FIXME: temporary and unsecure, must be changed when frontends # are refactored except exceptions.UnknownEntityError: return True # entity was not connected if entity in client._otr_context_manager.contexts: otrctx = client._otr_context_manager.get_context_for_user(entity) otrctx.disconnect() return True