diff --git a/libervia/backend/core/constants.py b/libervia/backend/core/constants.py index 969e6d1d..55ff1b94 100644 --- a/libervia/backend/core/constants.py +++ b/libervia/backend/core/constants.py @@ -176,6 +176,7 @@ class Const: MESS_UPDATE_EDIT = "EDIT" MESS_UPDATE_REACTION = "REACTION" MESS_UPDATE_RETRACT = "RETRACT" + MESS_UPDATE_SOURCES = "SOURCES" MESS_EXTRA_INFO = "info_type" EXTRA_INFO_DECR_ERR = "DECRYPTION_ERROR" diff --git a/libervia/backend/plugins/plugin_xep_0447.py b/libervia/backend/plugins/plugin_xep_0447.py index d78532f2..1676ed49 100644 --- a/libervia/backend/plugins/plugin_xep_0447.py +++ b/libervia/backend/plugins/plugin_xep_0447.py @@ -25,10 +25,12 @@ from typing import Any, ClassVar, NamedTuple, Self, cast from pydantic import BaseModel, Field import shortuuid +from sqlalchemy.orm.attributes import flag_modified from treq import collect from twisted.internet import defer from twisted.web.http_headers import Headers from twisted.web.iweb import IResponse +from twisted.words.protocols.jabber import jid from twisted.words.xish import domish from libervia.backend.core import exceptions @@ -42,10 +44,11 @@ from libervia.backend.plugins.plugin_xep_0103 import URLData, XEP_0103 from libervia.backend.plugins.plugin_xep_0334 import XEP_0334 from libervia.backend.plugins.plugin_xep_0358 import JinglePub, XEP_0358 from libervia.backend.plugins.plugin_xep_0363 import XEP_0363 -from libervia.backend.plugins.plugin_xep_0367 import XEP_0367 +from libervia.backend.plugins.plugin_xep_0367 import AttachedTo, XEP_0367 from libervia.backend.plugins.plugin_xep_0428 import XEP_0428 from libervia.backend.plugins.plugin_xep_0446 import FileMetadata, XEP_0446 from libervia.backend.tools import stream +from libervia.backend.tools.common import data_format log = getLogger(__name__) @@ -483,7 +486,9 @@ class XEP_0447: if file_id is not None: data["id"] = file_id try: - data["sources"] = self.parse_sources_elt(file_sharing_elt) + data["sources"] = [ + source.model_dump() for source in self.parse_sources_elt(file_sharing_elt) + ] except exceptions.NotFound: # XEP-0447 allows without (e.g. upload in progress) data["sources"] = [] @@ -630,7 +635,7 @@ class XEP_0447: # message_elt.addChild( - self._ma.get_attach_to_elt(uid=main_msg_id, stanza_id=main_msg_id) + AttachedTo(uid=main_msg_id, stanza_id=main_msg_id).to_element() ) # @@ -662,6 +667,11 @@ class XEP_0447: post_treat.addCallback( partial(self._add_file_sharing_attachments, client, message_elt) ) + post_treat.addCallback( + lambda data: defer.ensureDeferred( + self._handle_source_attachment(client, message_elt, data) + ) + ) return True def _on_connection_closed(self, client: SatXMPPEntity, reason: Any) -> bool: @@ -707,23 +717,117 @@ class XEP_0447: message_elt: domish.Element, data: dict[str, Any], ) -> bool: - """Block backward-compatibility messages from continuing the workflow. + """Message parsing trigger. - If the message contains both and , it is a legacy - compatibility message generated by our own multi-file sending. We raise - CancelError to stop the message from being stored in history or sent to - frontends. + Ignore backward-compatibility messages emitted by our own device. """ - ns_msg_att = "urn:xmpp:message-attaching:1" - attach_to_elt = next(message_elt.elements(ns_msg_att, "attach-to"), None) + attached_to = data["extra"].get("attached_to") sources_elt = next(message_elt.elements(NS_SFS, "sources"), None) - if attach_to_elt is not None and sources_elt is not None: - log.debug( - f"Ignoring backward-compatibility message for " - f"attach-to id={attach_to_elt['id']!r}" - ) - raise exceptions.CancelError( - "Cancelled by XEP_0447 (backward-compatibility message)" - ) + if attached_to is not None and sources_elt is not None: + attach_to_id = attached_to["stanza_id"] + # only block if this is our own backward-compatibility message (from our own + # JID). Incoming source attachments from other clients should be processed + # normally. + if message_elt["from"] == client.jid.full(): + log.debug( + f"Ignoring backward-compatibility message for " + f"attach-to id={attach_to_id!r}" + ) + raise exceptions.CancelError( + "Cancelled by XEP_0447 (backward-compatibility message)" + ) return True + + async def _handle_source_attachment( + self, client: SatXMPPEntity, message_elt: domish.Element, data: dict[str, Any] + ) -> dict[str, Any]: + """Handle deferred source attachment. + + When a message is attaching sources to an existing file-sharing message, find the + original message, merge the new sources, update storage, and emit a signal. + + @param client: Client instance. + @param message_elt: The element. + @param data: Parsed message data. + @return: Updated message data. + """ + attached_to = data["extra"].get("attached_to") + sources_elts = list(message_elt.elements(NS_SFS, "sources")) + file_sharing_elts = list(message_elt.elements(NS_SFS, "file-sharing")) + + # only process if we have attach-to + sources but no file-sharing + if attached_to is None or not sources_elts or file_sharing_elts: + return data + + if attached_to.get("uid") is None: + log.warning( + f"Received source attachment with unresolved attach-to id " + f"{attached_to['stanza_id']!r}, ignoring" + ) + return data + + # find the original message in history + message_type = data.get("type") + history = await self.host.memory.storage.get_history_from_xmpp_id( + client, attached_to["stanza_id"], message_type + ) + if history is None: + log.warning( + f"No history entry found for source attachment " + f"attach-to={attached_to['stanza_id']!r}" + ) + return data + + # verify sender: sources can only be added by the original sender + from_jid = jid.JID(message_elt["from"]) + if history.type == C.MESS_TYPE_GROUPCHAT: + if history.source_jid != from_jid: + log.warning( + f"Ignoring source attachment from {from_jid.full()}: " + f"not the original sender ({history.source_jid.full()})" + ) + return data + else: + if history.source_jid.userhostJID() != from_jid.userhostJID(): + log.warning( + f"Ignoring source attachment from {from_jid.full()}: " + f"not the original sender ({history.source_jid.full()})" + ) + return data + + # merge new sources into existing attachments + existing_attachments = history.extra.get(C.KEY_ATTACHMENTS, []) + new_sources_added = False + for sources_elt in sources_elts: + sources = self.parse_sources_elt(sources_elt) + file_id = sources_elt.getAttribute("id") + # find matching attachment by file-sharing id + for attachment in existing_attachments: + if file_id is None or attachment.get("id") == file_id: + for source in sources: + source_data = source.model_dump() + if source_data not in attachment["sources"]: + attachment["sources"].append(source_data) + new_sources_added = True + break + + if not new_sources_added: + log.debug( + f"No new sources to add for message uid={history.uid!r}, " + f"skipping update" + ) + else: + flag_modified(history, "extra") + await self.host.memory.storage.add(history) + + # notify frontends + update_data = MessageData(history.serialise()) + self.host.bridge.message_update( + history.uid, + C.MESS_UPDATE_SOURCES, + data_format.serialise(update_data), + client.profile, + ) + # we stop the workflow here to avoid storage and bridge signal + raise exceptions.CancelError("Source attachment merged, cancelling message")