diff libervia/backend/plugins/plugin_xep_0358.py @ 4335:430d5d99a740

plugin XEP-0358: "Publishing Available Jingle Sessions" implementation: rel 453
author Goffi <goffi@goffi.org>
date Tue, 03 Dec 2024 00:13:07 +0100
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/libervia/backend/plugins/plugin_xep_0358.py	Tue Dec 03 00:13:07 2024 +0100
@@ -0,0 +1,204 @@
+#!/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, Self
+
+from pydantic import BaseModel, Field
+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
+from zope.interface import implementer
+
+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.i18n import _
+from libervia.backend.core.log import getLogger
+from libervia.backend.models.types import DomishElementType, JIDType
+
+if TYPE_CHECKING:
+    from libervia.backend.core.main import LiberviaBackend
+
+log = getLogger(__name__)
+
+
+PLUGIN_INFO = {
+    C.PI_NAME: "Publishing Available Jingle Sessions",
+    C.PI_IMPORT_NAME: "XEP-0358",
+    C.PI_TYPE: "XEP",
+    C.PI_MODES: C.PLUG_MODE_BOTH,
+    C.PI_PROTOCOLS: [],
+    C.PI_DEPENDENCIES: [
+        "XEP-0234",
+    ],
+    C.PI_RECOMMENDATIONS: [],
+    C.PI_MAIN: "XEP_0358",
+    C.PI_HANDLER: "yes",
+    C.PI_DESCRIPTION: _("""Indicate available Jingle sessions."""),
+}
+
+NS_JINGLEPUB = "urn:xmpp:jinglepub:1"
+
+
+class Meta(BaseModel):
+    """Model for meta element in JinglePub element."""
+
+    lang: str | None = Field(None, description="The language of the meta data.")
+    title: str = Field(description="The title of the meta data.")
+    summary: str | None = Field(None, description="A summary of the meta data.")
+
+    def to_element(self) -> domish.Element:
+        """Build the <meta> element from this instance's data.
+
+        @return: <meta> element.
+        """
+        meta_elt = domish.Element((NS_JINGLEPUB, "meta"))
+        if self.lang:
+            meta_elt["xml:lang"] = self.lang
+        meta_elt["title"] = self.title
+        if self.summary:
+            meta_elt["summary"] = self.summary
+        return meta_elt
+
+    @classmethod
+    def from_element(cls, meta_elt: domish.Element) -> Self:
+        """Create a Meta instance from a <meta> element.
+
+        @param meta_elt: The <meta> element.
+        @return: Meta instance.
+        """
+        assert meta_elt.name == "meta" and meta_elt.uri == NS_JINGLEPUB
+        kwargs = {}
+        if meta_elt.hasAttribute("xml:lang"):
+            kwargs["lang"] = meta_elt["xml:lang"]
+        if meta_elt.hasAttribute("title"):
+            kwargs["title"] = meta_elt["title"]
+        if meta_elt.hasAttribute("summary"):
+            kwargs["summary"] = meta_elt["summary"]
+        return cls(**kwargs)
+
+
+class JinglePub(BaseModel):
+    """Model for JinglePub element."""
+
+    from_jid: JIDType = Field(description="The JID of the session owner.")
+    id: str = Field(description="The jinglepub identifier.")
+    uri: str | None = Field(
+        default=None, description="An alternate or informational URI of the content."
+    )
+    meta: list[Meta] = Field(
+        default_factory=list, description="Human-friendly information about the session."
+    )
+    descriptions: list[DomishElementType] = Field(
+        default_factory=list, description="Application format(s) for the Jingle session."
+    )
+
+    def set_attributes(self, jinglepub_elt: domish.Element) -> None:
+        """Set <jinglepub> element attributes from this instance's data."""
+        jinglepub_elt["from"] = str(self.from_jid)
+        jinglepub_elt["id"] = self.id
+
+    def set_children(self, jinglepub_elt: domish.Element) -> None:
+        """Set <jinglepub> element children from this instance's data."""
+        if self.uri:
+            uri_elt = jinglepub_elt.addElement((NS_JINGLEPUB, "uri"))
+            uri_elt.addContent(self.uri)
+        for meta in self.meta:
+            jinglepub_elt.addChild(meta.to_element())
+        for description in self.descriptions:
+            jinglepub_elt.addChild(description)
+
+    def to_element(self) -> domish.Element:
+        """Build the <jinglepub> element from this instance's data.
+
+        @return: <jinglepub> element.
+        """
+        jinglepub_elt = domish.Element((NS_JINGLEPUB, "jinglepub"))
+        self.set_attributes(jinglepub_elt)
+        self.set_children(jinglepub_elt)
+        return jinglepub_elt
+
+    @classmethod
+    def from_element(cls, element: domish.Element) -> Self:
+        """Create a JinglePub instance from a <jinglepub> element.
+
+        @param jinglepub_elt: The <jinglepub> element.
+        @return: JinglePub instance.
+        @raise exceptions.NotFound: If the <jinglepub> element is not found.
+        """
+        if element.uri != NS_JINGLEPUB or element.name != "jinglepub":
+            raise exceptions.NotFound("<jinglepub> element not found")
+
+        kwargs = {}
+        try:
+            kwargs["from_jid"] = jid.JID(element["from"])
+        except RuntimeError as e:
+            log.warning(f"Can't parse from_jid {e}: {element.toXml()}")
+            raise exceptions.NotFound("<jinglepub> element has invalid 'from' attribute")
+
+        if element.hasAttribute("id"):
+            kwargs["id"] = element["id"]
+
+        uri_elt = next(element.elements(NS_JINGLEPUB, "uri"), None)
+        if uri_elt:
+            kwargs["uri"] = str(uri_elt)
+
+        kwargs["meta"] = [
+            Meta.from_element(meta_elt)
+            for meta_elt in element.elements(NS_JINGLEPUB, "meta")
+        ]
+
+        kwargs["descriptions"] = [
+            child
+            for child in element.elements()
+            if child.uri != NS_JINGLEPUB and child.name == "description"
+        ]
+
+        return cls(**kwargs)
+
+
+class XEP_0358:
+    namespace = NS_JINGLEPUB
+
+    def __init__(self, host: "LiberviaBackend") -> None:
+        log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization")
+        self.host = host
+        host.register_namespace("jinglepub", NS_JINGLEPUB)
+
+    def get_handler(self, client: SatXMPPEntity) -> XMPPHandler:
+        return JinglePubHandler(self)
+
+
+@implementer(iwokkel.IDisco)
+class JinglePubHandler(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_JINGLEPUB),
+        ]
+
+    def getDiscoItems(
+        self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = ""
+    ) -> list[disco.DiscoItems]:
+        return []