From dbc3973642cc267e8045f585aba8f54292f83289 Mon Sep 17 00:00:00 2001 From: Goffi Date: Thu, 9 Jul 2026 13:00:20 +0200 Subject: [PATCH] plugin XEP-0447: handle deferred sources: XEP-0447 allow to send sources after the original message. Dino is doing that, and thus we can't display attachment immediately with only the original message. This patch handle that by updating the original message once the sources are received, and emitting a `message_update` signal to notify frontends. --- libervia/backend/core/constants.py | 1 + libervia/backend/plugins/plugin_xep_0447.py | 140 +++++++++++++++++--- 2 files changed, 123 insertions(+), 18 deletions(-) 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")