libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0384.py @ 4037:524856bd7b19

massive refactoring to switch from camelCase to snake_case: historically, Libervia (SàT before) was using camelCase as allowed by PEP8 when using a pre-PEP8 code, to use the same coding style as in Twisted. However, snake_case is more readable and it's better to follow PEP8 best practices, so it has been decided to move on full snake_case. Because Libervia has a huge codebase, this ended with a ugly mix of camelCase and snake_case. To fix that, this patch does a big refactoring by renaming every function and method (including bridge) that are not coming from Twisted or Wokkel, to use fully snake_case. This is a massive change, and may result in some bugs.

author	Goffi <goffi@goffi.org>
date	Sat, 08 Apr 2023 13:54:42 +0200
parents	1d5a81e3c9e8
children	c23cad65ae99

comparison

equal deleted inserted replaced

-:c4464d7ae97b
+:524856bd7b19
 KEY_ALL_JIDS = "ALL_JIDS"
 def __init__(self, profile: str, own_bare_jid: str) -> None:
 """
 @param profile: The profile this OMEMO data belongs to.
-@param own_bare_jid: The own bare JID, to return by the :meth:`loadOwnData` call.
+@param own_bare_jid: The own bare JID, to return by the :meth:`load_own_data` call.
 """
 self.__storage = persistent.LazyPersistentBinaryDict("XEP-0384", profile)
 self.__own_bare_jid = own_bare_jid
 namespace = oldmemo.oldmemo.NAMESPACE
 node = f"eu.siacs.conversations.axolotl.bundles:{device_id}"
 try:
-items, __ = await xep_0060.getItems(client, jid.JID(bare_jid), node, max_items=1)
+items, __ = await xep_0060.get_items(client, jid.JID(bare_jid), node, max_items=1)
 except Exception as e:
 raise omemo.BundleDownloadFailed(
 f"Bundle download failed for {bare_jid}: {device_id} under namespace"
 f" {namespace}"
 ) from e
 (NS_TM, "distrust"),
 content=serialized_identity_key
 )
 # Finally, encrypt and send the trust message!
-message_data = client.generateMessageXML(MessageData({
+message_data = client.generate_message_xml(MessageData({
 "from": own_jid,
 "to": recipient_jid,
 "uid": str(uuid.uuid4()),
 "message": {},
 "subject": {},
 @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.
 """
-client = sat.getClient(profile)
+client = sat.get_client(profile)
 xep_0060 = cast(XEP_0060, sat.plugins["XEP-0060"])
 class SessionManagerImpl(omemo.SessionManager):
 """
 Session manager implementation handling XMPP interactions and trust via an
 if isinstance(bundle, twomemo.twomemo.BundleImpl):
 element = twomemo.etree.serialize_bundle(bundle)
 node = "urn:xmpp:omemo:2:bundles"
 try:
-await xep_0060.sendItem(
+await xep_0060.send_item(
 client,
 client.jid.userhostJID(),
 node,
 xml_tools.et_elt_2_domish_elt(element),
 item_id=str(bundle.device_id),
 if isinstance(bundle, oldmemo.oldmemo.BundleImpl):
 element = oldmemo.etree.serialize_bundle(bundle)
 node = f"eu.siacs.conversations.axolotl.bundles:{bundle.device_id}"
 try:
-await xep_0060.sendItem(
+await xep_0060.send_item(
 client,
 client.jid.userhostJID(),
 node,
 xml_tools.et_elt_2_domish_elt(element),
 item_id=xep_0060.ID_SINGLETON,
 ) -> omemo.Bundle:
 if namespace == twomemo.twomemo.NAMESPACE:
 node = "urn:xmpp:omemo:2:bundles"
 try:
-items, __ = await xep_0060.getItems(
+items, __ = await xep_0060.get_items(
 client,
 jid.JID(bare_jid),
 node,
 item_ids=[ str(device_id) ]
 )
 async def _delete_bundle(namespace: str, device_id: int) -> None:
 if namespace == twomemo.twomemo.NAMESPACE:
 node = "urn:xmpp:omemo:2:bundles"
 try:
-await xep_0060.retractItems(
+await xep_0060.retract_items(
 client,
 client.jid.userhostJID(),
 node,
 [ str(device_id) ],
 notify=False
 if element is None or node is None:
 raise omemo.UnknownNamespace(f"Unknown namespace: {namespace}")
 try:
-await xep_0060.sendItem(
+await xep_0060.send_item(
 client,
 client.jid.userhostJID(),
 node,
 xml_tools.et_elt_2_domish_elt(element),
 item_id=xep_0060.ID_SINGLETON,
 if node is None:
 raise omemo.UnknownNamespace(f"Unknown namespace: {namespace}")
 try:
-items, __ = await xep_0060.getItems(client, jid.JID(bare_jid), node)
+items, __ = await xep_0060.get_items(client, jid.JID(bare_jid), node)
 except exceptions.NotFound:
 return {}
 except Exception as e:
 raise omemo.DeviceListDownloadFailed(
 f"Device list download failed for {bare_jid} under namespace"
 # 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.getParamA(
+trust_system = cast(str, sat.memory.param_get_a(
 PARAM_NAME,
 PARAM_CATEGORY,
 profile_key=profile
 ))
 element = oldmemo.etree.serialize_message(message)
 if element is None:
 raise omemo.UnknownNamespace(f"Unknown namespace: {message.namespace}")
-message_data = client.generateMessageXML(MessageData({
+message_data = client.generate_message_xml(MessageData({
 "from": client.jid,
 "to": jid.JID(bare_jid),
 "uid": str(uuid.uuid4()),
 "message": {},
 "subject": {},
 one message, the recipient JID. In case of a MUC message, the room JID.
 @raise TrustDecisionFailed: if the user cancels the prompt.
 """
 # This session manager handles encryption with both twomemo and oldmemo, but
-# both are currently registered as different plugins and the `deferXMLUI`
+# both are currently registered as different plugins and the `defer_xmlui`
 # below requires a single namespace identifying the encryption plugin. Thus,
 # get the namespace of the requested encryption method from the encryption
 # session using the feedback JID.
 encryption = client.encryption.getSession(feedback_jid)
 if encryption is None:
 "yourself. Do *not* validate a device if the fingerprint is wrong!"
 ))
 own_device, __ = await self.get_own_device_information()
-trust_ui.changeContainer("label")
+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(self.format_identity_key(own_device.identity_key)))
 trust_ui.addEmpty()
 trust_ui.addLabel(D_("Device ID"))
 trust_ui.addText(str(device.device_id))
 trust_ui.addLabel(D_("Fingerprint"))
 trust_ui.addText(" ".join(self.format_identity_key(device.identity_key)))
 trust_ui.addLabel(D_("Trust this device?"))
-trust_ui.addBool(f"trust_{index}", value=C.boolConst(False))
+trust_ui.addBool(f"trust_{index}", value=C.bool_const(False))
 trust_ui.addEmpty()
 trust_ui.addEmpty()
-trust_ui_result = await xml_tools.deferXMLUI(
+trust_ui_result = await xml_tools.defer_xmlui(
 sat,
 trust_ui,
 action_extra={ "meta_encryption_trust": namespace },
 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.XMLUIResult2DataFormResult(
+data_form_result = cast(Dict[str, str], xml_tools.xmlui_result_2_data_form_result(
 trust_ui_result
 ))
 trust_updates: Set[TrustUpdate] = set()
 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, sat.memory.getParamA(
+trust_system = cast(str, sat.memory.param_get_a(
 PARAM_NAME,
 PARAM_CATEGORY,
 profile_key=profile
 ))
 :meth:`~omemo.session_manager.SessionManager.create`.
 @raise DeviceListDownloadFailed: if a device list download failed. Forwarded from
 :meth:`~omemo.session_manager.SessionManager.create`.
 """
-client = sat.getClient(profile)
+client = sat.get_client(profile)
 xep_0060 = cast(XEP_0060, sat.plugins["XEP-0060"])
 storage = StorageImpl(profile)
 # TODO: Untested
 self.__sat = sat
 # Add configuration option to choose between manual trust and BTBV as the trust
 # model
-sat.memory.updateParams(DEFAULT_TRUST_MODEL_PARAM)
+sat.memory.update_params(DEFAULT_TRUST_MODEL_PARAM)
 # Plugins
 self.__xep_0045 = cast(Optional[XEP_0045], sat.plugins.get("XEP-0045"))
 self.__xep_0334 = cast(XEP_0334, sat.plugins["XEP-0334"])
 self.__xep_0359 = cast(Optional[XEP_0359], sat.plugins.get("XEP-0359"))
 "messageReceived",
 self._message_received_trigger,
 priority=100050
 )
 sat.trigger.add(
-"sendMessageData",
+"send_message_data",
 self.__send_message_data_trigger,
 priority=100050
 )
 # These triggers are used by twomemo, which does do SCE
 sat.trigger.add("send", self.__send_trigger, priority=0)
 # TODO: Add new triggers here for freshly received and about-to-be-sent stanzas,
 # including IQs.
 # Give twomemo a (slightly) higher priority than oldmemo
-sat.registerEncryptionPlugin(self, "TWOMEMO", twomemo.twomemo.NAMESPACE, 101)
+sat.register_encryption_plugin(self, "TWOMEMO", twomemo.twomemo.NAMESPACE, 101)
-sat.registerEncryptionPlugin(self, "OLDMEMO", oldmemo.oldmemo.NAMESPACE, 100)
+sat.register_encryption_plugin(self, "OLDMEMO", oldmemo.oldmemo.NAMESPACE, 100)
 xep_0163 = cast(XEP_0163, sat.plugins["XEP-0163"])
-xep_0163.addPEPEvent(
+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.addPEPEvent(
+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, sat.plugins[C.TEXT_CMDS])
 except KeyError:
 log.info(_("Text commands not available"))
 else:
-self.__text_commands.registerTextCommands(self)
+self.__text_commands.register_text_commands(self)
-def profileConnected(  # pylint: disable=invalid-name
+def profile_connected(  # pylint: disable=invalid-name
 self,
 client: SatXMPPClient
 ) -> None:
 """
 @param client: The client.
 @return: The constant value ``False``, indicating to the text commands plugin that
 the message is not supposed to be sent.
 """
 twomemo_requested = \
-client.encryption.isEncryptionRequested(mess_data, twomemo.twomemo.NAMESPACE)
+client.encryption.is_encryption_requested(mess_data, twomemo.twomemo.NAMESPACE)
 oldmemo_requested = \
-client.encryption.isEncryptionRequested(mess_data, oldmemo.oldmemo.NAMESPACE)
+client.encryption.is_encryption_requested(mess_data, oldmemo.oldmemo.NAMESPACE)
 if not (twomemo_requested or oldmemo_requested):
-self.__text_commands.feedBack(
+self.__text_commands.feed_back(
 client,
 _("You need to have OMEMO encryption activated to reset the session"),
 mess_data
 )
 return False
 for device in devices:
 log.debug(f"Replacing sessions with device {device}")
 await session_manager.replace_sessions(device)
-self.__text_commands.feedBack(
+self.__text_commands.feed_back(
 client,
 _("OMEMO session has been reset"),
 mess_data
 )
 return False
-async def getTrustUI(  # pylint: disable=invalid-name
+async def get_trust_ui(  # pylint: disable=invalid-name
 self,
 client: SatXMPPClient,
 entity: jid.JID
 ) -> xml_tools.XMLUI:
 """
 if entity.resource:
 raise ValueError("A bare JID is expected.")
 bare_jids: Set[str]
-if self.__xep_0045 is not None and self.__xep_0045.isJoinedRoom(client, entity):
+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() }
 session_manager = await self.get_session_manager(client.profile)
 if C.bool(data.get("cancelled", "false")):
 return {}
 data_form_result = cast(
 Dict[str, str],
-xml_tools.XMLUIResult2DataFormResult(data)
+xml_tools.xmlui_result_2_data_form_result(data)
 )
 trust_updates: Set[TrustUpdate] = set()
 for key, value in data_form_result.items():
 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.__sat.memory.getParamA(
+trust_system = cast(str, self.__sat.memory.param_get_a(
 PARAM_NAME,
 PARAM_CATEGORY,
 profile_key=profile
 ))
 frozenset(trust_updates)
 )
 return {}
-submit_id = self.__sat.registerCallback(callback, with_data=True, one_shot=True)
+submit_id = self.__sat.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
 " trust."
 ))
 own_device, __ = await session_manager.get_own_device_information()
-trust_ui.changeContainer("label")
+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(
 own_device.identity_key
 """
 bare_jids: Set[str] = set()
 try:
-room = cast(muc.Room, xep_0045.getRoom(client, room_jid))
+room = cast(muc.Room, xep_0045.get_room(client, room_jid))
 except exceptions.NotFound as e:
 raise exceptions.InternalError(
 "Participant list of unjoined MUC requested."
 ) from e
 return True
 room_jid = feedback_jid = sender_jid.userhostJID()
 try:
-room = cast(muc.Room, self.__xep_0045.getRoom(client, room_jid))
+room = cast(muc.Room, self.__xep_0045.get_room(client, room_jid))
 except exceptions.NotFound:
 log.warning(
 f"Ignoring MUC message from a room that has not been joined:"
 f" {room_jid}"
 )
 sender_jid = sender_user_jid
 message_uid: Optional[str] = None
 if self.__xep_0359 is not None:
-message_uid = self.__xep_0359.getOriginId(message_elt)
+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,
 # Mark the message as trusted or untrusted. Undecided counts as untrusted here.
 trust_level = \
 await session_manager._evaluate_custom_trust_level(device_information)
 if trust_level is omemo.TrustLevel.TRUSTED:
-post_treat.addCallback(client.encryption.markAsTrusted)
+post_treat.addCallback(client.encryption.mark_as_trusted)
 else:
-post_treat.addCallback(client.encryption.markAsUntrusted)
+post_treat.addCallback(client.encryption.mark_as_untrusted)
 # Mark the message as originally encrypted
 post_treat.addCallback(
-client.encryption.markAsEncrypted,
+client.encryption.mark_as_encrypted,
 namespace=message.namespace
 )
 # Handle potential ATM trust updates
 if affix_values is not None and affix_values.timestamp is not None:
 stanza.getAttribute("id", None)
 )
 # Add a store hint if this is a message stanza
 if stanza.name == "message":
-self.__xep_0334.addHintElements(stanza, [ "store" ])
+self.__xep_0334.add_hint_elements(stanza, [ "store" ])
 # Let the flow continue.
 return True
 async def __send_message_data_trigger(
 is_muc_message,
 stanza_id
 )
 # Add a store hint
-self.__xep_0334.addHintElements(stanza, [ "store" ])
+self.__xep_0334.add_hint_elements(stanza, [ "store" ])
 async def encrypt(
 self,
 client: SatXMPPClient,
 namespace: Literal["urn:xmpp:omemo:2", "eu.siacs.conversations.axolotl"],

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0384.py @ 4037:524856bd7b19