libervia-backend: libervia/backend/plugins/plugin_xep

comparison libervia/backend/plugins/plugin_xep_0384.py @ 4216:1a7a3e4b52a4

core (memory/migration): Update XEP-0384 and `fe3a02cb4bec_convert_legacypickle_columns_to_json.py` migration to properly handle (de)serialisation of `TrustMessageCacheEntry`.

author	Goffi <goffi@goffi.org>
date	Tue, 05 Mar 2024 17:31:12 +0100
parents	15482dc0b5d1
children	c0f3f29377f1

comparison

equal deleted inserted replaced

-:31c84a32c897
+:1a7a3e4b52a4
 target_jid: jid.JID
 target_key: bytes
 target_trust: bool
+def to_dict(self) -> dict[str, Any]:
+"""Convert the instance to a serialised dictionary"""
+data = {
+"target_jid": self.target_jid.full(),
+"target_key": self.target_key.hex(),
+"target_trust": self.target_trust,
+}
+return data
+@staticmethod
+def from_dict(data: dict[str, Any]) -> "TrustUpdate":
+"""Load a serialized dictionary"""
+data["target_jid"] = jid.JID(data["target_jid"])
+data["target_key"] = bytes.fromhex(data["target_key"])
+return TrustUpdate(**data)
 class TrustMessageCacheEntry(NamedTuple):
 # pylint: disable=invalid-name
 """
 An entry in the trust message cache used by ATM.
 sender_jid: jid.JID
 sender_key: bytes
 timestamp: datetime
 trust_update: TrustUpdate
+def to_dict(self) -> dict[str, Any]:
+"""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()
+}
+return data
+@staticmethod
+def from_dict(data: dict[str, Any]) -> "TrustMessageCacheEntry":
+"""Load a serialized dictionary"""
+data["sender_jid"] = jid.JID(data["sender_jid"])
+data["sender_key"] = bytes.fromhex(data["sender_key"])
+data["timestamp"] = datetime.fromisoformat(data["timestamp"])
+data["trust_update"] = TrustUpdate.from_dict(data["trust_update"])
+return TrustMessageCacheEntry(**data)
 class PartialTrustMessage(NamedTuple):
 # pylint: disable=invalid-name
 """
 "XEP-0384/TM",
 client.profile
 )
 # Load cache entries
-cache_entries = cast(
+cache_entries = {
-Set[TrustMessageCacheEntry],
+TrustMessageCacheEntry.from_dict(d)
-await trust_message_cache.get("cache", set())
+for d in await trust_message_cache.get("cache", [])
-)
+}
 # Expire cache entries that were overwritten by the applied trust updates
 cache_entries_by_target = {
 (
 cache_entry.trust_update.target_jid.userhostJID(),
 # Remove the corresponding cache entry
 cache_entries.remove(cache_entry)
 # Store the updated cache entries
-await trust_message_cache.force("cache", cache_entries)
+await trust_message_cache.force(
+"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
 to_policy=SCEAffixPolicy.OPTIONAL,
 from_policy=SCEAffixPolicy.OPTIONAL,
 custom_policies={}
 )
-def __init__(self, sat: LiberviaBackend) -> None:
+def __init__(self, host: LiberviaBackend) -> None:
 """
 @param sat: The SAT instance.
 """
-self.__sat = sat
+self.host = host
 # Add configuration option to choose between manual trust and BTBV as the trust
 # model
-sat.memory.update_params(DEFAULT_TRUST_MODEL_PARAM)
+host.memory.update_params(DEFAULT_TRUST_MODEL_PARAM)
 # Plugins
-self.__xep_0045 = cast(Optional[XEP_0045], sat.plugins.get("XEP-0045"))
+self._j = cast(XEP_0060, host.plugins["XEP-0060"])
-self.__xep_0334 = cast(XEP_0334, sat.plugins["XEP-0334"])
+self.__xep_0045 = cast(Optional[XEP_0045], host.plugins.get("XEP-0045"))
-self.__xep_0359 = cast(Optional[XEP_0359], sat.plugins.get("XEP-0359"))
+self.__xep_0334 = cast(XEP_0334, host.plugins["XEP-0334"])
-self.__xep_0420 = cast(XEP_0420, sat.plugins["XEP-0420"])
+self.__xep_0359 = cast(Optional[XEP_0359], host.plugins.get("XEP-0359"))
+self.__xep_0420 = cast(XEP_0420, host.plugins["XEP-0420"])
 # In contrast to one to one messages, MUC messages are reflected to the sender.
 # Thus, the sender does not add messages to their local message log when sending
 # them, but when the reflection is received. This approach does not pair well with
 # OMEMO, since for security reasons it is forbidden to encrypt messages for the
 self.__session_manager_waiters: Dict[str, List[defer.Deferred]] = {}
 # 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.
-sat.trigger.add(
+host.trigger.add(
 "message_received",
 self._message_received_trigger,
 priority=100050
 )
-sat.trigger.add("send", self.__send_trigger, priority=0)
+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.
 # Give twomemo a (slightly) higher priority than oldmemo
-sat.register_encryption_plugin(self, "OMEMO", twomemo.twomemo.NAMESPACE, 101)
+host.register_encryption_plugin(self, "OMEMO", twomemo.twomemo.NAMESPACE, 101)
-sat.register_encryption_plugin(
+host.register_encryption_plugin(
 self, "OMEMO_legacy", oldmemo.oldmemo.NAMESPACE, 100
 )
-xep_0163 = cast(XEP_0163, sat.plugins["XEP-0163"])
+xep_0163 = cast(XEP_0163, host.plugins["XEP-0163"])
 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)
 self.__on_device_list_update(items_event, profile)
 )
 )
 try:
-self.__text_commands = cast(TextCommands, sat.plugins[C.TEXT_CMDS])
+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)
 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.param_get_a(
+trust_system = cast(str, self.host.memory.param_get_a(
 PARAM_NAME,
 PARAM_CATEGORY,
 profile_key=profile
 ))
 frozenset(trust_updates)
 )
 return {}
-submit_id = self.__sat.register_callback(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_("OMEMO trust management"),
 submit_id=submit_id
 self.__session_manager_waiters[profile] = []
 # Build and store the session manager
 try:
 session_manager = await prepare_for_profile(
-self.__sat,
+self.host,
 profile,
 initial_own_label="Libervia"
 )
 except Exception as e:
 # In case of an error during initalization, notify the waiters accordingly
 target_trust=False
 )
 ))
 # Load existing cache entries
-existing_cache_entries = cast(
+existing_cache_entries = {
-Set[TrustMessageCacheEntry],
+TrustMessageCacheEntry.from_dict(d)
-await trust_message_cache.get("cache", set())
+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(),
 new_cache_entries.remove(cache_entry)
 # Store the remaining existing and new cache entries
 await trust_message_cache.force(
 "cache",
-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(

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0384.py @ 4216:1a7a3e4b52a4