libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0373.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	4836b81c5f31
children

comparison

equal deleted inserted replaced

-:c4464d7ae97b
+:524856bd7b19
 @raise UnknownKey: if the public key is not available.
 """
 @abstractmethod
 def import_public_key(self, packet: bytes) -> GPGPublicKey:
-"""Import a public key from a key material packet according to RFC 4880 §5.5.
+"""import a public key from a key material packet according to RFC 4880 §5.5.
 OpenPGP's ASCII Armor is not used.
 @param packet: A packet containing an exported public key.
 @return: The public key imported from the packet.
 self.host = host
 # Add configuration option to choose between manual trust and BTBV as the trust
 # model
-host.memory.updateParams(DEFAULT_TRUST_MODEL_PARAM)
+host.memory.update_params(DEFAULT_TRUST_MODEL_PARAM)
 self.__xep_0045 = cast(Optional[XEP_0045], host.plugins.get("XEP-0045"))
 self.__xep_0060 = cast(XEP_0060, host.plugins["XEP-0060"])
 self.__storage: Dict[str, persistent.LazyPersistentBinaryDict] = {}
 xep_0163 = cast(XEP_0163, host.plugins["XEP-0163"])
-xep_0163.addPEPEvent(
+xep_0163.add_pep_event(
 "OX_PUBLIC_KEYS_LIST",
 PUBLIC_KEYS_LIST_NODE,
 lambda items_event, profile: defer.ensureDeferred(
 self.__on_public_keys_list_update(items_event, profile)
 )
 )
-async def profileConnecting(self, client):
+async def profile_connecting(self, client):
 client.gpg_provider = get_gpg_provider(self.host, client)
-async def profileConnected(  # pylint: disable=invalid-name
+async def profile_connected(  # pylint: disable=invalid-name
 self,
 client: SatXMPPClient
 ) -> None:
 """
 @param client: The client.
 @param items_event: The event.
 @param profile: The profile this event belongs to.
 """
-client = self.host.getClient(profile)
+client = self.host.get_client(profile)
 sender = cast(jid.JID, items_event.sender)
 items = cast(List[domish.Element], items_event.items)
 if len(items) > 1:
 ))
 encryption_keys: Set[GPGPublicKey] = set()
 for recipient_jid in recipient_jids:
-# Import all keys of the recipient
+# import all keys of the recipient
 all_public_keys = await self.import_all_public_keys(client, recipient_jid)
 # Filter for keys that can encrypt
 encryption_keys |= set(filter(gpg_provider.can_encrypt, all_public_keys))
 decryption_keys = set(filter(
 lambda secret_key: gpg_provider.can_encrypt(secret_key.public_key),
 self.list_secret_keys(client)
 ))
-# Import all keys of the sender
+# import all keys of the sender
 all_public_keys = await self.import_all_public_keys(client, sender_jid)
 # Filter for keys that can sign
 verification_keys = set(filter(gpg_provider.can_sign, all_public_keys))
 pubkey_elt = domish.Element((NS_OX, "pubkey"))
 pubkey_elt.addElement("data", content=base64.b64encode(packet).decode("ASCII"))
 try:
-await self.__xep_0060.sendItem(
+await self.__xep_0060.send_item(
 client,
 client.jid.userhostJID(),
 node,
 pubkey_elt,
 format_datetime(),
 async def import_all_public_keys(
 self,
 client: SatXMPPClient,
 entity_jid: jid.JID
 ) -> Set[GPGPublicKey]:
-"""Import all public keys of a JID that have not been imported before.
+"""import all public keys of a JID that have not been imported before.
 @param client: The client.
 @param jid: The JID. Can be a bare JID.
 @return: The public keys.
 @note: Failure to import a key simply results in the key not being included in the
 available_public_keys.add(
 await self.import_public_key(client, entity_jid, missing_key.fingerprint)
 )
 except Exception as e:
 log.warning(
-f"Import of public key {missing_key.fingerprint} owned by"
+f"import of public key {missing_key.fingerprint} owned by"
 f" {entity_jid.userhost()} failed, ignoring: {e}"
 )
 return available_public_keys
 self,
 client: SatXMPPClient,
 jid: jid.JID,
 fingerprint: str
 ) -> GPGPublicKey:
-"""Import a public key.
+"""import a public key.
 @param client: The client.
 @param jid: The JID owning the public key. Can be a bare JID.
 @param fingerprint: The fingerprint of the public key.
 @return: The public key.
 gpg_provider = get_gpg_provider(self.host, client)
 node = f"urn:xmpp:openpgp:0:public-keys:{fingerprint}"
 try:
-items, __ = await self.__xep_0060.getItems(
+items, __ = await self.__xep_0060.get_items(
 client,
 jid.userhostJID(),
 node,
 max_items=1
 )
 pubkey_metadata_elt = public_keys_list_elt.addElement("pubkey-metadata")
 pubkey_metadata_elt["v4-fingerprint"] = public_key_metadata.fingerprint
 pubkey_metadata_elt["date"] = format_datetime(public_key_metadata.timestamp)
 try:
-await self.__xep_0060.sendItem(
+await self.__xep_0060.send_item(
 client,
 client.jid.userhostJID(),
 node,
 public_keys_list_elt,
 item_id=XEP_0060.ID_SINGLETON,
 """
 node = "urn:xmpp:openpgp:0:public-keys"
 try:
-items, __ = await self.__xep_0060.getItems(
+items, __ = await self.__xep_0060.get_items(
 client,
 jid.userhostJID(),
 node,
 max_items=1
 )
 protocols or protocol extensions.
 @raise XMPPInteractionFailed: if any interaction via XMPP failed.
 """
 try:
-infos = cast(DiscoInfo, await self.host.memory.disco.getInfos(
+infos = cast(DiscoInfo, await self.host.memory.disco.get_infos(
 client,
 client.jid.userhostJID()
 ))
 except Exception as e:
 raise XMPPInteractionFailed(
 # TODO: persistent-items is a SHOULD, how do we handle the feature missing?
 node = "urn:xmpp:openpgp:0:secret-key"
 try:
-items, __ = await self.__xep_0060.getItems(
+items, __ = await self.__xep_0060.get_items(
 client,
 client.jid.userhostJID(),
 node,
 max_items=1
 )
 secretkey_elt = domish.Element((NS_OX, "secretkey"))
 secretkey_elt.addContent(base64.b64encode(ciphertext).decode("ASCII"))
 try:
-await self.__xep_0060.sendItem(
+await self.__xep_0060.send_item(
 client,
 client.jid.userhostJID(),
 node,
 secretkey_elt
 )
 self,
 client: SatXMPPClient,
 ciphertext: bytes,
 backup_code: str
 ) -> Set[GPGSecretKey]:
-"""Import previously downloaded secret keys.
+"""import previously downloaded secret keys.
 The downloading and importing steps are separate since a backup code is required
 for the import and it should be possible to try multiple backup codes without
 redownloading the data every time. The first half of the import procedure is
 provided by :meth:`download_secret_keys`.
 # TODO: This should probably be a global helper somewhere
 bare_jids: Set[jid.JID] = 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
 key = f"/trust/{owner.userhost()}/{public_key.fingerprint}"
 await self.__storage[client.profile].force(key, trust_level.name)
-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[jid.JID]
-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.userhostJID() }
 all_public_keys = list({
 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)
 )
 for key, value in data_form_result.items():
 if not key.startswith("trust_"):
 continue
 if (await self.get_trust(client, public_key, owner)) is not trust:
 await self.set_trust(client, public_key, owner, value)
 return {}
-submit_id = self.host.registerCallback(callback, with_data=True, one_shot=True)
+submit_id = self.host.register_callback(callback, with_data=True, one_shot=True)
 result = xml_tools.XMLUI(
 panel_type=C.XMLUI_FORM,
 title=D_("OX trust management"),
 submit_id=submit_id
 "yourself. Do *not* validate a key if the fingerprint is wrong!"
 ))
 own_secret_keys = self.list_secret_keys(client)
-trust_ui.changeContainer("label")
+trust_ui.change_container("label")
 for index, secret_key in enumerate(own_secret_keys):
 trust_ui.addLabel(D_(f"Own secret key {index} fingerprint"))
 trust_ui.addText(secret_key.public_key.fingerprint)
 trust_ui.addEmpty()
 trust_ui.addEmpty()

Mercurial > libervia-backend

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