libervia-backend: libervia/backend/plugins/plugin_sec

comparison libervia/backend/plugins/plugin_sec_otr.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	4b842c1fb686
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 @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.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,
 "type": "chat",
 "extra": {},
 "timestamp": time.time(),
 }
 client.generate_message_xml(mess_data)
-xml = mess_data['xml']
+xml = mess_data["xml"]
 self._p_carbons.set_private(xml)
-self._p_hints.add_hint_elements(xml, [
+self._p_hints.add_hint_elements(
-self._p_hints.HINT_NO_COPY,
+xml, [self._p_hints.HINT_NO_COPY, self._p_hints.HINT_NO_PERMANENT_STORE]
-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)
 # 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':
+elif trusted == "trusted":
 return True
 else:
-log.error("Unexpected getCurrentTrust() value: {value}".format(
+log.error("Unexpected getCurrentTrust() value: {value}".format(value=trusted))
-value=trusted))
 return False
 def set_state(self, state):
 client = self.user.client
 old_state = self.state
 def host(self):
 return self.parent.host
 def start_context(self, other_jid):
 assert isinstance(other_jid, jid.JID)
-context = self.contexts.setdefault(
+context = self.contexts.setdefault(other_jid, Context(self, other_jid))
-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:
 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(
+"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
 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(
+priv_key = self.host.memory.decrypt_value(encrypted_priv_key, client.profile)
-encrypted_priv_key, client.profile
-)
 ctxMng.account.privkey = potr.crypt.PK.parsePrivateKey(
-unhexlify(priv_key.encode('utf-8'))
+unhexlify(priv_key.encode("utf-8"))
 )[0]
 else:
 ctxMng.account.privkey = None
 ctxMng.account.load_trusts()
 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:
 """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:
+if encrypted_session and encrypted_session["plugin"].namespace != NS_OTR:
-raise exceptions.ConflictError(_(
+raise exceptions.ConflictError(
-"Can't start an OTR session, there is already an encrypted session "
+_(
-"with {name}").format(name=encrypted_session['plugin'].name))
+"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"""
 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):
 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
 # 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()
 }
 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 = (
+message = next(
-next(iter(data["message"].values()))
+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})"
+"Received unencrypted message in an encrypted context (from {jid})".format(
-.format(jid=from_jid.full())
+jid=from_jid.full()
+)
 )
 feedback = (
 D_(
 "WARNING: received unencrypted data in a supposedly encrypted "
 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)
 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'])
+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)
 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:
+if encryption is None or encryption["plugin"].namespace != NS_OTR:
 return
-to_jid = mess_data['to']
+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.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.add_hint_elements(
-self._p_hints.HINT_NO_COPY,
+message_elt,
-self._p_hints.HINT_NO_PERMANENT_STORE])
+[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. "
 )
 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,
+def send_message_trigger(
-post_xml_treatments):
+self, client, mess_data, pre_xml_treatments, post_xml_treatments
+):
 if client.is_component:
 return True
 if mess_data["type"] == "groupchat":
 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()

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_sec_otr.py @ 4270:0d7bb4df2343