libervia-backend: sat/plugins/plugin_sec

comparison sat/plugins/plugin_sec_otr.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	3d735e0ab2fa
children	fee60f17ebac

comparison

equal deleted inserted replaced

-:ff5bcb12ae60
+:ab2696e34d29
-#!/usr/bin/env python2
+#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 # SAT plugin for OTR encryption
 # Copyright (C) 2009-2019 Jérôme Poisson (goffi@goffi.org)
 # 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 sat.core.i18n import _, D_
 from sat.core.constants import Const as C
 from sat.core.log import getLogger
 from sat.core import exceptions
-log = getLogger(__name__)
 from sat.tools import xml_tools
 from twisted.words.protocols.jabber import jid
 from twisted.python import failure
 from twisted.internet import defer
 from sat.memory import persistent
 import potr
-import copy
-import time
+log = getLogger(__name__)
-import uuid
 PLUGIN_INFO = {
-C.PI_NAME: u"OTR",
+C.PI_NAME: "OTR",
-C.PI_IMPORT_NAME: u"OTR",
+C.PI_IMPORT_NAME: "OTR",
-C.PI_TYPE: u"SEC",
+C.PI_TYPE: "SEC",
-C.PI_PROTOCOLS: [u"XEP-0364"],
+C.PI_PROTOCOLS: ["XEP-0364"],
-C.PI_DEPENDENCIES: [u"XEP-0280", u"XEP-0334"],
+C.PI_DEPENDENCIES: ["XEP-0280", "XEP-0334"],
-C.PI_MAIN: u"OTR",
+C.PI_MAIN: "OTR",
-C.PI_HANDLER: u"no",
+C.PI_HANDLER: "no",
-C.PI_DESCRIPTION: _(u"""Implementation of OTR"""),
+C.PI_DESCRIPTION: _("""Implementation of OTR"""),
 }
 NS_OTR = "urn:xmpp:otr:0"
 PRIVATE_KEY = "PRIVATE KEY"
-OTR_MENU = D_(u"OTR")
+OTR_MENU = D_("OTR")
 AUTH_TXT = D_(
-u"To authenticate your correspondent, you need to give your below fingerprint "
+"To authenticate your correspondent, you need to give your below fingerprint "
-u"*BY AN EXTERNAL CANAL* (i.e. not in this chat), and check that the one he gives "
+"*BY AN EXTERNAL CANAL* (i.e. not in this chat), and check that the one he gives "
-u"you is the same as below. If there is a mismatch, there can be a spy between you!"
+"you is the same as below. If there is a mismatch, there can be a spy between you!"
 )
 DROP_TXT = D_(
-u"You private key is used to encrypt messages for your correspondent, nobody except "
+"You private key is used to encrypt messages for your correspondent, nobody except "
-u"you must know it, if you are in doubt, you should drop it!\n\nAre you sure you "
+"you must know it, if you are in doubt, you should drop it!\n\nAre you sure you "
-u"want to drop your private key?"
+"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_(u"Some of advanced features are disabled !")
+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"
 @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")
+msg = msg_str
 client = self.user.client
-log.debug(u"injecting encrypted message to {to}".format(to=self.peer))
+log.debug("injecting encrypted message to {to}".format(to=self.peer))
 if appdata is None:
 mess_data = {
 "from": client.jid,
 "to": self.peer,
-"uid": unicode(uuid.uuid4()),
+"uid": str(uuid.uuid4()),
 "message": {"": msg},
 "subject": {},
 "type": "chat",
 "extra": {},
 "timestamp": time.time(),
 }
 client.generateMessageXML(mess_data)
-xml = mess_data[u'xml']
+xml = mess_data['xml']
 self._p_carbons.setPrivate(xml)
 self._p_hints.addHintElements(xml, [
 self._p_hints.HINT_NO_COPY,
 self._p_hints.HINT_NO_PERMANENT_STORE])
 client.send(mess_data["xml"])
 else:
-message_elt = appdata[u"xml"]
+message_elt = appdata["xml"]
-assert message_elt.name == u"message"
+assert message_elt.name == "message"
 message_elt.addElement("body", content=msg)
 def stopCb(self, __, feedback):
 client = self.user.client
 self.host.bridge.otrState(
 client.feedback(self.peer, feedback)
 def stopEb(self, failure_):
 # encryption may be already stopped in case of manual stop
 if not failure_.check(exceptions.NotFound):
-log.error(u"Error while stopping OTR encryption: {msg}".format(msg=failure_))
+log.error("Error while stopping OTR encryption: {msg}".format(msg=failure_))
 def isTrusted(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 == u'trusted':
+elif trusted == 'trusted':
 return True
 else:
-log.error(u"Unexpected getCurrentTrust() value: {value}".format(
+log.error("Unexpected getCurrentTrust() value: {value}".format(
 value=trusted))
 return False
 def setState(self, state):
 client = self.user.client
 old_state = self.state
 super(Context, self).setState(state)
-log.debug(u"setState: %s (old_state=%s)" % (state, old_state))
+log.debug("setState: %s (old_state=%s)" % (state, old_state))
 if state == potr.context.STATE_PLAINTEXT:
-feedback = _(u"/!\\ conversation with %(other_jid)s is now UNENCRYPTED") % {
+feedback = _("/!\\ conversation with %(other_jid)s is now UNENCRYPTED") % {
 "other_jid": self.peer.full()
 }
 d = client.encryption.stop(self.peer, NS_OTR)
 d.addCallback(self.stopCb, feedback=feedback)
 d.addErrback(self.stopEb)
 client.encryption.start(self.peer, NS_OTR)
 try:
 trusted = self.isTrusted()
 except TypeError:
 trusted = False
-trusted_str = _(u"trusted") if trusted else _(u"untrusted")
+trusted_str = _("trusted") if trusted else _("untrusted")
 if old_state == potr.context.STATE_ENCRYPTED:
 feedback = D_(
-u"{trusted} OTR conversation with {other_jid} REFRESHED"
+"{trusted} OTR conversation with {other_jid} REFRESHED"
 ).format(trusted=trusted_str, other_jid=self.peer.full())
 else:
 feedback = D_(
-u"{trusted} encrypted OTR conversation started with {other_jid}\n"
+"{trusted} encrypted OTR conversation started with {other_jid}\n"
-u"{extra_info}"
+"{extra_info}"
 ).format(
 trusted=trusted_str,
 other_jid=self.peer.full(),
 extra_info=NO_ADV_FEATURES,
 )
 self.host.bridge.otrState(
 OTR_STATE_ENCRYPTED, self.peer.full(), client.profile
 )
 elif state == potr.context.STATE_FINISHED:
-feedback = D_(u"OTR conversation with {other_jid} is FINISHED").format(
+feedback = D_("OTR conversation with {other_jid} is FINISHED").format(
 other_jid=self.peer.full()
 )
 d = client.encryption.stop(self.peer, NS_OTR)
 d.addCallback(self.stopCb, feedback=feedback)
 d.addErrback(self.stopEb)
 return
 else:
-log.error(D_(u"Unknown OTR state"))
+log.error(D_("Unknown OTR state"))
 return
 client.feedback(self.peer, feedback)
 def disconnect(self):
 #       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(u"new account: %s" % client.jid)
+log.debug("new account: %s" % client.jid)
 if not client.jid.resource:
 log.warning("Account created without resource")
-super(Account, self).__init__(unicode(client.jid), "xmpp", 1024)
+super(Account, self).__init__(str(client.jid), "xmpp", 1024)
 self.host = host
 self.client = client
 def loadPrivkey(self):
-log.debug(u"loadPrivkey")
+log.debug("loadPrivkey")
 return self.privkey
 def savePrivkey(self):
-log.debug(u"savePrivkey")
+log.debug("savePrivkey")
 if self.privkey is None:
-raise exceptions.InternalError(_(u"Save is called but privkey is None !"))
+raise exceptions.InternalError(_("Save is called but privkey is None !"))
-priv_key = self.privkey.serializePrivateKey().encode("hex")
+priv_key = hexlify(self.privkey.serializePrivateKey())
 d = self.host.memory.encryptValue(priv_key, self.client.profile)
 def save_encrypted_key(encrypted_priv_key):
 self.client._otr_data[PRIVATE_KEY] = encrypted_priv_key
 d.addCallback(save_encrypted_key)
 def loadTrusts(self):
 trust_data = self.client._otr_data.get("trust", {})
-for jid_, jid_data in trust_data.iteritems():
+for jid_, jid_data in trust_data.items():
-for fingerprint, trust_level in jid_data.iteritems():
+for fingerprint, trust_level in jid_data.items():
 log.debug(
-u'setting trust for {jid}: [{fingerprint}] = "{trust_level}"'.format(
+'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 saveTrusts(self):
-log.debug(u"saving trusts for {profile}".format(profile=self.client.profile))
+log.debug("saving trusts for {profile}".format(profile=self.client.profile))
-log.debug(u"trusts = {}".format(self.client._otr_data["trust"]))
+log.debug("trusts = {}".format(self.client._otr_data["trust"]))
 self.client._otr_data.force("trust")
 def setTrust(self, other_jid, fingerprint, trustLevel):
 try:
 trust_data = self.client._otr_data["trust"]
 other_jid, Context(self, other_jid)
 )
 return context
 def getContextForUser(self, other):
-log.debug(u"getContextForUser [%s]" % other)
+log.debug("getContextForUser [%s]" % other)
 if not other.resource:
-log.warning(u"getContextForUser called with a bare jid: %s" % other.full())
+log.warning("getContextForUser called with a bare jid: %s" % other.full())
 return self.startContext(other)
 class OTR(object):
 def __init__(self, host):
-log.info(_(u"OTR plugin initialization"))
+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[u"XEP-0334"]
+self._p_hints = host.plugins["XEP-0334"]
-self._p_carbons = host.plugins[u"XEP-0280"]
+self._p_carbons = host.plugins["XEP-0280"]
 host.trigger.add("MessageReceived", self.MessageReceivedTrigger, priority=100000)
 host.trigger.add("sendMessage", self.sendMessageTrigger, priority=100000)
 host.trigger.add("sendMessageData", self._sendMessageDataTrigger)
 host.bridge.addMethod(
 "skipOTR", ".plugin", in_sign="s", out_sign="", method=self._skipOTR
 #     self._dropPrivKey,
 #     security_limit=0,
 #     type_=C.MENU_SINGLE,
 # )
 host.trigger.add("presence_received", self._presenceReceivedTrigger)
-self.host.registerEncryptionPlugin(self, u"OTR", NS_OTR, directed=True)
+self.host.registerEncryptionPlugin(self, "OTR", NS_OTR, directed=True)
 def _skipOTR(self, profile):
 """Tell the backend to not handle OTR for this profile.
 @param profile (str): %(doc_profile)s
 if encrypted_priv_key is not None:
 priv_key = yield self.host.memory.decryptValue(
 encrypted_priv_key, client.profile
 )
 ctxMng.account.privkey = potr.crypt.PK.parsePrivateKey(
-priv_key.decode("hex")
+unhexlify(priv_key.encode('utf-8'))
 )[0]
 else:
 ctxMng.account.privkey = None
 ctxMng.account.loadTrusts()
 def profileDisconnected(self, client):
 if client.profile in self.skipped_profiles:
 self.skipped_profiles.remove(client.profile)
 return
-for context in client._otr_context_manager.contexts.values():
+for context in list(client._otr_context_manager.contexts.values()):
 context.disconnect()
 del client._otr_context_manager
 # encryption plugin methods
 dialog = xml_tools.XMLUI(
 C.XMLUI_DIALOG,
 dialog_opt={
 C.XMLUI_DATA_TYPE: C.XMLUI_DIALOG_MESSAGE,
 C.XMLUI_DATA_MESS: _(
-u"You have no private key yet, start an OTR conversation to "
+"You have no private key yet, start an OTR conversation to "
-u"have one"
+"have one"
 ),
 C.XMLUI_DATA_LVL: C.XMLUI_DATA_LVL_WARNING,
 },
-title=_(u"No private key"),
+title=_("No private key"),
 )
 return dialog
 other_fingerprint = otrctx.getCurrentKey()
 dialog = xml_tools.XMLUI(
 C.XMLUI_DIALOG,
 dialog_opt={
 C.XMLUI_DATA_TYPE: C.XMLUI_DIALOG_MESSAGE,
 C.XMLUI_DATA_MESS: _(
-u"Your fingerprint is:\n{fingerprint}\n\n"
+"Your fingerprint is:\n{fingerprint}\n\n"
-u"Start an OTR conversation to have your correspondent one."
+"Start an OTR conversation to have your correspondent one."
 ).format(fingerprint=priv_key),
 C.XMLUI_DATA_LVL: C.XMLUI_DATA_LVL_INFO,
 },
-title=_(u"Fingerprint"),
+title=_("Fingerprint"),
 )
 return dialog
 def setTrust(raw_data, profile):
 if xml_tools.isXMLUICancelled(raw_data):
 return {}
 # This method is called when authentication form is submited
 data = xml_tools.XMLUIResult2DataFormResult(raw_data)
 if data["match"] == "yes":
 otrctx.setCurrentTrust(OTR_STATE_TRUSTED)
-note_msg = _(u"Your correspondent {correspondent} is now TRUSTED")
+note_msg = _("Your correspondent {correspondent} is now TRUSTED")
 self.host.bridge.otrState(
 OTR_STATE_TRUSTED, entity_jid.full(), client.profile
 )
 else:
 otrctx.setCurrentTrust("")
-note_msg = _(u"Your correspondent {correspondent} is now UNTRUSTED")
+note_msg = _("Your correspondent {correspondent} is now UNTRUSTED")
 self.host.bridge.otrState(
 OTR_STATE_UNTRUSTED, entity_jid.full(), client.profile
 )
 note = xml_tools.XMLUI(
 C.XMLUI_DIALOG,
 submit_id = self.host.registerCallback(setTrust, with_data=True, one_shot=True)
 trusted = otrctx.isTrusted()
 xmlui = xml_tools.XMLUI(
 C.XMLUI_FORM,
-title=_(u"Authentication ({entity_jid})").format(entity_jid=entity_jid.full()),
+title=_("Authentication ({entity_jid})").format(entity_jid=entity_jid.full()),
 submit_id=submit_id,
 )
 xmlui.addText(_(AUTH_TXT))
 xmlui.addDivider()
 xmlui.addText(
-D_(u"Your own fingerprint is:\n{fingerprint}").format(fingerprint=priv_key)
+D_("Your own fingerprint is:\n{fingerprint}").format(fingerprint=priv_key)
 )
 xmlui.addText(
-D_(u"Your correspondent fingerprint should be:\n{fingerprint}").format(
+D_("Your correspondent fingerprint should be:\n{fingerprint}").format(
 fingerprint=other_fingerprint
 )
 )
 xmlui.addDivider("blank")
 xmlui.changeContainer("pairs")
-xmlui.addLabel(D_(u"Is your correspondent fingerprint the same as here ?"))
+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
 """
 client = self.host.getClient(profile)
 try:
 to_jid = jid.JID(menu_data["jid"])
 except KeyError:
-log.error(_(u"jid key is not present !"))
+log.error(_("jid key is not present !"))
 return defer.fail(exceptions.DataError)
 self.startRefresh(client, to_jid)
 return {}
 def startRefresh(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[u'plugin'].namespace != NS_OTR:
+if encrypted_session and encrypted_session['plugin'].namespace != NS_OTR:
 raise exceptions.ConflictError(_(
-u"Can't start an OTR session, there is already an encrypted session "
+"Can't start an OTR session, there is already an encrypted session "
-u"with {name}").format(name=encrypted_session[u'plugin'].name))
+"with {name}").format(name=encrypted_session['plugin'].name))
 if not to_jid.resource:
 to_jid.resource = self.host.memory.getMainResource(
 client, to_jid
 )  # FIXME: temporary and unsecure, must be changed when frontends
 #        are refactored
 """
 client = self.host.getClient(profile)
 try:
 to_jid = jid.JID(menu_data["jid"])
 except KeyError:
-log.error(_(u"jid key is not present !"))
+log.error(_("jid key is not present !"))
 return defer.fail(exceptions.DataError)
 self.endSession(client, to_jid)
 return {}
 def endSession(self, client, to_jid):
 """
 client = self.host.getClient(profile)
 try:
 to_jid = jid.JID(menu_data["jid"])
 except KeyError:
-log.error(_(u"jid key is not present !"))
+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"""
 to_jid.resource = self.host.memory.getMainResource(
 client, to_jid
 )  # FIXME: temporary and unsecure, must be changed when frontends
 #        are refactored
 except KeyError:
-log.error(_(u"jid key is not present !"))
+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(_(u"You don't have a private key yet !")).toXml()
+"xmlui": xml_tools.note(_("You don't have a private key yet !")).toXml()
 }
 def dropKey(data, profile):
 if C.bool(data["answer"]):
 # we end all sessions
-for context in ctxMng.contexts.values():
+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_(u"Your private key has been dropped")
+D_("Your private key has been dropped")
 ).toXml()
 }
 return {}
 submit_id = self.host.registerCallback(dropKey, with_data=True, one_shot=True)
 confirm = xml_tools.XMLUI(
 C.XMLUI_DIALOG,
-title=_(u"Confirm private key drop"),
+title=_("Confirm private key drop"),
 dialog_opt={"type": C.XMLUI_DIALOG_CONFIRM, "message": _(DROP_TXT)},
 submit_id=submit_id,
 )
 return {"xmlui": confirm.toXml()}
 def _receivedTreatment(self, data, client):
 from_jid = data["from"]
-log.debug(u"_receivedTreatment [from_jid = %s]" % from_jid)
+log.debug("_receivedTreatment [from_jid = %s]" % from_jid)
 otrctx = client._otr_context_manager.getContextForUser(from_jid)
 try:
 message = (
-data["message"].itervalues().next()
+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:
 encrypted = False
 if otrctx.state == potr.context.STATE_ENCRYPTED:
 log.warning(
-u"Received unencrypted message in an encrypted context (from {jid})"
+"Received unencrypted message in an encrypted context (from {jid})"
 .format(jid=from_jid.full())
 )
 feedback = (
 D_(
-u"WARNING: received unencrypted data in a supposedly encrypted "
+"WARNING: received unencrypted data in a supposedly encrypted "
-u"context"
+"context"
 ),
 )
 client.feedback(from_jid, feedback)
 except potr.context.NotEncryptedError:
-msg = D_(u"WARNING: received OTR encrypted data in an unencrypted context")
+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_(u"WARNING: received OTR error message: {msg}".format(msg=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_(u"Error while trying de decrypt OTR message: {msg}".format(msg=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:
 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].decode("utf-8")
+"": 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[u"history"]
+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
 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 = (
-data["message"].itervalues().next().encode("utf-8")
+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[u"history"] = C.HISTORY_SKIP
+data["history"] = C.HISTORY_SKIP
 return data
 def MessageReceivedTrigger(self, client, message_elt, post_treat):
 if message_elt.getAttribute("type") == C.MESS_TYPE_GROUPCHAT:
 # OTR is not possible in group chats
 client, to_jid
 )  # FIXME: temporary and unsecure, must be changed when frontends
 otrctx = client._otr_context_manager.getContextForUser(to_jid)
 message_elt = mess_data["xml"]
 if otrctx.state == potr.context.STATE_ENCRYPTED:
-log.debug(u"encrypting message")
+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
 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(u"No message found")
+log.warning("No message found")
 else:
 self._p_carbons.setPrivate(message_elt)
 self._p_hints.addHintElements(message_elt, [
 self._p_hints.HINT_NO_COPY,
 self._p_hints.HINT_NO_PERMANENT_STORE])
-otrctx.sendMessage(0, unicode(body).encode("utf-8"), appdata=mess_data)
+otrctx.sendMessage(0, str(body).encode("utf-8"), appdata=mess_data)
 else:
 feedback = D_(
-u"Your message was not sent because your correspondent closed the "
+"Your message was not sent because your correspondent closed the "
-u"encrypted conversation on his/her side. "
+"encrypted conversation on his/her side. "
-u"Either close your own side, or refresh the session."
+"Either close your own side, or refresh the session."
 )
-log.warning(_(u"Message discarded because closed encryption channel"))
+log.warning(_("Message discarded because closed encryption channel"))
 client.feedback(to_jid, feedback)
-raise failure.Failure(exceptions.CancelError(u"Cancelled by OTR plugin"))
+raise failure.Failure(exceptions.CancelError("Cancelled by OTR plugin"))
 def sendMessageTrigger(self, client, mess_data, pre_xml_treatments,
 post_xml_treatments):
 if mess_data["type"] == "groupchat":
 return True

Mercurial > libervia-backend

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