core (memory/migration):
Update XEP-0384 and `fe3a02cb4bec_convert_legacypickle_columns_to_json.py` migration to properly handle (de)serialisation of `TrustMessageCacheEntry`.
This commit is contained in:
Goffi
parent
d50ca26d12
commit
3694bb3991
2 changed files with 89 additions and 26 deletions
|
|
@ -10,6 +10,7 @@ import sqlalchemy as sa
|
|||
import pickle
|
||||
import json
|
||||
from libervia.backend.plugins.plugin_xep_0373 import PublicKeyMetadata
|
||||
from libervia.backend.plugins.plugin_xep_0384 import TrustMessageCacheEntry
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = "fe3a02cb4bec"
|
||||
|
|
@ -46,6 +47,21 @@ def convert_pickle_to_json(value, table, primary_keys):
|
|||
# `from_dict` methods.
|
||||
deserialized = [pkm.to_dict() for pkm in deserialized]
|
||||
|
||||
elif (
|
||||
table == "private_ind_bin"
|
||||
and primary_keys[0] == "XEP-0384/TM"
|
||||
and primary_keys[1] == "cache"
|
||||
):
|
||||
# Same issue and solution as for XEP-0373
|
||||
try:
|
||||
deserialized = [tm.to_dict() for tm in deserialized]
|
||||
except Exception as e:
|
||||
print(
|
||||
"Warning: Failed to convert Trust Management cache with value "
|
||||
f" {deserialized!r}, using empty array instead: {e}"
|
||||
)
|
||||
deserialized=[]
|
||||
|
||||
ret = json.dumps(deserialized, ensure_ascii=False, default=str)
|
||||
if table == 'history' and ret == "{}":
|
||||
# For history, we can remove empty data, but for other tables it may be
|
||||
|
|
@ -104,6 +120,14 @@ def convert_json_to_pickle(value, table, primary_keys):
|
|||
# Convert list of dicts back to set of PublicKeyMetadata objects
|
||||
if isinstance(deserialized, list):
|
||||
deserialized = {PublicKeyMetadata.from_dict(d) for d in deserialized}
|
||||
elif (
|
||||
table == "private_ind_bin"
|
||||
and primary_keys[0] == "XEP-0384/TM"
|
||||
and primary_keys[1] == "cache"
|
||||
):
|
||||
# Convert list of dicts back to set of TrustMessageCacheEntry objects
|
||||
if isinstance(deserialized, list):
|
||||
deserialized = {TrustMessageCacheEntry.from_dict(d) for d in deserialized}
|
||||
return pickle.dumps(deserialized, 0)
|
||||
except Exception as e:
|
||||
print(
|
||||
|
|
|
|||
|
|
@ -434,6 +434,22 @@ class TrustUpdate(NamedTuple):
|
|||
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
|
||||
|
|
@ -446,6 +462,25 @@ class TrustMessageCacheEntry(NamedTuple):
|
|||
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
|
||||
|
|
@ -478,10 +513,10 @@ async def manage_trust_message_cache(
|
|||
)
|
||||
|
||||
# Load cache entries
|
||||
cache_entries = cast(
|
||||
Set[TrustMessageCacheEntry],
|
||||
await trust_message_cache.get("cache", set())
|
||||
)
|
||||
cache_entries = {
|
||||
TrustMessageCacheEntry.from_dict(d)
|
||||
for d in await trust_message_cache.get("cache", [])
|
||||
}
|
||||
|
||||
# Expire cache entries that were overwritten by the applied trust updates
|
||||
cache_entries_by_target = {
|
||||
|
|
@ -534,7 +569,10 @@ async def manage_trust_message_cache(
|
|||
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?
|
||||
|
||||
|
|
@ -1540,22 +1578,23 @@ class OMEMO:
|
|||
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.__xep_0334 = cast(XEP_0334, sat.plugins["XEP-0334"])
|
||||
self.__xep_0359 = cast(Optional[XEP_0359], sat.plugins.get("XEP-0359"))
|
||||
self.__xep_0420 = cast(XEP_0420, sat.plugins["XEP-0420"])
|
||||
self._j = cast(XEP_0060, host.plugins["XEP-0060"])
|
||||
self.__xep_0045 = cast(Optional[XEP_0045], host.plugins.get("XEP-0045"))
|
||||
self.__xep_0334 = cast(XEP_0334, host.plugins["XEP-0334"])
|
||||
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
|
||||
|
|
@ -1579,23 +1618,23 @@ class OMEMO:
|
|||
# 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)
|
||||
sat.register_encryption_plugin(
|
||||
host.register_encryption_plugin(self, "OMEMO", twomemo.twomemo.NAMESPACE, 101)
|
||||
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,
|
||||
|
|
@ -1612,7 +1651,7 @@ class OMEMO:
|
|||
)
|
||||
|
||||
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:
|
||||
|
|
@ -1759,7 +1798,7 @@ class OMEMO:
|
|||
))
|
||||
|
||||
# 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
|
||||
|
|
@ -1781,7 +1820,7 @@ class OMEMO:
|
|||
|
||||
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,
|
||||
|
|
@ -1921,7 +1960,7 @@ class OMEMO:
|
|||
# Build and store the session manager
|
||||
try:
|
||||
session_manager = await prepare_for_profile(
|
||||
self.__sat,
|
||||
self.host,
|
||||
profile,
|
||||
initial_own_label="Libervia"
|
||||
)
|
||||
|
|
@ -2021,10 +2060,10 @@ class OMEMO:
|
|||
))
|
||||
|
||||
# Load existing cache entries
|
||||
existing_cache_entries = cast(
|
||||
Set[TrustMessageCacheEntry],
|
||||
await trust_message_cache.get("cache", set())
|
||||
)
|
||||
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 = {
|
||||
|
|
@ -2083,7 +2122,7 @@ class OMEMO:
|
|||
# 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
|
||||
|
|
|
|||
Loading…
Reference in a new issue