plugin XEP-0498: "Pubsub File Sharing" implementation:

Partial implementation of XEP-0498, necessary to implement the service part in email
gateway.

rel 453

libervia/backend/plugins/plugin_xep_0498.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+
+# Libervia plugin to jingle session publishing.
+# Copyright (C) 2009-2024 Jérôme Poisson (goffi@goffi.org)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# 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 TYPE_CHECKING, Final, Self
+
+from pydantic import BaseModel
+from twisted.words.protocols.jabber import jid
+from twisted.words.protocols.jabber.xmlstream import XMPPHandler
+from twisted.words.xish import domish
+from wokkel import disco, iwokkel, pubsub
+from zope.interface import implementer
+
+from libervia.backend.core.constants import Const as C
+from libervia.backend.core.core_types import SatXMPPEntity
+from libervia.backend.core.i18n import _
+from libervia.backend.core.log import getLogger
+from libervia.backend.plugins.plugin_xep_0234 import NS_JINGLE_FT
+from libervia.backend.plugins.plugin_xep_0358 import JinglePub
+from libervia.backend.plugins.plugin_xep_0446 import FileMetadata
+from libervia.backend.plugins.plugin_xep_0447 import FileSharing, JinglePubSource
+
+if TYPE_CHECKING:
+    from libervia.backend.core.main import LiberviaBackend
+
+log = getLogger(__name__)
+
+
+PLUGIN_INFO = {
+    C.PI_NAME: "Pubsub File Sharing",
+    C.PI_IMPORT_NAME: "XEP-0498",
+    C.PI_TYPE: "XEP",
+    C.PI_MODES: C.PLUG_MODE_BOTH,
+    C.PI_PROTOCOLS: [],
+    C.PI_DEPENDENCIES: [
+        "XEP-0060",
+        "XEP-0447",
+    ],
+    C.PI_RECOMMENDATIONS: [],
+    C.PI_MAIN: "XEP_0498",
+    C.PI_HANDLER: "yes",
+    C.PI_DESCRIPTION: _("""Share and retrieve files via Pubsub."""),
+}
+
+NS_PUBSUB_FILE_SHARING: Final = "urn:xmpp:pubsub-file-sharing:0"
+
+
+class NodeData(BaseModel):
+    """Model for JinglePub element."""
+
+    files: dict[str, FileSharing]
+
+    def to_elements(self) -> list[domish.Element]:
+        """Return the list of item elements corresponding to this model"""
+        items = []
+        for item_id, file_sharing in self.files.items():
+            item_elt = pubsub.Item(id=item_id, payload=file_sharing.to_element())
+            items.append(item_elt)
+
+        return items
+
+    @classmethod
+    def from_files_data(cls, source_jid: jid.JID, files_data: list[dict]) -> Self:
+        """Generate from list of file data as returned by ``memory.get_files``.
+
+        @param files_data: list of files data as returned by ``memory.get_files``.
+        @return: Instance of ``NodeData``.
+        """
+        kw = {}
+        for file_data in files_data:
+            file_metadata = FileMetadata.from_filedata_dict(file_data)
+            source = JinglePubSource(
+                from_jid=source_jid,
+                id=file_data["id"],
+                descriptions=[domish.Element((NS_JINGLE_FT, "description"))],
+            )
+            # We don't know if names are unique, so we add ID to be sure.
+            key = f'{file_data["name"]}_{file_data["id"]}'
+            kw[key] = FileSharing(file=file_metadata, sources=[source])
+
+        return cls(files=kw)
+
+
+class XEP_0498:
+    namespace = NS_PUBSUB_FILE_SHARING
+
+    def __init__(self, host: "LiberviaBackend") -> None:
+        log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
+        self.host = host
+        host.register_namespace("pubsub-file-sharing", NS_PUBSUB_FILE_SHARING)
+
+    def get_handler(self, client: SatXMPPEntity) -> XMPPHandler:
+        return PubsubFileSharingHandler(self)
+
+
+@implementer(iwokkel.IDisco)
+class PubsubFileSharingHandler(XMPPHandler):
+
+    def __init__(self, plugin_parent):
+        self.plugin_parent = plugin_parent
+
+    def getDiscoInfo(
+        self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = ""
+    ) -> list[disco.DiscoFeature]:
+        return [
+            disco.DiscoFeature(NS_PUBSUB_FILE_SHARING),
+        ]
+
+    def getDiscoItems(
+        self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = ""
+    ) -> list[disco.DiscoItems]:
+        return []