erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0447: Mutiple-files support:

Browse source 
- multiple-files support now that the XEP has them. The legacy message compatibility layer
  is implemented too, with the new `message_sent` trigger for 1:1, and with follow-up
  messages sent on message echo for MUC.
- usual modernisation (type hints, `get_plugin_data`)
- (core) new `message_sent` trigger point
This commit is contained in:
Goffi 2026-07-05 00:59:06 +02:00
commit 31296c6108
4 changed files with 334 additions and 100 deletions
Download patch file Download diff file Expand all files Collapse all files

10
libervia/backend/core/core_types.py
View file

@ -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":

3
libervia/backend/core/xmpp.py
View file

@ -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(

10
libervia/backend/plugins/plugin_xep_0428.py
View file

@ -15,8 +15,6 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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 []

409
libervia/backend/plugins/plugin_xep_0447.py
View file

@ -16,33 +16,34 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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 <sources/> 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 <attach-to/> 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 <sources> 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 <file-sharing/> element
@param extra: extra metadata describing how to access the URL
@return: ``<sfs/>`` 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: ``<file-sharing/>`` 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 <sources/> element
@param sources_elt: <sources/> 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 <file-sharing/> element and return file-sharing data
@param file_sharing_elt: <file-sharing/> 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 <file-sharing/> without <sources/> (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 <message> 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
# <file-sharing> element in a message.
self, client: SatXMPPEntity, message_elt: domish.Element, data: dict[str, Any]
) -> dict[str, Any]:
"""Check <message> 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 <message/>, 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
<file-sharing/> 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 <file-sharing/> 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 <file-sharing/>
@param file_id: Id of the <file-sharing/> 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"]
# <attach-to/>
message_elt.addChild(
self._ma.get_attach_to_elt(uid=main_msg_id, stanza_id=main_msg_id)
)
# <sources/>
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)
# <fallback/>
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 <attach-to/> and <sources/>, 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