libervia-backend: libervia/backend/plugins/plugin_xep

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

Reformatted code base using black.

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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 import base64
 from datetime import datetime
 import enum
 import logging
 import time
-from typing import \
+from typing import (
-Any, Dict, FrozenSet, Iterable, List, Literal, NamedTuple, Optional, Set, Type, Union, cast
+Any,
+Dict,
+FrozenSet,
+Iterable,
+List,
+Literal,
+NamedTuple,
+Optional,
+Set,
+Type,
+Union,
+cast,
+)
 import uuid
 import xml.etree.ElementTree as ET
 from xml.sax.saxutils import quoteattr
 from typing_extensions import Final, Never, assert_never
 from libervia.backend.plugins.plugin_xep_0045 import XEP_0045
 from libervia.backend.plugins.plugin_xep_0060 import XEP_0060
 from libervia.backend.plugins.plugin_xep_0163 import XEP_0163
 from libervia.backend.plugins.plugin_xep_0334 import XEP_0334
 from libervia.backend.plugins.plugin_xep_0359 import XEP_0359
-from libervia.backend.plugins.plugin_xep_0420 import \
+from libervia.backend.plugins.plugin_xep_0420 import (
-XEP_0420, SCEAffixPolicy, SCEAffixValues, SCEProfile
+XEP_0420,
+SCEAffixPolicy,
+SCEAffixValues,
+SCEProfile,
+)
 from libervia.backend.tools import xml_tools
 from twisted.internet import defer
 from twisted.words.protocols.jabber import error, jid
 from twisted.words.xish import domish
 " download/install the pip packages 'omemo', 'twomemo', 'oldmemo' and"
 f" 'xmlschema'.\nexception: {import_error}"
 ) from import_error
-__all__ = [  # pylint: disable=unused-variable
+__all__ = ["PLUGIN_INFO", "OMEMO"]  # pylint: disable=unused-variable
-"PLUGIN_INFO",
-"OMEMO"
-]
 log = cast(Logger, getLogger(__name__))  # type: ignore[no-untyped-call]
 PLUGIN_INFO = {
 C.PI_NAME: "OMEMO",
 C.PI_IMPORT_NAME: "XEP-0384",
 C.PI_TYPE: "SEC",
-C.PI_PROTOCOLS: [ "XEP-0384" ],
+C.PI_PROTOCOLS: ["XEP-0384"],
-C.PI_DEPENDENCIES: [ "XEP-0163", "XEP-0280", "XEP-0334", "XEP-0060", "XEP-0420" ],
+C.PI_DEPENDENCIES: ["XEP-0163", "XEP-0280", "XEP-0334", "XEP-0060", "XEP-0420"],
-C.PI_RECOMMENDATIONS: [ "XEP-0045", "XEP-0359", C.TEXT_CMDS ],
+C.PI_RECOMMENDATIONS: ["XEP-0045", "XEP-0359", C.TEXT_CMDS],
 C.PI_MAIN: "OMEMO",
 C.PI_HANDLER: "no",
 C.PI_DESCRIPTION: _("""Implementation of OMEMO"""),
 }
 own_device_id = await self.__storage.get(LegacyStorageImpl.KEY_DEVICE_ID, None)
 if own_device_id is None:
 return None
 return oldmemo.migrations.OwnData(
-own_bare_jid=self.__own_bare_jid,
+own_bare_jid=self.__own_bare_jid, own_device_id=own_device_id
-own_device_id=own_device_id
 )
 async def deleteOwnData(self) -> None:
 try:
 await self.__storage.remove(LegacyStorageImpl.KEY_DEVICE_ID)
 pass
 async def loadState(self) -> Optional[oldmemo.migrations.State]:
 return cast(
 Optional[oldmemo.migrations.State],
-await self.__storage.get(LegacyStorageImpl.KEY_STATE, None)
+await self.__storage.get(LegacyStorageImpl.KEY_STATE, None),
 )
 async def deleteState(self) -> None:
 try:
 await self.__storage.remove(LegacyStorageImpl.KEY_STATE)
 except KeyError:
 pass
 async def loadSession(
-self,
+self, bare_jid: str, device_id: int
-bare_jid: str,
-device_id: int
 ) -> Optional[oldmemo.migrations.Session]:
-key = "\n".join([ LegacyStorageImpl.KEY_SESSION, bare_jid, str(device_id) ])
+key = "\n".join([LegacyStorageImpl.KEY_SESSION, bare_jid, str(device_id)])
 return cast(
-Optional[oldmemo.migrations.Session],
+Optional[oldmemo.migrations.Session], await self.__storage.get(key, None)
-await self.__storage.get(key, None)
 )
 async def deleteSession(self, bare_jid: str, device_id: int) -> None:
-key = "\n".join([ LegacyStorageImpl.KEY_SESSION, bare_jid, str(device_id) ])
+key = "\n".join([LegacyStorageImpl.KEY_SESSION, bare_jid, str(device_id)])
 try:
 await self.__storage.remove(key)
 except KeyError:
 pass
 async def loadActiveDevices(self, bare_jid: str) -> Optional[List[int]]:
-key = "\n".join([ LegacyStorageImpl.KEY_ACTIVE_DEVICES, bare_jid ])
+key = "\n".join([LegacyStorageImpl.KEY_ACTIVE_DEVICES, bare_jid])
-return cast(
+return cast(Optional[List[int]], await self.__storage.get(key, None))
-Optional[List[int]],
-await self.__storage.get(key, None)
-)
 async def loadInactiveDevices(self, bare_jid: str) -> Optional[Dict[int, int]]:
-key = "\n".join([ LegacyStorageImpl.KEY_INACTIVE_DEVICES, bare_jid ])
+key = "\n".join([LegacyStorageImpl.KEY_INACTIVE_DEVICES, bare_jid])
-return cast(
+return cast(Optional[Dict[int, int]], await self.__storage.get(key, None))
-Optional[Dict[int, int]],
-await self.__storage.get(key, None)
-)
 async def deleteActiveDevices(self, bare_jid: str) -> None:
-key = "\n".join([ LegacyStorageImpl.KEY_ACTIVE_DEVICES, bare_jid ])
+key = "\n".join([LegacyStorageImpl.KEY_ACTIVE_DEVICES, bare_jid])
 try:
 await self.__storage.remove(key)
 except KeyError:
 pass
 async def deleteInactiveDevices(self, bare_jid: str) -> None:
-key = "\n".join([ LegacyStorageImpl.KEY_INACTIVE_DEVICES, bare_jid ])
+key = "\n".join([LegacyStorageImpl.KEY_INACTIVE_DEVICES, bare_jid])
 try:
 await self.__storage.remove(key)
 except KeyError:
 pass
 async def loadTrust(
-self,
+self, bare_jid: str, device_id: int
-bare_jid: str,
-device_id: int
 ) -> Optional[oldmemo.migrations.Trust]:
-key = "\n".join([ LegacyStorageImpl.KEY_TRUST, bare_jid, str(device_id) ])
+key = "\n".join([LegacyStorageImpl.KEY_TRUST, bare_jid, str(device_id)])
 return cast(
-Optional[oldmemo.migrations.Trust],
+Optional[oldmemo.migrations.Trust], await self.__storage.get(key, None)
-await self.__storage.get(key, None)
 )
 async def deleteTrust(self, bare_jid: str, device_id: int) -> None:
-key = "\n".join([ LegacyStorageImpl.KEY_TRUST, bare_jid, str(device_id) ])
+key = "\n".join([LegacyStorageImpl.KEY_TRUST, bare_jid, str(device_id)])
 try:
 await self.__storage.remove(key)
 except KeyError:
 pass
 except KeyError:
 pass
 async def download_oldmemo_bundle(
-client: SatXMPPClient,
+client: SatXMPPClient, xep_0060: XEP_0060, bare_jid: str, device_id: int
-xep_0060: XEP_0060,
-bare_jid: str,
-device_id: int
 ) -> oldmemo.oldmemo.BundleImpl:
 """Download the oldmemo bundle corresponding to a specific device.
 @param client: The client.
 @param xep_0060: The XEP-0060 plugin instance to use for pubsub interactions.
 raise omemo.BundleDownloadFailed(
 f"Bundle download failed for {bare_jid}: {device_id} under namespace"
 f" {namespace}: Unexpected number of items retrieved: {len(items)}."
 )
-element = \
+element = next(
-next(iter(xml_tools.domish_elt_2_et_elt(cast(domish.Element, items[0]))), None)
+iter(xml_tools.domish_elt_2_et_elt(cast(domish.Element, items[0]))), None
+)
 if element is None:
 raise omemo.BundleDownloadFailed(
 f"Bundle download failed for {bare_jid}: {device_id} under namespace"
 f" {namespace}: Item download succeeded but parsing failed: {element}."
 )
 # add ATM to the OMEMO plugin directly instead of having it a separate Libervia plugin.
 NS_TM: Final = "urn:xmpp:tm:1"
 NS_ATM: Final = "urn:xmpp:atm:1"
-TRUST_MESSAGE_SCHEMA = xmlschema.XMLSchema("""<?xml version='1.0' encoding='UTF-8'?>
+TRUST_MESSAGE_SCHEMA = xmlschema.XMLSchema(
+"""<?xml version='1.0' encoding='UTF-8'?>
 <xs:schema xmlns:xs='http://www.w3.org/2001/XMLSchema'
 targetNamespace='urn:xmpp:tm:1'
 xmlns='urn:xmpp:tm:1'
 elementFormDefault='qualified'>
 </xs:sequence>
 <xs:attribute name='jid' type='xs:string' use='required'/>
 </xs:complexType>
 </xs:element>
 </xs:schema>
-""")
+"""
+)
 # This is compatible with omemo:2's SCE profile
 TM_SCE_PROFILE = SCEProfile(
 rpad_policy=SCEAffixPolicy.REQUIRED,
 time_policy=SCEAffixPolicy.REQUIRED,
 to_policy=SCEAffixPolicy.OPTIONAL,
 from_policy=SCEAffixPolicy.OPTIONAL,
-custom_policies={}
+custom_policies={},
 )
 class TrustUpdate(NamedTuple):
 # pylint: disable=invalid-name
 """Convert the instance to a serialised dictionary"""
 data = {
 "sender_jid": self.sender_jid.full(),
 "sender_key": self.sender_key.hex(),
 "timestamp": self.timestamp.isoformat(),
-"trust_update": self.trust_update.to_dict()
+"trust_update": self.trust_update.to_dict(),
 }
 return data
 @staticmethod
 def from_dict(data: dict[str, Any]) -> "TrustMessageCacheEntry":
 async def manage_trust_message_cache(
 client: SatXMPPClient,
 session_manager: omemo.SessionManager,
-applied_trust_updates: FrozenSet[TrustUpdate]
+applied_trust_updates: FrozenSet[TrustUpdate],
 ) -> None:
 """Manage the ATM trust message cache after trust updates have been applied.
 @param client: The client this operation runs under.
 @param session_manager: The session manager to use.
 @param applied_trust_updates: The trust updates that have already been applied,
 triggering this cache management run.
 """
 trust_message_cache = persistent.LazyPersistentBinaryDict(
-"XEP-0384/TM",
+"XEP-0384/TM", client.profile
-client.profile
 )
 # Load cache entries
 cache_entries = {
 TrustMessageCacheEntry.from_dict(d)
 # Expire cache entries that were overwritten by the applied trust updates
 cache_entries_by_target = {
 (
 cache_entry.trust_update.target_jid.userhostJID(),
-cache_entry.trust_update.target_key
+cache_entry.trust_update.target_key,
 ): cache_entry
-for cache_entry
+for cache_entry in cache_entries
-in cache_entries
 }
 for trust_update in applied_trust_updates:
 cache_entry = cache_entries_by_target.get(
-(trust_update.target_jid.userhostJID(), trust_update.target_key),
+(trust_update.target_jid.userhostJID(), trust_update.target_key), None
-None
 )
 if cache_entry is not None:
 cache_entries.remove(cache_entry)
 # Apply the trust update
 await session_manager.set_trust(
 cache_entry.trust_update.target_jid.userhost(),
 cache_entry.trust_update.target_key,
-trust_level.name
+trust_level.name,
 )
 # Track the fact that this trust update has been applied
 new_trust_updates.add(cache_entry.trust_update)
 # Remove the corresponding cache entry
 cache_entries.remove(cache_entry)
 # Store the updated cache entries
-await trust_message_cache.force(
+await trust_message_cache.force("cache", [tm.to_dict() for tm in cache_entries])
-"cache",
-[tm.to_dict() for tm in cache_entries]
-)
 # TODO: Notify the user ("feedback") about automatically updated trust?
 if len(new_trust_updates) > 0:
 # If any trust has been updated, recursively perform another run of cache
 # management
 await manage_trust_message_cache(
-client,
+client, session_manager, frozenset(new_trust_updates)
-session_manager,
-frozenset(new_trust_updates)
 )
 async def get_trust_as_trust_updates(
-session_manager: omemo.SessionManager,
+session_manager: omemo.SessionManager, target_jid: jid.JID
-target_jid: jid.JID
 ) -> FrozenSet[TrustUpdate]:
 """Get the trust status of all known keys of a JID as trust updates for use with ATM.
 @param session_manager: The session manager to load the trust from.
 @param target_jid: The JID to load the trust for.
 target_trust = False
 else:
 # Skip devices that are not explicitly trusted or distrusted
 continue
-trust_updates.add(TrustUpdate(
+trust_updates.add(
-target_jid=target_jid.userhostJID(),
+TrustUpdate(
-target_key=device.identity_key,
+target_jid=target_jid.userhostJID(),
-target_trust=target_trust
+target_key=device.identity_key,
-))
+target_trust=target_trust,
+)
+)
 return frozenset(trust_updates)
 async def send_trust_messages(
 client: SatXMPPClient,
 session_manager: omemo.SessionManager,
-applied_trust_updates: FrozenSet[TrustUpdate]
+applied_trust_updates: FrozenSet[TrustUpdate],
 ) -> None:
 """Send information about updated trust to peers via ATM (XEP-0450).
 @param client: The client.
 @param session_manager: The session manager.
 # Send Trust Messages for newly trusted and distrusted devices
 own_jid = client.jid.userhostJID()
 own_trust_updates = await get_trust_as_trust_updates(session_manager, own_jid)
 # JIDs of which at least one device's trust has been updated
-updated_jids = frozenset({
+updated_jids = frozenset(
-trust_update.target_jid.userhostJID()
+{trust_update.target_jid.userhostJID() for trust_update in applied_trust_updates}
-for trust_update
+)
-in applied_trust_updates
-})
 trust_messages: Set[PartialTrustMessage] = set()
 for updated_jid in updated_jids:
 # Get the trust updates for that JID
-trust_updates = frozenset({
+trust_updates = frozenset(
-trust_update for trust_update in applied_trust_updates
+{
-if trust_update.target_jid.userhostJID() == updated_jid
+trust_update
-})
+for trust_update in applied_trust_updates
+if trust_update.target_jid.userhostJID() == updated_jid
+}
+)
 if updated_jid == own_jid:
 # If the own JID is updated, _all_ peers have to be notified
 # TODO: Using my author's privilege here to shamelessly access private fields
 # and storage keys until I've added public API to get a list of peers to
 # python-omemo.
 storage: omemo.Storage = getattr(session_manager, "_SessionManager__storage")
-peer_jids = frozenset({
+peer_jids = frozenset(
-jid.JID(bare_jid).userhostJID() for bare_jid in (await storage.load_list(
+{
-f"/{OMEMO.NS_TWOMEMO}/bare_jids",
+jid.JID(bare_jid).userhostJID()
-str
+for bare_jid in (
-)).maybe([])
+await storage.load_list(f"/{OMEMO.NS_TWOMEMO}/bare_jids", str)
-})
+).maybe([])
+}
+)
 if len(peer_jids) == 0:
 # If there are no peers to notify, notify our other devices about the
 # changes directly
-trust_messages.add(PartialTrustMessage(
+trust_messages.add(
-recipient_jid=own_jid,
+PartialTrustMessage(
-updated_jid=own_jid,
+recipient_jid=own_jid,
-trust_updates=trust_updates
+updated_jid=own_jid,
-))
+trust_updates=trust_updates,
+)
+)
 else:
 # Otherwise, notify all peers about the changes in trust and let carbons
 # handle the copy to our own JID
 for peer_jid in peer_jids:
-trust_messages.add(PartialTrustMessage(
+trust_messages.add(
-recipient_jid=peer_jid,
+PartialTrustMessage(
-updated_jid=own_jid,
+recipient_jid=peer_jid,
-trust_updates=trust_updates
+updated_jid=own_jid,
-))
+trust_updates=trust_updates,
+)
+)
 # Also send full trust information about _every_ peer to our newly
 # trusted devices
-peer_trust_updates = \
+peer_trust_updates = await get_trust_as_trust_updates(
-await get_trust_as_trust_updates(session_manager, peer_jid)
+session_manager, peer_jid
+)
-trust_messages.add(PartialTrustMessage(
-recipient_jid=own_jid,
+trust_messages.add(
-updated_jid=peer_jid,
+PartialTrustMessage(
-trust_updates=peer_trust_updates
+recipient_jid=own_jid,
-))
+updated_jid=peer_jid,
+trust_updates=peer_trust_updates,
+)
+)
 # Send information about our own devices to our newly trusted devices
-trust_messages.add(PartialTrustMessage(
+trust_messages.add(
-recipient_jid=own_jid,
+PartialTrustMessage(
-updated_jid=own_jid,
+recipient_jid=own_jid,
-trust_updates=own_trust_updates
+updated_jid=own_jid,
-))
+trust_updates=own_trust_updates,
+)
+)
 else:
 # Notify our other devices about the changes in trust
-trust_messages.add(PartialTrustMessage(
+trust_messages.add(
-recipient_jid=own_jid,
+PartialTrustMessage(
-updated_jid=updated_jid,
+recipient_jid=own_jid,
-trust_updates=trust_updates
+updated_jid=updated_jid,
-))
+trust_updates=trust_updates,
+)
+)
 # Send a summary of our own trust to newly trusted devices
-trust_messages.add(PartialTrustMessage(
+trust_messages.add(
-recipient_jid=updated_jid,
+PartialTrustMessage(
-updated_jid=own_jid,
+recipient_jid=updated_jid,
-trust_updates=own_trust_updates
+updated_jid=own_jid,
-))
+trust_updates=own_trust_updates,
+)
+)
 # All trust messages prepared. Merge all trust messages directed at the same
 # recipient.
-recipient_jids = { trust_message.recipient_jid for trust_message in trust_messages }
+recipient_jids = {trust_message.recipient_jid for trust_message in trust_messages}
 for recipient_jid in recipient_jids:
 updated: Dict[jid.JID, Set[TrustUpdate]] = {}
 for trust_message in trust_messages:
 # Merge trust messages directed at that recipient
 if trust_message.recipient_jid == recipient_jid:
 # Merge the trust updates
-updated[trust_message.updated_jid] = \
+updated[trust_message.updated_jid] = updated.get(
-updated.get(trust_message.updated_jid, set())
+trust_message.updated_jid, set()
+)
 updated[trust_message.updated_jid] |= trust_message.trust_updates
 # Build the trust message
 trust_message_elt = domish.Element((NS_TM, "trust-message"))
 for updated_jid, trust_updates in updated.items():
 key_owner_elt = trust_message_elt.addElement((NS_TM, "key-owner"))
 key_owner_elt["jid"] = updated_jid.userhost()
 for trust_update in trust_updates:
-serialized_identity_key = \
+serialized_identity_key = base64.b64encode(
-base64.b64encode(trust_update.target_key).decode("ASCII")
+trust_update.target_key
+).decode("ASCII")
 if trust_update.target_trust:
 key_owner_elt.addElement(
-(NS_TM, "trust"),
+(NS_TM, "trust"), content=serialized_identity_key
-content=serialized_identity_key
 )
 else:
 key_owner_elt.addElement(
-(NS_TM, "distrust"),
+(NS_TM, "distrust"), content=serialized_identity_key
-content=serialized_identity_key
 )
 # Finally, encrypt and send the trust message!
-message_data = client.generate_message_xml(MessageData({
+message_data = client.generate_message_xml(
-"from": own_jid,
+MessageData(
-"to": recipient_jid,
+{
-"uid": str(uuid.uuid4()),
+"from": own_jid,
-"message": {},
+"to": recipient_jid,
-"subject": {},
+"uid": str(uuid.uuid4()),
-"type": C.MESS_TYPE_CHAT,
+"message": {},
-"extra": {},
+"subject": {},
-"timestamp": time.time()
+"type": C.MESS_TYPE_CHAT,
-}))
+"extra": {},
+"timestamp": time.time(),
+}
+)
+)
 message_data["xml"].addChild(trust_message_elt)
 plaintext = XEP_0420.pack_stanza(TM_SCE_PROFILE, message_data["xml"])
 feedback_jid = recipient_jid
 # TODO: The following is mostly duplicate code
 try:
 messages, encryption_errors = await session_manager.encrypt(
-frozenset({ own_jid.userhost(), recipient_jid.userhost() }),
+frozenset({own_jid.userhost(), recipient_jid.userhost()}),
-{ OMEMO.NS_TWOMEMO: plaintext },
+{OMEMO.NS_TWOMEMO: plaintext},
-backend_priority_order=[ OMEMO.NS_TWOMEMO ],
+backend_priority_order=[OMEMO.NS_TWOMEMO],
-identifier=feedback_jid.userhost()
+identifier=feedback_jid.userhost(),
 )
 except Exception as e:
 msg = _(
 # pylint: disable=consider-using-f-string
 "Can't encrypt message for {entities}: {reason}".format(
-entities=', '.join({ own_jid.userhost(), recipient_jid.userhost() }),
+entities=", ".join({own_jid.userhost(), recipient_jid.userhost()}),
-reason=e
+reason=e,
 )
 )
 log.warning(msg)
-client.feedback(feedback_jid, msg, {
+client.feedback(feedback_jid, msg, {C.MESS_EXTRA_INFO: C.EXTRA_INFO_ENCR_ERR})
-C.MESS_EXTRA_INFO: C.EXTRA_INFO_ENCR_ERR
-})
 raise e
 if len(encryption_errors) > 0:
 log.warning(
 f"Ignored the following non-critical encryption errors:"
 f" {encryption_errors}"
 )
-encrypted_errors_stringified = ", ".join([
+encrypted_errors_stringified = ", ".join(
-f"device {err.device_id} of {err.bare_jid} under namespace"
+[
-f" {err.namespace}"
+f"device {err.device_id} of {err.bare_jid} under namespace"
-for err
+f" {err.namespace}"
-in encryption_errors
+for err in encryption_errors
-])
+]
+)
 client.feedback(
 feedback_jid,
 D_(
 "There were non-critical errors during encryption resulting in some"
 " of your destinees' devices potentially not receiving the message."
 " This happens when the encryption data/key material of a device is"
 " incomplete or broken, which shouldn't happen for actively used"
 " devices, and can usually be ignored. The following devices are"
 f" affected: {encrypted_errors_stringified}."
-)
+),
 )
 message = next(
-message for message in messages
+message for message in messages if message.namespace == OMEMO.NS_TWOMEMO
-if message.namespace == OMEMO.NS_TWOMEMO
 )
 # Add the encrypted element
-message_data["xml"].addChild(xml_tools.et_elt_2_domish_elt(
+message_data["xml"].addChild(
-twomemo.etree.serialize_message(message)
+xml_tools.et_elt_2_domish_elt(twomemo.etree.serialize_message(message))
-))
+)
 await client.a_send(message_data["xml"])
-def make_session_manager(sat: LiberviaBackend, profile: str) -> Type[omemo.SessionManager]:
+def make_session_manager(
+sat: LiberviaBackend, profile: str
+) -> Type[omemo.SessionManager]:
 """
 @param sat: The SAT instance.
 @param profile: The profile.
 @return: A non-abstract subclass of :class:`~omemo.session_manager.SessionManager`
 with XMPP interactions and trust handled via the SAT instance.
 xml_tools.et_elt_2_domish_elt(element),
 item_id=str(bundle.device_id),
 extra={
 XEP_0060.EXTRA_PUBLISH_OPTIONS: {
 XEP_0060.OPT_ACCESS_MODEL: "open",
-XEP_0060.OPT_MAX_ITEMS: "max"
+XEP_0060.OPT_MAX_ITEMS: "max",
 },
-XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "raise"
+XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "raise",
-}
+},
 )
 except (error.StanzaError, Exception) as e:
 if (
 isinstance(e, error.StanzaError)
 and e.condition == "conflict"
 xml_tools.et_elt_2_domish_elt(element),
 item_id=xep_0060.ID_SINGLETON,
 extra={
 XEP_0060.EXTRA_PUBLISH_OPTIONS: {
 XEP_0060.OPT_ACCESS_MODEL: "open",
-XEP_0060.OPT_MAX_ITEMS: 1
+XEP_0060.OPT_MAX_ITEMS: 1,
 },
-XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options"
+XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options",
-}
+},
 )
 except Exception as e:
 raise omemo.BundleUploadFailed(
 f"Bundle upload failed: {bundle}"
 ) from e
 raise omemo.UnknownNamespace(f"Unknown namespace: {bundle.namespace}")
 @staticmethod
 async def _download_bundle(
-namespace: str,
+namespace: str, bare_jid: str, device_id: int
-bare_jid: str,
-device_id: int
 ) -> omemo.Bundle:
 if namespace == twomemo.twomemo.NAMESPACE:
 node = "urn:xmpp:omemo:2:bundles"
 try:
 items, __ = await xep_0060.get_items(
-client,
+client, jid.JID(bare_jid), node, item_ids=[str(device_id)]
-jid.JID(bare_jid),
-node,
-item_ids=[ str(device_id) ]
 )
 except Exception as e:
 raise omemo.BundleDownloadFailed(
 f"Bundle download failed for {bare_jid}: {device_id} under"
 f" namespace {namespace}"
 f" {len(items)}."
 )
 element = next(
 iter(xml_tools.domish_elt_2_et_elt(cast(domish.Element, items[0]))),
-None
+None,
 )
 if element is None:
 raise omemo.BundleDownloadFailed(
 f"Bundle download failed for {bare_jid}: {device_id} under"
 f" namespace {namespace}: Item download succeeded but parsing"
 f" namespace {namespace}"
 ) from e
 if namespace == oldmemo.oldmemo.NAMESPACE:
 return await download_oldmemo_bundle(
-client,
+client, xep_0060, bare_jid, device_id
-xep_0060,
-bare_jid,
-device_id
 )
 raise omemo.UnknownNamespace(f"Unknown namespace: {namespace}")
 @staticmethod
 try:
 await xep_0060.retract_items(
 client,
 client.jid.userhostJID(),
 node,
-[ str(device_id) ],
+[str(device_id)],
-notify=False
+notify=False,
 )
 except Exception as e:
 raise omemo.BundleDeletionFailed(
 f"Bundle deletion failed for {device_id} under namespace"
 f" {namespace}"
 raise omemo.UnknownNamespace(f"Unknown namespace: {namespace}")
 @staticmethod
 async def _upload_device_list(
-namespace: str,
+namespace: str, device_list: Dict[int, Optional[str]]
-device_list: Dict[int, Optional[str]]
 ) -> None:
 element: Optional[ET.Element] = None
 node: Optional[str] = None
 if namespace == twomemo.twomemo.NAMESPACE:
 xml_tools.et_elt_2_domish_elt(element),
 item_id=xep_0060.ID_SINGLETON,
 extra={
 XEP_0060.EXTRA_PUBLISH_OPTIONS: {
 XEP_0060.OPT_MAX_ITEMS: 1,
-XEP_0060.OPT_ACCESS_MODEL: "open"
+XEP_0060.OPT_ACCESS_MODEL: "open",
 },
-XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "raise"
+XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "raise",
-}
+},
 )
 except (error.StanzaError, Exception) as e:
 if (
 isinstance(e, error.StanzaError)
 and e.condition == "conflict"
 f"Device list upload failed for namespace {namespace}"
 ) from e
 @staticmethod
 async def _download_device_list(
-namespace: str,
+namespace: str, bare_jid: str
-bare_jid: str
 ) -> Dict[int, Optional[str]]:
 node: Optional[str] = None
 if namespace == twomemo.twomemo.NAMESPACE:
 node = TWOMEMO_DEVICE_LIST_NODE
 f"Device list download failed for {bare_jid} under namespace"
 f" {namespace}: Unexpected number of items retrieved: {len(items)}."
 )
 element = next(
-iter(xml_tools.domish_elt_2_et_elt(cast(domish.Element, items[0]))),
+iter(xml_tools.domish_elt_2_et_elt(cast(domish.Element, items[0]))), None
-None
 )
 if element is None:
 raise omemo.DeviceListDownloadFailed(
 f"Device list download failed for {bare_jid} under namespace"
 ) from e
 raise omemo.UnknownNamespace(f"Unknown namespace: {namespace}")
 async def _evaluate_custom_trust_level(
-self,
+self, device: omemo.DeviceInformation
-device: omemo.DeviceInformation
 ) -> omemo.TrustLevel:
 # Get the custom trust level
 try:
 trust_level = TrustLevel(device.trust_level_name)
 except ValueError as e:
 # The blindly trusted case is more complicated, since its evaluation depends
 # on the trust system and phase
 if trust_level is TrustLevel.BLINDLY_TRUSTED:
 # Get the name of the active trust system
-trust_system = cast(str, sat.memory.param_get_a(
+trust_system = cast(
-PARAM_NAME,
+str,
-PARAM_CATEGORY,
+sat.memory.param_get_a(
-profile_key=profile
+PARAM_NAME, PARAM_CATEGORY, profile_key=profile
-))
+),
+)
 # If the trust model is BTBV, blind trust is always enabled
 if trust_system == "btbv":
 return omemo.TrustLevel.TRUSTED
 # and counts as undecided
 if trust_system == "atm":
 # Find out whether we are in phase one or two
 devices = await self.get_device_information(device.bare_jid)
-phase_one = all(TrustLevel(device.trust_level_name) in {
+phase_one = all(
-TrustLevel.UNDECIDED,
+TrustLevel(device.trust_level_name)
-TrustLevel.BLINDLY_TRUSTED
+in {TrustLevel.UNDECIDED, TrustLevel.BLINDLY_TRUSTED}
-} for device in devices)
+for device in devices
+)
 if phase_one:
 return omemo.TrustLevel.TRUSTED
 return omemo.TrustLevel.UNDECIDED
 )
 assert_never(trust_level)
 async def _make_trust_decision(
-self,
+self, undecided: FrozenSet[omemo.DeviceInformation], identifier: Optional[str]
-undecided: FrozenSet[omemo.DeviceInformation],
-identifier: Optional[str]
 ) -> None:
 if identifier is None:
 raise omemo.TrustDecisionFailed(
 "The identifier must contain the feedback JID."
 )
 # Both the ATM and the BTBV trust models work with blind trust before the
 # first manual verification is performed. Thus, we can separate bare JIDs into
 # two pools here, one pool of bare JIDs for which blind trust is active, and
 # one pool of bare JIDs for which manual trust is used instead.
-bare_jids = { device.bare_jid for device in undecided }
+bare_jids = {device.bare_jid for device in undecided}
 blind_trust_bare_jids: Set[str] = set()
 manual_trust_bare_jids: Set[str] = set()
 # For each bare JID, decide whether blind trust applies
 # Get all known devices belonging to the bare JID
 devices = await self.get_device_information(bare_jid)
 # If the trust levels of all devices correspond to those used by blind
 # trust, blind trust applies. Otherwise, fall back to manual trust.
-if all(TrustLevel(device.trust_level_name) in {
+if all(
-TrustLevel.UNDECIDED,
+TrustLevel(device.trust_level_name)
-TrustLevel.BLINDLY_TRUSTED
+in {TrustLevel.UNDECIDED, TrustLevel.BLINDLY_TRUSTED}
-} for device in devices):
+for device in devices
+):
 blind_trust_bare_jids.add(bare_jid)
 else:
 manual_trust_bare_jids.add(bare_jid)
 # With the JIDs sorted into their respective pools, the undecided devices can
 # be categorized too
-blindly_trusted_devices = \
+blindly_trusted_devices = {
-{ dev for dev in undecided if dev.bare_jid in blind_trust_bare_jids }
+dev for dev in undecided if dev.bare_jid in blind_trust_bare_jids
-manually_trusted_devices = \
+}
-{ dev for dev in undecided if dev.bare_jid in manual_trust_bare_jids }
+manually_trusted_devices = {
+dev for dev in undecided if dev.bare_jid in manual_trust_bare_jids
+}
 # Blindly trust devices handled by blind trust
 if len(blindly_trusted_devices) > 0:
 for device in blindly_trusted_devices:
 await self.set_trust(
 device.bare_jid,
 device.identity_key,
-TrustLevel.BLINDLY_TRUSTED.name
+TrustLevel.BLINDLY_TRUSTED.name,
 )
-blindly_trusted_devices_stringified = ", ".join([
+blindly_trusted_devices_stringified = ", ".join(
-f"device {device.device_id} of {device.bare_jid} under namespace"
+[
-f" {device.namespaces}"
+f"device {device.device_id} of {device.bare_jid} under namespace"
-for device
+f" {device.namespaces}"
-in blindly_trusted_devices
+for device in blindly_trusted_devices
-])
+]
+)
 client.feedback(
 feedback_jid,
 D_(
 "Not all destination devices are trusted, unknown devices will be"
 " blindly trusted.\nFollowing devices have been automatically"
 f" trusted: {blindly_trusted_devices_stringified}."
-)
+),
 )
 # Prompt the user for manual trust decisions on the devices handled by manual
 # trust
 if len(manually_trusted_devices) > 0:
 D_(
 "Not all destination devices are trusted, we can't encrypt"
 " message in such a situation. Please indicate if you trust"
 " those devices or not in the trust manager before we can"
 " send this message."
-)
+),
 )
 await self.__prompt_manual_trust(
-frozenset(manually_trusted_devices),
+frozenset(manually_trusted_devices), feedback_jid
-feedback_jid
 )
 @staticmethod
 async def _send_message(message: omemo.Message, bare_jid: str) -> None:
 element: Optional[ET.Element] = None
 element = oldmemo.etree.serialize_message(message)
 if element is None:
 raise omemo.UnknownNamespace(f"Unknown namespace: {message.namespace}")
-message_data = client.generate_message_xml(MessageData({
+message_data = client.generate_message_xml(
-"from": client.jid,
+MessageData(
-"to": jid.JID(bare_jid),
+{
-"uid": str(uuid.uuid4()),
+"from": client.jid,
-"message": {},
+"to": jid.JID(bare_jid),
-"subject": {},
+"uid": str(uuid.uuid4()),
-"type": C.MESS_TYPE_CHAT,
+"message": {},
-"extra": {},
+"subject": {},
-"timestamp": time.time()
+"type": C.MESS_TYPE_CHAT,
-}))
+"extra": {},
+"timestamp": time.time(),
+}
+)
+)
 message_data["xml"].addChild(xml_tools.et_elt_2_domish_elt(element))
 try:
 await client.a_send(message_data["xml"])
 except Exception as e:
 raise omemo.MessageSendingFailed() from e
 async def __prompt_manual_trust(
-self,
+self, undecided: FrozenSet[omemo.DeviceInformation], feedback_jid: jid.JID
-undecided: FrozenSet[omemo.DeviceInformation],
-feedback_jid: jid.JID
 ) -> None:
 """Asks the user to decide on the manual trust level of a set of devices.
 Blocks until the user has made a decision and updates the trust levels of all
 devices using :meth:`set_trust`.
 namespace = encryption["plugin"].namespace
 # Casting this to Any, otherwise all calls on the variable cause type errors
 # pylint: disable=no-member
-trust_ui = cast(Any, xml_tools.XMLUI(
+trust_ui = cast(
-panel_type=C.XMLUI_FORM,
+Any,
-title=D_("OMEMO trust management"),
+xml_tools.XMLUI(
-submit_id=""
+panel_type=C.XMLUI_FORM,
-))
+title=D_("OMEMO trust management"),
-trust_ui.addText(D_(
+submit_id="",
-"This is OMEMO trusting system. You'll see below the devices of your "
+),
-"contacts, and a checkbox to trust them or not. A trusted device "
+)
-"can read your messages in plain text, so be sure to only validate "
+trust_ui.addText(
-"devices that you are sure are belonging to your contact. It's better "
+D_(
-"to do this when you are next to your contact and their device, so "
+"This is OMEMO trusting system. You'll see below the devices of your "
-"you can check the \"fingerprint\" (the number next to the device) "
+"contacts, and a checkbox to trust them or not. A trusted device "
-"yourself. Do *not* validate a device if the fingerprint is wrong!"
+"can read your messages in plain text, so be sure to only validate "
-))
+"devices that you are sure are belonging to your contact. It's better "
+"to do this when you are next to your contact and their device, so "
+'you can check the "fingerprint" (the number next to the device) '
+"yourself. Do *not* validate a device if the fingerprint is wrong!"
+)
+)
 own_device, __ = await self.get_own_device_information()
 trust_ui.change_container("label")
 trust_ui.addLabel(D_("This device ID"))
 trust_ui.addEmpty()
 trust_ui_result = await xml_tools.defer_xmlui(
 sat,
 trust_ui,
-action_extra={ "meta_encryption_trust": namespace },
+action_extra={"meta_encryption_trust": namespace},
-profile=profile
+profile=profile,
 )
 if C.bool(trust_ui_result.get("cancelled", "false")):
 raise omemo.TrustDecisionFailed("Trust UI cancelled.")
-data_form_result = cast(Dict[str, str], xml_tools.xmlui_result_2_data_form_result(
+data_form_result = cast(
-trust_ui_result
+Dict[str, str], xml_tools.xmlui_result_2_data_form_result(trust_ui_result)
-))
+)
 trust_updates: Set[TrustUpdate] = set()
 for key, value in data_form_result.items():
 if not key.startswith("trust_"):
 continue
-device = undecided_ordered[int(key[len("trust_"):])]
+device = undecided_ordered[int(key[len("trust_") :])]
 target_trust = C.bool(value)
-trust_level = \
+trust_level = (
 TrustLevel.TRUSTED if target_trust else TrustLevel.DISTRUSTED
+)
 await self.set_trust(
-device.bare_jid,
+device.bare_jid, device.identity_key, trust_level.name
-device.identity_key,
+)
-trust_level.name
-)
+trust_updates.add(
+TrustUpdate(
-trust_updates.add(TrustUpdate(
+target_jid=jid.JID(device.bare_jid).userhostJID(),
-target_jid=jid.JID(device.bare_jid).userhostJID(),
+target_key=device.identity_key,
-target_key=device.identity_key,
+target_trust=target_trust,
-target_trust=target_trust
+)
-))
+)
 # Check whether ATM is enabled and handle everything in case it is
-trust_system = cast(str, sat.memory.param_get_a(
+trust_system = cast(
-PARAM_NAME,
+str,
-PARAM_CATEGORY,
+sat.memory.param_get_a(PARAM_NAME, PARAM_CATEGORY, profile_key=profile),
-profile_key=profile
+)
-))
 if trust_system == "atm":
 await manage_trust_message_cache(client, self, frozenset(trust_updates))
 await send_trust_messages(client, self, frozenset(trust_updates))
 profile: str,
 initial_own_label: Optional[str],
 signed_pre_key_rotation_period: int = 7 * 24 * 60 * 60,
 pre_key_refill_threshold: int = 99,
 max_num_per_session_skipped_keys: int = 1000,
-max_num_per_message_skipped_keys: Optional[int] = None
+max_num_per_message_skipped_keys: Optional[int] = None,
 ) -> omemo.SessionManager:
 """Prepare the OMEMO library (storage, backends, core) for a specific profile.
 @param sat: The SAT instance.
 @param profile: The profile.
 # TODO: Do we want BLINDLY_TRUSTED or TRUSTED here?
 TrustLevel.BLINDLY_TRUSTED.name,
 TrustLevel.UNDECIDED.name,
 TrustLevel.DISTRUSTED.name,
 lambda bare_jid, device_id: download_oldmemo_bundle(
-client,
+client, xep_0060, bare_jid, device_id
-xep_0060,
+),
-bare_jid,
-device_id
-)
 )
 session_manager = await make_session_manager(sat, profile).create(
 [
 twomemo.Twomemo(
 storage,
 max_num_per_session_skipped_keys,
-max_num_per_message_skipped_keys
+max_num_per_message_skipped_keys,
 ),
 oldmemo.Oldmemo(
 storage,
 max_num_per_session_skipped_keys,
-max_num_per_message_skipped_keys
+max_num_per_message_skipped_keys,
-)
+),
 ],
 storage,
 client.jid.userhost(),
 initial_own_label,
 TrustLevel.UNDECIDED.value,
 signed_pre_key_rotation_period,
 pre_key_refill_threshold,
-omemo.AsyncFramework.TWISTED
+omemo.AsyncFramework.TWISTED,
 )
 # This shouldn't hurt here since we're not running on overly constrainted devices.
 # TODO: Consider ensuring data consistency regularly/in response to certain events
 await session_manager.ensure_data_consistency()
 ``urn:xmpp:omemo:2`` namespace and the (legacy) ``eu.siacs.conversations.axolotl``
 namespace. Both versions of the protocol are handled by this plugin and compatibility
 between the two is maintained. MUC messages are supported next to one to one messages.
 For trust management, the two trust models "ATM" and "BTBV" are supported.
 """
 NS_TWOMEMO = twomemo.twomemo.NAMESPACE
 NS_OLDMEMO = oldmemo.oldmemo.NAMESPACE
 # For MUC/MIX message stanzas, the <to/> affix is a MUST
 SCE_PROFILE_GROUPCHAT = SCEProfile(
 rpad_policy=SCEAffixPolicy.REQUIRED,
 time_policy=SCEAffixPolicy.OPTIONAL,
 to_policy=SCEAffixPolicy.REQUIRED,
 from_policy=SCEAffixPolicy.OPTIONAL,
-custom_policies={}
+custom_policies={},
 )
 # For everything but MUC/MIX message stanzas, the <to/> affix is a MAY
 SCE_PROFILE = SCEProfile(
 rpad_policy=SCEAffixPolicy.REQUIRED,
 time_policy=SCEAffixPolicy.OPTIONAL,
 to_policy=SCEAffixPolicy.OPTIONAL,
 from_policy=SCEAffixPolicy.OPTIONAL,
-custom_policies={}
+custom_policies={},
 )
 def __init__(self, host: LiberviaBackend) -> None:
 """
 @param sat: The SAT instance.
 # These triggers are used by oldmemo, which doesn't do SCE and only applies to
 # messages. Temporarily, until a more fitting trigger for SCE-based encryption is
 # added, the message_received trigger is also used for twomemo.
 host.trigger.add(
-"message_received",
+"message_received", self._message_received_trigger, priority=100050
-self._message_received_trigger,
-priority=100050
 )
 host.trigger.add("send", self.__send_trigger, priority=0)
 # TODO: Add new triggers here for freshly received and about-to-be-sent stanzas,
 # including IQs.
 xep_0163.add_pep_event(
 "TWOMEMO_DEVICES",
 TWOMEMO_DEVICE_LIST_NODE,
 lambda items_event, profile: defer.ensureDeferred(
 self.__on_device_list_update(items_event, profile)
-)
+),
 )
 xep_0163.add_pep_event(
 "OLDMEMO_DEVICES",
 OLDMEMO_DEVICE_LIST_NODE,
 lambda items_event, profile: defer.ensureDeferred(
 self.__on_device_list_update(items_event, profile)
-)
+),
 )
 try:
 self.__text_commands = cast(TextCommands, host.plugins[C.TEXT_CMDS])
 except KeyError:
 log.info(_("Text commands not available"))
 else:
 self.__text_commands.register_text_commands(self)
 def profile_connected(  # pylint: disable=invalid-name
-self,
+self, client: SatXMPPClient
-client: SatXMPPClient
 ) -> None:
 """
 @param client: The client.
 """
-defer.ensureDeferred(self.get_session_manager(
+defer.ensureDeferred(self.get_session_manager(cast(str, client.profile)))
-cast(str, client.profile)
-))
 async def cmd_omemo_reset(
-self,
+self, client: SatXMPPClient, mess_data: MessageData
-client: SatXMPPClient,
-mess_data: MessageData
 ) -> Literal[False]:
 """Reset all sessions of devices that belong to the recipient of ``mess_data``.
 This must only be callable manually by the user. Use this when a session is
 apparently broken, i.e. sending and receiving encrypted messages doesn't work and
 reset all sessions with.
 @return: The constant value ``False``, indicating to the text commands plugin that
 the message is not supposed to be sent.
 """
-twomemo_requested = \
+twomemo_requested = client.encryption.is_encryption_requested(
-client.encryption.is_encryption_requested(mess_data, twomemo.twomemo.NAMESPACE)
+mess_data, twomemo.twomemo.NAMESPACE
-oldmemo_requested = \
+)
-client.encryption.is_encryption_requested(mess_data, oldmemo.oldmemo.NAMESPACE)
+oldmemo_requested = client.encryption.is_encryption_requested(
+mess_data, oldmemo.oldmemo.NAMESPACE
+)
 if not (twomemo_requested or oldmemo_requested):
 self.__text_commands.feed_back(
 client,
 _("You need to have OMEMO encryption activated to reset the session"),
-mess_data
+mess_data,
 )
 return False
 bare_jid = mess_data["to"].userhost()
 for device in devices:
 log.debug(f"Replacing sessions with device {device}")
 await session_manager.replace_sessions(device)
 self.__text_commands.feed_back(
-client,
+client, _("OMEMO session has been reset"), mess_data
-_("OMEMO session has been reset"),
-mess_data
 )
 return False
 async def get_trust_ui(  # pylint: disable=invalid-name
-self,
+self, client: SatXMPPClient, entity: jid.JID
-client: SatXMPPClient,
-entity: jid.JID
 ) -> xml_tools.XMLUI:
 """
 @param client: The client.
 @param entity: The entity whose device trust levels to manage.
 @return: An XMLUI instance which opens a form to manage the trust level of all
 bare_jids: Set[str]
 if self.__xep_0045 is not None and self.__xep_0045.is_joined_room(client, entity):
 bare_jids = self.__get_joined_muc_users(client, self.__xep_0045, entity)
 else:
-bare_jids = { entity.userhost() }
+bare_jids = {entity.userhost()}
 session_manager = await self.get_session_manager(client.profile)
 # At least sort the devices by bare JID such that they aren't listed completely
 # random
-devices = sorted(cast(Set[omemo.DeviceInformation], set()).union(*[
+devices = sorted(
-await session_manager.get_device_information(bare_jid)
+cast(Set[omemo.DeviceInformation], set()).union(
-for bare_jid
+*[
-in bare_jids
+await session_manager.get_device_information(bare_jid)
-]), key=lambda device: device.bare_jid)
+for bare_jid in bare_jids
+]
-async def callback(
+),
-data: Any,
+key=lambda device: device.bare_jid,
-profile: str
+)
-) -> Dict[Never, Never]:
+async def callback(data: Any, profile: str) -> Dict[Never, Never]:
 """
 @param data: The XMLUI result produces by the trust UI form.
 @param profile: The profile.
 @return: An empty dictionary. The type of the return value was chosen
 conservatively since the exact options are neither known not needed here.
 if C.bool(data.get("cancelled", "false")):
 return {}
 data_form_result = cast(
-Dict[str, str],
+Dict[str, str], xml_tools.xmlui_result_2_data_form_result(data)
-xml_tools.xmlui_result_2_data_form_result(data)
 )
 trust_updates: Set[TrustUpdate] = set()
 for key, value in data_form_result.items():
 if not key.startswith("trust_"):
 continue
-device = devices[int(key[len("trust_"):])]
+device = devices[int(key[len("trust_") :])]
 trust_level_name = value
 if device.trust_level_name != trust_level_name:
 await session_manager.set_trust(
-device.bare_jid,
+device.bare_jid, device.identity_key, trust_level_name
-device.identity_key,
-trust_level_name
 )
 target_trust: Optional[bool] = None
 if TrustLevel(trust_level_name) is TrustLevel.TRUSTED:
 target_trust = True
 if TrustLevel(trust_level_name) is TrustLevel.DISTRUSTED:
 target_trust = False
 if target_trust is not None:
-trust_updates.add(TrustUpdate(
+trust_updates.add(
-target_jid=jid.JID(device.bare_jid).userhostJID(),
+TrustUpdate(
-target_key=device.identity_key,
+target_jid=jid.JID(device.bare_jid).userhostJID(),
-target_trust=target_trust
+target_key=device.identity_key,
-))
+target_trust=target_trust,
+)
+)
 # Check whether ATM is enabled and handle everything in case it is
-trust_system = cast(str, self.host.memory.param_get_a(
+trust_system = cast(
-PARAM_NAME,
+str,
-PARAM_CATEGORY,
+self.host.memory.param_get_a(
-profile_key=profile
+PARAM_NAME, PARAM_CATEGORY, profile_key=profile
-))
+),
+)
 if trust_system == "atm":
 if len(trust_updates) > 0:
 await manage_trust_message_cache(
-client,
+client, session_manager, frozenset(trust_updates)
-session_manager,
-frozenset(trust_updates)
 )
 await send_trust_messages(
-client,
+client, session_manager, frozenset(trust_updates)
-session_manager,
-frozenset(trust_updates)
 )
 return {}
 submit_id = self.host.register_callback(callback, with_data=True, one_shot=True)
 result = xml_tools.XMLUI(
 panel_type=C.XMLUI_FORM,
 title=D_("OMEMO trust management"),
-submit_id=submit_id
+submit_id=submit_id,
 )
 # Casting this to Any, otherwise all calls on the variable cause type errors
 # pylint: disable=no-member
 trust_ui = cast(Any, result)
-trust_ui.addText(D_(
+trust_ui.addText(
-"This is OMEMO trusting system. You'll see below the devices of your"
+D_(
-" contacts, and a list selection to trust them or not. A trusted device"
+"This is OMEMO trusting system. You'll see below the devices of your"
-" can read your messages in plain text, so be sure to only validate"
+" contacts, and a list selection to trust them or not. A trusted device"
-" devices that you are sure are belonging to your contact. It's better"
+" can read your messages in plain text, so be sure to only validate"
-" to do this when you are next to your contact and their device, so"
+" devices that you are sure are belonging to your contact. It's better"
-" you can check the \"fingerprint\" (the number next to the device)"
+" to do this when you are next to your contact and their device, so"
-" yourself. Do *not* validate a device if the fingerprint is wrong!"
+' you can check the "fingerprint" (the number next to the device)'
-" Note that manually validating a fingerprint disables any form of automatic"
+" yourself. Do *not* validate a device if the fingerprint is wrong!"
-" trust."
+" Note that manually validating a fingerprint disables any form of automatic"
-))
+" trust."
+)
+)
 own_device, __ = await session_manager.get_own_device_information()
 trust_ui.change_container("label")
 trust_ui.addLabel(D_("This device ID"))
 trust_ui.addText(str(own_device.device_id))
 trust_ui.addLabel(D_("This device's fingerprint"))
-trust_ui.addText(" ".join(session_manager.format_identity_key(
+trust_ui.addText(
-own_device.identity_key
+" ".join(session_manager.format_identity_key(own_device.identity_key))
-)))
+)
 trust_ui.addEmpty()
 trust_ui.addEmpty()
 for index, device in enumerate(devices):
 trust_ui.addLabel(D_("Contact"))
 trust_ui.addJid(jid.JID(device.bare_jid))
 trust_ui.addLabel(D_("Device ID"))
 trust_ui.addText(str(device.device_id))
 trust_ui.addLabel(D_("Fingerprint"))
-trust_ui.addText(" ".join(session_manager.format_identity_key(
+trust_ui.addText(
-device.identity_key
+" ".join(session_manager.format_identity_key(device.identity_key))
-)))
+)
 trust_ui.addLabel(D_("Trust this device?"))
 current_trust_level = TrustLevel(device.trust_level_name)
-avaiable_trust_levels = \
+avaiable_trust_levels = {
-{ TrustLevel.DISTRUSTED, TrustLevel.TRUSTED, current_trust_level }
+TrustLevel.DISTRUSTED,
+TrustLevel.TRUSTED,
+current_trust_level,
+}
 trust_ui.addList(
 f"trust_{index}",
-options=[ trust_level.name for trust_level in avaiable_trust_levels ],
+options=[trust_level.name for trust_level in avaiable_trust_levels],
 selected=current_trust_level.name,
-styles=[ "inline" ]
+styles=["inline"],
 )
 twomemo_active = dict(device.active).get(twomemo.twomemo.NAMESPACE)
 if twomemo_active is None:
 trust_ui.addEmpty()
 return result
 @staticmethod
 def __get_joined_muc_users(
-client: SatXMPPClient,
+client: SatXMPPClient, xep_0045: XEP_0045, room_jid: jid.JID
-xep_0045: XEP_0045,
-room_jid: jid.JID
 ) -> Set[str]:
 """
 @param client: The client.
 @param xep_0045: A MUC plugin instance.
 @param room_jid: The room JID.
 self.__session_manager_waiters[profile] = []
 # Build and store the session manager
 try:
 session_manager = await prepare_for_profile(
-self.host,
+self.host, profile, initial_own_label="Libervia"
-profile,
-initial_own_label="Libervia"
 )
 except Exception as e:
 # In case of an error during initalization, notify the waiters accordingly
 # and delete them
 for waiter in self.__session_manager_waiters[profile]:
 self,
 client: SatXMPPClient,
 message_elt: domish.Element,
 session_manager: omemo.SessionManager,
 sender_device_information: omemo.DeviceInformation,
-timestamp: datetime
+timestamp: datetime,
 ) -> None:
 """Check a newly decrypted message stanza for ATM content and perform ATM in case.
 @param client: The client which received the message.
 @param message_elt: The message element. Can be modified.
 the message.
 @param timestamp: Timestamp extracted from the SCE time affix.
 """
 trust_message_cache = persistent.LazyPersistentBinaryDict(
-"XEP-0384/TM",
+"XEP-0384/TM", client.profile
-client.profile
 )
 new_cache_entries: Set[TrustMessageCacheEntry] = set()
 for trust_message_elt in message_elt.elements(NS_TM, "trust-message"):
 key_owner_jid = jid.JID(key_owner_elt["jid"]).userhostJID()
 for trust_elt in key_owner_elt.elements(NS_TM, "trust"):
 assert isinstance(trust_elt, domish.Element)
-new_cache_entries.add(TrustMessageCacheEntry(
+new_cache_entries.add(
-sender_jid=jid.JID(sender_device_information.bare_jid),
+TrustMessageCacheEntry(
-sender_key=sender_device_information.identity_key,
+sender_jid=jid.JID(sender_device_information.bare_jid),
-timestamp=timestamp,
+sender_key=sender_device_information.identity_key,
-trust_update=TrustUpdate(
+timestamp=timestamp,
-target_jid=key_owner_jid,
+trust_update=TrustUpdate(
-target_key=base64.b64decode(str(trust_elt)),
+target_jid=key_owner_jid,
-target_trust=True
+target_key=base64.b64decode(str(trust_elt)),
+target_trust=True,
+),
 )
-))
+)
 for distrust_elt in key_owner_elt.elements(NS_TM, "distrust"):
 assert isinstance(distrust_elt, domish.Element)
-new_cache_entries.add(TrustMessageCacheEntry(
+new_cache_entries.add(
-sender_jid=jid.JID(sender_device_information.bare_jid),
+TrustMessageCacheEntry(
-sender_key=sender_device_information.identity_key,
+sender_jid=jid.JID(sender_device_information.bare_jid),
-timestamp=timestamp,
+sender_key=sender_device_information.identity_key,
-trust_update=TrustUpdate(
+timestamp=timestamp,
-target_jid=key_owner_jid,
+trust_update=TrustUpdate(
-target_key=base64.b64decode(str(distrust_elt)),
+target_jid=key_owner_jid,
-target_trust=False
+target_key=base64.b64decode(str(distrust_elt)),
+target_trust=False,
+),
 )
-))
+)
 # Load existing cache entries
 existing_cache_entries = {
 TrustMessageCacheEntry.from_dict(d)
 for d in await trust_message_cache.get("cache", [])
 # Discard cache entries by timestamp comparison
 existing_by_target = {
 (
 cache_entry.trust_update.target_jid.userhostJID(),
-cache_entry.trust_update.target_key
+cache_entry.trust_update.target_key,
 ): cache_entry
-for cache_entry
+for cache_entry in existing_cache_entries
-in existing_cache_entries
 }
 # Iterate over a copy here, such that new_cache_entries can be modified
 for new_cache_entry in set(new_cache_entries):
 existing_cache_entry = existing_by_target.get(
 (
 new_cache_entry.trust_update.target_jid.userhostJID(),
-new_cache_entry.trust_update.target_key
+new_cache_entry.trust_update.target_key,
 ),
-None
+None,
 )
 if existing_cache_entry is not None:
 if existing_cache_entry.timestamp > new_cache_entry.timestamp:
 # If the existing cache entry is newer than the new cache entry,
 )
 await session_manager.set_trust(
 trust_update.target_jid.userhost(),
 trust_update.target_key,
-trust_level.name
+trust_level.name,
 )
 applied_trust_updates.add(trust_update)
 new_cache_entries.remove(cache_entry)
 # Store the remaining existing and new cache entries
 await trust_message_cache.force(
-"cache",
+"cache", [tm.to_dict() for tm in existing_cache_entries | new_cache_entries]
-[tm.to_dict() for tm in existing_cache_entries | new_cache_entries]
 )
 # If the trust of at least one device was modified, run the ATM cache update logic
 if len(applied_trust_updates) > 0:
 await manage_trust_message_cache(
-client,
+client, session_manager, frozenset(applied_trust_updates)
-session_manager,
-frozenset(applied_trust_updates)
 )
 async def _message_received_trigger(
 self,
 client: SatXMPPClient,
 message_elt: domish.Element,
-post_treat: defer.Deferred
+post_treat: defer.Deferred,
 ) -> bool:
 """
 @param client: The client which received the message.
 @param message_elt: The message element. Can be modified.
 @param post_treat: A deferred which evaluates to a :class:`MessageData` once the
 message_uid = self.__xep_0359.get_origin_id(message_elt)
 if message_uid is None:
 message_uid = message_elt.getAttribute("id")
 if message_uid is not None:
 muc_plaintext_cache_key = MUCPlaintextCacheKey(
-client,
+client, room_jid, message_uid
-room_jid,
-message_uid
 )
 else:
 # I'm not sure why this check is required, this code is copied from the old
 # plugin.
 if sender_jid.userhostJID() == client.jid.userhostJID():
 sender_bare_jid = sender_jid.userhost()
 message: Optional[omemo.Message] = None
 encrypted_elt: Optional[domish.Element] = None
-twomemo_encrypted_elt = cast(Optional[domish.Element], next(
+twomemo_encrypted_elt = cast(
-message_elt.elements(twomemo.twomemo.NAMESPACE, "encrypted"),
+Optional[domish.Element],
-None
+next(message_elt.elements(twomemo.twomemo.NAMESPACE, "encrypted"), None),
-))
+)
-oldmemo_encrypted_elt = cast(Optional[domish.Element], next(
+oldmemo_encrypted_elt = cast(
-message_elt.elements(oldmemo.oldmemo.NAMESPACE, "encrypted"),
+Optional[domish.Element],
-None
+next(message_elt.elements(oldmemo.oldmemo.NAMESPACE, "encrypted"), None),
-))
+)
 try:
 session_manager = await self.get_session_manager(cast(str, client.profile))
 except Exception as e:
 log.error(f"error while preparing profile for {client.profile}: {e}")
 return True
 if twomemo_encrypted_elt is not None:
 try:
 message = twomemo.etree.parse_message(
-xml_tools.domish_elt_2_et_elt(twomemo_encrypted_elt),
+xml_tools.domish_elt_2_et_elt(twomemo_encrypted_elt), sender_bare_jid
-sender_bare_jid
 )
 except (ValueError, XMLSchemaValidationError):
 log.warning(
 f"Ingoring malformed encrypted message for namespace"
 f" {twomemo.twomemo.NAMESPACE}: {twomemo_encrypted_elt.toXml()}"
 try:
 message = await oldmemo.etree.parse_message(
 xml_tools.domish_elt_2_et_elt(oldmemo_encrypted_elt),
 sender_bare_jid,
 client.jid.userhost(),
-session_manager
+session_manager,
 )
 except (ValueError, XMLSchemaValidationError):
 log.warning(
 f"Ingoring malformed encrypted message for namespace"
 f" {oldmemo.oldmemo.NAMESPACE}: {oldmemo_encrypted_elt.toXml()}"
 feedback_jid,
 D_(
 f"An OMEMO message from {sender_jid.full()} has not been"
 f" encrypted for our device, we can't decrypt it."
 ),
-{ C.MESS_EXTRA_INFO: C.EXTRA_INFO_DECR_ERR }
+{C.MESS_EXTRA_INFO: C.EXTRA_INFO_DECR_ERR},
 )
 log.warning("Message not encrypted for us.")
 else:
 log.debug("Message not encrypted for us.")
 # No point in further processing this message.
 return False
 except Exception as e:
-log.warning(_("Can't decrypt message: {reason}\n{xml}").format(
+log.warning(
-reason=e,
+_("Can't decrypt message: {reason}\n{xml}").format(
-xml=message_elt.toXml()
+reason=e, xml=message_elt.toXml()
-))
+)
+)
 client.feedback(
 feedback_jid,
 D_(
 f"An OMEMO message from {sender_jid.full()} can't be decrypted:"
 f" {e}"
 ),
-{ C.MESS_EXTRA_INFO: C.EXTRA_INFO_DECR_ERR }
+{C.MESS_EXTRA_INFO: C.EXTRA_INFO_DECR_ERR},
 )
 # No point in further processing this message
 return False
 affix_values: Optional[SCEAffixValues] = None
 if message.namespace == twomemo.twomemo.NAMESPACE:
 if plaintext is not None:
 # XEP_0420.unpack_stanza handles the whole unpacking, including the
 # relevant modifications to the element
-sce_profile = \
+sce_profile = (
 OMEMO.SCE_PROFILE_GROUPCHAT if is_muc_message else OMEMO.SCE_PROFILE
+)
 try:
 affix_values = self.__xep_0420.unpack_stanza(
-sce_profile,
+sce_profile, message_elt, plaintext
-message_elt,
-plaintext
 )
 except Exception as e:
-log.warning(D_(
+log.warning(
-f"Error unpacking SCE-encrypted message: {e}\n{plaintext}"
+D_(f"Error unpacking SCE-encrypted message: {e}\n{plaintext}")
-))
+)
 client.feedback(
 feedback_jid,
 D_(
 f"An OMEMO message from {sender_jid.full()} was rejected:"
 f" {e}"
 ),
-{ C.MESS_EXTRA_INFO: C.EXTRA_INFO_DECR_ERR }
+{C.MESS_EXTRA_INFO: C.EXTRA_INFO_DECR_ERR},
 )
 # No point in further processing this message
 return False
 else:
 if affix_values.timestamp is not None:
 if plaintext is not None:
 # Add the decrypted body
 message_elt.addElement("body", content=plaintext.decode("utf-8"))
 # Mark the message as trusted or untrusted. Undecided counts as untrusted here.
-trust_level = \
+trust_level = await session_manager._evaluate_custom_trust_level(
-await session_manager._evaluate_custom_trust_level(device_information)
+device_information
+)
 if trust_level is omemo.TrustLevel.TRUSTED:
 post_treat.addCallback(client.encryption.mark_as_trusted)
 else:
 post_treat.addCallback(client.encryption.mark_as_untrusted)
 # Mark the message as originally encrypted
 post_treat.addCallback(
-client.encryption.mark_as_encrypted,
+client.encryption.mark_as_encrypted, namespace=message.namespace
-namespace=message.namespace
 )
 # Handle potential ATM trust updates
 if affix_values is not None and affix_values.timestamp is not None:
 await self.__message_received_trigger_atm(
 client,
 message_elt,
 session_manager,
 device_information,
-affix_values.timestamp
+affix_values.timestamp,
 )
 # Message processed successfully, continue with the flow
 return True
 @param stanza: The stanza that is about to be sent. Can be modified.
 @return: Whether the send message flow should continue or not.
 """
 # SCE is only applicable to message and IQ stanzas
 # FIXME: temporary disabling IQ stanza encryption
-if stanza.name not in { "message" }:  # , "iq" }:
+if stanza.name not in {"message"}:  # , "iq" }:
 return True
 # Get the intended recipient
 recipient = stanza.getAttribute("to", None)
 if recipient is None:
 client,
 encryption_ns,
 stanza,
 recipient_bare_jid,
 stanza.getAttribute("type", C.MESS_TYPE_NORMAL) == C.MESS_TYPE_GROUPCHAT,
-stanza.getAttribute("id", None)
+stanza.getAttribute("id", None),
 )
 else:
 # Encryption is requested for this recipient, but not with twomemo
 return True
 # Add a store hint if this is a message stanza
 if stanza.name == "message":
-self.__xep_0334.add_hint_elements(stanza, [ "store" ])
+self.__xep_0334.add_hint_elements(stanza, ["store"])
 # Let the flow continue.
 return True
 async def download_missing_device_lists(
 not_in_roster = [j for j in recipients if not client.roster.is_jid_in_roster(j)]
 for bare_jid in not_in_roster:
 device_information = await session_manager.get_device_information(
 bare_jid.userhost()
 )
-if (
+if not device_information or not all(
-not device_information
+namespace in di.namespaces for di in device_information
-or not all(namespace in di.namespaces for di in device_information)
 ):
 if namespace == self.NS_TWOMEMO:
 algo, node = "OMEMO", TWOMEMO_DEVICE_LIST_NODE
 elif namespace == self.NS_OLDMEMO:
 algo, node = "OMEMO_legacy", OLDMEMO_DEVICE_LIST_NODE
 client: SatXMPPClient,
 namespace: NamespaceType,
 stanza: domish.Element,
 recipient_jids: Union[jid.JID, Set[jid.JID]],
 is_muc_message: bool,
-stanza_id: Optional[str]
+stanza_id: Optional[str],
 ) -> None:
 """
 @param client: The client.
 @param namespace: The namespace of the OMEMO version to use.
 @param stanza: The stanza. Twomemo will encrypt the whole stanza using SCE,
 )
 room_jid = feedback_jid = recipient_jid.userhostJID()
 recipient_bare_jids = self.__get_joined_muc_users(
-client,
+client, self.__xep_0045, room_jid
-self.__xep_0045,
-room_jid
 )
 muc_plaintext_cache_key = MUCPlaintextCacheKey(
-client=client,
+client=client, room_jid=room_jid, message_uid=stanza_id
-room_jid=room_jid,
-message_uid=stanza_id
 )
 else:
 recipient_bare_jids = {r.userhost() for r in recipient_jids}
 feedback_jid = recipient_jid.userhostJID()
 """
 if namespace == twomemo.twomemo.NAMESPACE:
 return self.__xep_0420.pack_stanza(
 OMEMO.SCE_PROFILE_GROUPCHAT if is_muc_message else OMEMO.SCE_PROFILE,
-stanza
+stanza,
 )
 if namespace == oldmemo.oldmemo.NAMESPACE:
 plaintext: Optional[bytes] = None
 for child in stanza.elements():
 if child.name == "body" and plaintext is None:
 plaintext = str(child).encode("utf-8")
 # Any other sensitive elements to remove here?
-if child.name in { "body", "html" }:
+if child.name in {"body", "html"}:
 stanza.children.remove(child)
 if plaintext is None:
 log.warning(
 "No body found in intercepted message to be encrypted with"
 return
 log.debug(f"Plaintext to encrypt: {plaintext}")
 session_manager = await self.get_session_manager(client.profile)
-await self.download_missing_device_lists(client, namespace, recipient_jids, session_manager)
+await self.download_missing_device_lists(
+client, namespace, recipient_jids, session_manager
+)
 try:
 messages, encryption_errors = await session_manager.encrypt(
 frozenset(recipient_bare_jids),
-{ namespace: plaintext },
+{namespace: plaintext},
-backend_priority_order=[ namespace ],
+backend_priority_order=[namespace],
-identifier=feedback_jid.userhost()
+identifier=feedback_jid.userhost(),
 )
 except Exception as e:
 msg = _(
 # pylint: disable=consider-using-f-string
 "Can't encrypt message for {entities}: {reason}".format(
-entities=', '.join(recipient_bare_jids),
+entities=", ".join(recipient_bare_jids), reason=e
-reason=e
 )
 )
 log.warning(msg)
-client.feedback(feedback_jid, msg, {
+client.feedback(feedback_jid, msg, {C.MESS_EXTRA_INFO: C.EXTRA_INFO_ENCR_ERR})
-C.MESS_EXTRA_INFO: C.EXTRA_INFO_ENCR_ERR
-})
 raise e
 if len(encryption_errors) > 0:
 log.warning(
 f"Ignored the following non-critical encryption errors:"
 f" {encryption_errors}"
 )
-encrypted_errors_stringified = ", ".join([
+encrypted_errors_stringified = ", ".join(
-f"device {err.device_id} of {err.bare_jid} under namespace"
+[
-f" {err.namespace}"
+f"device {err.device_id} of {err.bare_jid} under namespace"
-for err
+f" {err.namespace}"
-in encryption_errors
+for err in encryption_errors
-])
+]
+)
 client.feedback(
 feedback_jid,
 D_(
 "There were non-critical errors during encryption resulting in some"
 " of your destinees' devices potentially not receiving the message."
 " This happens when the encryption data/key material of a device is"
 " incomplete or broken, which shouldn't happen for actively used"
 " devices, and can usually be ignored. The following devices are"
 f" affected: {encrypted_errors_stringified}."
-)
+),
 )
 message = next(message for message in messages if message.namespace == namespace)
 if namespace == twomemo.twomemo.NAMESPACE:
 # Add the encrypted element
-stanza.addChild(xml_tools.et_elt_2_domish_elt(
+stanza.addChild(
-twomemo.etree.serialize_message(message)
+xml_tools.et_elt_2_domish_elt(twomemo.etree.serialize_message(message))
-))
+)
 if namespace == oldmemo.oldmemo.NAMESPACE:
 # Add the encrypted element
-stanza.addChild(xml_tools.et_elt_2_domish_elt(
+stanza.addChild(
-oldmemo.etree.serialize_message(message)
+xml_tools.et_elt_2_domish_elt(oldmemo.etree.serialize_message(message))
-))
+)
 if muc_plaintext_cache_key is not None:
 self.__muc_plaintext_cache[muc_plaintext_cache_key] = plaintext
 async def __on_device_list_update(
-self,
+self, items_event: pubsub.ItemsEvent, profile: str
-items_event: pubsub.ItemsEvent,
-profile: str
 ) -> None:
 """Handle device list updates fired by PEP.
 @param items_event: The event.
 @param profile: The profile this event belongs to.
 items = cast(List[domish.Element], items_event.items)
 client = self.host.get_client(profile)
 await self._update_device_list(client, sender, items)
 async def _update_device_list(
-self,
+self, client: SatXMPPEntity, sender: jid.JID, items: list[domish.Element]
-client: SatXMPPEntity,
-sender: jid.JID,
-items: list[domish.Element]
 ) -> None:
 if len(items) > 1:
 log.warning("Ignoring device list update with more than one element.")
 return
 return
 session_manager = await self.get_session_manager(client.profile)
 await session_manager.update_device_list(
-namespace,
+namespace, sender.userhost(), device_list
-sender.userhost(),
+)
-device_list
-)

Mercurial > libervia-backend

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