diff --git a/libervia/backend/core/core_types.py b/libervia/backend/core/core_types.py index ca9cc3a7..f91cc1d0 100644 --- a/libervia/backend/core/core_types.py +++ b/libervia/backend/core/core_types.py @@ -44,6 +44,7 @@ if TYPE_CHECKING: SatDiscoProtocol, SatMessageProtocol, ) + from libervia.backend.memory.encryption import EncryptionHandler from libervia.backend.models.tor import TorMode from treq.client import HTTPClient from txtorcon import Tor @@ -143,6 +144,15 @@ class SatXMPPEntity(ABC): raise exceptions.ConfigError("Plugin data already stored in {import_name!r}") self.plugin_data[import_name] = plugin_data + @property + @abstractmethod + def encryption(self) -> "EncryptionHandler": + """Get the encryption handler for the client. + + @return (EncryptionHandler): the encryption handler + """ + raise NotImplementedError + @property @abstractmethod def treq(self) -> "HTTPClient": diff --git a/libervia/backend/core/xmpp.py b/libervia/backend/core/xmpp.py index 143abd79..763c61a8 100644 --- a/libervia/backend/core/xmpp.py +++ b/libervia/backend/core/xmpp.py @@ -870,6 +870,9 @@ class SatXMPPEntity(core_types.SatXMPPEntity): "send_message_data", self, mess_data, triggers_no_cancel=True ) await self.a_send(mess_data["xml"]) + await self.host_app.trigger.async_point( + "message_sent", self, mess_data, triggers_no_cancel=True + ) return mess_data def sendMessage( diff --git a/libervia/backend/plugins/plugin_xep_0428.py b/libervia/backend/plugins/plugin_xep_0428.py index c66aa54b..6f833e44 100644 --- a/libervia/backend/plugins/plugin_xep_0428.py +++ b/libervia/backend/plugins/plugin_xep_0428.py @@ -15,8 +15,6 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from typing import Optional - from twisted.words.protocols.jabber import xmlstream from twisted.words.xish import domish from tx_xmpp import disco @@ -46,7 +44,7 @@ NS_FALLBACK = "urn:xmpp:fallback:0" class XEP_0428: def __init__(self, host): - log.info(_("XEP-0428 (Fallback Indication) plugin initialization")) + log.info(f"Plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization.") host.register_namespace("fallback", NS_FALLBACK) def add_fallback_elt( @@ -86,8 +84,10 @@ class XEP_0428: @implementer(disco.IDisco) class XEP_0428_handler(xmlstream.XMPPHandler): - def getDiscoInfo(self, __, target, nodeIdentifier=""): + def getDiscoInfo(self, requestor: str, target: str, nodeIdentifier: str = "") -> list: return [disco.DiscoFeature(NS_FALLBACK)] - def getDiscoItems(self, requestor, target, nodeIdentifier=""): + def getDiscoItems( + self, requestor: str, target: str, nodeIdentifier: str = "" + ) -> list: return [] diff --git a/libervia/backend/plugins/plugin_xep_0447.py b/libervia/backend/plugins/plugin_xep_0447.py index 57882f7c..d78532f2 100644 --- a/libervia/backend/plugins/plugin_xep_0447.py +++ b/libervia/backend/plugins/plugin_xep_0447.py @@ -16,33 +16,34 @@ # along with this program. If not, see . from abc import ABC, abstractmethod +from dataclasses import dataclass, field from functools import partial import mimetypes from pathlib import Path -from typing import ( - Any, - ClassVar, - Dict, - List, - Optional, - Self, - Tuple, - Union, - cast, -) +import time +from typing import Any, ClassVar, NamedTuple, Self, cast from pydantic import BaseModel, Field +import shortuuid 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.xish import domish from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C -from libervia.backend.core.core_types import SatXMPPEntity +from libervia.backend.core.core_types import MessageData, SatXMPPEntity from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger +from libervia.backend.plugins.plugin_misc_attach import AttachPlugin +from libervia.backend.plugins.plugin_xep_0066 import XEP_0066 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_0428 import XEP_0428 from libervia.backend.plugins.plugin_xep_0446 import FileMetadata, XEP_0446 from libervia.backend.tools import stream @@ -63,7 +64,7 @@ PLUGIN_INFO = { "ATTACH", "DOWNLOAD", ], - C.PI_RECOMMENDATIONS: ["XEP-0363"], + C.PI_RECOMMENDATIONS: ["XEP-0066", "XEP-0363", "XEP-0367", "XEP-0428"], C.PI_MAIN: "XEP_0447", C.PI_HANDLER: "no", C.PI_DESCRIPTION: _("""Implementation of XEP-0447 (Stateless File Sharing)"""), @@ -193,6 +194,39 @@ class JinglePubSource(JinglePub, Source): return super().to_element() +class LegacyAttachment(NamedTuple): + """Single attachment entry for a backward-compatibility message. + + Keys used when building elements for legacy clients. + """ + + file_id: str + url: str + + +class LegacyAttachments(NamedTuple): + """Data stored for a pending MUC legacy attach message. + + Kept until the MUC echo arrives with a stanza-id, after which backward-compatibility + messages are sent using that stanza-id as the target. + """ + + mess_data: dict[str, Any] + attachments: list[LegacyAttachment] + timestamp: float + + +@dataclass +class SFSClientData: + """Per-client data for XEP-0447 plugin. + + Stores pending legacy attach entries for MUC messages (waiting for stanza-id echo). + """ + + # origin_id -> LegacyEntry + pending_legacy: dict[str, LegacyAttachments] = field(default_factory=dict) + + class XEP_0447: namespace = NS_SFS @@ -204,10 +238,13 @@ class XEP_0447: self._sources_handlers: dict[tuple[str, str], type[Source]] = {} self._u = cast(XEP_0103, host.plugins["XEP-0103"]) self._jp = cast(XEP_0358, host.plugins["XEP-0358"]) - self._hints = host.plugins["XEP-0334"] + self._hints = cast(XEP_0334, host.plugins["XEP-0334"]) self._m = cast(XEP_0446, host.plugins["XEP-0446"]) - self._http_upload = host.plugins.get("XEP-0363") - self._attach = host.plugins["ATTACH"] + self._http_upload = cast(XEP_0363 | None, host.plugins.get("XEP-0363")) + self._oob = cast(XEP_0066 | None, host.plugins.get("XEP-0066")) + self._ma = cast(XEP_0367 | None, host.plugins.get("XEP-0367")) + self._fb = cast(XEP_0428 | None, host.plugins.get("XEP-0428")) + self._attach = cast(AttachPlugin, host.plugins["ATTACH"]) self._attach.register(self.can_handle_attachment, self.attach, priority=1000) self.register_source( self._u.namespace, @@ -223,6 +260,9 @@ class XEP_0447: self._u.namespace, self.download ) host.trigger.add("message_received", self._message_received_trigger) + host.trigger.add("message_parse", self._message_parse_trigger) + host.trigger.add("message_sent", self._message_sent_trigger) + host.trigger.add("disconnected", self._on_connection_closed) def register_source( self, @@ -250,11 +290,11 @@ class XEP_0447: async def download( self, client: SatXMPPEntity, - attachment: Dict[str, Any], - source: Dict[str, Any], - dest_path: Union[Path, str], - extra: Optional[Dict[str, Any]] = None, - ) -> Tuple[str, defer.Deferred]: + attachment: dict[str, Any], + source: dict[str, Any], + dest_path: Path | str, + extra: dict[str, Any] | None = None, + ) -> tuple[str, defer.Deferred]: # TODO: handle url-data headers if extra is None: extra = {} @@ -272,8 +312,15 @@ class XEP_0447: try: file_size = int(attachment["size"]) except (KeyError, ValueError): - head_data = await treq_client.head(download_url) - file_size = int(head_data.headers.getRawHeaders("content-length")[0]) + head_data = cast(IResponse, await treq_client.head(download_url)) + content_length = cast(Headers, head_data.headers).getRawHeaders( + "content-length" + ) + if content_length is None: + raise ValueError( + f"No content-length header in HEAD response for {download_url}" + ) + file_size = int(content_length[0]) file_obj = stream.SatFile( self.host, @@ -285,7 +332,7 @@ class XEP_0447: progress_id = file_obj.uid - resp = await treq_client.get(download_url, unbuffered=True) + resp = cast(IResponse, await treq_client.get(download_url, unbuffered=True)) if resp.code == 200: d = collect(resp, file_obj.write) d.addCallback(lambda __: file_obj.close()) @@ -294,7 +341,9 @@ class XEP_0447: self.host.plugins["DOWNLOAD"].errback_download(file_obj, d, resp) return progress_id, d - async def can_handle_attachment(self, client, data): + async def can_handle_attachment( + self, client: SatXMPPEntity, data: dict[str, Any] + ) -> bool: if self._http_upload is None: return False try: @@ -305,7 +354,7 @@ class XEP_0447: return True def get_sources_elt( - self, children: Optional[List[domish.Element]] = None + self, children: list[domish.Element] | None = None ) -> domish.Element: """Generate element""" sources_elt = domish.Element((NS_SFS, "sources")) @@ -316,28 +365,43 @@ class XEP_0447: def get_file_sharing_elt( self, - sources: List[Dict[str, Any]], - disposition: Optional[str] = None, - name: Optional[str] = None, - media_type: Optional[str] = None, - desc: Optional[str] = None, - size: Optional[int] = None, - file_hash: Optional[Tuple[str, str]] = None, - date: Optional[Union[float, int]] = None, - width: Optional[int] = None, - height: Optional[int] = None, - length: Optional[int] = None, - thumbnail: Optional[str] = None, + sources: list[dict[str, Any]], + disposition: str | None = None, + file_id: str | None = None, + name: str | None = None, + media_type: str | None = None, + desc: str | None = None, + size: int | None = None, + file_hash: tuple[str, str] | None = None, + date: float | int | None = None, + width: int | None = None, + height: int | None = None, + length: int | None = None, + thumbnail: str | None = None, **kwargs, ) -> domish.Element: """Generate the element - @param extra: extra metadata describing how to access the URL - @return: ```` element + @param sources: list of source data dicts + @param disposition: disposition of the file + @param file_id: unique identifier for the file-sharing element + @param name: filename + @param media_type: media type + @param desc: description + @param size: file size + @param file_hash: tuple of (hash_algo, hash) + @param date: file date + @param width: width (for images) + @param height: height (for images) + @param length: length (for audio/video) + @param thumbnail: thumbnail data + @return: ```` element """ file_sharing_elt = domish.Element((NS_SFS, "file-sharing")) if disposition is not None: file_sharing_elt["disposition"] = disposition + if file_id is not None: + file_sharing_elt["id"] = file_id if media_type is None and name: media_type = mimetypes.guess_type(name, strict=False)[0] file_sharing_elt.addChild( @@ -366,7 +430,7 @@ class XEP_0447: return file_sharing_elt - def parse_sources_elt(self, sources_elt: domish.Element) -> List[Source]: + def parse_sources_elt(self, sources_elt: domish.Element) -> list[Source]: """Parse element @param sources_elt: element, or a direct parent element @@ -396,7 +460,7 @@ class XEP_0447: sources.append(source) return sources - def parse_file_sharing_elt(self, file_sharing_elt: domish.Element) -> Dict[str, Any]: + def parse_file_sharing_elt(self, file_sharing_elt: domish.Element) -> dict[str, Any]: """Parse element and return file-sharing data @param file_sharing_elt: element @@ -415,20 +479,61 @@ class XEP_0447: disposition = file_sharing_elt.getAttribute("disposition") if disposition is not None: data["disposition"] = disposition + file_id = file_sharing_elt.getAttribute("id") + if file_id is not None: + data["id"] = file_id try: data["sources"] = self.parse_sources_elt(file_sharing_elt) - except exceptions.NotFound as e: - raise ValueError(str(e)) + except exceptions.NotFound: + # XEP-0447 allows without (e.g. upload in progress) + data["sources"] = [] return data def _add_file_sharing_attachments( - self, client: SatXMPPEntity, message_elt: domish.Element, data: Dict[str, Any] - ) -> Dict[str, Any]: - """Check for a shared file, and add it as an attachment""" - # XXX: XEP-0447 doesn't support several attachments in a single message, for now - # however that should be fixed in future version, and so we accept several - # element in a message. + self, client: SatXMPPEntity, message_elt: domish.Element, data: dict[str, Any] + ) -> dict[str, Any]: + """Check for shared files, and add them as attachments.""" + # check for pending legacy attachments + origin_id = data["extra"].get("origin_id") + if origin_id: + sfs_data = client.get_plugin_data("XEP-0447", SFSClientData) + if origin_id in sfs_data.pending_legacy: + stanza_id = data["extra"].get("stanza_id") + if stanza_id: + entry_data = sfs_data.pending_legacy.pop(origin_id) + for entry in entry_data.attachments: + d = defer.ensureDeferred( + self._send_legacy_attach_message( + client, + entry_data.mess_data, + stanza_id, + entry.file_id, + entry.url, + ) + ) + d.addErrback( + lambda __: log.exception( + f"Error sending legacy attach message" + ) + ) + else: + log.warning( + f"MUC message for origin_id={origin_id!r} has no stanza-id, " + "skipping legacy attach messages." + ) + del sfs_data.pending_legacy[origin_id] + + # purge expired entries + now = time.monotonic() + expired = [ + k for k, v in sfs_data.pending_legacy.items() if now - v.timestamp > 60 + ] + if expired: + log.warning(f"Purging {len(expired)} expired legacy attach entries") + for k in expired: + del sfs_data.pending_legacy[k] + for file_sharing_elt in message_elt.elements(NS_SFS, "file-sharing"): attachment = self.parse_file_sharing_elt(file_sharing_elt) @@ -445,50 +550,46 @@ class XEP_0447: return data - async def attach(self, client, data): - # XXX: for now, XEP-0447 only allow to send one file per , thus we need - # to send each file in a separate message + async def attach(self, client: SatXMPPEntity, data: dict[str, Any]) -> None: + """Attach files to a message. + + All files are uploaded and included in a single message with multiple + elements, each identified by a unique id. + Backward-compatible attach messages are sent for legacy clients. + """ attachments = data["extra"][C.KEY_ATTACHMENTS] - if not data["message"] or data["message"] == {"": ""}: - extra_attachments = attachments[1:] - del attachments[1:] - else: - # we have a message, we must send first attachment separately - extra_attachments = attachments[:] - attachments.clear() - del data["extra"][C.KEY_ATTACHMENTS] - if attachments: - if len(attachments) > 1: - raise exceptions.InternalError( - "There should not be more that one attachment at this point" - ) - await self._attach.upload_files(client, data) - self._hints.add_hint_elements(data["xml"], [self._hints.HINT_STORE]) - for attachment in attachments: - try: - file_hash = (attachment["hash_algo"], attachment["hash"]) - except KeyError: - file_hash = None - file_sharing_elt = self.get_file_sharing_elt( - [{"url": attachment["url"]}], - name=attachment.get("name"), - size=attachment.get("size"), - desc=attachment.get("desc"), - media_type=attachment.get("media_type"), - file_hash=file_hash, - ) - data["xml"].addChild(file_sharing_elt) + # upload all files + await self._attach.upload_files(client, data) + self._hints.add_hint_elements(data["xml"], [self._hints.HINT_STORE]) - for attachment in extra_attachments: - # we send all remaining attachment in a separate message - await client.sendMessage( - to_jid=data["to"], - message={"": ""}, - subject=data["subject"], - mess_type=data["type"], - extra={C.KEY_ATTACHMENTS: [attachment]}, + # build element for each attachment + for attachment in attachments: + file_id = attachment.get("name") or shortuuid.uuid() + try: + file_hash = (attachment["hash_algo"], attachment["hash"]) + except KeyError: + file_hash = None + file_sharing_elt = self.get_file_sharing_elt( + [{"url": attachment["url"]}], + file_id=file_id, + name=attachment.get("name"), + size=attachment.get("size"), + desc=attachment.get("desc"), + media_type=attachment.get("media_type"), + file_hash=file_hash, ) + data["xml"].addChild(file_sharing_elt) + + # store legacy info for deferred sending after main message + if self._oob is not None and self._ma is not None and self._fb is not None: + legacy_attachments: list[LegacyAttachment] = [] + for attachment in attachments: + file_id = attachment.get("name") or shortuuid.uuid() + legacy_attachments.append( + LegacyAttachment(file_id=file_id, url=attachment["url"]) + ) + data["extra"]["_sfs_legacy"] = legacy_attachments if ( not data["extra"] @@ -498,11 +599,131 @@ class XEP_0447: # nothing left to send, we can cancel the message raise exceptions.CancelError("Cancelled by XEP_0447 attachment handling") - def _message_received_trigger(self, client, message_elt, post_treat): - # we use a post_treat callback instead of "message_parse" trigger because we need - # to check if the "encrypted" flag is set to decide if we add the same flag to the - # attachment + async def _send_legacy_attach_message( + self, + client: SatXMPPEntity, + data: dict[str, Any], + main_msg_id: str, + file_id: str, + url: str, + ) -> None: + """Send a backward-compatible attach message for legacy clients. + + @param client: Client instance. + @param data: Original message data (for to, type) + @param main_msg_id: Uid of the main message containing + @param file_id: Id of the element this attaches to + @param url: Download url for the file + """ + assert self._oob is not None and self._ma is not None and self._fb is not None + mess_data: dict[str, Any] = { + "from": client.jid, + "to": data["to"], + "uid": "", + "message": {"": url}, + "type": data["type"], + "subject": {}, + "extra": {}, + } + client.generate_message_xml(cast(MessageData, mess_data)) + message_elt = mess_data["xml"] + + # + message_elt.addChild( + self._ma.get_attach_to_elt(uid=main_msg_id, stanza_id=main_msg_id) + ) + + # + sources_elt = self.get_sources_elt() + sources_elt["id"] = file_id + sources_elt.addChild(self._u.generate_url_data(url=url).to_element()) + message_elt.addChild(sources_elt) + + # + self._fb.add_fallback_elt(message_elt, NS_SFS, url) + + # OOB + message_elt.addChild(self._oob.generate_oob(url)) + + await client.send_message_data(cast(MessageData, mess_data)) + + def _message_received_trigger( + self, + client: SatXMPPEntity, + message_elt: domish.Element, + post_treat: defer.Deferred, + ) -> bool: + """Register post-treatment to add file-sharing attachments to message data. + + We use a post_treat callback instead of "message_parse" trigger because we need + to check if the "encrypted" flag is set to decide if we add the same flag to the + attachment. + """ post_treat.addCallback( partial(self._add_file_sharing_attachments, client, message_elt) ) return True + + def _on_connection_closed(self, client: SatXMPPEntity, reason: Any) -> bool: + """Clean up pending legacy entries when a client disconnects.""" + sfs_data = client.get_plugin_data("XEP-0447", SFSClientData) + if sfs_data.pending_legacy: + log.warning( + f"Clearing {len(sfs_data.pending_legacy)} pending legacy entries for " + f"{client.jid.userhost()}" + ) + sfs_data.pending_legacy.clear() + return True + + async def _message_sent_trigger( + self, client: SatXMPPEntity, mess_data: dict[str, Any] + ) -> None: + """Send backward-compatibility messages after the main message. + + For non-groupchat, send immediately using origin_id. + For groupchat, defer until we receive the MUC echo with stanza-id. + """ + legacy_attachments = mess_data["extra"].get("_sfs_legacy") + if legacy_attachments is None: + return + origin_id = mess_data["extra"].get("origin_id") or mess_data["uid"] + if mess_data["type"] == C.MESS_TYPE_GROUPCHAT: + # defer: wait for MUC echo to get stanza-id + sfs_data = client.get_plugin_data("XEP-0447", SFSClientData) + sfs_data.pending_legacy[origin_id] = LegacyAttachments( + mess_data=mess_data, + attachments=legacy_attachments, + timestamp=time.monotonic(), + ) + return + for entry in legacy_attachments: + await self._send_legacy_attach_message( + client, mess_data, origin_id, entry.file_id, entry.url + ) + + def _message_parse_trigger( + self, + client: SatXMPPEntity, + message_elt: domish.Element, + data: dict[str, Any], + ) -> bool: + """Block backward-compatibility messages from continuing the workflow. + + 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. + """ + ns_msg_att = "urn:xmpp:message-attaching:1" + attach_to_elt = next(message_elt.elements(ns_msg_att, "attach-to"), None) + 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)" + ) + return True