libervia-backend: sat/plugins/plugin_sec

comparison sat/plugins/plugin_sec_otr.py @ 2643:189e38fb11ff

core: style improvments (90 chars limit)

author	Goffi <goffi@goffi.org>
date	Sun, 29 Jul 2018 18:44:27 +0200
parents	56f94936df1e
children	7213caa5c5d0

comparison

equal deleted inserted replaced

-:755a0b8643bd
+:189e38fb11ff
 # 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)
+# 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
 from sat.core.i18n import _, D_
 from sat.core.constants import Const as C
 from sat.core.log import getLogger
 NS_OTR = "otr_plugin"
 PRIVATE_KEY = "PRIVATE KEY"
 OTR_MENU = D_(u"OTR")
 AUTH_TXT = D_(
-u"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!"
+u"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 "
+u"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 must know it, if you are in doubt, you should drop it!\n\nAre you sure you want to drop your private key?"
+u"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 "
+u"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 !")
 DEFAULT_POLICY_FLAGS = {"ALLOW_V1": False, "ALLOW_V2": True, "REQUIRE_ENCRYPTION": True}
 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
+stanza will be injected directly if appdata is None,
-and follow normal workflow
+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)
 feedback = D_(
 u"{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{extra_info}"
+u"{trusted} encrypted OTR conversation started with {other_jid}\n"
+u"{extra_info}"
 ).format(
 trusted=trusted_str,
 other_jid=self.peer.full(),
 extra_info=NO_ADV_FEATURES,
 )
 """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."""
+"""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 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(u"new account: %s" % client.jid)
 if not client.jid.resource:
 @param to_jid(jid.JID): jid to start encrypted session with
 """
 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
+)  # FIXME: temporary and unsecure, must be changed when frontends
+#        are refactored
 otrctx = client._otr_context_manager.getContextForUser(to_jid)
 query = otrctx.sendMessage(0, "?OTRv?")
 otrctx.inject(query)
 def _otrSessionEnd(self, menu_data, profile):
 def endSession(self, client, to_jid):
 """End an OTR session"""
 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
+)  # FIXME: temporary and unsecure, must be changed when frontends
+#        are refactored
 otrctx = client._otr_context_manager.getContextForUser(to_jid)
 otrctx.disconnect()
 return {}
 def _otrAuthenticate(self, menu_data, profile):
 def authenticate(self, client, to_jid):
 """Authenticate other user and see our own fingerprint"""
 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
+)  # FIXME: temporary and unsecure, must be changed when frontends
+#        are refactored
 ctxMng = client._otr_context_manager
 otrctx = ctxMng.getContextForUser(to_jid)
 priv_key = ctxMng.account.privkey
 if priv_key is None:
 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 have one"
+u"You have no private key yet, start an OTR conversation to "
+u"have one"
 ),
 C.XMLUI_DATA_LVL: C.XMLUI_DATA_LVL_WARNING,
 },
 title=_(u"No private key"),
 )
 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\nStart an OTR conversation to have your correspondent one."
+u"Your fingerprint is:\n{fingerprint}\n\n"
+u"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"),
 )
 try:
 to_jid = jid.JID(menu_data["jid"])
 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
+)  # FIXME: temporary and unsecure, must be changed when frontends
+#        are refactored
 except KeyError:
 log.error(_(u"jid key is not present !"))
 return defer.fail(exceptions.DataError)
 ctxMng = client._otr_context_manager
 if C.bool(data["answer"]):
 # we end all sessions
 for context in 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
+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")
 ).toXml()
 }
 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})".format(
+u"Received unencrypted message in an encrypted context (from {jid})"
-jid=from_jid.full()
+.format(jid=from_jid.full())
-)
 )
 feedback = (
 D_(
-u"WARNING: received unencrypted data in a supposedly encrypted context"
+u"WARNING: received unencrypted data in a supposedly encrypted "
+u"context"
 ),
 )
 client.feedback(from_jid, feedback)
 except StopIteration:
 return data
 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
+# receiveMessage() will return a tuple,
+# the first part of which will be the decrypted message
 data["message"] = {
 "": res[0].decode("utf-8")
 }  # 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
+# we want to keep message in history, even if no store is
+# requested in message hints
 del data[u"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
+# data[u'history'] = C.HISTORY_SKIP # we send the decrypted message to
+# frontends, but we don't want it in
+# history
 else:
 log.warning(
 u"An encrypted message was expected, but got {}".format(
 data["message"]
 )
 data["message"].itervalues().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
+#  FIXME: it may be better to cancel the message and send it direclty to
-#        this is used by Libervia, but this may send garbage message to other frontends
+#         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
 return data
 else:
 self._p_carbons.setPrivate(message_elt)
 otrctx.sendMessage(0, unicode(body).encode("utf-8"), appdata=mess_data)
 else:
 feedback = D_(
-u"Your message was not sent because your correspondent closed the encrypted conversation on his/her side. "
+u"Your message was not sent because your correspondent closed the "
+u"encrypted conversation on his/her side. "
 u"Either close your own side, or refresh the session."
 )
 log.warning(_(u"Message discarded because closed encryption channel"))
 client.feedback(to_jid, feedback)
 raise failure.Failure(exceptions.CancelError(u"Cancelled by OTR plugin"))
-def sendMessageTrigger(
+def sendMessageTrigger(self, client, mess_data, pre_xml_treatments,
-self, client, mess_data, pre_xml_treatments, post_xml_treatments
+post_xml_treatments):
-):
 if mess_data["type"] == "groupchat":
 return True
-if (
-client.profile in self.skipped_profiles
+if client.profile in self.skipped_profiles:
-):  #  FIXME: should not be done on a per-profile basis
+#  FIXME: should not be done on a per-profile basis
 return True
 to_jid = copy.copy(mess_data["to"])
 if not to_jid.resource:
 to_jid.resource = self.host.memory.getMainResource(
 client, to_jid
 )  # FIXME: full jid may not be known
 otrctx = client._otr_context_manager.getContextForUser(to_jid)
 if otrctx.state != potr.context.STATE_PLAINTEXT:
 self._p_hints.addHint(mess_data, self._p_hints.HINT_NO_COPY)
 self._p_hints.addHint(mess_data, self._p_hints.HINT_NO_PERMANENT_STORE)
-mess_data[
+mess_data["OTR"] = (otrctx)  #  this indicate that encryption is needed in
-"OTR"
+#  sendMessageData trigger
-] = (
+if not mess_data["to"].resource:
-otrctx
+# if not resource was given, we force it here
-)  #  this indicate that encryption is needed in sendMessageData trigger
-if not mess_data[
-"to"
-].resource:  #  if not resource was given, we force it here
 mess_data["to"] = to_jid
 return True
 def _presenceReceivedTrigger(self, entity, show, priority, statuses, profile):
 if show != C.PRESENCE_UNAVAILABLE:
 client = self.host.getClient(profile)
 if not entity.resource:
 try:
 entity.resource = self.host.memory.getMainResource(
 client, entity
-)  # FIXME: temporary and unsecure, must be changed when frontends are refactored
+)  # 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.getContextForUser(entity)
 otrctx.disconnect()

Mercurial > libervia-backend

comparison sat/plugins/plugin_sec_otr.py @ 2643:189e38fb11ff